Tirsvad
37f3cc436a
retry(func, retries=7, description=None), RetriesExhaustedError, resilient login, book_class and get_my_bookings that re-read the card before a repeat and count a booking only when the page confirms it, --retries option, step-9 console image, seeded chaos tests at 50 percent failure and container tests against failing local fixtures. Task: MIL-007#1 Task: MIL-007#2 Task: MIL-007#3 Task: MIL-007#4
177 lines
6.5 KiB
Python
177 lines
6.5 KiB
Python
"""@file fakes.py
|
|
@brief A fake of the GymPort for the unit tests.
|
|
"""
|
|
|
|
import random
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
|
|
from snack_and_lift import constants
|
|
from snack_and_lift.domain.booking_entry import BookingEntry
|
|
from snack_and_lift.domain.errors import GymError, LoginFailedError
|
|
from snack_and_lift.domain.models import ButtonState, ClassCard
|
|
|
|
_PRESS_RESULTS = {
|
|
constants.BUTTON_TEXT_BOOK: constants.BUTTON_TEXT_BOOKED,
|
|
constants.BUTTON_TEXT_JOIN_WAITLIST: constants.BUTTON_TEXT_WAITLISTED,
|
|
}
|
|
|
|
|
|
_WAITLISTED_BY_STATE = {
|
|
ButtonState.BOOK: False,
|
|
ButtonState.BOOKED: False,
|
|
ButtonState.JOIN_WAITLIST: True,
|
|
ButtonState.WAITLISTED: True,
|
|
}
|
|
|
|
|
|
def _entry_of(card: ClassCard) -> BookingEntry | None:
|
|
"""@brief The My Bookings entry that a card ends in, or None when it is not booked."""
|
|
flag = _WAITLISTED_BY_STATE.get(card.state)
|
|
if flag is None:
|
|
return None
|
|
time_text = card.time_text.removeprefix(constants.TIME_TEXT_PREFIX).strip()
|
|
return BookingEntry(card.name, f"{card.date_text}, {time_text}", flag)
|
|
|
|
|
|
NETWORK_ERROR_TEXT = "Network request failed. Please try again."
|
|
|
|
|
|
class FakeGym:
|
|
"""@brief Records the calls; can refuse the login, serve cards and fail a click.
|
|
|
|
Unless `bookings` is given, the My Bookings page shows every card that is, or by the
|
|
click becomes, booked or waitlisted. A successful click turns the card's button into its
|
|
new state, so a card read again shows what the click did.
|
|
|
|
`fail_first` is a deterministic failure schedule: it maps the name of a port method
|
|
(`log_in`, `press_class_button`, `open_my_bookings`, `list_my_bookings`) to the number of
|
|
its first calls that fail like a dropped request. With `click_goes_through`, a failing
|
|
`press_class_button` still turns the button (the click arrived, the answer did not).
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
login_error: str | None = None,
|
|
cards: list[ClassCard] | None = None,
|
|
press_result: str | None = None,
|
|
press_error: str | None = None,
|
|
bookings: list[BookingEntry] | None = None,
|
|
bookings_error: str | None = None,
|
|
fail_first: dict[str, int] | None = None,
|
|
click_goes_through: bool = False,
|
|
) -> None:
|
|
self.login_error = login_error
|
|
self.cards = cards or []
|
|
self.press_result = press_result
|
|
self.press_error = press_error
|
|
self.bookings = bookings
|
|
self.bookings_error = bookings_error
|
|
self.click_goes_through = click_goes_through
|
|
self._failures_left = dict(fail_first or {})
|
|
self.calls: list[str] = []
|
|
self.logins: list[tuple[str, str]] = []
|
|
self.screenshots: list[Path] = []
|
|
self.text_screenshots: list[tuple[str, Path]] = []
|
|
self.pressed: list[str] = []
|
|
self.closed = False
|
|
|
|
def open_site(self) -> None:
|
|
self.calls.append("open_site")
|
|
|
|
def _should_fail(self, name: str) -> bool:
|
|
"""@brief Whether this call is one of the scheduled failures; counts it."""
|
|
left = self._failures_left.get(name, 0)
|
|
if left > 0:
|
|
self._failures_left[name] = left - 1
|
|
return True
|
|
return False
|
|
|
|
def _turn_button(self, card_id: str) -> str:
|
|
"""@brief Turn the button of a card to the state its click gives; return the new text."""
|
|
index = next(i for i, card in enumerate(self.cards) if card.card_id == card_id)
|
|
card = self.cards[index]
|
|
new_text = _PRESS_RESULTS[card.button_text]
|
|
self.cards[index] = replace(card, button_text=new_text)
|
|
return new_text
|
|
|
|
def log_in(self, email: str, password: str) -> None:
|
|
self.calls.append("log_in")
|
|
self.logins.append((email, password))
|
|
if self.login_error is not None:
|
|
raise LoginFailedError(self.login_error)
|
|
if self._should_fail("log_in"):
|
|
raise LoginFailedError(NETWORK_ERROR_TEXT)
|
|
|
|
def list_class_cards(self) -> list[ClassCard]:
|
|
self.calls.append("list_class_cards")
|
|
return list(self.cards)
|
|
|
|
def press_class_button(self, card_id: str) -> str:
|
|
self.calls.append("press_class_button")
|
|
self.pressed.append(card_id)
|
|
if self.press_error is not None:
|
|
raise GymError(self.press_error)
|
|
if self.press_result is not None:
|
|
return self.press_result
|
|
if self._should_fail("press_class_button"):
|
|
if self.click_goes_through:
|
|
self._turn_button(card_id)
|
|
raise GymError(NETWORK_ERROR_TEXT)
|
|
return self._turn_button(card_id)
|
|
|
|
def open_my_bookings(self) -> None:
|
|
self.calls.append("open_my_bookings")
|
|
if self._should_fail("open_my_bookings"):
|
|
raise GymError(NETWORK_ERROR_TEXT)
|
|
|
|
def list_my_bookings(self) -> list[BookingEntry]:
|
|
self.calls.append("list_my_bookings")
|
|
if self._should_fail("list_my_bookings"):
|
|
raise GymError(NETWORK_ERROR_TEXT)
|
|
if self.bookings_error is not None:
|
|
raise GymError(self.bookings_error)
|
|
if self.bookings is not None:
|
|
return list(self.bookings)
|
|
entries = (_entry_of(card) for card in self.cards)
|
|
return [entry for entry in entries if entry is not None]
|
|
|
|
def save_text_screenshot(self, text: str, path: Path) -> None:
|
|
self.calls.append("save_text_screenshot")
|
|
self.text_screenshots.append((text, path))
|
|
|
|
def save_screenshot(self, path: Path) -> None:
|
|
self.calls.append("save_screenshot")
|
|
self.screenshots.append(path)
|
|
|
|
def close(self) -> None:
|
|
self.calls.append("close")
|
|
self.closed = True
|
|
|
|
|
|
class ChaosGym(FakeGym):
|
|
"""@brief A gym whose login, click and bookings read each fail with a set probability.
|
|
|
|
The site's network simulation fails up to 50% of these requests and writes nothing when
|
|
one fails. The random numbers come from a seeded generator, so a run is repeatable.
|
|
"""
|
|
|
|
_FAILING_CALLS = frozenset({"log_in", "press_class_button", "list_my_bookings"})
|
|
|
|
def __init__(
|
|
self,
|
|
rng: random.Random,
|
|
failure_probability: float = 0.5,
|
|
cards: list[ClassCard] | None = None,
|
|
) -> None:
|
|
super().__init__(cards=cards)
|
|
self._rng = rng
|
|
self._failure_probability = failure_probability
|
|
|
|
def _should_fail(self, name: str) -> bool:
|
|
return name in self._FAILING_CALLS and self._rng.random() < self._failure_probability
|
|
|
|
|
|
def no_sleep(seconds: float) -> None:
|
|
"""@brief A `sleep` that returns at once, so a retry test never waits."""
|