Compare commits

..
Author SHA1 Message Date
Tirsvad 6871e54333 Resolve pending commit links 2026-10-28 18:14:19 +08:00
Tirsvad 5d28e682b4 Add review record RC-021 for MIL-007 2026-10-28 18:14:18 +08:00
Tirsvad 37f3cc436a Add retry wrapper and resilient login, booking and bookings read (step 9)
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
2026-10-28 18:14:13 +08:00
Tirsvad a830e3ae5f Merge pull request 'MIL-006: verify bookings and time travel QA' (#45) from mil-006-verify-and-time-travel into main
TirSystem/git-workflows: Sync GitHub mirror metadata / sync-metadata (push) Successful in 7s
2026-10-28 11:03:00 +01:00
Tirsvad 944ba204cf Resolve pending commit links 2026-10-28 18:02:52 +08:00
Tirsvad f57040e136 Add review record RC-020 for MIL-006 2026-10-28 18:02:50 +08:00
Tirsvad 8c6cc148f5 Add My Bookings verification and date-independent booking (steps 7 and 8)
Booking entries parsed from the My Bookings page, comparison of expected
and found bookings with the lecture's output, NoBookingsFoundError as the
counterpart of NoSuchElementException, exit code 3 on a mismatch, tests for
every starting weekday of the 8-day window, container tests against local
fixtures.

Task: MIL-006#1
Task: MIL-006#2
Task: MIL-006#3
Task: MIL-006#4
2026-10-28 18:02:44 +08:00
Tirsvad 0a665ee9ac Merge pull request 'MIL-005: counters and every Tuesday and Thursday class' (#44) from mil-005-counters-and-both-days into main
TirSystem/git-workflows: Sync GitHub mirror metadata / sync-metadata (push) Successful in 6s
2026-10-28 10:54:26 +01:00
29 changed files with 2477 additions and 498 deletions
+1 -1
View File
@@ -16,7 +16,7 @@ document of a type. `Primary File` may contain a glob (e.g.
| PP | Project Plan | docs/project-plan.md | 002 | | PP | Project Plan | docs/project-plan.md | 002 |
| MIL | Milestone / Gateway | docs/milestones/*.md | 009 | | MIL | Milestone / Gateway | docs/milestones/*.md | 009 |
| ADR | Architecture Decision Record | docs/adr/adr-NNNN-*.md | 0004 | | ADR | Architecture Decision Record | docs/adr/adr-NNNN-*.md | 0004 |
| RC | SQA Review Record | docs/sqa/reviews/rc-*.md | 020 | | RC | SQA Review Record | docs/sqa/reviews/rc-*.md | 022 |
| TM | Traceability Matrix | docs/sqa/traceability-matrix.md | 002 | | TM | Traceability Matrix | docs/sqa/traceability-matrix.md | 002 |
| DICT | Domain Dictionary | docs/dictionary.md | 002 | | DICT | Domain Dictionary | docs/dictionary.md | 002 |
@@ -0,0 +1,62 @@
# SQA Review Record
## Metadata
| Key | Value |
| --- | --- |
| ID | RC-020 |
| CrossReference | [MIL-006], [QC-PY-001] |
## Version History
| Date | Status | Author | Reviewer | Change | Commit |
| --- | --- | --- | --- | --- | --- |
| 2026-10-28 | Accepted | Jens Tirsvad Nielsen | AI as stand-in for S02 | Review record, full review of the source and tests of MIL-006 | [f57040e] |
---
## Artifact Under Review
- Instance reviewed: [MIL-006]
- Checklist used: [QC-PY-001] (`QC-PY-001`)
- Scope: full review of the source and tests of MIL-006
- Language and domain: n/a
- Language reviewer: none
- Mode: autonomous, not independent (the reviewer is `AI as stand-in for S02`)
## Checklist Results
| # | Criterion | Status | Evidence/Notes |
| --- | --- | --- | --- |
| 1 | Packages, modules, functions, variables, classes and constants follow PEP 8 casing (`snake_case`, `PascalCase`, `UPPER_SNAKE`) | Pass | Modules, functions and variables are `snake_case`, classes `PascalCase`, constants `UPPER_SNAKE`; `ruff check` with the N rules passes. |
| 2 | Names state purpose in the domain's language; no unexplained abbreviations, no single-letter names outside tiny scopes | Pass | Names state purpose (for example `retry`, `login`, `book_class`, `get_my_bookings`); no single-letter names outside comprehensions. |
| 3 | Code is produced by the project's formatter and passes its linter with no unexplained suppressions | Pass | `ruff format` leaves 0 files changed and `ruff check` reports 0 findings; there is no suppression. |
| 4 | Every function and method signature is type-annotated, including `-> None` | Pass | Every function and method has annotations (ruff ANN rules pass), including `-> None`. |
| 5 | No bare `except:`, no swallowed exceptions; specific exceptions are raised and the cause is kept (`raise ... from`) | Pass | No bare `except`; specific exceptions are caught, and re-raised with `raise ... from` where the cause matters. |
| 6 | No mutable default arguments and no shadowed builtins | Pass | No mutable default arguments; no builtin is shadowed (ruff B and A-style checks pass). |
| 7 | Files, locks and connections are managed with context managers | Pass | Files and the browser are managed with `with`/`try-finally`; no unmanaged file handle exists. |
| 8 | Public modules, classes and functions have docstrings that say what, not how | Pass | Public modules, classes and functions have Doxygen docstrings (`@brief`, `@param`, `@return`). |
| 9 | Logging uses `logging`, not `print`; no secrets or personal data in log output | Pass | No print calls; messages go through `logging`. |
| 10 | Classes and operations trace to the Design Class Diagram they implement; deviations are recorded | N-A | No Design Class Diagram exists; ADR-0001 records that design artifacts are skipped. Classes follow ADR-0002. |
| 11 | Tests exist for new behaviour, are named for the behaviour, and do not depend on order or the network | Pass | 174 unit tests (booking entries, verification, date independence for all seven starting weekdays, command line) need no browser or network; 15 container tests run the adapter against local fixtures; coverage of the non-infrastructure modules is 100%. |
| 12 | Type checker runs in strict mode without errors; `Any` is justified in a comment | N-A | No strict type checker is configured; annotations are complete and checked by ruff. |
| 13 | Dependencies are declared and pinned in the project's dependency file, none unused | Pass | `pyproject.toml` declares Selenium (runtime) and pytest, pytest-cov, ruff (dev) with lower bounds; none is unused. |
| 14 | Each piece of knowledge (a business rule, constant, format, validation or query) is defined in one place; code is merged only where it expresses the same knowledge, not where it merely looks alike | Pass | Each rule, constant and message exists once (constants in `constants.py`, rules in `domain/`). |
| 15 | Imports point inward: business-rule modules import no framework, database, UI or delivery-mechanism code, and there are no import cycles | Pass | `domain` and `application` import no Selenium; only `infrastructure` imports it; no import cycle exists. |
| 16 | SOLID holds: a class or module has one reason to change; new behaviour is added by extension, not by editing a type test in several places; subclasses honour the contract of their base; protocols are narrow; high-level code depends on protocols or abstract types and receives its implementations | Pass | One reason to change per module; the adapter hides Selenium behind a port; fakes honour the port contract. |
| 17 | Source is laid out in the layers and folders of the clean architecture folder structure (`domain`, `application`, `infrastructure`, `presentation` packages under `src/`, and `unit`, `integration` and `e2e` under `tests/`), in `snake_case`, and imports between the layers point inward | Pass | Layout follows `src/snack_and_lift/{domain,application,infrastructure}` and `tests/{unit,integration,e2e}`; there is no presentation layer for a command-line bot (ADR-0002). |
| 18 | File and folder names are English, contain no space and no character outside letters, digits, `-`, `_` and `.`, and follow the casing of the language for files | Pass | All file and folder names are English lowercase with `_`, `-` and `.` only. |
## Overall Verdict
Go-with-conditions - the code of MIL-006 meets every Mandatory criterion of QC-PY-001 and there is no blocking finding, but part of the gate evidence cannot be produced by the stand-in. Gate criterion 1 (unit tests of the comparison and the mismatch message) is met, and the date independence is tested for every starting weekday (no hardcoded date, no list-position rule). Criteria 2 and 3 (container runs on the live site, including after the admin advances 3 days) are not verified, and criterion 4 is not met: `step-7.png` and `step-8.png` come from a run with an account. The framework forbids the stand-in to enter passwords or register on a third-party site; the admin time controls also need a login. The review is not independent: the AI wrote and reviewed it as stand-in for S02, and S01 can challenge it in the end-of-run summary.
## Action Items
| Action | Owner | Due |
| --- | --- | --- |
| Run the bot with your own account (also after advancing the site 3 days as admin) and keep `docs/images/step-7.png` and `step-8.png` | S01 | when S01 has time |
---
[MIL-006]: ../../milestones/mil-006-verify-and-time-travel.md
[QC-PY-001]: ../../../framework/qc/code/qc-programming-python.md
[f57040e]: https://git.tirsystem.com/Tirsvad-Udemy-100-days-of-code/049-snack-and-lift/commit/f57040e136feac01f10d99279038f48992518f43
@@ -0,0 +1,62 @@
# SQA Review Record
## Metadata
| Key | Value |
| --- | --- |
| ID | RC-021 |
| CrossReference | [MIL-007], [QC-PY-001] |
## Version History
| Date | Status | Author | Reviewer | Change | Commit |
| --- | --- | --- | --- | --- | --- |
| 2026-10-28 | Accepted | Jens Tirsvad Nielsen | AI as stand-in for S02 | Review record, full review of the source and tests of MIL-007 | [5d28e68] |
---
## Artifact Under Review
- Instance reviewed: [MIL-007]
- Checklist used: [QC-PY-001] (`QC-PY-001`)
- Scope: full review of the source and tests of MIL-007
- Language and domain: n/a
- Language reviewer: none
- Mode: autonomous, not independent (the reviewer is `AI as stand-in for S02`)
## Checklist Results
| # | Criterion | Status | Evidence/Notes |
| --- | --- | --- | --- |
| 1 | Packages, modules, functions, variables, classes and constants follow PEP 8 casing (`snake_case`, `PascalCase`, `UPPER_SNAKE`) | Pass | Modules, functions and variables are `snake_case`, classes `PascalCase`, constants `UPPER_SNAKE`; `ruff check` with the N rules passes. |
| 2 | Names state purpose in the domain's language; no unexplained abbreviations, no single-letter names outside tiny scopes | Pass | Names state purpose (for example `retry`, `login`, `book_class`, `get_my_bookings`); no single-letter names outside comprehensions. |
| 3 | Code is produced by the project's formatter and passes its linter with no unexplained suppressions | Pass | `ruff format` leaves 0 files changed and `ruff check` reports 0 findings; there is no suppression. |
| 4 | Every function and method signature is type-annotated, including `-> None` | Pass | Every function and method has annotations (ruff ANN rules pass), including `-> None`. |
| 5 | No bare `except:`, no swallowed exceptions; specific exceptions are raised and the cause is kept (`raise ... from`) | Pass | No bare `except`; specific exceptions are caught, and re-raised with `raise ... from` where the cause matters. |
| 6 | No mutable default arguments and no shadowed builtins | Pass | No mutable default arguments; no builtin is shadowed (ruff B and A-style checks pass). |
| 7 | Files, locks and connections are managed with context managers | Pass | Files and the browser are managed with `with`/`try-finally`; no unmanaged file handle exists. |
| 8 | Public modules, classes and functions have docstrings that say what, not how | Pass | Public modules, classes and functions have Doxygen docstrings (`@brief`, `@param`, `@return`). |
| 9 | Logging uses `logging`, not `print`; no secrets or personal data in log output | Pass | No print calls; messages go through `logging`. |
| 10 | Classes and operations trace to the Design Class Diagram they implement; deviations are recorded | N-A | No Design Class Diagram exists; ADR-0001 records that design artifacts are skipped. Classes follow ADR-0002. |
| 11 | Tests exist for new behaviour, are named for the behaviour, and do not depend on order or the network | Pass | 237 unit tests (retry, resilient login, booking and bookings read, a seeded chaos test at 50% failure: 97 of 100 seeds complete, seeds 0 to 17 must all pass) need no browser or network; 20 container tests run the adapter against failing local fixtures; coverage of the non-infrastructure modules is 100%. |
| 12 | Type checker runs in strict mode without errors; `Any` is justified in a comment | N-A | No strict type checker is configured; annotations are complete and checked by ruff. |
| 13 | Dependencies are declared and pinned in the project's dependency file, none unused | Pass | `pyproject.toml` declares Selenium (runtime) and pytest, pytest-cov, ruff (dev) with lower bounds; none is unused. |
| 14 | Each piece of knowledge (a business rule, constant, format, validation or query) is defined in one place; code is merged only where it expresses the same knowledge, not where it merely looks alike | Pass | Each rule, constant and message exists once (constants in `constants.py`, rules in `domain/`). |
| 15 | Imports point inward: business-rule modules import no framework, database, UI or delivery-mechanism code, and there are no import cycles | Pass | `domain` and `application` import no Selenium; only `infrastructure` imports it; no import cycle exists. |
| 16 | SOLID holds: a class or module has one reason to change; new behaviour is added by extension, not by editing a type test in several places; subclasses honour the contract of their base; protocols are narrow; high-level code depends on protocols or abstract types and receives its implementations | Pass | One reason to change per module; the adapter hides Selenium behind a port; fakes honour the port contract. |
| 17 | Source is laid out in the layers and folders of the clean architecture folder structure (`domain`, `application`, `infrastructure`, `presentation` packages under `src/`, and `unit`, `integration` and `e2e` under `tests/`), in `snake_case`, and imports between the layers point inward | Pass | Layout follows `src/snack_and_lift/{domain,application,infrastructure}` and `tests/{unit,integration,e2e}`; there is no presentation layer for a command-line bot (ADR-0002). |
| 18 | File and folder names are English, contain no space and no character outside letters, digits, `-`, `_` and `.`, and follow the casing of the language for files | Pass | All file and folder names are English lowercase with `_`, `-` and `.` only. |
## Overall Verdict
Go-with-conditions - the code of MIL-007 meets every Mandatory criterion of QC-PY-001 and there is no blocking finding, but part of the gate evidence cannot be produced by the stand-in. Gate criteria 1 (retry unit tests) and 2 (a booking only counts when the page confirms it) are met in tests. Criterion 3 (three container runs on the live site at 50% simulated failure) is not verified, and criterion 4 is not met: `step-9.png` comes from a run with an account. The framework forbids the stand-in to enter passwords or register on a third-party site, and the network simulation is an admin control behind a login. Known behaviour: a wrong password or an empty bookings page is also retried seven times (about 3 seconds) before exit code 1. The review is not independent: the AI wrote and reviewed it as stand-in for S02, and S01 can challenge it in the end-of-run summary.
## Action Items
| Action | Owner | Due |
| --- | --- | --- |
| Enable the network simulation at 50% as admin, run the bot three times with your own account and keep `docs/images/step-9.png` | S01 | when S01 has time |
---
[MIL-007]: ../../milestones/mil-007-resilience.md
[QC-PY-001]: ../../../framework/qc/code/qc-programming-python.md
[5d28e68]: https://git.tirsystem.com/Tirsvad-Udemy-100-days-of-code/049-snack-and-lift/commit/5d28e682b41e1a03b0cac1a6b2ea4977750d45b8
+2
View File
@@ -40,6 +40,8 @@ Tracks backward and forward links between artifact instances so that the Busines
| Source and tests of [MIL-003] | Python source code | - | - | [MIL-003], [ADR-0002], [ADR-0003] | - | RC-017 (Go-with-conditions) | | Source and tests of [MIL-003] | Python source code | - | - | [MIL-003], [ADR-0002], [ADR-0003] | - | RC-017 (Go-with-conditions) |
| Source and tests of [MIL-004] | Python source code | - | - | [MIL-004], [ADR-0002] | - | RC-018 (Go-with-conditions) | | Source and tests of [MIL-004] | Python source code | - | - | [MIL-004], [ADR-0002] | - | RC-018 (Go-with-conditions) |
| Source and tests of [MIL-005] | Python source code | - | - | [MIL-005], [ADR-0002] | - | RC-019 (Go-with-conditions) | | Source and tests of [MIL-005] | Python source code | - | - | [MIL-005], [ADR-0002] | - | RC-019 (Go-with-conditions) |
| Source and tests of [MIL-006] | Python source code | - | - | [MIL-006], [ADR-0002] | - | RC-020 (Go-with-conditions) |
| Source and tests of [MIL-007] | Python source code | - | - | [MIL-007], [ADR-0002] | - | RC-021 (Go-with-conditions) |
## Concern Coverage ## Concern Coverage
+69 -6
View File
@@ -3,10 +3,12 @@
""" """
import logging import logging
import time
from collections.abc import Callable from collections.abc import Callable
from snack_and_lift import constants from snack_and_lift import constants
from snack_and_lift.application.ports import GymPort from snack_and_lift.application.ports import GymPort
from snack_and_lift.application.retry import Sleep, retry
from snack_and_lift.domain.errors import GymError from snack_and_lift.domain.errors import GymError
from snack_and_lift.domain.models import ButtonState, ClassCard, ClassResult from snack_and_lift.domain.models import ButtonState, ClassCard, ClassResult
from snack_and_lift.domain.rules import ( from snack_and_lift.domain.rules import (
@@ -41,12 +43,68 @@ def book_class(gym: GymPort, card: ClassCard) -> ClassResult:
return ClassResult(card, outcome) return ClassResult(card, outcome)
def book_next_tuesday(gym: GymPort) -> ClassResult | None: def _read_current_card(gym: GymPort, card: ClassCard) -> ClassCard:
"""@brief Read a class again from the page, to see what an earlier attempt did.
@param gym The gym site.
@param card The class, as it was read before.
@return The class with the state the page shows now.
@exception GymError The schedule could not be read or the class is no longer on it.
"""
for current in gym.list_class_cards():
if current.card_id == card.card_id:
return current
raise GymError(constants.MESSAGE_CLASS_NOT_ON_SCHEDULE.format(card_id=card.card_id))
def book_class_with_retry(
gym: GymPort,
card: ClassCard,
retries: int = constants.RETRY_COUNT,
sleep: Sleep = time.sleep,
) -> ClassResult:
"""@brief Book a class, repeating it when the site drops the request.
Before each repeat the class is read from the page again. When the button now shows the
state that the click should give, the click did go through although its confirmation
failed, so the class counts as newly booked or waitlisted and is not clicked again.
@param gym The gym site.
@param card The class, as read from the page.
@param retries The most attempts to make.
@param sleep Waits between two attempts.
@return The class and what happened.
@exception RetriesExhaustedError Every attempt failed.
"""
attempts = 0
def attempt() -> ClassResult:
nonlocal attempts
attempts += 1
if attempts == 1:
return book_class(gym, card)
current = _read_current_card(gym, card)
action = decide_outcome_action(card)
if CONFIRMED_STATE.get(action) is current.state:
outcome = OUTCOME_OF_ACTION[action]
logger.info(format_outcome(outcome, card))
return ClassResult(card, outcome)
return book_class(gym, current)
description = constants.RETRY_DESCRIPTION_BOOK.format(name=card.name, day=card.day_label)
return retry(attempt, retries, description, sleep=sleep)
def book_next_tuesday(
gym: GymPort, retries: int = constants.RETRY_COUNT, sleep: Sleep = time.sleep
) -> ClassResult | None:
"""@brief Book the next Tuesday class at the target time. """@brief Book the next Tuesday class at the target time.
@param gym The gym site. @param gym The gym site.
@param retries The most attempts to book the class.
@param sleep Waits between two attempts.
@return The result, or None when the schedule has no such class. @return The result, or None when the schedule has no such class.
@exception GymError The page could not be read or the booking failed. @exception GymError The page could not be read or the booking failed on every attempt.
""" """
card = find_next_class(gym.list_class_cards(), constants.NEXT_CLASS_WEEKDAY) card = find_next_class(gym.list_class_cards(), constants.NEXT_CLASS_WEEKDAY)
if card is None: if card is None:
@@ -57,11 +115,14 @@ def book_next_tuesday(gym: GymPort) -> ClassResult | None:
) )
) )
return None return None
return book_class(gym, card) return book_class_with_retry(gym, card, retries, sleep)
def book_target_classes( def book_target_classes(
gym: GymPort, on_result: Callable[[ClassResult], None] | None = None gym: GymPort,
on_result: Callable[[ClassResult], None] | None = None,
retries: int = constants.RETRY_COUNT,
sleep: Sleep = time.sleep,
) -> list[ClassResult]: ) -> list[ClassResult]:
"""@brief Book every Tuesday and Thursday class at the target time, in page order. """@brief Book every Tuesday and Thursday class at the target time, in page order.
@@ -69,8 +130,10 @@ def book_target_classes(
@param gym The gym site. @param gym The gym site.
@param on_result Called after each class was handled, for example to take a screenshot. @param on_result Called after each class was handled, for example to take a screenshot.
@param retries The most attempts to book one class.
@param sleep Waits between two attempts.
@return What happened to each class; empty when the schedule has none. @return What happened to each class; empty when the schedule has none.
@exception GymError The page could not be read or a booking failed. @exception GymError The page could not be read or a booking failed on every attempt.
""" """
cards = find_target_classes(gym.list_class_cards()) cards = find_target_classes(gym.list_class_cards())
if not cards: if not cards:
@@ -78,7 +141,7 @@ def book_target_classes(
return [] return []
results: list[ClassResult] = [] results: list[ClassResult] = []
for card in cards: for card in cards:
result = book_class(gym, card) result = book_class_with_retry(gym, card, retries, sleep)
results.append(result) results.append(result)
if on_result is not None: if on_result is not None:
on_result(result) on_result(result)
+82 -66
View File
@@ -1,66 +1,82 @@
"""@file ports.py """@file ports.py
@brief The ports the use cases need from the outside world. @brief The ports the use cases need from the outside world.
""" """
from pathlib import Path from pathlib import Path
from typing import Protocol from typing import Protocol
from snack_and_lift.domain.models import ClassCard from snack_and_lift.domain.booking_entry import BookingEntry
from snack_and_lift.domain.models import ClassCard
class GymPort(Protocol):
"""@brief The gym site as the use cases see it.""" class GymPort(Protocol):
"""@brief The gym site as the use cases see it."""
def open_site(self) -> None:
"""@brief Open the home page and wait until it is ready. def open_site(self) -> None:
"""@brief Open the home page and wait until it is ready.
@exception GymError The page did not load.
""" @exception GymError The page did not load.
... """
...
def log_in(self, email: str, password: str) -> None:
"""@brief Log in and wait until the class schedule is shown. def log_in(self, email: str, password: str) -> None:
"""@brief Log in and wait until the class schedule is shown.
@param email The account e-mail.
@param password The account password. @param email The account e-mail.
@exception LoginFailedError The site refused the login. @param password The account password.
@exception GymError The page did not behave as expected. @exception LoginFailedError The site refused the login.
""" @exception GymError The page did not behave as expected.
... """
...
def list_class_cards(self) -> list[ClassCard]:
"""@brief Read every class card of the schedule, in page order. def list_class_cards(self) -> list[ClassCard]:
"""@brief Read every class card of the schedule, in page order.
@return The classes, each with the day of its group.
@exception GymError The schedule could not be read. @return The classes, each with the day of its group.
""" @exception GymError The schedule could not be read.
... """
...
def press_class_button(self, card_id: str) -> str:
"""@brief Click the button of a class and wait until the request has finished. def press_class_button(self, card_id: str) -> str:
"""@brief Click the button of a class and wait until the request has finished.
@param card_id The id of the card without its prefix.
@return The new text of the button. @param card_id The id of the card without its prefix.
@exception GymError The site shows an error for the class (the message is its text) @return The new text of the button.
or nothing settled in time. @exception GymError The site shows an error for the class (the message is its text)
""" or nothing settled in time.
... """
...
def save_text_screenshot(self, text: str, path: Path) -> None:
"""@brief Save an image of console text. def open_my_bookings(self) -> None:
"""@brief Open the My Bookings page and wait until it is shown.
@param text The lines to draw, in a dark monospace style.
@param path The PNG file to write; its folders are created. @exception GymError The page did not load.
""" """
... ...
def save_screenshot(self, path: Path) -> None: def list_my_bookings(self) -> list[BookingEntry]:
"""@brief Save a screenshot of the current page. """@brief Read every confirmed booking and waitlist entry of the My Bookings page.
@param path The PNG file to write; its folders are created. @return The entries, confirmed ones first; empty when the page shows its empty state.
""" @exception GymError The page shows an error (the message is its text) or did not settle.
... """
...
def close(self) -> None:
"""@brief Close the browser.""" def save_text_screenshot(self, text: str, path: Path) -> None:
... """@brief Save an image of console text.
@param text The lines to draw, in a dark monospace style.
@param path The PNG file to write; its folders are created.
"""
...
def save_screenshot(self, path: Path) -> None:
"""@brief Save a screenshot of the current page.
@param path The PNG file to write; its folders are created.
"""
...
def close(self) -> None:
"""@brief Close the browser."""
...
+67
View File
@@ -0,0 +1,67 @@
"""@file retry.py
@brief Lecture step 9: repeat a step that failed because the site dropped a request.
"""
import logging
import time
from collections.abc import Callable
from snack_and_lift import constants
from snack_and_lift.domain.errors import GymError, RetriesExhaustedError
logger = logging.getLogger(__name__)
## Waits for a number of seconds; tests pass a recorder so that they never really sleep.
Sleep = Callable[[float], None]
def retry[T](
func: Callable[[], T],
retries: int = constants.RETRY_COUNT,
description: str | None = None,
*,
delay: float = constants.RETRY_DELAY_SECONDS,
sleep: Sleep = time.sleep,
) -> T:
"""@brief Call a function until it succeeds, at most `retries` times.
Only a `GymError` is repeated: it is what a dropped request looks like. Any other error,
such as missing credentials, is a defect and passes through at once.
@param func The step, called without arguments.
@param retries The most attempts to make, at least 1.
@param description What the step does, for the log (never put a secret in it).
@param delay Seconds to wait between two attempts.
@param sleep Waits for a number of seconds.
@return What the function returned on its first successful attempt.
@exception ValueError `retries` is below 1.
@exception RetriesExhaustedError Every attempt raised a `GymError`; the last one is the cause.
"""
if retries < 1:
raise ValueError(constants.MESSAGE_RETRIES_INVALID.format(retries=retries))
name = description or constants.RETRY_DEFAULT_DESCRIPTION
for attempt in range(1, retries + 1):
try:
result = func()
except GymError as error:
if attempt == retries:
message = constants.MESSAGE_RETRIES_EXHAUSTED.format(
description=name, retries=retries, error=error
)
raise RetriesExhaustedError(message) from error
logger.warning(
constants.MESSAGE_RETRY_FAILED.format(
attempt=attempt,
retries=retries,
description=name,
error=str(error).rstrip("."),
)
)
sleep(delay)
else:
if attempt > 1:
logger.info(
constants.MESSAGE_RETRY_SUCCEEDED.format(description=name, attempts=attempt)
)
return result
raise AssertionError("unreachable: the last attempt returns or raises") # pragma: no cover
+25
View File
@@ -3,9 +3,12 @@
""" """
import logging import logging
import time
from snack_and_lift import constants
from snack_and_lift.application.credentials import Credentials from snack_and_lift.application.credentials import Credentials
from snack_and_lift.application.ports import GymPort from snack_and_lift.application.ports import GymPort
from snack_and_lift.application.retry import Sleep, retry
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -19,3 +22,25 @@ def login(gym: GymPort, credentials: Credentials) -> None:
""" """
gym.log_in(credentials.email, credentials.password) gym.log_in(credentials.email, credentials.password)
logger.info("✓ Logged in: the class schedule is shown") logger.info("✓ Logged in: the class schedule is shown")
def login_with_retry(
gym: GymPort,
credentials: Credentials,
retries: int = constants.RETRY_COUNT,
sleep: Sleep = time.sleep,
) -> None:
"""@brief Log in, repeating the login when the site drops the request.
@param gym The gym site.
@param credentials The account to log in with.
@param retries The most attempts to make.
@param sleep Waits between two attempts.
@exception RetriesExhaustedError Every attempt failed.
"""
retry(
lambda: login(gym, credentials),
retries,
constants.RETRY_DESCRIPTION_LOGIN,
sleep=sleep,
)
@@ -0,0 +1,73 @@
"""@file verification.py
@brief Lecture steps 7 and 8: check the bookings on the My Bookings page.
"""
import logging
import time
from collections.abc import Sequence
from snack_and_lift import constants
from snack_and_lift.application.ports import GymPort
from snack_and_lift.application.retry import Sleep, retry
from snack_and_lift.domain.booking_entry import BookingEntry, is_target_entry
from snack_and_lift.domain.errors import NoBookingsFoundError
from snack_and_lift.domain.models import ClassResult
from snack_and_lift.domain.verification import VerificationResult, build_verification
logger = logging.getLogger(__name__)
def get_my_bookings(gym: GymPort) -> list[BookingEntry]:
"""@brief Open the My Bookings page and read all its entries.
@param gym The gym site.
@return Every confirmed booking and waitlist entry, not only the target classes.
@exception NoBookingsFoundError The page shows no entry at all.
@exception GymError The page could not be opened or read.
"""
gym.open_my_bookings()
entries = gym.list_my_bookings()
if not entries:
raise NoBookingsFoundError(constants.MESSAGE_NO_BOOKINGS_FOUND)
return entries
def get_my_bookings_with_retry(
gym: GymPort, retries: int = constants.RETRY_COUNT, sleep: Sleep = time.sleep
) -> list[BookingEntry]:
"""@brief Read the My Bookings page, repeating it when the page fails to load.
@param gym The gym site.
@param retries The most attempts to make.
@param sleep Waits between two attempts.
@return Every confirmed booking and waitlist entry.
@exception RetriesExhaustedError Every attempt failed.
"""
return retry(
lambda: get_my_bookings(gym), retries, constants.RETRY_DESCRIPTION_BOOKINGS, sleep=sleep
)
def verify_bookings(
gym: GymPort,
results: Sequence[ClassResult],
retries: int = constants.RETRY_COUNT,
sleep: Sleep = time.sleep,
) -> VerificationResult:
"""@brief Compare what the bot booked with the My Bookings page.
The page is not opened when the bot handled no class.
@param gym The gym site.
@param results What happened to each class.
@param retries The most attempts to read the page.
@param sleep Waits between two attempts.
@return The comparison of the expected entries with the Tuesday and Thursday 6pm entries.
@exception RetriesExhaustedError The page could not be read on any attempt.
"""
if not results:
return build_verification(results, [])
entries = [
entry for entry in get_my_bookings_with_retry(gym, retries, sleep) if is_target_entry(entry)
]
return build_verification(results, entries)
+261 -209
View File
@@ -1,209 +1,261 @@
"""@file cli.py """@file cli.py
@brief Command-line entry point (the composition root of the bot). @brief Command-line entry point (the composition root of the bot).
""" """
import argparse import argparse
import logging import logging
import os import os
import sys import sys
from collections.abc import Callable, Sequence import time
from pathlib import Path from collections.abc import Callable, Sequence
from pathlib import Path
from snack_and_lift import constants
from snack_and_lift.application.booking import book_next_tuesday, book_target_classes from snack_and_lift import constants
from snack_and_lift.application.credentials import Credentials, load_credentials from snack_and_lift.application.booking import book_next_tuesday, book_target_classes
from snack_and_lift.application.ports import GymPort from snack_and_lift.application.credentials import Credentials, load_credentials
from snack_and_lift.application.session import login from snack_and_lift.application.ports import GymPort
from snack_and_lift.domain.errors import GymError, LoginFailedError, MissingCredentialsError from snack_and_lift.application.retry import Sleep
from snack_and_lift.domain.models import ClassResult from snack_and_lift.application.session import login_with_retry
from snack_and_lift.domain.report import format_detail_list, format_summary from snack_and_lift.application.verification import verify_bookings
from snack_and_lift.infrastructure.browser import create_driver, default_profile_dir from snack_and_lift.domain.errors import GymError, LoginFailedError, MissingCredentialsError
from snack_and_lift.infrastructure.console_recorder import ConsoleRecorder from snack_and_lift.domain.models import ClassResult
from snack_and_lift.infrastructure.selenium_gym import SeleniumGym from snack_and_lift.domain.report import format_detail_list, format_summary
from snack_and_lift.domain.verification import format_verification
## Builds the gym site from (headless, profile folder, keep open). from snack_and_lift.infrastructure.browser import create_driver, default_profile_dir
GymFactory = Callable[[bool, Path, bool], GymPort] from snack_and_lift.infrastructure.console_recorder import ConsoleRecorder
from snack_and_lift.infrastructure.selenium_gym import SeleniumGym
logger = logging.getLogger(__name__)
## Builds the gym site from (headless, profile folder, keep open).
GymFactory = Callable[[bool, Path, bool], GymPort]
def selenium_gym_factory(headless: bool, profile_dir: Path, keep_open: bool) -> GymPort:
"""@brief Start Chrome and wrap it as the gym site. logger = logging.getLogger(__name__)
@param headless True to run without a window.
@param profile_dir The Chrome user data folder. def selenium_gym_factory(headless: bool, profile_dir: Path, keep_open: bool) -> GymPort:
@param keep_open True to leave the browser open when the program ends. """@brief Start Chrome and wrap it as the gym site.
@return The gym site.
""" @param headless True to run without a window.
return SeleniumGym(create_driver(headless, profile_dir, keep_open)) @param profile_dir The Chrome user data folder.
@param keep_open True to leave the browser open when the program ends.
@return The gym site.
def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: """
"""@brief Read the command line. return SeleniumGym(create_driver(headless, profile_dir, keep_open))
@param argv Arguments without the program name; None reads `sys.argv`.
@return The parsed options; `keep_open` is True unless `--headless` is given. def _at_least_one(text: str) -> int:
""" """@brief Read a whole number of at least 1 for an option.
parser = argparse.ArgumentParser(
prog="snack-and-lift", @param text The text of the option.
description="Books the Tuesday and Thursday 6pm classes at the Snack and Lift gym.", @return The number.
) @exception argparse.ArgumentTypeError The text is not a whole number of at least 1.
parser.add_argument( """
"--headless", try:
action="store_true", number = int(text)
help="run Chrome without a window (the Docker image does this)", except ValueError:
) number = 0
parser.add_argument( if number < 1:
"--keep-open", raise argparse.ArgumentTypeError(f"{text!r} is not a whole number of at least 1")
action="store_true", return number
default=None,
help="leave the browser open at the end (default when not headless)",
) def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
parser.add_argument( """@brief Read the command line.
"--profile-dir",
type=Path, @param argv Arguments without the program name; None reads `sys.argv`.
default=None, @return The parsed options; `keep_open` is True unless `--headless` is given.
help="Chrome profile folder (default: ./chrome_profile)", """
) parser = argparse.ArgumentParser(
parser.add_argument( prog="snack-and-lift",
"--screenshot-dir", description="Books the Tuesday and Thursday 6pm classes at the Snack and Lift gym.",
type=Path, epilog=(
default=None, "exit codes: 0 done, 1 a step failed (also when every retry failed), "
help="save a screenshot after each step into this folder", "2 credentials missing, 3 the My Bookings page does not match the bookings made"
) ),
parser.add_argument( )
"--first-tuesday-only", parser.add_argument(
action="store_true", "--headless",
help="book only the next Tuesday 6pm class (lecture steps 3 and 4)", action="store_true",
) help="run Chrome without a window (the Docker image does this)",
args = parser.parse_args(argv) )
if args.keep_open is None: parser.add_argument(
args.keep_open = not args.headless "--keep-open",
if args.profile_dir is None: action="store_true",
args.profile_dir = default_profile_dir() default=None,
return args help="leave the browser open at the end (default when not headless)",
)
parser.add_argument(
def _use_utf8_output() -> None: "--profile-dir",
"""@brief Make stdout UTF-8 so the check mark of the messages prints on Windows.""" type=Path,
reconfigure = getattr(sys.stdout, "reconfigure", None) default=None,
if reconfigure is not None: help="Chrome profile folder (default: ./chrome_profile)",
reconfigure(encoding="utf-8") )
parser.add_argument(
"--screenshot-dir",
def _save_step(gym: GymPort, screenshot_dir: Path | None, name: str) -> None: type=Path,
"""@brief Save a screenshot when a folder was asked for. default=None,
help="save a screenshot after each step into this folder",
@param gym The gym site. )
@param screenshot_dir The folder, or None for no screenshots. parser.add_argument(
@param name The file name. "--first-tuesday-only",
""" action="store_true",
if screenshot_dir is not None: help="book only the next Tuesday 6pm class (lecture steps 3 and 4); skips the verification",
gym.save_screenshot(screenshot_dir / name) )
parser.add_argument(
"--retries",
def _save_console(gym: GymPort, screenshot_dir: Path | None, text: str, name: str) -> None: type=_at_least_one,
"""@brief Save console text as an image when a folder was asked for. default=constants.RETRY_COUNT,
metavar="N",
@param gym The gym site. help=(
@param screenshot_dir The folder, or None for no screenshots. "attempts for the login, each booking and the My Bookings page when the site "
@param text The console text to draw. f"drops a request (default: {constants.RETRY_COUNT}, at least 1)"
@param name The file name. ),
""" )
if screenshot_dir is not None: args = parser.parse_args(argv)
gym.save_text_screenshot(text, screenshot_dir / name) if args.keep_open is None:
args.keep_open = not args.headless
if args.profile_dir is None:
def _run( args.profile_dir = default_profile_dir()
gym: GymPort, args: argparse.Namespace, recorder: ConsoleRecorder, credentials: Credentials return args
) -> None:
"""@brief Open the site, log in and book the classes, saving the step images.
def _use_utf8_output() -> None:
Books every Tuesday and Thursday 6pm class and logs the summary and the detailed list; """@brief Make stdout UTF-8 so the check mark of the messages prints on Windows."""
with `--first-tuesday-only` it books the next Tuesday class only. reconfigure = getattr(sys.stdout, "reconfigure", None)
if reconfigure is not None:
reconfigure(encoding="utf-8")
@param gym The gym site.
@param args The parsed options.
@param recorder The recorded console lines, drawn for step 4. def _save_step(gym: GymPort, screenshot_dir: Path | None, name: str) -> None:
@param credentials The account to log in with. """@brief Save a screenshot when a folder was asked for.
@exception GymError A step failed.
""" @param gym The gym site.
gym.open_site() @param screenshot_dir The folder, or None for no screenshots.
_save_step(gym, args.screenshot_dir, "step-1.png") @param name The file name.
login(gym, credentials) """
_save_step(gym, args.screenshot_dir, "step-2.png") if screenshot_dir is not None:
if args.first_tuesday_only: gym.save_screenshot(screenshot_dir / name)
book_next_tuesday(gym)
_save_step(gym, args.screenshot_dir, constants.SCREENSHOT_STEP_3)
_save_console(gym, args.screenshot_dir, recorder.text, constants.SCREENSHOT_STEP_4) def _save_console(gym: GymPort, screenshot_dir: Path | None, text: str, name: str) -> None:
return """@brief Save console text as an image when a folder was asked for.
window_saved = False
@param gym The gym site.
def save_window_once(_: ClassResult) -> None: @param screenshot_dir The folder, or None for no screenshots.
nonlocal window_saved @param text The console text to draw.
if not window_saved: @param name The file name.
_save_step(gym, args.screenshot_dir, constants.SCREENSHOT_STEP_3) """
window_saved = True if screenshot_dir is not None:
gym.save_text_screenshot(text, screenshot_dir / name)
results = book_target_classes(gym, save_window_once)
if not window_saved:
_save_step(gym, args.screenshot_dir, constants.SCREENSHOT_STEP_3) def _run(
_save_console(gym, args.screenshot_dir, recorder.text, constants.SCREENSHOT_STEP_4) gym: GymPort,
summary = format_summary(results) args: argparse.Namespace,
logger.info("%s", summary) recorder: ConsoleRecorder,
_save_console(gym, args.screenshot_dir, summary, constants.SCREENSHOT_STEP_5) credentials: Credentials,
detail = format_detail_list(results) sleep: Sleep,
logger.info("%s", detail) ) -> bool:
_save_console(gym, args.screenshot_dir, f"{summary}\n{detail}", constants.SCREENSHOT_STEP_6) """@brief Open the site, log in, book the classes and verify them, saving the step images.
Books every Tuesday and Thursday 6pm class, logs the summary and the detailed list and
def main( verifies the bookings on the My Bookings page; with `--first-tuesday-only` it books the
argv: Sequence[str] | None = None, next Tuesday class only and does not verify.
gym_factory: GymFactory = selenium_gym_factory,
) -> int: @param gym The gym site.
"""@brief Start the bot: open the site, log in and book the next Tuesday 6pm class. @param args The parsed options.
@param recorder The recorded console lines, drawn for step 4.
@param argv Arguments without the program name; None reads `sys.argv`. @param credentials The account to log in with.
@param gym_factory Builds the gym site; tests pass a fake. @param sleep Waits between two attempts of a retried step.
@return Process exit code: 0 done, 1 a step failed, 2 credentials missing. @return False when the My Bookings page does not match the bookings made, else True.
""" @exception GymError A step failed on every attempt.
logging.basicConfig(level=logging.INFO, format="%(message)s") """
root = logging.getLogger() gym.open_site()
root.setLevel(logging.INFO) _save_step(gym, args.screenshot_dir, "step-1.png")
recorder = ConsoleRecorder() login_with_retry(gym, credentials, args.retries, sleep)
root.addHandler(recorder) _save_step(gym, args.screenshot_dir, "step-2.png")
try: if args.first_tuesday_only:
return _main(argv, gym_factory, recorder) book_next_tuesday(gym, args.retries, sleep)
finally: _save_step(gym, args.screenshot_dir, constants.SCREENSHOT_STEP_3)
root.removeHandler(recorder) _save_console(gym, args.screenshot_dir, recorder.text, constants.SCREENSHOT_STEP_4)
return True
window_saved = False
def _main(argv: Sequence[str] | None, gym_factory: GymFactory, recorder: ConsoleRecorder) -> int:
"""@brief The body of `main`, run while the recorder is attached. def save_window_once(_: ClassResult) -> None:
nonlocal window_saved
@param argv Arguments without the program name. if not window_saved:
@param gym_factory Builds the gym site. _save_step(gym, args.screenshot_dir, constants.SCREENSHOT_STEP_3)
@param recorder The console recorder. window_saved = True
@return Process exit code.
""" results = book_target_classes(gym, save_window_once, args.retries, sleep)
_use_utf8_output() if not window_saved:
args = parse_args(argv) _save_step(gym, args.screenshot_dir, constants.SCREENSHOT_STEP_3)
try: _save_console(gym, args.screenshot_dir, recorder.text, constants.SCREENSHOT_STEP_4)
credentials = load_credentials(os.environ) summary = format_summary(results)
except MissingCredentialsError as error: logger.info("%s", summary)
logger.error("%s", error) _save_console(gym, args.screenshot_dir, summary, constants.SCREENSHOT_STEP_5)
return 2 detail = format_detail_list(results)
gym = gym_factory(args.headless, args.profile_dir, args.keep_open) logger.info("%s", detail)
try: _save_console(gym, args.screenshot_dir, f"{summary}\n{detail}", constants.SCREENSHOT_STEP_6)
_run(gym, args, recorder, credentials) verification = verify_bookings(gym, results, args.retries, sleep)
except LoginFailedError as error: _save_step(gym, args.screenshot_dir, constants.SCREENSHOT_STEP_7)
logger.error("Login failed: %s", error) verification_text = format_verification(verification)
return 1 logger.info("%s", verification_text)
except GymError as error: _save_console(gym, args.screenshot_dir, verification_text, constants.SCREENSHOT_STEP_8)
logger.error("Failed: %s", error) _save_console(gym, args.screenshot_dir, recorder.text, constants.SCREENSHOT_STEP_9)
return 1 return verification.is_success
finally:
if not args.keep_open:
gym.close() def main(
return 0 argv: Sequence[str] | None = None,
gym_factory: GymFactory = selenium_gym_factory,
sleep: Sleep = time.sleep,
) -> int:
"""@brief Start the bot: open the site, log in, book the classes and verify them.
@param argv Arguments without the program name; None reads `sys.argv`.
@param gym_factory Builds the gym site; tests pass a fake.
@param sleep Waits between two attempts of a retried step; tests pass a recorder.
@return Process exit code: 0 done, 1 a step failed, 2 credentials missing, 3 mismatch.
"""
logging.basicConfig(level=logging.INFO, format="%(message)s")
root = logging.getLogger()
root.setLevel(logging.INFO)
recorder = ConsoleRecorder()
root.addHandler(recorder)
try:
return _main(argv, gym_factory, recorder, sleep)
finally:
root.removeHandler(recorder)
def _main(
argv: Sequence[str] | None, gym_factory: GymFactory, recorder: ConsoleRecorder, sleep: Sleep
) -> int:
"""@brief The body of `main`, run while the recorder is attached.
@param argv Arguments without the program name.
@param gym_factory Builds the gym site.
@param recorder The console recorder.
@param sleep Waits between two attempts of a retried step.
@return Process exit code.
"""
_use_utf8_output()
args = parse_args(argv)
try:
credentials = load_credentials(os.environ)
except MissingCredentialsError as error:
logger.error("%s", error)
return 2
gym = gym_factory(args.headless, args.profile_dir, args.keep_open)
try:
verified = _run(gym, args, recorder, credentials, sleep)
except LoginFailedError as error:
logger.error("Login failed: %s", error)
return 1
except GymError as error:
logger.error("Failed: %s", error)
return 1
finally:
if not args.keep_open:
gym.close()
return 0 if verified else constants.EXIT_VERIFICATION_MISMATCH
+76
View File
@@ -18,6 +18,31 @@ WAIT_TIMEOUT_SECONDS: Final[int] = 10
## Attempts of the `retry` wrapper before it gives up (lecture step 9). ## Attempts of the `retry` wrapper before it gives up (lecture step 9).
RETRY_COUNT: Final[int] = 7 RETRY_COUNT: Final[int] = 7
## Seconds to wait between two attempts of the `retry` wrapper.
RETRY_DELAY_SECONDS: Final[float] = 0.5
## Name of a retried step that was given no description.
RETRY_DEFAULT_DESCRIPTION: Final[str] = "operation"
## Descriptions of the retried steps, as they appear in the log.
RETRY_DESCRIPTION_LOGIN: Final[str] = "login"
RETRY_DESCRIPTION_BOOK: Final[str] = "book {name} on {day}"
RETRY_DESCRIPTION_BOOKINGS: Final[str] = "get my bookings"
## Messages of the `retry` wrapper.
MESSAGE_RETRY_FAILED: Final[str] = (
"⚠ Attempt {attempt}/{retries} failed: {description}: {error}. Retrying..."
)
MESSAGE_RETRY_SUCCEEDED: Final[str] = "✓ {description} succeeded after {attempts} attempts"
MESSAGE_RETRIES_EXHAUSTED: Final[str] = "{description} failed after {retries} attempts: {error}"
MESSAGE_RETRIES_INVALID: Final[str] = "retries must be at least 1, got {retries}"
## Error raised when a class card is no longer on the schedule while a booking is repeated.
MESSAGE_CLASS_NOT_ON_SCHEDULE: Final[str] = "class {card_id} not found on the schedule"
## File name of the screenshot of lecture step 9 (the console with the retry warnings).
SCREENSHOT_STEP_9: Final[str] = "step-9.png"
## Weekdays of the classes the bot books, as the site abbreviates them. ## Weekdays of the classes the bot books, as the site abbreviates them.
TARGET_WEEKDAYS: Final[tuple[str, ...]] = ("Tue", "Thu") TARGET_WEEKDAYS: Final[tuple[str, ...]] = ("Tue", "Thu")
@@ -152,3 +177,54 @@ CONSOLE_PAGE_TEMPLATE: Final[str] = (
## Prefix of the `data:` URL that carries the console page. ## Prefix of the `data:` URL that carries the console page.
CONSOLE_PAGE_URL_PREFIX: Final[str] = "data:text/html;charset=utf-8," CONSOLE_PAGE_URL_PREFIX: Final[str] = "data:text/html;charset=utf-8,"
## File names of the screenshots of lecture steps 7 and 8.
SCREENSHOT_STEP_7: Final[str] = "step-7.png"
SCREENSHOT_STEP_8: Final[str] = "step-8.png"
## Process exit code when the My Bookings page does not match what the bot did.
EXIT_VERIFICATION_MISMATCH: Final[int] = 3
## Id of the navigation link that opens the My Bookings page.
MY_BOOKINGS_LINK_ID: Final[str] = "my-bookings-link"
## Id of the My Bookings page, shown after its bookings were loaded.
MY_BOOKINGS_PAGE_ID: Final[str] = "my-bookings-page"
## Id of the element shown when the account has no booking and no waitlist entry.
EMPTY_BOOKINGS_STATE_ID: Final[str] = "empty-bookings-state"
## CSS selectors of the confirmed booking cards and the waitlist cards.
BOOKING_CARD_CSS: Final[str] = "#confirmed-bookings-section div[id^='booking-card-']"
WAITLIST_CARD_CSS: Final[str] = "#waitlist-section div[id^='waitlist-card-']"
## Prefixes of the element ids of a booking card and of a waitlist card.
BOOKING_CARD_ID_PREFIX: Final[str] = "booking-card-"
BOOKING_NAME_ID_PREFIX: Final[str] = "booking-class-name-"
WAITLIST_CARD_ID_PREFIX: Final[str] = "waitlist-card-"
WAITLIST_NAME_ID_PREFIX: Final[str] = "waitlist-class-name-"
## XPath, relative to a card, of the paragraph "When: Tue, Aug 12, 6:00 PM".
WHEN_XPATH: Final[str] = ".//p[strong[starts-with(normalize-space(), 'When:')]]"
## Prefix of the date and time text of a booking card.
WHEN_TEXT_PREFIX: Final[str] = "When:"
## Suffix the My Bookings page adds to the name of a waitlist entry.
WAITLIST_NAME_SUFFIX: Final[str] = " (Waitlist)"
## Message of the error raised when the My Bookings page shows no booking at all.
MESSAGE_NO_BOOKINGS_FOUND: Final[str] = "The My Bookings page shows no booking or waitlist entry"
## Lines of the verification on the My Bookings page (lecture steps 7 and 8).
VERIFY_TITLE: Final[str] = "--- VERIFYING ON MY BOOKINGS PAGE ---"
VERIFY_LINE_VERIFIED: Final[str] = " ✓ Verified: {name}"
VERIFY_RESULT_TITLE: Final[str] = "--- VERIFICATION RESULT ---"
VERIFY_EXPECTED: Final[str] = "Expected: {count} {noun}"
VERIFY_FOUND: Final[str] = "Found: {count} {noun}"
VERIFY_NOUN_SINGULAR: Final[str] = "booking"
VERIFY_NOUN_PLURAL: Final[str] = "bookings"
VERIFY_SUCCESS: Final[str] = "✅ SUCCESS: All bookings verified!"
VERIFY_MISMATCH: Final[str] = "❌ MISMATCH: Missing {count} bookings"
VERIFY_LINE_MISSING: Final[str] = " ✗ Missing: {entry}"
VERIFY_LINE_UNEXPECTED: Final[str] = " ? Unexpected: {entry}"
@@ -0,0 +1,88 @@
"""@file booking_entry.py
@brief One entry of the My Bookings page: a confirmed booking or a waitlist entry.
"""
import re
from dataclasses import dataclass
from snack_and_lift import constants
from snack_and_lift.domain.day_label import parse_day_label
from snack_and_lift.domain.errors import PageFormatError
from snack_and_lift.domain.rules import parse_time
_WHEN = re.compile(r"(?P<date>[A-Z][a-z]{2}, [A-Z][a-z]{2} \d{1,2}), (?P<time>.+)")
@dataclass(frozen=True, slots=True)
class BookingEntry:
"""@brief A booking or waitlist entry, as read from the My Bookings page.
`name` is the class name WITHOUT the ` (Waitlist)` suffix the page adds to a waitlist
entry; `waitlisted` carries that fact. `when_text` is the page text without the
`When:` prefix, for example `Tue, Aug 12, 6:00 PM`.
"""
name: str
when_text: str
waitlisted: bool
@classmethod
def from_page(cls, heading: str, when_text: str, waitlisted: bool) -> "BookingEntry":
"""@brief Build an entry from the texts of its card.
@param heading The class name as the page shows it, possibly with the waitlist suffix.
@param when_text The date and time paragraph, with or without the `When:` prefix.
@param waitlisted True for a card of the waitlist section.
@return The entry, with the suffix and the prefix removed.
"""
name = heading.strip().removesuffix(constants.WAITLIST_NAME_SUFFIX.strip()).strip()
when = when_text.strip().removeprefix(constants.WHEN_TEXT_PREFIX).strip()
return cls(name, when, waitlisted)
def _split_when(self) -> tuple[str, str]:
"""@brief Split the date text from the time text.
@return The date text (`Tue, Aug 12`) and the time text (`6:00 PM`).
@exception PageFormatError The text has no date and time.
"""
match = _WHEN.fullmatch(self.when_text)
if match is None:
raise PageFormatError(f"Unexpected booking date and time: {self.when_text!r}")
return match["date"], match["time"]
@property
def weekday(self) -> str:
"""@brief The weekday of the entry (`Tue`).
@exception PageFormatError The date and time text is not valid.
"""
return parse_day_label(self._split_when()[0])[0]
@property
def date_text(self) -> str:
"""@brief The date of the entry (`Tue, Aug 12`), comparable with a class card's date.
@exception PageFormatError The date and time text is not valid.
"""
return parse_day_label(self._split_when()[0])[1]
@property
def time(self) -> tuple[int, int]:
"""@brief The hour (0-23) and the minute of the entry.
@exception PageFormatError The date and time text is not valid.
"""
return parse_time(self._split_when()[1])
def is_target_entry(entry: BookingEntry) -> bool:
"""@brief Tell whether the entry is a class the bot books: a target weekday at the target time.
@param entry The entry.
@return True for Tuesday or Thursday at `TARGET_HOUR`:`TARGET_MINUTE`.
@exception PageFormatError The date and time text is not valid.
"""
return entry.weekday in constants.TARGET_WEEKDAYS and entry.time == (
constants.TARGET_HOUR,
constants.TARGET_MINUTE,
)
+12
View File
@@ -21,3 +21,15 @@ class UnknownButtonError(GymError):
class PageFormatError(GymError): class PageFormatError(GymError):
"""@brief A text read from the page (a day label or a time) has an unexpected format.""" """@brief A text read from the page (a day label or a time) has an unexpected format."""
class RetriesExhaustedError(GymError):
"""@brief A step failed on every attempt; the cause is the error of the last attempt."""
class NoBookingsFoundError(GymError):
"""@brief The My Bookings page shows no booking where the bot expects some.
The project's counterpart of Selenium's `NoSuchElementException`, which the lecture asks
the student to raise when the bookings that should be there are missing.
"""
+108
View File
@@ -0,0 +1,108 @@
"""@file verification.py
@brief Lecture steps 7 and 8: compare what the bot did with the My Bookings page.
"""
from collections import Counter
from collections.abc import Sequence
from dataclasses import dataclass
from snack_and_lift import constants
from snack_and_lift.domain.booking_entry import BookingEntry
from snack_and_lift.domain.models import ClassResult, Outcome
_WAITLISTED_OUTCOMES = (Outcome.NEW_WAITLIST, Outcome.ALREADY_WAITLISTED)
## An entry is identified by its class name, its date text and whether it is waitlisted.
_Key = tuple[str, str, bool]
@dataclass(frozen=True, slots=True)
class VerificationResult:
"""@brief What the My Bookings page shows compared with what the bot did.
`verified`, `missing` and `unexpected` hold display names, with the waitlist suffix for a
waitlist entry; `missing` and `unexpected` also say the day (`Spin Class on Tue, Aug 12`).
"""
expected: int
found: int
verified: list[str]
missing: list[str]
unexpected: list[str]
@property
def is_success(self) -> bool:
"""@brief True when every expected entry was found and nothing else was."""
return self.expected == self.found and not self.missing and not self.unexpected
def _display_name(name: str, waitlisted: bool) -> str:
"""@brief The name as the page shows it."""
return name + constants.WAITLIST_NAME_SUFFIX if waitlisted else name
def _expected_keys(results: Sequence[ClassResult]) -> list[_Key]:
"""@brief The entries the bot's results say must be on the page."""
return [(r.card.name, r.card.date_text, r.outcome in _WAITLISTED_OUTCOMES) for r in results]
def _entry_keys(entries: Sequence[BookingEntry]) -> list[_Key]:
"""@brief The identity of each entry found on the page."""
return [(e.name, e.date_text, e.waitlisted) for e in entries]
def _describe(key: _Key) -> str:
"""@brief Write an entry for the missing and unexpected lines."""
name, day, waitlisted = key
return f"{_display_name(name, waitlisted)} on {day}"
def build_verification(
results: Sequence[ClassResult], entries: Sequence[BookingEntry]
) -> VerificationResult:
"""@brief Compare the results of the booking with the target entries of the page.
@param results What happened to each class; waitlisted when the outcome is a waitlist one.
@param entries The Tuesday and Thursday 6pm entries found on the page.
@return The comparison; the counts are the numbers of results and of entries.
@exception PageFormatError An entry or a card has no valid date.
"""
expected = _expected_keys(results)
remaining = Counter(_entry_keys(entries))
verified: list[str] = []
missing: list[str] = []
for key in expected:
if remaining[key] > 0:
remaining[key] -= 1
verified.append(_display_name(key[0], key[2]))
else:
missing.append(_describe(key))
unexpected = [_describe(key) for key in remaining.elements()]
return VerificationResult(len(expected), len(entries), verified, missing, unexpected)
def _count_text(template: str, count: int) -> str:
"""@brief Write a count with `booking` in the singular for 1."""
noun = constants.VERIFY_NOUN_SINGULAR if count == 1 else constants.VERIFY_NOUN_PLURAL
return template.format(count=count, noun=noun)
def format_verification(result: VerificationResult) -> str:
"""@brief Write the verification block.
@param result The comparison.
@return The block, starting with `--- VERIFYING ON MY BOOKINGS PAGE ---`.
"""
lines = [constants.VERIFY_TITLE]
lines.extend(constants.VERIFY_LINE_VERIFIED.format(name=name) for name in result.verified)
lines.append("")
lines.append(constants.VERIFY_RESULT_TITLE)
lines.append(_count_text(constants.VERIFY_EXPECTED, result.expected))
lines.append(_count_text(constants.VERIFY_FOUND, result.found))
if result.is_success:
lines.append(constants.VERIFY_SUCCESS)
return "\n".join(lines)
lines.append(constants.VERIFY_MISMATCH.format(count=result.expected - result.found))
lines.extend(constants.VERIFY_LINE_MISSING.format(entry=text) for text in result.missing)
lines.extend(constants.VERIFY_LINE_UNEXPECTED.format(entry=text) for text in result.unexpected)
return "\n".join(lines)
+262 -194
View File
@@ -1,194 +1,262 @@
"""@file selenium_gym.py """@file selenium_gym.py
@brief The Selenium adapter of the GymPort. @brief The Selenium adapter of the GymPort.
""" """
import functools import functools
import html import html
import logging import logging
import urllib.parse import urllib.parse
from collections.abc import Callable from collections.abc import Callable
from pathlib import Path from pathlib import Path
from string import Template from string import Template
from typing import Concatenate, Literal from typing import Concatenate, Literal
from selenium.common.exceptions import ( from selenium.common.exceptions import (
ElementClickInterceptedException, ElementClickInterceptedException,
NoSuchElementException, NoSuchElementException,
StaleElementReferenceException, StaleElementReferenceException,
TimeoutException, TimeoutException,
) )
from selenium.webdriver.common.by import By from selenium.webdriver.common.by import By
from selenium.webdriver.remote.webdriver import WebDriver from selenium.webdriver.remote.webdriver import WebDriver
from selenium.webdriver.remote.webelement import WebElement from selenium.webdriver.remote.webelement import WebElement
from selenium.webdriver.support import expected_conditions as ec from selenium.webdriver.support import expected_conditions as ec
from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support.ui import WebDriverWait
from snack_and_lift import constants from snack_and_lift import constants
from snack_and_lift.domain.errors import GymError, LoginFailedError from snack_and_lift.domain.booking_entry import BookingEntry
from snack_and_lift.domain.models import ButtonState, ClassCard from snack_and_lift.domain.errors import GymError, LoginFailedError
from snack_and_lift.domain.models import ButtonState, ClassCard
logger = logging.getLogger(__name__)
logger = logging.getLogger(__name__)
_SELENIUM_ERRORS = (
TimeoutException, _SELENIUM_ERRORS = (
NoSuchElementException, TimeoutException,
StaleElementReferenceException, NoSuchElementException,
ElementClickInterceptedException, StaleElementReferenceException,
) ElementClickInterceptedException,
)
def _translate_errors[S, **P, R](
method: Callable[Concatenate[S, P], R], def _translate_errors[S, **P, R](
) -> Callable[Concatenate[S, P], R]: method: Callable[Concatenate[S, P], R],
"""@brief Turn Selenium errors of a method into GymError. ) -> Callable[Concatenate[S, P], R]:
"""@brief Turn Selenium errors of a method into GymError.
@param method The adapter method to wrap.
@return The wrapped method. @param method The adapter method to wrap.
""" @return The wrapped method.
"""
@functools.wraps(method)
def wrapper(self: S, *args: P.args, **kwargs: P.kwargs) -> R: @functools.wraps(method)
try: def wrapper(self: S, *args: P.args, **kwargs: P.kwargs) -> R:
return method(self, *args, **kwargs) try:
except _SELENIUM_ERRORS as error: return method(self, *args, **kwargs)
raise GymError(f"{method.__name__} failed: {type(error).__name__}") from error except _SELENIUM_ERRORS as error:
raise GymError(f"{method.__name__} failed: {type(error).__name__}") from error
return wrapper
return wrapper
class SeleniumGym:
"""@brief The gym site driven by a Selenium WebDriver.""" class SeleniumGym:
"""@brief The gym site driven by a Selenium WebDriver."""
def __init__(self, driver: WebDriver, base_url: str = constants.GYM_URL) -> None:
"""@brief Wrap a driver. def __init__(self, driver: WebDriver, base_url: str = constants.GYM_URL) -> None:
"""@brief Wrap a driver.
@param driver A running WebDriver.
@param base_url The address of the home page. @param driver A running WebDriver.
""" @param base_url The address of the home page.
self._driver = driver """
self._base_url = base_url self._driver = driver
self._wait = WebDriverWait(driver, constants.WAIT_TIMEOUT_SECONDS) self._base_url = base_url
self._wait = WebDriverWait(driver, constants.WAIT_TIMEOUT_SECONDS)
@_translate_errors
def open_site(self) -> None: @_translate_errors
"""@brief Open the home page and wait for the login button. def open_site(self) -> None:
"""@brief Open the home page and wait for the login button.
@exception GymError The page did not load.
""" @exception GymError The page did not load.
self._driver.get(self._base_url) """
self._wait.until(ec.presence_of_element_located((By.ID, constants.NAV_LOGIN_BUTTON_ID))) self._driver.get(self._base_url)
self._wait.until(ec.presence_of_element_located((By.ID, constants.NAV_LOGIN_BUTTON_ID)))
@_translate_errors
def log_in(self, email: str, password: str) -> None: @_translate_errors
"""@brief Fill in the login form and wait for the schedule or an error. def log_in(self, email: str, password: str) -> None:
"""@brief Fill in the login form and wait for the schedule or an error.
@param email The account e-mail.
@param password The account password. @param email The account e-mail.
@exception LoginFailedError The site shows a login error. @param password The account password.
@exception GymError The page did not behave as expected. @exception LoginFailedError The site shows a login error.
""" @exception GymError The page did not behave as expected.
self._wait.until(ec.element_to_be_clickable((By.ID, constants.NAV_LOGIN_BUTTON_ID))).click() """
self._wait.until( # After a failed login the form is still shown, so a repeated login skips the nav button.
ec.presence_of_element_located((By.ID, constants.LOGIN_EMAIL_INPUT_ID)) if not self._driver.find_elements(By.ID, constants.LOGIN_EMAIL_INPUT_ID):
).send_keys(email) self._wait.until(
self._driver.find_element(By.ID, constants.LOGIN_PASSWORD_INPUT_ID).send_keys(password) ec.element_to_be_clickable((By.ID, constants.NAV_LOGIN_BUTTON_ID))
self._driver.find_element(By.ID, constants.LOGIN_SUBMIT_BUTTON_ID).click() ).click()
self._wait.until( email_input = self._wait.until(
ec.any_of( ec.presence_of_element_located((By.ID, constants.LOGIN_EMAIL_INPUT_ID))
ec.presence_of_element_located((By.ID, constants.SCHEDULE_PAGE_ID)), )
ec.presence_of_element_located((By.ID, constants.ERROR_MESSAGE_ID)), password_input = self._driver.find_element(By.ID, constants.LOGIN_PASSWORD_INPUT_ID)
) for field, value in ((email_input, email), (password_input, password)):
) field.clear()
errors = self._driver.find_elements(By.ID, constants.ERROR_MESSAGE_ID) field.send_keys(value)
if errors: self._driver.find_element(By.ID, constants.LOGIN_SUBMIT_BUTTON_ID).click()
raise LoginFailedError(errors[0].text) self._wait.until(
ec.any_of(
@_translate_errors ec.presence_of_element_located((By.ID, constants.SCHEDULE_PAGE_ID)),
def list_class_cards(self) -> list[ClassCard]: ec.presence_of_element_located((By.ID, constants.ERROR_MESSAGE_ID)),
"""@brief Read every class card; the day comes from the ancestor day group. )
)
@return The classes in page order. errors = self._driver.find_elements(By.ID, constants.ERROR_MESSAGE_ID)
@exception GymError The schedule could not be read. if errors:
""" raise LoginFailedError(errors[0].text)
self._wait.until(
ec.presence_of_element_located((By.CSS_SELECTOR, constants.CLASS_CARD_CSS)) @_translate_errors
) def list_class_cards(self) -> list[ClassCard]:
elements = self._driver.find_elements(By.CSS_SELECTOR, constants.CLASS_CARD_CSS) """@brief Read every class card; the day comes from the ancestor day group.
return [self._read_card(element) for element in elements]
@return The classes in page order.
def _read_card(self, element: WebElement) -> ClassCard: @exception GymError The schedule could not be read.
"""@brief Turn one card element into plain data. """
self._wait.until(
@param element The card. ec.presence_of_element_located((By.CSS_SELECTOR, constants.CLASS_CARD_CSS))
@return The class. )
""" elements = self._driver.find_elements(By.CSS_SELECTOR, constants.CLASS_CARD_CSS)
card_id = (element.get_attribute("id") or "").removeprefix(constants.CLASS_CARD_ID_PREFIX) return [self._read_card(element) for element in elements]
buttons = element.find_elements(By.ID, constants.BOOK_BUTTON_ID_PREFIX + card_id)
labels = element.find_elements(By.ID, constants.COMPLETED_LABEL_ID_PREFIX + card_id) def _read_card(self, element: WebElement) -> ClassCard:
control = (buttons or labels or [None])[0] """@brief Turn one card element into plain data.
return ClassCard(
card_id=card_id, @param element The card.
name=element.find_element(By.ID, constants.CLASS_NAME_ID_PREFIX + card_id).text, @return The class.
day_label=element.find_element(By.XPATH, constants.DAY_TITLE_XPATH).text, """
time_text=element.find_element(By.ID, constants.CLASS_TIME_ID_PREFIX + card_id).text, card_id = (element.get_attribute("id") or "").removeprefix(constants.CLASS_CARD_ID_PREFIX)
button_text=control.text if control is not None else "", buttons = element.find_elements(By.ID, constants.BOOK_BUTTON_ID_PREFIX + card_id)
) labels = element.find_elements(By.ID, constants.COMPLETED_LABEL_ID_PREFIX + card_id)
control = (buttons or labels or [None])[0]
@_translate_errors return ClassCard(
def press_class_button(self, card_id: str) -> str: card_id=card_id,
"""@brief Scroll to the button, click it and wait for the result or the error box. name=element.find_element(By.ID, constants.CLASS_NAME_ID_PREFIX + card_id).text,
day_label=element.find_element(By.XPATH, constants.DAY_TITLE_XPATH).text,
@param card_id The id of the card without its prefix. time_text=element.find_element(By.ID, constants.CLASS_TIME_ID_PREFIX + card_id).text,
@return The new text of the button. button_text=control.text if control is not None else "",
@exception GymError The site shows an error for the class, or nothing settled in time. )
"""
button_id = constants.BOOK_BUTTON_ID_PREFIX + card_id @_translate_errors
button = self._wait.until(ec.presence_of_element_located((By.ID, button_id))) def press_class_button(self, card_id: str) -> str:
self._driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", button) """@brief Scroll to the button, click it and wait for the result or the error box.
before = button.text
self._wait.until(ec.element_to_be_clickable((By.ID, button_id))).click() @param card_id The id of the card without its prefix.
@return The new text of the button.
def settled(driver: WebDriver) -> tuple[str, str] | Literal[False]: @exception GymError The site shows an error for the class, or nothing settled in time.
errors = driver.find_elements(By.ID, constants.CLASS_ERROR_ID_PREFIX + card_id) """
text = driver.find_element(By.ID, button_id).text button_id = constants.BOOK_BUTTON_ID_PREFIX + card_id
if errors: button = self._wait.until(ec.presence_of_element_located((By.ID, button_id)))
return text, errors[0].text or "class error" self._driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", button)
if text != before and ButtonState.from_text(text) is not ButtonState.BUSY: before = button.text
return text, "" self._wait.until(ec.element_to_be_clickable((By.ID, button_id))).click()
return False
def settled(driver: WebDriver) -> tuple[str, str] | Literal[False]:
text, error = self._wait.until(settled) errors = driver.find_elements(By.ID, constants.CLASS_ERROR_ID_PREFIX + card_id)
if error: text = driver.find_element(By.ID, button_id).text
raise GymError(error) if errors:
return text return text, errors[0].text or "class error"
if text != before and ButtonState.from_text(text) is not ButtonState.BUSY:
@_translate_errors return text, ""
def save_text_screenshot(self, text: str, path: Path) -> None: return False
"""@brief Draw console text in a temporary tab and save a screenshot of it.
text, error = self._wait.until(settled)
A tab is used so that the schedule page keeps its state. if error:
raise GymError(error)
@param text The lines to draw. return text
@param path The PNG file to write; its folders are created.
""" @_translate_errors
page = Template(constants.CONSOLE_PAGE_TEMPLATE).substitute(text=html.escape(text)) def open_my_bookings(self) -> None:
original = self._driver.current_window_handle """@brief Click the My Bookings link and wait for the page.
self._driver.switch_to.new_window("tab")
try: @exception GymError The page did not load.
self._driver.get(constants.CONSOLE_PAGE_URL_PREFIX + urllib.parse.quote(page)) """
self.save_screenshot(path) self._wait.until(ec.element_to_be_clickable((By.ID, constants.MY_BOOKINGS_LINK_ID))).click()
finally: self._wait.until(ec.presence_of_element_located((By.ID, constants.MY_BOOKINGS_PAGE_ID)))
self._driver.close()
self._driver.switch_to.window(original) @_translate_errors
def list_my_bookings(self) -> list[BookingEntry]:
def save_screenshot(self, path: Path) -> None: """@brief Wait for the bookings to load, then read the confirmed and waitlist cards.
"""@brief Save a screenshot of the current page.
@return The entries, confirmed ones first; empty when the page shows its empty state.
@param path The PNG file to write; its folders are created. @exception GymError The page shows an error box, or nothing was shown in time.
""" """
path.parent.mkdir(parents=True, exist_ok=True) self._wait.until(
self._driver.save_screenshot(str(path)) ec.any_of(
ec.presence_of_element_located((By.CSS_SELECTOR, constants.BOOKING_CARD_CSS)),
def close(self) -> None: ec.presence_of_element_located((By.CSS_SELECTOR, constants.WAITLIST_CARD_CSS)),
"""@brief Quit the browser.""" ec.presence_of_element_located((By.ID, constants.EMPTY_BOOKINGS_STATE_ID)),
self._driver.quit() ec.presence_of_element_located((By.ID, constants.ERROR_MESSAGE_ID)),
)
)
errors = self._driver.find_elements(By.ID, constants.ERROR_MESSAGE_ID)
if errors:
raise GymError(errors[0].text or "My Bookings error")
confirmed = self._driver.find_elements(By.CSS_SELECTOR, constants.BOOKING_CARD_CSS)
waitlist = self._driver.find_elements(By.CSS_SELECTOR, constants.WAITLIST_CARD_CSS)
return [
self._read_entry(
element, constants.BOOKING_CARD_ID_PREFIX, constants.BOOKING_NAME_ID_PREFIX, False
)
for element in confirmed
] + [
self._read_entry(
element, constants.WAITLIST_CARD_ID_PREFIX, constants.WAITLIST_NAME_ID_PREFIX, True
)
for element in waitlist
]
@staticmethod
def _read_entry(
element: WebElement, card_prefix: str, name_prefix: str, waitlisted: bool
) -> BookingEntry:
"""@brief Turn one booking or waitlist card into plain data.
@param element The card.
@param card_prefix The prefix of the card id.
@param name_prefix The prefix of the id of the card's heading.
@param waitlisted True for a card of the waitlist section.
@return The entry.
"""
card_id = (element.get_attribute("id") or "").removeprefix(card_prefix)
return BookingEntry.from_page(
heading=element.find_element(By.ID, name_prefix + card_id).text,
when_text=element.find_element(By.XPATH, constants.WHEN_XPATH).text,
waitlisted=waitlisted,
)
@_translate_errors
def save_text_screenshot(self, text: str, path: Path) -> None:
"""@brief Draw console text in a temporary tab and save a screenshot of it.
A tab is used so that the schedule page keeps its state.
@param text The lines to draw.
@param path The PNG file to write; its folders are created.
"""
page = Template(constants.CONSOLE_PAGE_TEMPLATE).substitute(text=html.escape(text))
original = self._driver.current_window_handle
self._driver.switch_to.new_window("tab")
try:
self._driver.get(constants.CONSOLE_PAGE_URL_PREFIX + urllib.parse.quote(page))
self.save_screenshot(path)
finally:
self._driver.close()
self._driver.switch_to.window(original)
def save_screenshot(self, path: Path) -> None:
"""@brief Save a screenshot of the current page.
@param path The PNG file to write; its folders are created.
"""
path.parent.mkdir(parents=True, exist_ok=True)
self._driver.save_screenshot(str(path))
def close(self) -> None:
"""@brief Quit the browser."""
self._driver.quit()
+11
View File
@@ -12,8 +12,19 @@
// Dummy account of this fixture only; it is not a real account. // Dummy account of this fixture only; it is not a real account.
const ACCEPTED_EMAIL = "fixture@example.invalid"; const ACCEPTED_EMAIL = "fixture@example.invalid";
const ACCEPTED_PASSWORD = "fixture-password"; const ACCEPTED_PASSWORD = "fixture-password";
// Imitates the network simulation: ?fail=N drops the first N submits after a short delay.
let failuresLeft = parseInt(new URLSearchParams(window.location.search).get("fail") || "0", 10);
document.getElementById("login-form").addEventListener("submit", function (event) { document.getElementById("login-form").addEventListener("submit", function (event) {
event.preventDefault(); event.preventDefault();
document.getElementById("message-area").innerHTML = "";
if (failuresLeft > 0) {
failuresLeft -= 1;
setTimeout(function () {
document.getElementById("message-area").innerHTML =
'<div id="error-message">Network request failed. Please try again.</div>';
}, 100);
return;
}
const email = document.getElementById("email-input").value; const email = document.getElementById("email-input").value;
const password = document.getElementById("password-input").value; const password = document.getElementById("password-input").value;
if (email === ACCEPTED_EMAIL && password === ACCEPTED_PASSWORD) { if (email === ACCEPTED_EMAIL && password === ACCEPTED_PASSWORD) {
+53
View File
@@ -0,0 +1,53 @@
<!DOCTYPE html>
<html lang="en">
<head><meta charset="utf-8"><title>Fixture gym my bookings</title></head>
<body>
<nav><a id="my-bookings-link" href="my_bookings.html">My Bookings</a></nav>
<div id="my-bookings-page" data-bookings-count="0" data-waitlist-count="0" data-loading="true">
<h1>My Bookings</h1>
<p id="loading-placeholder">Loading your bookings...</p>
<div id="content"></div>
</div>
<script>
// Imitates the site: a loading text, then the page for the state in the query
// (?state=empty or ?state=error; anything else shows two bookings and a waitlist entry).
const params = new URLSearchParams(window.location.search);
const state = params.get("state");
// ?fail=N shows the error for the first N loads of the page in this tab (the network
// simulation); the count lives in sessionStorage because every load starts a new page.
const loads = parseInt(sessionStorage.getItem("loads") || "0", 10);
const dropped = loads < parseInt(params.get("fail") || "0", 10);
sessionStorage.setItem("loads", String(loads + 1));
// The nav link keeps the state, so opening the page from here shows the same state again.
document.getElementById("my-bookings-link").href = "my_bookings.html" + window.location.search;
const page = document.getElementById("my-bookings-page");
const content = document.getElementById("content");
setTimeout(function () {
document.getElementById("loading-placeholder").remove();
page.dataset.loading = "false";
if (state === "error" || dropped) {
content.innerHTML = '<div id="error-message">Failed to load bookings</div>';
} else if (state === "empty") {
content.innerHTML = '<div id="empty-bookings-state">You have no bookings yet</div>';
} else {
page.dataset.bookingsCount = "2";
page.dataset.waitlistCount = "1";
content.innerHTML =
'<div id="confirmed-bookings-section">' +
'<div id="booking-card-11" data-booking-status="confirmed">' +
'<h3 id="booking-class-name-11">Spin Class</h3>' +
'<p><strong>When:</strong> Tue, Aug 12, 6:00 PM</p></div>' +
'<div id="booking-card-12" data-booking-status="confirmed">' +
'<h3 id="booking-class-name-12">Yoga Class</h3>' +
'<p><strong>When:</strong> Wed, Aug 13, 7:00 AM</p></div>' +
'</div>' +
'<div id="waitlist-section">' +
'<div id="waitlist-card-21" data-booking-status="waitlisted">' +
'<h3 id="waitlist-class-name-21">HIIT Class (Waitlist)</h3>' +
'<p><strong>When:</strong> Thu, Aug 14, 6:00 PM</p></div>' +
'</div>';
}
}, 200);
</script>
</body>
</html>
+9 -3
View File
@@ -2,6 +2,7 @@
<html lang="en"> <html lang="en">
<head><meta charset="utf-8"><title>Fixture gym schedule</title></head> <head><meta charset="utf-8"><title>Fixture gym schedule</title></head>
<body> <body>
<nav><a id="my-bookings-link" href="my_bookings.html">My Bookings</a></nav>
<div id="schedule-page"> <div id="schedule-page">
<h1>Class Schedule (Today: Wed, Aug 13)</h1> <h1>Class Schedule (Today: Wed, Aug 13)</h1>
<select id="day-filter"><option>Next 7 Days</option></select> <select id="day-filter"><option>Next 7 Days</option></select>
@@ -45,7 +46,7 @@
<p id="class-time-301"><strong>Time:</strong> 7:00 AM</p> <p id="class-time-301"><strong>Time:</strong> 7:00 AM</p>
<button id="book-button-301">Book Class</button> <button id="book-button-301">Book Class</button>
</div> </div>
<div id="class-card-302" data-class-type="spin" data-user-booked="false" data-fail-once="true"> <div id="class-card-302" data-class-type="spin" data-user-booked="false" data-fail-times="1">
<h3 id="class-name-302">Spin Class</h3> <h3 id="class-name-302">Spin Class</h3>
<p id="class-time-302"><strong>Time:</strong> 6:00 PM</p> <p id="class-time-302"><strong>Time:</strong> 6:00 PM</p>
<button id="book-button-302">Book Class</button> <button id="book-button-302">Book Class</button>
@@ -54,6 +55,11 @@
</div> </div>
<script> <script>
// Imitates the site: a busy text, then the new state; card 302 fails on its first click. // Imitates the site: a busy text, then the new state; card 302 fails on its first click.
// ?fail=N makes every button fail on its first N clicks (the network simulation).
const failParam = parseInt(new URLSearchParams(window.location.search).get("fail") || "0", 10);
document.querySelectorAll("div[id^='class-card-']").forEach(function (card) {
card.dataset.failLeft = String(Math.max(failParam, parseInt(card.dataset.failTimes || "0", 10)));
});
document.querySelectorAll("button[id^='book-button-']").forEach(function (button) { document.querySelectorAll("button[id^='book-button-']").forEach(function (button) {
button.addEventListener("click", function () { button.addEventListener("click", function () {
const id = button.id.replace("book-button-", ""); const id = button.id.replace("book-button-", "");
@@ -65,8 +71,8 @@
button.textContent = waitlist ? "Joining..." : "Booking..."; button.textContent = waitlist ? "Joining..." : "Booking...";
button.disabled = true; button.disabled = true;
setTimeout(function () { setTimeout(function () {
if (card.dataset.failOnce === "true") { if (parseInt(card.dataset.failLeft, 10) > 0) {
card.dataset.failOnce = "false"; card.dataset.failLeft = String(parseInt(card.dataset.failLeft, 10) - 1);
button.textContent = oldText; button.textContent = oldText;
button.disabled = false; button.disabled = false;
const box = document.createElement("div"); const box = document.createElement("div");
+152 -8
View File
@@ -3,13 +3,22 @@
""" """
import os import os
from collections.abc import Iterator from collections.abc import Callable, Iterator
from pathlib import Path from pathlib import Path
import pytest import pytest
from snack_and_lift.application.booking import book_target_classes from snack_and_lift.application.booking import book_class_with_retry, book_target_classes
from snack_and_lift.domain.errors import GymError, LoginFailedError from snack_and_lift.application.credentials import Credentials
from snack_and_lift.application.session import login_with_retry
from snack_and_lift.application.verification import get_my_bookings, get_my_bookings_with_retry
from snack_and_lift.domain.booking_entry import BookingEntry
from snack_and_lift.domain.errors import (
GymError,
LoginFailedError,
NoBookingsFoundError,
RetriesExhaustedError,
)
from snack_and_lift.domain.models import Outcome from snack_and_lift.domain.models import Outcome
from snack_and_lift.infrastructure.browser import create_driver from snack_and_lift.infrastructure.browser import create_driver
from snack_and_lift.infrastructure.selenium_gym import SeleniumGym from snack_and_lift.infrastructure.selenium_gym import SeleniumGym
@@ -117,12 +126,147 @@ def test_save_text_screenshot_writes_a_png_and_keeps_the_page(
def test_book_target_classes_books_every_tuesday_and_thursday_class( def test_book_target_classes_books_every_tuesday_and_thursday_class(
schedule_gym: SeleniumGym, schedule_gym: SeleniumGym,
) -> None: ) -> None:
with pytest.raises(GymError, match="Network error"): # Card 302 drops its first click; the booking repeats it and the page confirms the second.
book_target_classes(schedule_gym) results = book_target_classes(schedule_gym, sleep=lambda _: None)
results = book_target_classes(schedule_gym)
assert [(r.card.card_id, r.outcome) for r in results] == [ assert [(r.card.card_id, r.outcome) for r in results] == [
("201", Outcome.ALREADY_WAITLISTED), ("201", Outcome.NEW_WAITLIST),
("302", Outcome.NEW_BOOKING), ("302", Outcome.NEW_BOOKING),
] ]
FIXTURE_MY_BOOKINGS = (Path(__file__).parent / "fixtures" / "my_bookings.html").as_uri()
@pytest.fixture
def open_my_bookings_page(tmp_path: Path) -> Iterator[Callable[[str], SeleniumGym]]:
"""@brief Opens the My Bookings fixture in the given state (a query such as `?state=empty`)."""
driver = create_driver(headless=True, profile_dir=tmp_path / "profile", keep_open=False)
def open_page(query: str = "") -> SeleniumGym:
driver.get(FIXTURE_MY_BOOKINGS + query)
return SeleniumGym(driver, base_url=FIXTURE_MY_BOOKINGS)
try:
yield open_page
finally:
driver.quit()
def test_open_my_bookings_follows_the_nav_link(schedule_gym: SeleniumGym) -> None:
schedule_gym.open_my_bookings()
assert [entry.name for entry in schedule_gym.list_my_bookings()] == [
"Spin Class",
"Yoga Class",
"HIIT Class",
]
def test_list_my_bookings_reads_confirmed_and_waitlist_entries(
open_my_bookings_page: Callable[[str], SeleniumGym],
) -> None:
entries = open_my_bookings_page("").list_my_bookings()
assert entries == [
BookingEntry("Spin Class", "Tue, Aug 12, 6:00 PM", False),
BookingEntry("Yoga Class", "Wed, Aug 13, 7:00 AM", False),
BookingEntry("HIIT Class", "Thu, Aug 14, 6:00 PM", True),
]
def test_list_my_bookings_returns_nothing_for_the_empty_state(
open_my_bookings_page: Callable[[str], SeleniumGym],
) -> None:
assert open_my_bookings_page("?state=empty").list_my_bookings() == []
def test_get_my_bookings_raises_for_the_empty_state(
open_my_bookings_page: Callable[[str], SeleniumGym],
) -> None:
with pytest.raises(NoBookingsFoundError):
get_my_bookings(open_my_bookings_page("?state=empty"))
def test_list_my_bookings_raises_with_the_text_of_the_error_box(
open_my_bookings_page: Callable[[str], SeleniumGym],
) -> None:
with pytest.raises(GymError, match="Failed to load bookings"):
open_my_bookings_page("?state=error").list_my_bookings()
FIXTURE_LOGIN = (Path(__file__).parent / "fixtures" / "login.html").as_uri()
@pytest.fixture
def open_fixture(tmp_path: Path) -> Iterator[Callable[[str], SeleniumGym]]:
"""@brief Opens a fixture page by its address, with the query that sets its failures."""
driver = create_driver(headless=True, profile_dir=tmp_path / "profile", keep_open=False)
def open_page(url: str) -> SeleniumGym:
driver.get(url)
return SeleniumGym(driver, base_url=url)
try:
yield open_page
finally:
driver.quit()
def test_login_with_retry_logs_in_after_two_dropped_requests(
open_fixture: Callable[[str], SeleniumGym],
) -> None:
gym = open_fixture(FIXTURE_LOGIN + "?fail=2")
sleeps: list[float] = []
login_with_retry(gym, Credentials(FIXTURE_EMAIL, FIXTURE_PASSWORD), sleep=sleeps.append)
assert len(sleeps) == 2
def test_login_with_retry_gives_up_when_every_request_is_dropped(
open_fixture: Callable[[str], SeleniumGym],
) -> None:
gym = open_fixture(FIXTURE_LOGIN + "?fail=5")
with pytest.raises(RetriesExhaustedError, match="Network request failed"):
login_with_retry(
gym, Credentials(FIXTURE_EMAIL, FIXTURE_PASSWORD), retries=3, sleep=lambda _: None
)
def test_book_class_with_retry_books_after_two_dropped_clicks(
open_fixture: Callable[[str], SeleniumGym],
) -> None:
gym = open_fixture(FIXTURE_SCHEDULE + "?fail=2")
card = next(card for card in gym.list_class_cards() if card.card_id == "302")
result = book_class_with_retry(gym, card, sleep=lambda _: None)
assert result.outcome is Outcome.NEW_BOOKING
assert next(c for c in gym.list_class_cards() if c.card_id == "302").button_text == "Booked"
def test_book_target_classes_completes_when_every_class_drops_two_clicks(
open_fixture: Callable[[str], SeleniumGym],
) -> None:
gym = open_fixture(FIXTURE_SCHEDULE + "?fail=2")
results = book_target_classes(gym, sleep=lambda _: None)
assert [(r.card.card_id, r.outcome) for r in results] == [
("201", Outcome.NEW_WAITLIST),
("302", Outcome.NEW_BOOKING),
]
def test_get_my_bookings_with_retry_reads_the_page_after_two_failed_loads(
open_fixture: Callable[[str], SeleniumGym],
) -> None:
gym = open_fixture(FIXTURE_MY_BOOKINGS + "?fail=2")
sleeps: list[float] = []
entries = get_my_bookings_with_retry(gym, sleep=sleeps.append)
assert [entry.name for entry in entries] == ["Spin Class", "Yoga Class", "HIIT Class"]
assert sleeps
+109 -4
View File
@@ -2,11 +2,14 @@
@brief A fake of the GymPort for the unit tests. @brief A fake of the GymPort for the unit tests.
""" """
import random
from dataclasses import replace
from pathlib import Path from pathlib import Path
from snack_and_lift import constants 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.errors import GymError, LoginFailedError
from snack_and_lift.domain.models import ClassCard from snack_and_lift.domain.models import ButtonState, ClassCard
_PRESS_RESULTS = { _PRESS_RESULTS = {
constants.BUTTON_TEXT_BOOK: constants.BUTTON_TEXT_BOOKED, constants.BUTTON_TEXT_BOOK: constants.BUTTON_TEXT_BOOKED,
@@ -14,8 +17,38 @@ _PRESS_RESULTS = {
} }
_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: class FakeGym:
"""@brief Records the calls; can refuse the login, serve cards and fail a click.""" """@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__( def __init__(
self, self,
@@ -23,11 +56,19 @@ class FakeGym:
cards: list[ClassCard] | None = None, cards: list[ClassCard] | None = None,
press_result: str | None = None, press_result: str | None = None,
press_error: 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: ) -> None:
self.login_error = login_error self.login_error = login_error
self.cards = cards or [] self.cards = cards or []
self.press_result = press_result self.press_result = press_result
self.press_error = press_error 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.calls: list[str] = []
self.logins: list[tuple[str, str]] = [] self.logins: list[tuple[str, str]] = []
self.screenshots: list[Path] = [] self.screenshots: list[Path] = []
@@ -38,11 +79,29 @@ class FakeGym:
def open_site(self) -> None: def open_site(self) -> None:
self.calls.append("open_site") 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: def log_in(self, email: str, password: str) -> None:
self.calls.append("log_in") self.calls.append("log_in")
self.logins.append((email, password)) self.logins.append((email, password))
if self.login_error is not None: if self.login_error is not None:
raise LoginFailedError(self.login_error) raise LoginFailedError(self.login_error)
if self._should_fail("log_in"):
raise LoginFailedError(NETWORK_ERROR_TEXT)
def list_class_cards(self) -> list[ClassCard]: def list_class_cards(self) -> list[ClassCard]:
self.calls.append("list_class_cards") self.calls.append("list_class_cards")
@@ -55,8 +114,27 @@ class FakeGym:
raise GymError(self.press_error) raise GymError(self.press_error)
if self.press_result is not None: if self.press_result is not None:
return self.press_result return self.press_result
card = next(card for card in self.cards if card.card_id == card_id) if self._should_fail("press_class_button"):
return _PRESS_RESULTS[card.button_text] 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: def save_text_screenshot(self, text: str, path: Path) -> None:
self.calls.append("save_text_screenshot") self.calls.append("save_text_screenshot")
@@ -69,3 +147,30 @@ class FakeGym:
def close(self) -> None: def close(self) -> None:
self.calls.append("close") self.calls.append("close")
self.closed = True 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."""
@@ -0,0 +1,83 @@
"""@file test_application_verification.py
@brief Reading the My Bookings page and verifying the bookings, against a fake gym.
"""
import pytest
from tests.unit.fakes import FakeGym, no_sleep
from tests.unit.test_booking_entry import make_entry
from tests.unit.test_models import make_card
from snack_and_lift.application.booking import book_target_classes
from snack_and_lift.application.verification import get_my_bookings, verify_bookings
from snack_and_lift.domain.errors import GymError, NoBookingsFoundError, RetriesExhaustedError
from snack_and_lift.domain.models import ClassResult, Outcome
def test_get_my_bookings_opens_the_page_and_returns_all_entries() -> None:
entries = [make_entry("Wed, Aug 13, 7:00 AM"), make_entry(waitlisted=True)]
gym = FakeGym(bookings=entries)
assert get_my_bookings(gym) == entries
assert gym.calls == ["open_my_bookings", "list_my_bookings"]
def test_get_my_bookings_raises_when_the_page_has_no_entry() -> None:
with pytest.raises(NoBookingsFoundError):
get_my_bookings(FakeGym(bookings=[]))
def test_get_my_bookings_propagates_the_page_error() -> None:
with pytest.raises(GymError, match="Failed to load"):
get_my_bookings(FakeGym(bookings_error="Failed to load bookings"))
def test_verify_bookings_succeeds_for_the_booked_classes() -> None:
gym = FakeGym(
cards=[
make_card("Tue, Aug 12", card_id="1"),
make_card("Thu, Aug 14", card_id="2", button_text="Join Waitlist"),
]
)
results = book_target_classes(gym)
verification = verify_bookings(gym, results)
assert verification.is_success
assert verification.verified == ["Spin Class", "Spin Class (Waitlist)"]
def test_verify_bookings_ignores_entries_that_are_not_target_classes() -> None:
gym = FakeGym(bookings=[make_entry(), make_entry("Wed, Aug 13, 6:00 PM")])
results = [ClassResult(make_card("Tue, Aug 12"), Outcome.NEW_BOOKING)]
assert verify_bookings(gym, results).is_success
def test_verify_bookings_reports_a_missing_booking() -> None:
gym = FakeGym(bookings=[make_entry("Wed, Aug 13, 7:00 AM")])
results = [ClassResult(make_card("Tue, Aug 12"), Outcome.NEW_BOOKING)]
verification = verify_bookings(gym, results)
assert not verification.is_success
assert verification.missing == ["Spin Class on Tue, Aug 12"]
def test_verify_bookings_raises_when_classes_were_booked_but_the_page_is_empty() -> None:
gym = FakeGym(bookings=[])
results = [ClassResult(make_card("Tue, Aug 12"), Outcome.NEW_BOOKING)]
with pytest.raises(RetriesExhaustedError) as caught:
verify_bookings(gym, results, sleep=no_sleep)
assert isinstance(caught.value.__cause__, NoBookingsFoundError)
def test_verify_bookings_skips_the_page_when_nothing_is_expected() -> None:
gym = FakeGym()
verification = verify_bookings(gym, [])
assert verification.is_success
assert verification.expected == 0
assert gym.calls == []
+2 -2
View File
@@ -5,7 +5,7 @@
import logging import logging
import pytest import pytest
from tests.unit.fakes import FakeGym from tests.unit.fakes import FakeGym, no_sleep
from tests.unit.test_models import make_card from tests.unit.test_models import make_card
from snack_and_lift.application.booking import ( from snack_and_lift.application.booking import (
@@ -147,4 +147,4 @@ def test_book_target_classes_propagates_a_failing_class() -> None:
gym = FakeGym(cards=[make_card("Thu, Aug 14")], press_error="Network down") gym = FakeGym(cards=[make_card("Thu, Aug 14")], press_error="Network down")
with pytest.raises(GymError, match="Network down"): with pytest.raises(GymError, match="Network down"):
book_target_classes(gym) book_target_classes(gym, sleep=no_sleep)
+65
View File
@@ -0,0 +1,65 @@
"""@file test_booking_entry.py
@brief The entries of the My Bookings page: parsing and the target test.
"""
import dataclasses
import pytest
from snack_and_lift.domain.booking_entry import BookingEntry, is_target_entry
from snack_and_lift.domain.errors import GymError, NoBookingsFoundError, PageFormatError
def make_entry(
when_text: str = "Tue, Aug 12, 6:00 PM", name: str = "Spin Class", waitlisted: bool = False
) -> BookingEntry:
return BookingEntry(name, when_text, waitlisted)
def test_entry_derives_weekday_date_and_time() -> None:
entry = make_entry("Thu, Aug 14, 6:00 PM")
assert entry.weekday == "Thu"
assert entry.date_text == "Thu, Aug 14"
assert entry.time == (18, 0)
@pytest.mark.parametrize(
"when_text", ["", "Someday", "Tue, Aug 12", "Tue, Aug 12, noon", "6:00 PM"]
)
def test_entry_with_bad_when_text_raises(when_text: str) -> None:
with pytest.raises(PageFormatError):
is_target_entry(make_entry(when_text))
def test_from_page_removes_the_waitlist_suffix_and_the_when_prefix() -> None:
entry = BookingEntry.from_page("Spin Class (Waitlist)", "When: Tue, Aug 12, 6:00 PM", True)
assert entry == BookingEntry("Spin Class", "Tue, Aug 12, 6:00 PM", True)
def test_from_page_keeps_a_plain_name() -> None:
assert BookingEntry.from_page("Spin Class", "Tue, Aug 12, 6:00 PM", False).name == "Spin Class"
@pytest.mark.parametrize(
("when_text", "expected"),
[
("Tue, Aug 12, 6:00 PM", True),
("Thu, Aug 14, 6:00 PM", True),
("Wed, Aug 13, 6:00 PM", False),
("Tue, Aug 12, 6:00 AM", False),
("Thu, Aug 14, 6:30 PM", False),
],
)
def test_is_target_entry(when_text: str, expected: bool) -> None:
assert is_target_entry(make_entry(when_text)) is expected
def test_entry_is_immutable() -> None:
with pytest.raises(dataclasses.FrozenInstanceError):
make_entry().name = "Other" # type: ignore[misc]
def test_no_bookings_found_is_a_gym_error() -> None:
assert issubclass(NoBookingsFoundError, GymError)
+63 -5
View File
@@ -6,7 +6,8 @@ import logging
from pathlib import Path from pathlib import Path
import pytest import pytest
from tests.unit.fakes import FakeGym from tests.unit.fakes import FakeGym, no_sleep
from tests.unit.test_booking_entry import make_entry
from tests.unit.test_models import make_card from tests.unit.test_models import make_card
from snack_and_lift import constants from snack_and_lift import constants
@@ -67,11 +68,14 @@ def test_main_saves_both_screenshots_on_success(tmp_path: Path) -> None:
tmp_path / "step-1.png", tmp_path / "step-1.png",
tmp_path / "step-2.png", tmp_path / "step-2.png",
tmp_path / "step-3.png", tmp_path / "step-3.png",
tmp_path / "step-7.png",
] ]
assert [path for _, path in gym.text_screenshots] == [ assert [path for _, path in gym.text_screenshots] == [
tmp_path / "step-4.png", tmp_path / "step-4.png",
tmp_path / "step-5.png", tmp_path / "step-5.png",
tmp_path / "step-6.png", tmp_path / "step-6.png",
tmp_path / "step-8.png",
tmp_path / "step-9.png",
] ]
assert gym.calls == [ assert gym.calls == [
"open_site", "open_site",
@@ -83,6 +87,9 @@ def test_main_saves_both_screenshots_on_success(tmp_path: Path) -> None:
"save_text_screenshot", "save_text_screenshot",
"save_text_screenshot", "save_text_screenshot",
"save_text_screenshot", "save_text_screenshot",
"save_screenshot",
"save_text_screenshot",
"save_text_screenshot",
] ]
@@ -140,15 +147,64 @@ def test_main_saves_step_3_after_the_first_class_and_the_report_images(tmp_path:
"save_text_screenshot", "save_text_screenshot",
"save_text_screenshot", "save_text_screenshot",
"save_text_screenshot", "save_text_screenshot",
"open_my_bookings",
"list_my_bookings",
"save_screenshot",
"save_text_screenshot",
"save_text_screenshot",
] ]
assert gym.screenshots[-1] == tmp_path / "step-3.png" assert gym.screenshots[-2:] == [tmp_path / "step-3.png", tmp_path / "step-7.png"]
(_, _), (step_5, _), (step_6, _) = gym.text_screenshots (_, _), (step_5, _), (step_6, _), (step_8, step_8_path), (step_9, step_9_path) = (
gym.text_screenshots
)
assert step_8_path == tmp_path / "step-8.png"
assert step_8.startswith("--- VERIFYING ON MY BOOKINGS PAGE ---")
assert step_8.endswith("✅ SUCCESS: All bookings verified!")
assert step_9_path == tmp_path / "step-9.png"
assert step_9.startswith("✓ Logged in")
assert step_9.endswith("✅ SUCCESS: All bookings verified!")
assert step_5.startswith("--- BOOKING SUMMARY ---\nNew bookings: 2") assert step_5.startswith("--- BOOKING SUMMARY ---\nNew bookings: 2")
assert "--- DETAILED CLASS LIST ---" not in step_5 assert "--- DETAILED CLASS LIST ---" not in step_5
assert step_6.startswith(step_5) assert step_6.startswith(step_5)
assert "--- DETAILED CLASS LIST ---\n • [New Booking] Spin Class on Thu, Aug 14" in step_6 assert "--- DETAILED CLASS LIST ---\n • [New Booking] Spin Class on Thu, Aug 14" in step_6
@pytest.mark.usefixtures("credentials_env")
def test_main_logs_the_verification_and_returns_0_on_success(
caplog: pytest.LogCaptureFixture,
) -> None:
gym = FakeGym(cards=[make_card("Tue, Aug 19")])
with caplog.at_level(logging.INFO):
assert main([], lambda *_: gym) == 0
assert "--- VERIFYING ON MY BOOKINGS PAGE ---\n ✓ Verified: Spin Class" in caplog.text
assert "Expected: 1 booking\nFound: 1 booking\n✅ SUCCESS" in caplog.text
@pytest.mark.usefixtures("credentials_env")
def test_main_returns_3_on_a_mismatch(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
gym = FakeGym(cards=[make_card("Tue, Aug 19")], bookings=[make_entry("Wed, Aug 20, 7:00 AM")])
with caplog.at_level(logging.INFO):
code = main(["--headless", "--screenshot-dir", str(tmp_path)], lambda *_: gym)
assert code == 3
assert "❌ MISMATCH: Missing 1 bookings" in caplog.text
assert " ✗ Missing: Spin Class on Tue, Aug 19" in caplog.text
assert gym.screenshots[-1] == tmp_path / "step-7.png"
assert gym.text_screenshots[-2][1] == tmp_path / "step-8.png"
assert gym.text_screenshots[-1][1] == tmp_path / "step-9.png"
assert gym.closed is True
@pytest.mark.usefixtures("credentials_env")
def test_main_returns_1_when_the_bookings_page_is_empty() -> None:
gym = FakeGym(cards=[make_card("Tue, Aug 19")], bookings=[])
assert main([], lambda *_: gym, no_sleep) == 1
@pytest.mark.usefixtures("credentials_env") @pytest.mark.usefixtures("credentials_env")
def test_main_logs_an_empty_report_without_target_classes( def test_main_logs_an_empty_report_without_target_classes(
caplog: pytest.LogCaptureFixture, caplog: pytest.LogCaptureFixture,
@@ -176,6 +232,8 @@ def test_main_first_tuesday_only_books_one_class_and_no_report(
assert code == 0 assert code == 0
assert gym.pressed == ["2"] assert gym.pressed == ["2"]
assert "BOOKING SUMMARY" not in caplog.text assert "BOOKING SUMMARY" not in caplog.text
assert "VERIFYING" not in caplog.text
assert "open_my_bookings" not in gym.calls
assert [path.name for path in gym.screenshots] == ["step-1.png", "step-2.png", "step-3.png"] assert [path.name for path in gym.screenshots] == ["step-1.png", "step-2.png", "step-3.png"]
assert [path.name for _, path in gym.text_screenshots] == ["step-4.png"] assert [path.name for _, path in gym.text_screenshots] == ["step-4.png"]
@@ -190,7 +248,7 @@ def test_first_tuesday_only_flag_is_read() -> None:
def test_main_returns_1_when_the_booking_fails() -> None: def test_main_returns_1_when_the_booking_fails() -> None:
gym = FakeGym(cards=[make_card("Tue, Aug 19")], press_error="Network down") gym = FakeGym(cards=[make_card("Tue, Aug 19")], press_error="Network down")
assert main(["--headless"], lambda *_: gym) == 1 assert main(["--headless"], lambda *_: gym, no_sleep) == 1
assert gym.closed is True assert gym.closed is True
@@ -220,7 +278,7 @@ def test_main_returns_2_when_credentials_are_missing(
def test_main_returns_1_when_login_fails() -> None: def test_main_returns_1_when_login_fails() -> None:
gym = FakeGym(login_error="Invalid credentials") gym = FakeGym(login_error="Invalid credentials")
assert main(["--headless"], lambda *_: gym) == 1 assert main(["--headless"], lambda *_: gym, no_sleep) == 1
assert gym.closed is True assert gym.closed is True
+4
View File
@@ -29,3 +29,7 @@ def test_retry_count_is_seven() -> None:
def test_bot_books_tuesday_and_thursday() -> None: def test_bot_books_tuesday_and_thursday() -> None:
assert constants.TARGET_WEEKDAYS == ("Tue", "Thu") assert constants.TARGET_WEEKDAYS == ("Tue", "Thu")
def test_retry_delay_is_half_a_second() -> None:
assert constants.RETRY_DELAY_SECONDS == 0.5
+87
View File
@@ -0,0 +1,87 @@
"""@file test_date_independence.py
@brief The bot picks classes by weekday and time only, whatever the first visible day is.
"""
from datetime import date, timedelta
import pytest
from tests.unit.fakes import FakeGym
from snack_and_lift.application.booking import book_target_classes
from snack_and_lift.application.verification import verify_bookings
from snack_and_lift.domain.models import ClassCard
from snack_and_lift.domain.rules import find_target_classes
## A Monday; the schedules below start on this day plus an offset, so every weekday occurs.
_A_MONDAY = date(2025, 8, 11)
_DAYS_SHOWN = 8
_TUESDAY, _THURSDAY = 1, 3
_TIMES = (
("Time: 6:00 AM", "Book Class"),
("Time: 6:00 PM", "Book Class"),
("Time: 7:00 PM", "Book Class"),
)
def _day_text(day: date) -> str:
return f"{day:%a}, {day:%b} {day.day}"
def _label(index: int, day: date) -> str:
if index == 0:
return f"Today ({_day_text(day)})"
if index == 1:
return f"Tomorrow ({_day_text(day)})"
return _day_text(day)
def make_schedule(start_offset: int) -> tuple[list[ClassCard], set[str]]:
"""@brief Cards of 8 days from a start weekday, and the ids of the Tue/Thu 6pm ones."""
cards: list[ClassCard] = []
wanted: set[str] = set()
for index in range(_DAYS_SHOWN):
day = _A_MONDAY + timedelta(days=start_offset + index)
for slot, (time_text, button) in enumerate(_TIMES):
card_id = f"{index}{slot}"
cards.append(ClassCard(card_id, "Spin Class", _label(index, day), time_text, button))
if day.weekday() in (_TUESDAY, _THURSDAY) and slot == 1:
wanted.add(card_id)
return cards, wanted
@pytest.mark.parametrize("start_offset", range(7))
def test_find_target_classes_for_every_starting_weekday(start_offset: int) -> None:
cards, wanted = make_schedule(start_offset)
assert {card.card_id for card in find_target_classes(cards)} == wanted
assert wanted
@pytest.mark.parametrize("start_offset", range(7))
def test_booking_and_verification_for_every_starting_weekday(start_offset: int) -> None:
cards, wanted = make_schedule(start_offset)
gym = FakeGym(cards=cards)
results = book_target_classes(gym)
assert set(gym.pressed) == wanted
assert len(gym.pressed) == len(wanted)
assert verify_bookings(gym, results).is_success
def test_schedule_whose_first_visible_day_is_a_thursday_includes_today() -> None:
cards, _ = make_schedule(3)
assert cards[0].day_label.startswith("Today (Thu,")
chosen = find_target_classes(cards)
assert [card.weekday for card in chosen] == ["Thu", "Tue", "Thu"]
assert chosen[0].card_id == "01"
def test_card_order_does_not_change_which_classes_are_chosen() -> None:
cards, wanted = make_schedule(2)
chosen = find_target_classes(list(reversed(cards)))
assert {card.card_id for card in chosen} == wanted
+316
View File
@@ -0,0 +1,316 @@
"""@file test_resilience.py
@brief Lecture step 9: login, booking and bookings read survive a site that drops requests.
"""
import logging
import random
from pathlib import Path
import pytest
from tests.unit.fakes import ChaosGym, FakeGym, no_sleep
from tests.unit.test_booking_entry import make_entry
from tests.unit.test_models import make_card
from snack_and_lift import constants
from snack_and_lift.application.booking import (
book_class_with_retry,
book_next_tuesday,
book_target_classes,
)
from snack_and_lift.application.credentials import Credentials
from snack_and_lift.application.session import login_with_retry
from snack_and_lift.application.verification import get_my_bookings_with_retry, verify_bookings
from snack_and_lift.cli import main, parse_args
from snack_and_lift.domain.errors import GymError, LoginFailedError, RetriesExhaustedError
from snack_and_lift.domain.models import ButtonState, ClassResult, Outcome
DUMMY_EMAIL = "dummy@example.invalid"
DUMMY_PASSWORD = "dummy-password"
CREDENTIALS = Credentials(DUMMY_EMAIL, DUMMY_PASSWORD)
@pytest.fixture
def credentials_env(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv(constants.ACCOUNT_EMAIL_ENV, DUMMY_EMAIL)
monkeypatch.setenv(constants.ACCOUNT_PASSWORD_ENV, DUMMY_PASSWORD)
def test_login_with_retry_logs_in_after_two_dropped_requests(
caplog: pytest.LogCaptureFixture,
) -> None:
gym = FakeGym(fail_first={"log_in": 2})
sleeps: list[float] = []
with caplog.at_level(logging.INFO):
login_with_retry(gym, CREDENTIALS, sleep=sleeps.append)
assert gym.calls == ["log_in"] * 3
assert sleeps == [constants.RETRY_DELAY_SECONDS] * 2
assert "⚠ Attempt 2/7 failed: login: Network request failed. Please try again." in caplog.text
assert "✓ login succeeded after 3 attempts" in caplog.text
def test_login_with_retry_never_logs_the_password(caplog: pytest.LogCaptureFixture) -> None:
gym = FakeGym(fail_first={"log_in": 2})
with caplog.at_level(logging.DEBUG):
login_with_retry(gym, CREDENTIALS, sleep=no_sleep)
assert DUMMY_PASSWORD not in caplog.text
assert DUMMY_EMAIL not in caplog.text
def test_login_with_retry_gives_up_after_the_set_attempts() -> None:
gym = FakeGym(fail_first={"log_in": 10})
with pytest.raises(RetriesExhaustedError, match="login failed after 3 attempts") as caught:
login_with_retry(gym, CREDENTIALS, retries=3, sleep=no_sleep)
assert isinstance(caught.value.__cause__, LoginFailedError)
assert len(gym.logins) == 3
def test_book_class_with_retry_repeats_a_dropped_click() -> None:
card = make_card()
gym = FakeGym(cards=[card], fail_first={"press_class_button": 2})
result = book_class_with_retry(gym, card, sleep=no_sleep)
assert result.outcome is Outcome.NEW_BOOKING
assert gym.pressed == ["123", "123", "123"]
assert gym.calls.count("list_class_cards") == 2
def test_book_class_with_retry_reads_nothing_again_when_the_first_attempt_works() -> None:
card = make_card()
gym = FakeGym(cards=[card])
book_class_with_retry(gym, card, sleep=no_sleep)
assert gym.calls == ["press_class_button"]
@pytest.mark.parametrize(
("button_text", "outcome", "final_state"),
[
("Book Class", Outcome.NEW_BOOKING, ButtonState.BOOKED),
("Join Waitlist", Outcome.NEW_WAITLIST, ButtonState.WAITLISTED),
],
)
def test_book_class_with_retry_counts_a_click_that_went_through_as_new(
button_text: str,
outcome: Outcome,
final_state: ButtonState,
caplog: pytest.LogCaptureFixture,
) -> None:
card = make_card(button_text=button_text)
gym = FakeGym(cards=[card], fail_first={"press_class_button": 1}, click_goes_through=True)
with caplog.at_level(logging.INFO):
result = book_class_with_retry(gym, card, sleep=no_sleep)
assert result.outcome is outcome
assert gym.pressed == ["123"]
assert gym.cards[0].state is final_state
assert "Already" not in caplog.text
def test_book_class_with_retry_keeps_already_booked_as_it_is() -> None:
card = make_card(button_text="Booked")
gym = FakeGym(cards=[card])
result = book_class_with_retry(gym, card, sleep=no_sleep)
assert result.outcome is Outcome.ALREADY_BOOKED
assert gym.pressed == []
def test_book_class_with_retry_does_not_count_a_click_without_confirmation() -> None:
card = make_card()
gym = FakeGym(cards=[card], press_result="Book Class")
with pytest.raises(RetriesExhaustedError, match="click not confirmed"):
book_class_with_retry(gym, card, retries=2, sleep=no_sleep)
assert gym.pressed == ["123", "123"]
def test_book_class_with_retry_fails_when_the_class_left_the_schedule() -> None:
card = make_card()
gym = FakeGym(cards=[card], fail_first={"press_class_button": 1})
gym.cards.clear()
with pytest.raises(RetriesExhaustedError, match="not found on the schedule"):
book_class_with_retry(gym, card, retries=2, sleep=no_sleep)
def test_book_class_with_retry_gives_up_after_the_set_attempts() -> None:
card = make_card()
gym = FakeGym(cards=[card], fail_first={"press_class_button": 9})
with pytest.raises(RetriesExhaustedError, match="book Spin Class on Tomorrow"):
book_class_with_retry(gym, card, retries=3, sleep=no_sleep)
assert len(gym.pressed) == 3
def test_book_next_tuesday_repeats_a_dropped_click() -> None:
gym = FakeGym(cards=[make_card("Tue, Aug 19")], fail_first={"press_class_button": 1})
result = book_next_tuesday(gym, sleep=no_sleep)
assert result is not None
assert result.outcome is Outcome.NEW_BOOKING
def test_book_target_classes_repeats_each_class_on_its_own() -> None:
gym = FakeGym(
cards=[make_card("Thu, Aug 14", card_id="1"), make_card("Tue, Aug 19", card_id="2")],
fail_first={"press_class_button": 3},
)
results = book_target_classes(gym, sleep=no_sleep)
assert [result.outcome for result in results] == [Outcome.NEW_BOOKING] * 2
assert gym.pressed == ["1", "1", "1", "1", "2"]
def test_get_my_bookings_with_retry_opens_the_page_again_after_a_failure() -> None:
entries = [make_entry("Wed, Aug 13, 7:00 AM")]
gym = FakeGym(bookings=entries, fail_first={"open_my_bookings": 1, "list_my_bookings": 1})
assert get_my_bookings_with_retry(gym, sleep=no_sleep) == entries
assert gym.calls == [
"open_my_bookings",
"open_my_bookings",
"list_my_bookings",
"open_my_bookings",
"list_my_bookings",
]
def test_get_my_bookings_with_retry_gives_up_after_the_set_attempts() -> None:
gym = FakeGym(bookings_error="Failed to load bookings")
with pytest.raises(RetriesExhaustedError, match="Failed to load bookings"):
get_my_bookings_with_retry(gym, retries=2, sleep=no_sleep)
def test_verify_bookings_reads_the_page_through_retry() -> None:
card = make_card("Tue, Aug 12")
gym = FakeGym(cards=[card], fail_first={"list_my_bookings": 2})
results = [ClassResult(card, Outcome.NEW_BOOKING)]
verification = verify_bookings(gym, results, sleep=no_sleep)
assert verification.is_success
def test_main_completes_a_run_with_dropped_requests_and_draws_them_in_step_9(
credentials_env: None, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
gym = FakeGym(
cards=[make_card("Thu, Aug 14", card_id="1"), make_card("Tue, Aug 19", card_id="2")],
fail_first={"log_in": 1, "press_class_button": 2, "list_my_bookings": 1},
)
with caplog.at_level(logging.INFO):
code = main(["--screenshot-dir", str(tmp_path)], lambda *_: gym, no_sleep)
assert code == 0
step_9, step_9_path = gym.text_screenshots[-1]
assert step_9_path == tmp_path / "step-9.png"
assert "⚠ Attempt 1/7 failed: login" in step_9
assert "succeeded after 2 attempts" in step_9
assert step_9.endswith("✅ SUCCESS: All bookings verified!")
def test_main_returns_1_when_every_retry_failed(
credentials_env: None, caplog: pytest.LogCaptureFixture
) -> None:
gym = FakeGym(fail_first={"log_in": 5})
with caplog.at_level(logging.INFO):
code = main(["--retries", "2"], lambda *_: gym, no_sleep)
assert code == 1
assert len(gym.logins) == 2
assert "Failed: login failed after 2 attempts" in caplog.text
def test_main_uses_one_attempt_when_asked() -> None:
assert parse_args(["--retries", "1"]).retries == 1
def test_retries_option_defaults_to_the_constant() -> None:
assert parse_args([]).retries == constants.RETRY_COUNT
@pytest.mark.parametrize("value", ["0", "-3", "many", "1.5"])
def test_retries_option_rejects_less_than_one(value: str) -> None:
with pytest.raises(SystemExit):
parse_args(["--retries", value])
def test_help_epilog_mentions_retries(capsys: pytest.CaptureFixture[str]) -> None:
with pytest.raises(SystemExit):
parse_args(["--help"])
out = capsys.readouterr().out
assert "--retries N" in out
assert "every retry failed" in out
## Chance that a request fails in the chaos test: the site's maximum (50%).
CHAOS_FAILURE_PROBABILITY = 0.5
## Chaos runs, and how many must complete. With 7 attempts an operation fails for good with
## probability 0.5**7 (about 0.8%). A run has four such operations (login, two clicks, the
## bookings read), so about 3% of runs fail and about 97 of 100 complete; 90 leaves room for
## the spread of the seeds without hiding a real defect. The seeds are fixed, so it is stable.
CHAOS_RUNS = 100
CHAOS_MIN_COMPLETED = 90
## Seeds that were checked to complete fully (seed 18 is the first that runs out of attempts);
## a change that breaks one of them is a defect.
CHAOS_SEEDS_THAT_MUST_PASS = tuple(range(18))
def _run_chaos(seed: int) -> tuple[int, ChaosGym]:
gym = ChaosGym(
random.Random(seed),
CHAOS_FAILURE_PROBABILITY,
cards=[make_card("Thu, Aug 14", card_id="1"), make_card("Tue, Aug 19", card_id="2")],
)
return main(["--retries", "7"], lambda *_: gym, no_sleep), gym
def _is_consistent(gym: ChaosGym) -> bool:
"""@brief True when every target class is booked on the page after the run."""
return all(card.state is ButtonState.BOOKED for card in gym.cards)
@pytest.mark.usefixtures("credentials_env")
def test_chaos_runs_with_half_of_the_requests_failing_mostly_complete() -> None:
completed = 0
for seed in range(CHAOS_RUNS):
code, gym = _run_chaos(seed)
assert code in (0, 1), f"seed {seed}: a verified run never ends in a mismatch"
if code == 0:
assert _is_consistent(gym), f"seed {seed}: verified but not booked"
completed += 1
assert completed >= CHAOS_MIN_COMPLETED
@pytest.mark.usefixtures("credentials_env")
@pytest.mark.parametrize("seed", CHAOS_SEEDS_THAT_MUST_PASS)
def test_chaos_seed_completes_with_a_consistent_verified_result(seed: int) -> None:
code, gym = _run_chaos(seed)
assert code == 0
assert _is_consistent(gym)
def test_retry_error_is_a_gym_error_for_the_exit_code() -> None:
assert issubclass(RetriesExhaustedError, GymError)
+167
View File
@@ -0,0 +1,167 @@
"""@file test_retry.py
@brief The retry wrapper: success, repeats, exhaustion, what it logs and what it never retries.
"""
import logging
import pytest
from snack_and_lift import constants
from snack_and_lift.application.retry import retry
from snack_and_lift.domain.errors import (
GymError,
LoginFailedError,
MissingCredentialsError,
PageFormatError,
RetriesExhaustedError,
UnknownButtonError,
)
class Flaky:
"""@brief A step that raises the given errors in turn and then returns a value."""
def __init__(self, errors: list[Exception], value: str = "done") -> None:
self._errors = list(errors)
self._value = value
self.calls = 0
def __call__(self) -> str:
self.calls += 1
if self._errors:
raise self._errors.pop(0)
return self._value
def test_retry_returns_the_result_on_the_first_attempt() -> None:
step = Flaky([])
sleeps: list[float] = []
assert retry(step, sleep=sleeps.append) == "done"
assert step.calls == 1
assert sleeps == []
def test_retry_succeeds_after_some_failures(caplog: pytest.LogCaptureFixture) -> None:
step = Flaky([GymError("a"), GymError("b")])
with caplog.at_level(logging.INFO):
result = retry(step, 7, "login", sleep=lambda _: None)
assert result == "done"
assert step.calls == 3
assert "✓ login succeeded after 3 attempts" in caplog.text
def test_retry_does_not_log_success_after_a_single_attempt(
caplog: pytest.LogCaptureFixture,
) -> None:
with caplog.at_level(logging.INFO):
retry(Flaky([]), 7, "login", sleep=lambda _: None)
assert "succeeded" not in caplog.text
def test_retry_raises_after_the_last_attempt_with_the_last_error_as_cause() -> None:
last = GymError("third")
step = Flaky([GymError("first"), GymError("second"), last])
with pytest.raises(RetriesExhaustedError, match="login failed after 3 attempts") as caught:
retry(step, 3, "login", sleep=lambda _: None)
assert caught.value.__cause__ is last
assert "third" in str(caught.value)
assert step.calls == 3
def test_retries_exhausted_is_a_gym_error() -> None:
with pytest.raises(GymError):
retry(Flaky([GymError("x")]), 1, sleep=lambda _: None)
def test_retry_with_one_attempt_does_not_repeat_or_sleep() -> None:
step = Flaky([GymError("x"), GymError("y")])
sleeps: list[float] = []
with pytest.raises(RetriesExhaustedError):
retry(step, 1, sleep=sleeps.append)
assert step.calls == 1
assert sleeps == []
def test_retry_with_one_attempt_returns_on_success() -> None:
assert retry(Flaky([]), 1, sleep=lambda _: None) == "done"
def test_retry_waits_the_delay_between_attempts_but_not_after_the_last() -> None:
sleeps: list[float] = []
with pytest.raises(RetriesExhaustedError):
retry(Flaky([GymError("x")] * 4), 4, delay=1.5, sleep=sleeps.append)
assert sleeps == [1.5, 1.5, 1.5]
def test_retry_uses_the_default_delay_and_attempts() -> None:
sleeps: list[float] = []
step = Flaky([GymError("x")] * constants.RETRY_COUNT)
with pytest.raises(RetriesExhaustedError):
retry(step, sleep=sleeps.append)
assert step.calls == constants.RETRY_COUNT
assert sleeps == [constants.RETRY_DELAY_SECONDS] * (constants.RETRY_COUNT - 1)
def test_retry_logs_a_warning_with_attempt_description_and_error(
caplog: pytest.LogCaptureFixture,
) -> None:
step = Flaky([GymError("Network request failed. Please try again.")])
with caplog.at_level(logging.WARNING):
retry(step, 7, "login", sleep=lambda _: None)
assert (
"⚠ Attempt 1/7 failed: login: Network request failed. Please try again. Retrying..."
in caplog.text
)
def test_retry_names_a_step_without_description(caplog: pytest.LogCaptureFixture) -> None:
with caplog.at_level(logging.WARNING):
retry(Flaky([GymError("x")]), sleep=lambda _: None)
assert f"failed: {constants.RETRY_DEFAULT_DESCRIPTION}: x" in caplog.text
@pytest.mark.parametrize(
"error",
[LoginFailedError("no"), UnknownButtonError("busy"), PageFormatError("format")],
)
def test_retry_repeats_every_kind_of_gym_error(error: GymError) -> None:
step = Flaky([error])
assert retry(step, 2, sleep=lambda _: None) == "done"
assert step.calls == 2
def test_retry_lets_other_errors_pass_untouched() -> None:
error = MissingCredentialsError("missing")
step = Flaky([error])
with pytest.raises(MissingCredentialsError) as caught:
retry(step, 7, sleep=lambda _: None)
assert caught.value is error
assert step.calls == 1
@pytest.mark.parametrize("retries", [0, -1])
def test_retry_rejects_fewer_than_one_attempt(retries: int) -> None:
step = Flaky([])
with pytest.raises(ValueError, match="at least 1"):
retry(step, retries)
assert step.calls == 0
+106
View File
@@ -0,0 +1,106 @@
"""@file test_verification.py
@brief The comparison of the bot's results with the My Bookings page, and its text.
"""
from tests.unit.test_booking_entry import make_entry
from tests.unit.test_models import make_card
from snack_and_lift.domain.models import ClassResult, Outcome
from snack_and_lift.domain.verification import build_verification, format_verification
BOOKED = ClassResult(make_card("Tue, Aug 12"), Outcome.NEW_BOOKING)
WAITLISTED = ClassResult(make_card("Thu, Aug 14", card_id="2"), Outcome.NEW_WAITLIST)
def test_everything_found_is_a_success() -> None:
result = build_verification(
[BOOKED, WAITLISTED],
[make_entry("Tue, Aug 12, 6:00 PM"), make_entry("Thu, Aug 14, 6:00 PM", waitlisted=True)],
)
assert result.is_success
assert result.verified == ["Spin Class", "Spin Class (Waitlist)"]
assert (result.expected, result.found) == (2, 2)
def test_success_text_is_exact() -> None:
result = build_verification(
[BOOKED, WAITLISTED],
[make_entry("Tue, Aug 12, 6:00 PM"), make_entry("Thu, Aug 14, 6:00 PM", waitlisted=True)],
)
assert format_verification(result) == (
"--- VERIFYING ON MY BOOKINGS PAGE ---\n"
" ✓ Verified: Spin Class\n"
" ✓ Verified: Spin Class (Waitlist)\n"
"\n"
"--- VERIFICATION RESULT ---\n"
"Expected: 2 bookings\n"
"Found: 2 bookings\n"
"✅ SUCCESS: All bookings verified!"
)
def test_one_booking_is_singular() -> None:
text = format_verification(build_verification([BOOKED], [make_entry("Tue, Aug 12, 6:00 PM")]))
assert "Expected: 1 booking\nFound: 1 booking\n" in text
def test_missing_entry_is_a_mismatch_with_a_line() -> None:
result = build_verification([BOOKED, WAITLISTED], [make_entry("Tue, Aug 12, 6:00 PM")])
text = format_verification(result)
assert not result.is_success
assert result.missing == ["Spin Class (Waitlist) on Thu, Aug 14"]
assert "Expected: 2 bookings\nFound: 1 booking\n❌ MISMATCH: Missing 1 bookings" in text
assert text.endswith(" ✗ Missing: Spin Class (Waitlist) on Thu, Aug 14")
def test_unexpected_entry_gives_a_negative_missing_count() -> None:
result = build_verification(
[BOOKED], [make_entry("Tue, Aug 12, 6:00 PM"), make_entry("Thu, Aug 14, 6:00 PM")]
)
text = format_verification(result)
assert result.unexpected == ["Spin Class on Thu, Aug 14"]
assert "❌ MISMATCH: Missing -1 bookings" in text
assert text.endswith(" ? Unexpected: Spin Class on Thu, Aug 14")
def test_waitlist_flag_must_match() -> None:
result = build_verification([BOOKED], [make_entry("Tue, Aug 12, 6:00 PM", waitlisted=True)])
assert result.missing == ["Spin Class on Tue, Aug 12"]
assert result.unexpected == ["Spin Class (Waitlist) on Tue, Aug 12"]
assert "Missing 0 bookings" in format_verification(result)
def test_date_must_match() -> None:
result = build_verification([BOOKED], [make_entry("Tue, Aug 19, 6:00 PM")])
assert not result.is_success
def test_already_waitlisted_is_expected_as_waitlist() -> None:
results = [ClassResult(make_card("Thu, Aug 14"), Outcome.ALREADY_WAITLISTED)]
assert build_verification(
results, [make_entry("Thu, Aug 14, 6:00 PM", waitlisted=True)]
).is_success
def test_nothing_expected_and_nothing_found_is_a_success() -> None:
result = build_verification([], [])
assert result.is_success
assert "Expected: 0 bookings\nFound: 0 bookings\n✅ SUCCESS" in format_verification(result)
def test_same_class_twice_is_matched_one_to_one() -> None:
result = build_verification([BOOKED, BOOKED], [make_entry("Tue, Aug 12, 6:00 PM")])
assert result.verified == ["Spin Class"]
assert result.missing == ["Spin Class on Tue, Aug 12"]