Refactor Event Polling Mechanism in Test Code

[coderabbitai[bot]]

There is a need to refactor the event polling mechanism in the test code to avoid the necessity of advancing the clock twice to capture all events. This issue was identified during the review of PR #482. Further details can be found in the discussion here: https://github.com/makerspace/makeradmin/pull/482#discussion_r1564086953

[emanuelen5]

@coderabbitai Can you update the description with examples of the code that needs refactoring, as well as link to it?

[coderabbitai[bot]]

[!TIP] For best results, initiate chat on the files or code changes.

@emanuelen5, I've located the relevant sections of the test code that need refactoring related to the event polling mechanism. Below are the examples from the codebase:

Example 1: Triggering Stripe Source Event

def trigger_stripe_source_event(self, source_id, expected_event_count=1):
    """Make server fetch events and filter it on source and type, do this until one event was processed."""
    for i in range(10):
        event_count = self.post(
            f"/webshop/process_stripe_events",
            dict(
                start=self.test_start_timestamp,
                type=f"{EventType.SOURCE}*",
            )
        )
        if event_count >= expected_event_count:
            return
        time.sleep(1)
    raise AssertionError(f"failed to get source events for {source_id}")

Example 2: Advancing Test Clocks

class FakeClock:
    def __init__(self, date: datetime) -> None:
        self.date = date
        self.stripe_clock = stripe.test_helpers.TestClock.create(frozen_time=int(date.timestamp()), name="TestClock")

    def advance(self, to: datetime) -> None:
        assert to >= self.date
        self.date = to
        stripe.test_helpers.TestClock.advance(self.stripe_clock.id, frozen_time=int(self.date.timestamp()))

Example 3: Polling Stripe Events

def poll_stripe_events(self, min_time: datetime, until_seen_type: str, test_clock: str) -> None:
    batched_events: List[stripe.Event] = []
    done = 0
    while not done:
        time.sleep(1)
        try:
            with db_session.begin_nested():
                events = stripe.Event.list(created={"gte": int(min_time.timestamp())})
                for ev in events.auto_paging_iter():
                    if ev.id in self.seen_event_ids:
                        continue
                    if test_clock_for_event(ev) != test_clock:
                        continue
                    self.seen_event_ids.add(ev.id)
                    if not ev.type.startswith("test_helpers.test_clock."):
                        batched_events.append(ev)
                    if ev.type == until_seen_type:
                        done = 1
                        break
        except stripe.RateLimitError:
            time.sleep(1 + random.random() * 5)
            continue
        except Exception as e:
            logger.error("Error while polling for stripe events: %s", e)
            raise

These examples illustrate the current implementation of event polling and clock management in the test suite. The code can be found in the test files within the repository, specifically those that handle Stripe integration tests.

Please let me know if you need further details or modifications to these examples!

Analysis chain

Scripts executed

The following scripts were executed for the analysis: --- Script: ```shell #!/bin/bash # Find files related to event polling in test code fd 'test' --type f --exec rg -C 3 'event|poll|clock' {} ``` Length of output: 36096