MIL-004: book a class and check its state #43

Merged
Tirsvad merged 2 commits from mil-004-book-class into main 2026-10-28 10:49:18 +01:00
18 changed files with 1398 additions and 296 deletions
Showing only changes of commit 08ba464ec9 - Show all commits
+58
View File
@@ -0,0 +1,58 @@
"""@file booking.py
@brief Lecture steps 3 and 4: book the next Tuesday 6pm class and report its state.
"""
import logging
from snack_and_lift import constants
from snack_and_lift.application.ports import GymPort
from snack_and_lift.domain.errors import GymError
from snack_and_lift.domain.models import ButtonState, ClassCard, ClassResult
from snack_and_lift.domain.rules import (
CONFIRMED_STATE,
OUTCOME_OF_ACTION,
decide_outcome_action,
find_next_class,
format_outcome,
)
logger = logging.getLogger(__name__)
def book_class(gym: GymPort, card: ClassCard) -> ClassResult:
"""@brief Book or waitlist a class, or report that it already is.
@param gym The gym site.
@param card The class, as read from the page.
@return The class and what happened.
@exception UnknownButtonError The button is busy, completed or unknown.
@exception GymError The click failed or was not confirmed by the button.
"""
action = decide_outcome_action(card)
expected = CONFIRMED_STATE.get(action)
if expected is not None:
new_text = gym.press_class_button(card.card_id)
if ButtonState.from_text(new_text) is not expected:
raise GymError(f"click not confirmed: the button says {new_text!r}")
outcome = OUTCOME_OF_ACTION[action]
logger.info(format_outcome(outcome, card))
return ClassResult(card, outcome)
def book_next_tuesday(gym: GymPort) -> ClassResult | None:
"""@brief Book the next Tuesday class at the target time.
@param gym The gym site.
@return The result, or None when the schedule has no such class.
@exception GymError The page could not be read or the booking failed.
"""
card = find_next_class(gym.list_class_cards(), constants.NEXT_CLASS_WEEKDAY)
if card is None:
logger.warning(
constants.MESSAGE_NO_CLASS_FOUND.format(
weekday=constants.NEXT_CLASS_WEEKDAY,
time=f"{constants.TARGET_HOUR:02d}:{constants.TARGET_MINUTE:02d}",
)
)
return None
return book_class(gym, card)
+66 -38
View File
@@ -1,38 +1,66 @@
"""@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
class GymPort(Protocol):
"""@brief The gym site as the use cases see it."""
class GymPort(Protocol):
def open_site(self) -> None: """@brief The gym site as the use cases see it."""
"""@brief Open the home page and wait until it is ready.
def open_site(self) -> None:
@exception GymError The page did not load. """@brief Open the home page and wait until it is ready.
"""
... @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:
@param email The account e-mail. """@brief Log in and wait until the class schedule is shown.
@param password The account password.
@exception LoginFailedError The site refused the login. @param email The account e-mail.
@exception GymError The page did not behave as expected. @param password The account password.
""" @exception LoginFailedError The site refused the login.
... @exception GymError The page did not behave as expected.
"""
def save_screenshot(self, path: Path) -> None: ...
"""@brief Save a screenshot of the current page.
def list_class_cards(self) -> list[ClassCard]:
@param path The PNG file to write; its folders are created. """@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.
def close(self) -> None: """
"""@brief Close the browser.""" ...
...
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.
@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.
@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."""
...
+167 -125
View File
@@ -1,125 +1,167 @@
"""@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 from collections.abc import Callable, Sequence
from pathlib import Path from pathlib import Path
from snack_and_lift.application.credentials import load_credentials from snack_and_lift import constants
from snack_and_lift.application.ports import GymPort from snack_and_lift.application.booking import book_next_tuesday
from snack_and_lift.application.session import login from snack_and_lift.application.credentials import Credentials, load_credentials
from snack_and_lift.domain.errors import LoginFailedError, MissingCredentialsError from snack_and_lift.application.ports import GymPort
from snack_and_lift.infrastructure.browser import create_driver, default_profile_dir from snack_and_lift.application.session import login
from snack_and_lift.infrastructure.selenium_gym import SeleniumGym from snack_and_lift.domain.errors import GymError, LoginFailedError, MissingCredentialsError
from snack_and_lift.infrastructure.browser import create_driver, default_profile_dir
## Builds the gym site from (headless, profile folder, keep open). from snack_and_lift.infrastructure.console_recorder import ConsoleRecorder
GymFactory = Callable[[bool, Path, bool], GymPort] 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: logger = logging.getLogger(__name__)
"""@brief Start Chrome and wrap it as the gym site.
@param headless True to run without a window. def selenium_gym_factory(headless: bool, profile_dir: Path, keep_open: bool) -> GymPort:
@param profile_dir The Chrome user data folder. """@brief Start Chrome and wrap it as the gym site.
@param keep_open True to leave the browser open when the program ends.
@return The gym site. @param headless True to run without a window.
""" @param profile_dir The Chrome user data folder.
return SeleniumGym(create_driver(headless, profile_dir, keep_open)) @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: return SeleniumGym(create_driver(headless, profile_dir, keep_open))
"""@brief Read the command line.
@param argv Arguments without the program name; None reads `sys.argv`. def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
@return The parsed options; `keep_open` is True unless `--headless` is given. """@brief Read the command line.
"""
parser = argparse.ArgumentParser( @param argv Arguments without the program name; None reads `sys.argv`.
prog="snack-and-lift", @return The parsed options; `keep_open` is True unless `--headless` is given.
description="Books the Tuesday and Thursday 6pm classes at the Snack and Lift gym.", """
) parser = argparse.ArgumentParser(
parser.add_argument( prog="snack-and-lift",
"--headless", description="Books the Tuesday and Thursday 6pm classes at the Snack and Lift gym.",
action="store_true", )
help="run Chrome without a window (the Docker image does this)", parser.add_argument(
) "--headless",
parser.add_argument( action="store_true",
"--keep-open", help="run Chrome without a window (the Docker image does this)",
action="store_true", )
default=None, parser.add_argument(
help="leave the browser open at the end (default when not headless)", "--keep-open",
) action="store_true",
parser.add_argument( default=None,
"--profile-dir", help="leave the browser open at the end (default when not headless)",
type=Path, )
default=None, parser.add_argument(
help="Chrome profile folder (default: ./chrome_profile)", "--profile-dir",
) type=Path,
parser.add_argument( default=None,
"--screenshot-dir", help="Chrome profile folder (default: ./chrome_profile)",
type=Path, )
default=None, parser.add_argument(
help="save a screenshot after each step into this folder", "--screenshot-dir",
) type=Path,
args = parser.parse_args(argv) default=None,
if args.keep_open is None: help="save a screenshot after each step into this folder",
args.keep_open = not args.headless )
if args.profile_dir is None: args = parser.parse_args(argv)
args.profile_dir = default_profile_dir() if args.keep_open is None:
return args args.keep_open = not args.headless
if args.profile_dir is None:
args.profile_dir = default_profile_dir()
def _use_utf8_output() -> None: return args
"""@brief Make stdout UTF-8 so the check mark of the messages prints on Windows."""
reconfigure = getattr(sys.stdout, "reconfigure", None)
if reconfigure is not None: def _use_utf8_output() -> None:
reconfigure(encoding="utf-8") """@brief Make stdout UTF-8 so the check mark of the messages prints on Windows."""
reconfigure = getattr(sys.stdout, "reconfigure", None)
if reconfigure is not None:
def _save_step(gym: GymPort, screenshot_dir: Path | None, name: str) -> None: reconfigure(encoding="utf-8")
"""@brief Save a screenshot when a folder was asked for.
@param gym The gym site. def _save_step(gym: GymPort, screenshot_dir: Path | None, name: str) -> None:
@param screenshot_dir The folder, or None for no screenshots. """@brief Save a screenshot when a folder was asked for.
@param name The file name.
""" @param gym The gym site.
if screenshot_dir is not None: @param screenshot_dir The folder, or None for no screenshots.
gym.save_screenshot(screenshot_dir / name) @param name The file name.
"""
if screenshot_dir is not None:
def main( gym.save_screenshot(screenshot_dir / name)
argv: Sequence[str] | None = None,
gym_factory: GymFactory = selenium_gym_factory,
) -> int: def _run(
"""@brief Start the bot: open the site and log in. gym: GymPort, args: argparse.Namespace, recorder: ConsoleRecorder, credentials: Credentials
) -> None:
@param argv Arguments without the program name; None reads `sys.argv`. """@brief Open the site, log in and book the next Tuesday class, saving the step images.
@param gym_factory Builds the gym site; tests pass a fake.
@return Process exit code: 0 done, 1 login failed, 2 credentials missing. @param gym The gym site.
""" @param args The parsed options.
logging.basicConfig(level=logging.INFO, format="%(message)s") @param recorder The recorded console lines, drawn for step 4.
_use_utf8_output() @param credentials The account to log in with.
args = parse_args(argv) @exception GymError A step failed.
try: """
credentials = load_credentials(os.environ) gym.open_site()
except MissingCredentialsError as error: _save_step(gym, args.screenshot_dir, "step-1.png")
logger.error("%s", error) login(gym, credentials)
return 2 _save_step(gym, args.screenshot_dir, "step-2.png")
gym = gym_factory(args.headless, args.profile_dir, args.keep_open) book_next_tuesday(gym)
try: _save_step(gym, args.screenshot_dir, constants.SCREENSHOT_STEP_3)
gym.open_site() if args.screenshot_dir is not None:
_save_step(gym, args.screenshot_dir, "step-1.png") gym.save_text_screenshot(recorder.text, args.screenshot_dir / constants.SCREENSHOT_STEP_4)
login(gym, credentials)
_save_step(gym, args.screenshot_dir, "step-2.png")
except LoginFailedError as error: def main(
logger.error("Login failed: %s", error) argv: Sequence[str] | None = None,
return 1 gym_factory: GymFactory = selenium_gym_factory,
finally: ) -> int:
if not args.keep_open: """@brief Start the bot: open the site, log in and book the next Tuesday 6pm class.
gym.close()
return 0 @param argv Arguments without the program name; None reads `sys.argv`.
@param gym_factory Builds the gym site; tests pass a fake.
@return Process exit code: 0 done, 1 a step failed, 2 credentials missing.
"""
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)
finally:
root.removeHandler(recorder)
def _main(argv: Sequence[str] | None, gym_factory: GymFactory, recorder: ConsoleRecorder) -> 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.
@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:
_run(gym, args, recorder, credentials)
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
+62
View File
@@ -64,3 +64,65 @@ HEADLESS_BROWSER_ARGUMENTS: Final[tuple[str, ...]] = (
"--disable-dev-shm-usage", "--disable-dev-shm-usage",
"--disable-gpu", "--disable-gpu",
) )
## Weekday of the class that lecture steps 3 and 4 book, as the site abbreviates it.
NEXT_CLASS_WEEKDAY: Final[str] = "Tue"
## Hour (24-hour clock) of the class the bot books.
TARGET_HOUR: Final[int] = 18
## Minute of the class the bot books.
TARGET_MINUTE: Final[int] = 0
## CSS selector of every class card on the schedule.
CLASS_CARD_CSS: Final[str] = "div[id^='class-card-']"
## Prefix of the element ids of one card; the card id (for example "123") follows it.
CLASS_CARD_ID_PREFIX: Final[str] = "class-card-"
CLASS_NAME_ID_PREFIX: Final[str] = "class-name-"
CLASS_TIME_ID_PREFIX: Final[str] = "class-time-"
BOOK_BUTTON_ID_PREFIX: Final[str] = "book-button-"
COMPLETED_LABEL_ID_PREFIX: Final[str] = "completed-label-"
CLASS_ERROR_ID_PREFIX: Final[str] = "class-error-"
## XPath, relative to a card, of the title of the day group the card is in (the "ancestor" hint).
DAY_TITLE_XPATH: Final[str] = (
"ancestor::div[starts-with(@id,'day-group-')]//h2[starts-with(@id,'day-title-')]"
)
## Texts of the button of a card.
BUTTON_TEXT_BOOK: Final[str] = "Book Class"
BUTTON_TEXT_JOIN_WAITLIST: Final[str] = "Join Waitlist"
BUTTON_TEXT_BOOKED: Final[str] = "Booked"
BUTTON_TEXT_WAITLISTED: Final[str] = "Waitlisted"
BUTTON_TEXT_BOOKING: Final[str] = "Booking..."
BUTTON_TEXT_JOINING: Final[str] = "Joining..."
BUTTON_TEXT_COMPLETED: Final[str] = "Completed"
## Prefix of the time text of a card ("Time: 6:00 PM").
TIME_TEXT_PREFIX: Final[str] = "Time:"
## Day labels the site wraps around the date of the next two days.
DAY_LABEL_WRAPPERS: Final[tuple[str, ...]] = ("Today", "Tomorrow")
## Console messages of the booking, with the class name and the day label as the site shows it.
MESSAGE_BOOKED: Final[str] = "✓ Successfully booked: {name} on {day}"
MESSAGE_WAITLISTED: Final[str] = "✓ Joined waitlist for: {name} on {day}"
MESSAGE_ALREADY_BOOKED: Final[str] = "✓ Already booked: {name} on {day}"
MESSAGE_ALREADY_WAITLISTED: Final[str] = "✓ Already on waitlist: {name} on {day}"
MESSAGE_NO_CLASS_FOUND: Final[str] = "No {weekday} class at {time} found on the schedule"
## File names of the screenshots of lecture steps 3 and 4.
SCREENSHOT_STEP_3: Final[str] = "step-3.png"
SCREENSHOT_STEP_4: Final[str] = "step-4.png"
## `string.Template` of the page that renders console text for a screenshot.
CONSOLE_PAGE_TEMPLATE: Final[str] = (
"<!DOCTYPE html><html><head><meta charset='utf-8'></head>"
"<body style='margin:0;background:#1e1e1e;color:#d4d4d4'>"
"<pre style='margin:0;padding:16px;font:16px/1.5 monospace;white-space:pre-wrap'>"
"$text</pre></body></html>"
)
## Prefix of the `data:` URL that carries the console page.
CONSOLE_PAGE_URL_PREFIX: Final[str] = "data:text/html;charset=utf-8,"
+29
View File
@@ -0,0 +1,29 @@
"""@file day_label.py
@brief Reads the title of a day group of the schedule ("Today (Tue, Aug 12)").
Kept apart from `rules.py` so that `models.py` can use it without a circular import.
"""
import re
from snack_and_lift import constants
from snack_and_lift.domain.errors import PageFormatError
_DATE = re.compile(r"(?P<weekday>[A-Z][a-z]{2}), [A-Z][a-z]{2} \d{1,2}")
_WRAPPER = re.compile(rf"(?:{'|'.join(constants.DAY_LABEL_WRAPPERS)}) \((?P<inner>.+)\)")
def parse_day_label(label: str) -> tuple[str, str]:
"""@brief Split a day label into the weekday and the date text.
@param label `Today (Tue, Aug 12)`, `Tomorrow (Wed, Aug 13)` or `Thu, Aug 14`.
@return The weekday (`Tue`) and the date text (`Tue, Aug 12`).
@exception PageFormatError The label has none of these forms.
"""
text = label.strip()
wrapped = _WRAPPER.fullmatch(text)
date_text = wrapped["inner"] if wrapped else text
match = _DATE.fullmatch(date_text)
if match is None:
raise PageFormatError(f"Unexpected day label: {label!r}")
return match["weekday"], date_text
+23 -15
View File
@@ -1,15 +1,23 @@
"""@file errors.py """@file errors.py
@brief The errors of the bot. @brief The errors of the bot.
""" """
class GymError(Exception): class GymError(Exception):
"""@brief The gym site did not behave as the bot expected (a page, element or timeout).""" """@brief The gym site did not behave as the bot expected (a page, element or timeout)."""
class LoginFailedError(GymError): class LoginFailedError(GymError):
"""@brief The site refused the login; the message is the text the site shows.""" """@brief The site refused the login; the message is the text the site shows."""
class MissingCredentialsError(Exception): class MissingCredentialsError(Exception):
"""@brief The account e-mail or password is not set in the environment.""" """@brief The account e-mail or password is not set in the environment."""
class UnknownButtonError(GymError):
"""@brief A class button shows a text the bot cannot act on (busy, completed or unknown)."""
class PageFormatError(GymError):
"""@brief A text read from the page (a day label or a time) has an unexpected format."""
+94
View File
@@ -0,0 +1,94 @@
"""@file models.py
@brief Plain data of the schedule: the button state, a class card and the result of a booking.
"""
from dataclasses import dataclass
from enum import Enum, auto
from snack_and_lift import constants
from snack_and_lift.domain.day_label import parse_day_label
class ButtonState(Enum):
"""@brief What the button of a class card says."""
BOOK = auto()
JOIN_WAITLIST = auto()
BOOKED = auto()
WAITLISTED = auto()
BUSY = auto()
COMPLETED = auto()
UNKNOWN = auto()
@classmethod
def from_text(cls, text: str) -> "ButtonState":
"""@brief Map the text of a button to its state.
@param text The visible text of the button (or the label of a past class).
@return The state; UNKNOWN for a text the bot does not know.
"""
return _STATE_BY_TEXT.get(text.strip(), cls.UNKNOWN)
_STATE_BY_TEXT: dict[str, ButtonState] = {
constants.BUTTON_TEXT_BOOK: ButtonState.BOOK,
constants.BUTTON_TEXT_JOIN_WAITLIST: ButtonState.JOIN_WAITLIST,
constants.BUTTON_TEXT_BOOKED: ButtonState.BOOKED,
constants.BUTTON_TEXT_WAITLISTED: ButtonState.WAITLISTED,
constants.BUTTON_TEXT_BOOKING: ButtonState.BUSY,
constants.BUTTON_TEXT_JOINING: ButtonState.BUSY,
constants.BUTTON_TEXT_COMPLETED: ButtonState.COMPLETED,
}
@dataclass(frozen=True, slots=True)
class ClassCard:
"""@brief One class of the schedule, as read from the page.
`card_id` is the id of the card without the `class-card-` prefix (for example
"123"); the site builds the ids of the card's other elements from the same suffix.
"""
card_id: str
name: str
day_label: str
time_text: str
button_text: str
@property
def state(self) -> ButtonState:
"""@brief The state of the button."""
return ButtonState.from_text(self.button_text)
@property
def weekday(self) -> str:
"""@brief The weekday of the day group (`Tue`).
@exception PageFormatError The day label is not in a known form.
"""
return parse_day_label(self.day_label)[0]
@property
def date_text(self) -> str:
"""@brief The date of the day group (`Tue, Aug 12`).
@exception PageFormatError The day label is not in a known form.
"""
return parse_day_label(self.day_label)[1]
class Outcome(Enum):
"""@brief What the booking did, or found, for a class."""
NEW_BOOKING = auto()
NEW_WAITLIST = auto()
ALREADY_BOOKED = auto()
ALREADY_WAITLISTED = auto()
@dataclass(frozen=True, slots=True)
class ClassResult:
"""@brief The class that was handled and what happened to it."""
card: ClassCard
outcome: Outcome
+139
View File
@@ -0,0 +1,139 @@
"""@file rules.py
@brief The booking rules: which class is the target, what to do for a button state, what to say.
"""
import re
from collections.abc import Iterable
from enum import Enum, auto
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, UnknownButtonError
from snack_and_lift.domain.models import ButtonState, ClassCard, Outcome
__all__ = [
"CONFIRMED_STATE",
"OUTCOME_OF_ACTION",
"Action",
"decide_outcome_action",
"find_next_class",
"format_outcome",
"is_target_time",
"parse_day_label",
"parse_time",
]
_TIME = re.compile(
rf"(?:{re.escape(constants.TIME_TEXT_PREFIX)}\s*)?(\d{{1,2}}):(\d{{2}})\s*(AM|PM)",
re.IGNORECASE,
)
_MAX_CLOCK_HOUR = 12
_MAX_MINUTE = 59
class Action(Enum):
"""@brief What the bot does for a class card."""
CLICK_BOOK = auto()
CLICK_WAITLIST = auto()
ALREADY_BOOKED = auto()
ALREADY_WAITLISTED = auto()
## The outcome that follows from each action.
OUTCOME_OF_ACTION: dict[Action, Outcome] = {
Action.CLICK_BOOK: Outcome.NEW_BOOKING,
Action.CLICK_WAITLIST: Outcome.NEW_WAITLIST,
Action.ALREADY_BOOKED: Outcome.ALREADY_BOOKED,
Action.ALREADY_WAITLISTED: Outcome.ALREADY_WAITLISTED,
}
## The button state that proves a click worked.
CONFIRMED_STATE: dict[Action, ButtonState] = {
Action.CLICK_BOOK: ButtonState.BOOKED,
Action.CLICK_WAITLIST: ButtonState.WAITLISTED,
}
_ACTION_OF_STATE: dict[ButtonState, Action] = {
ButtonState.BOOK: Action.CLICK_BOOK,
ButtonState.JOIN_WAITLIST: Action.CLICK_WAITLIST,
ButtonState.BOOKED: Action.ALREADY_BOOKED,
ButtonState.WAITLISTED: Action.ALREADY_WAITLISTED,
}
_MESSAGE_OF_OUTCOME: dict[Outcome, str] = {
Outcome.NEW_BOOKING: constants.MESSAGE_BOOKED,
Outcome.NEW_WAITLIST: constants.MESSAGE_WAITLISTED,
Outcome.ALREADY_BOOKED: constants.MESSAGE_ALREADY_BOOKED,
Outcome.ALREADY_WAITLISTED: constants.MESSAGE_ALREADY_WAITLISTED,
}
def parse_time(text: str) -> tuple[int, int]:
"""@brief Read a 12-hour time such as `6:00 PM`.
@param text The time, with or without the `Time:` prefix.
@return The hour (0-23) and the minute.
@exception PageFormatError The text is not a valid time.
"""
match = _TIME.fullmatch(text.strip())
if match is None:
raise PageFormatError(f"Unexpected time text: {text!r}")
hour, minute = int(match[1]), int(match[2])
if not 1 <= hour <= _MAX_CLOCK_HOUR or minute > _MAX_MINUTE:
raise PageFormatError(f"Time out of range: {text!r}")
hour %= _MAX_CLOCK_HOUR
if match[3].upper() == "PM":
hour += _MAX_CLOCK_HOUR
return hour, minute
def is_target_time(card: ClassCard) -> bool:
"""@brief Tell whether the class starts at the time the bot books.
@param card The class.
@return True at `TARGET_HOUR`:`TARGET_MINUTE`.
@exception PageFormatError The time text is not valid.
"""
return parse_time(card.time_text) == (constants.TARGET_HOUR, constants.TARGET_MINUTE)
def find_next_class(cards: Iterable[ClassCard], weekday: str) -> ClassCard | None:
"""@brief Find the first class of a weekday at the target time, in page order.
The weekday is read from each card's day label; the position in the list means nothing.
Classes already completed are skipped.
@param cards The classes in the order of the page.
@param weekday The weekday as the site abbreviates it (`Tue`).
@return The class, or None when there is none.
"""
for card in cards:
if card.state is ButtonState.COMPLETED:
continue
if card.weekday == weekday and is_target_time(card):
return card
return None
def decide_outcome_action(card: ClassCard) -> Action:
"""@brief Decide what to do from the state of the button.
@param card The class.
@return The action.
@exception UnknownButtonError The button is busy, completed or unknown; `retry` can repeat.
"""
action = _ACTION_OF_STATE.get(card.state)
if action is None:
raise UnknownButtonError(f"Cannot act on button {card.button_text!r} of {card.name}")
return action
def format_outcome(outcome: Outcome, card: ClassCard) -> str:
"""@brief Build the console message of an outcome.
@param outcome What happened.
@param card The class; its day label is shown as the site writes it.
@return For example `✓ Successfully booked: Spin Class on Tomorrow (Thu, Aug 14)`.
"""
return _MESSAGE_OF_OUTCOME[outcome].format(name=card.name, day=card.day_label)
@@ -0,0 +1,35 @@
"""@file console_recorder.py
@brief Keeps the lines the bot logs, so they can be drawn as the lecture's console screenshot.
"""
import logging
class ConsoleRecorder(logging.Handler):
"""@brief A logging handler that stores the formatted message of every record."""
def __init__(self, level: int = logging.NOTSET) -> None:
"""@brief Create an empty recorder.
@param level The lowest level to keep.
"""
super().__init__(level)
self.setFormatter(logging.Formatter("%(message)s"))
self._lines: list[str] = []
@property
def lines(self) -> list[str]:
"""@brief The recorded lines, oldest first (a copy)."""
return list(self._lines)
@property
def text(self) -> str:
"""@brief The recorded lines joined by newlines."""
return "\n".join(self._lines)
def emit(self, record: logging.LogRecord) -> None:
"""@brief Store one record.
@param record The log record.
"""
self._lines.append(self.format(record))
+194 -111
View File
@@ -1,111 +1,194 @@
"""@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 logging import html
from collections.abc import Callable import logging
from pathlib import Path import urllib.parse
from typing import Concatenate from collections.abc import Callable
from pathlib import Path
from selenium.common.exceptions import ( from string import Template
ElementClickInterceptedException, from typing import Concatenate, Literal
NoSuchElementException,
StaleElementReferenceException, from selenium.common.exceptions import (
TimeoutException, ElementClickInterceptedException,
) NoSuchElementException,
from selenium.webdriver.common.by import By StaleElementReferenceException,
from selenium.webdriver.remote.webdriver import WebDriver TimeoutException,
from selenium.webdriver.support import expected_conditions as ec )
from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.common.by import By
from selenium.webdriver.remote.webdriver import WebDriver
from snack_and_lift import constants from selenium.webdriver.remote.webelement import WebElement
from snack_and_lift.domain.errors import GymError, LoginFailedError from selenium.webdriver.support import expected_conditions as ec
from selenium.webdriver.support.ui import WebDriverWait
logger = logging.getLogger(__name__)
from snack_and_lift import constants
_SELENIUM_ERRORS = ( from snack_and_lift.domain.errors import GymError, LoginFailedError
TimeoutException, from snack_and_lift.domain.models import ButtonState, ClassCard
NoSuchElementException,
StaleElementReferenceException, logger = logging.getLogger(__name__)
ElementClickInterceptedException,
) _SELENIUM_ERRORS = (
TimeoutException,
NoSuchElementException,
def _translate_errors[S, **P, R]( StaleElementReferenceException,
method: Callable[Concatenate[S, P], R], ElementClickInterceptedException,
) -> Callable[Concatenate[S, P], R]: )
"""@brief Turn Selenium errors of a method into GymError.
@param method The adapter method to wrap. def _translate_errors[S, **P, R](
@return The wrapped method. method: Callable[Concatenate[S, P], R],
""" ) -> Callable[Concatenate[S, P], R]:
"""@brief Turn Selenium errors of a method into GymError.
@functools.wraps(method)
def wrapper(self: S, *args: P.args, **kwargs: P.kwargs) -> R: @param method The adapter method to wrap.
try: @return The wrapped method.
return method(self, *args, **kwargs) """
except _SELENIUM_ERRORS as error:
raise GymError(f"{method.__name__} failed: {type(error).__name__}") from error @functools.wraps(method)
def wrapper(self: S, *args: P.args, **kwargs: P.kwargs) -> R:
return wrapper try:
return method(self, *args, **kwargs)
except _SELENIUM_ERRORS as error:
class SeleniumGym: raise GymError(f"{method.__name__} failed: {type(error).__name__}") from error
"""@brief The gym site driven by a Selenium WebDriver."""
return wrapper
def __init__(self, driver: WebDriver, base_url: str = constants.GYM_URL) -> None:
"""@brief Wrap a driver.
class SeleniumGym:
@param driver A running WebDriver. """@brief The gym site driven by a Selenium WebDriver."""
@param base_url The address of the home page.
""" def __init__(self, driver: WebDriver, base_url: str = constants.GYM_URL) -> None:
self._driver = driver """@brief Wrap a driver.
self._base_url = base_url
self._wait = WebDriverWait(driver, constants.WAIT_TIMEOUT_SECONDS) @param driver A running WebDriver.
@param base_url The address of the home page.
@_translate_errors """
def open_site(self) -> None: self._driver = driver
"""@brief Open the home page and wait for the login button. self._base_url = base_url
self._wait = WebDriverWait(driver, constants.WAIT_TIMEOUT_SECONDS)
@exception GymError The page did not load.
""" @_translate_errors
self._driver.get(self._base_url) def open_site(self) -> None:
self._wait.until(ec.presence_of_element_located((By.ID, constants.NAV_LOGIN_BUTTON_ID))) """@brief Open the home page and wait for the login button.
@_translate_errors @exception GymError The page did not load.
def log_in(self, email: str, password: str) -> None: """
"""@brief Fill in the login form and wait for the schedule or an error. self._driver.get(self._base_url)
self._wait.until(ec.presence_of_element_located((By.ID, constants.NAV_LOGIN_BUTTON_ID)))
@param email The account e-mail.
@param password The account password. @_translate_errors
@exception LoginFailedError The site shows a login error. def log_in(self, email: str, password: str) -> None:
@exception GymError The page did not behave as expected. """@brief Fill in the login form and wait for the schedule or an error.
"""
self._wait.until(ec.element_to_be_clickable((By.ID, constants.NAV_LOGIN_BUTTON_ID))).click() @param email The account e-mail.
self._wait.until( @param password The account password.
ec.presence_of_element_located((By.ID, constants.LOGIN_EMAIL_INPUT_ID)) @exception LoginFailedError The site shows a login error.
).send_keys(email) @exception GymError The page did not behave as expected.
self._driver.find_element(By.ID, constants.LOGIN_PASSWORD_INPUT_ID).send_keys(password) """
self._driver.find_element(By.ID, constants.LOGIN_SUBMIT_BUTTON_ID).click() self._wait.until(ec.element_to_be_clickable((By.ID, constants.NAV_LOGIN_BUTTON_ID))).click()
self._wait.until( 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)), ).send_keys(email)
ec.presence_of_element_located((By.ID, constants.ERROR_MESSAGE_ID)), self._driver.find_element(By.ID, constants.LOGIN_PASSWORD_INPUT_ID).send_keys(password)
) self._driver.find_element(By.ID, constants.LOGIN_SUBMIT_BUTTON_ID).click()
) self._wait.until(
errors = self._driver.find_elements(By.ID, constants.ERROR_MESSAGE_ID) ec.any_of(
if errors: ec.presence_of_element_located((By.ID, constants.SCHEDULE_PAGE_ID)),
raise LoginFailedError(errors[0].text) ec.presence_of_element_located((By.ID, constants.ERROR_MESSAGE_ID)),
)
def save_screenshot(self, path: Path) -> None: )
"""@brief Save a screenshot of the current page. errors = self._driver.find_elements(By.ID, constants.ERROR_MESSAGE_ID)
if errors:
@param path The PNG file to write; its folders are created. raise LoginFailedError(errors[0].text)
"""
path.parent.mkdir(parents=True, exist_ok=True) @_translate_errors
self._driver.save_screenshot(str(path)) def list_class_cards(self) -> list[ClassCard]:
"""@brief Read every class card; the day comes from the ancestor day group.
def close(self) -> None:
"""@brief Quit the browser.""" @return The classes in page order.
self._driver.quit() @exception GymError The schedule could not be read.
"""
self._wait.until(
ec.presence_of_element_located((By.CSS_SELECTOR, constants.CLASS_CARD_CSS))
)
elements = self._driver.find_elements(By.CSS_SELECTOR, constants.CLASS_CARD_CSS)
return [self._read_card(element) for element in elements]
def _read_card(self, element: WebElement) -> ClassCard:
"""@brief Turn one card element into plain data.
@param element The card.
@return The class.
"""
card_id = (element.get_attribute("id") or "").removeprefix(constants.CLASS_CARD_ID_PREFIX)
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]
return ClassCard(
card_id=card_id,
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,
time_text=element.find_element(By.ID, constants.CLASS_TIME_ID_PREFIX + card_id).text,
button_text=control.text if control is not None else "",
)
@_translate_errors
def press_class_button(self, card_id: str) -> str:
"""@brief Scroll to the button, click it and wait for the result or the error box.
@param card_id The id of the card without its prefix.
@return The new text of the button.
@exception GymError The site shows an error for the class, or nothing settled in time.
"""
button_id = constants.BOOK_BUTTON_ID_PREFIX + card_id
button = self._wait.until(ec.presence_of_element_located((By.ID, button_id)))
self._driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", button)
before = button.text
self._wait.until(ec.element_to_be_clickable((By.ID, button_id))).click()
def settled(driver: WebDriver) -> tuple[str, str] | Literal[False]:
errors = driver.find_elements(By.ID, constants.CLASS_ERROR_ID_PREFIX + card_id)
text = driver.find_element(By.ID, button_id).text
if errors:
return text, errors[0].text or "class error"
if text != before and ButtonState.from_text(text) is not ButtonState.BUSY:
return text, ""
return False
text, error = self._wait.until(settled)
if error:
raise GymError(error)
return text
@_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()
+78 -1
View File
@@ -2,6 +2,83 @@
<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>
<div id="schedule-page">Class schedule</div> <div id="schedule-page">
<h1>Class Schedule (Today: Wed, Aug 13)</h1>
<select id="day-filter"><option>Next 7 Days</option></select>
<div id="day-group-wed-aug-13">
<h2 id="day-title-wed-aug-13">Today (Wed, Aug 13)</h2>
<div id="class-card-101" data-class-type="yoga" data-user-booked="false">
<h3 id="class-name-101">Yoga Class</h3>
<p id="class-time-101"><strong>Time:</strong> 6:00 AM</p>
<span id="completed-label-101">Completed</span>
</div>
<div id="class-card-102" data-class-type="spin" data-user-booked="false">
<h3 id="class-name-102">Spin Class</h3>
<p id="class-time-102"><strong>Time:</strong> 6:00 PM</p>
<button id="book-button-102">Book Class</button>
</div>
</div>
<div id="day-group-thu-aug-14">
<h2 id="day-title-thu-aug-14">Tomorrow (Thu, Aug 14)</h2>
<div id="class-card-201" data-class-type="hiit" data-user-booked="false">
<h3 id="class-name-201">HIIT Class</h3>
<p id="class-time-201"><strong>Time:</strong> 6:00 PM</p>
<button id="book-button-201">Join Waitlist</button>
</div>
</div>
<div id="day-group-fri-aug-15">
<h2 id="day-title-fri-aug-15">Fri, Aug 15</h2>
<div id="class-card-401" data-class-type="hiit" data-user-booked="true">
<h3 id="class-name-401">HIIT Class</h3>
<p id="class-time-401"><strong>Time:</strong> 5:00 PM</p>
<button id="book-button-401" disabled>Booked</button>
</div>
</div>
<div id="day-group-tue-aug-19">
<h2 id="day-title-tue-aug-19">Tue, Aug 19</h2>
<div id="class-card-301" data-class-type="yoga" data-user-booked="false">
<h3 id="class-name-301">Yoga Class</h3>
<p id="class-time-301"><strong>Time:</strong> 7:00 AM</p>
<button id="book-button-301">Book Class</button>
</div>
<div id="class-card-302" data-class-type="spin" data-user-booked="false" data-fail-once="true">
<h3 id="class-name-302">Spin Class</h3>
<p id="class-time-302"><strong>Time:</strong> 6:00 PM</p>
<button id="book-button-302">Book Class</button>
</div>
</div>
</div>
<script>
// Imitates the site: a busy text, then the new state; card 302 fails on its first click.
document.querySelectorAll("button[id^='book-button-']").forEach(function (button) {
button.addEventListener("click", function () {
const id = button.id.replace("book-button-", "");
const card = document.getElementById("class-card-" + id);
const oldText = button.textContent;
const stale = document.getElementById("class-error-" + id);
if (stale) { stale.remove(); }
const waitlist = oldText === "Join Waitlist";
button.textContent = waitlist ? "Joining..." : "Booking...";
button.disabled = true;
setTimeout(function () {
if (card.dataset.failOnce === "true") {
card.dataset.failOnce = "false";
button.textContent = oldText;
button.disabled = false;
const box = document.createElement("div");
box.id = "class-error-" + id;
box.textContent = "Network error, please try again";
card.appendChild(box);
return;
}
button.textContent = waitlist ? "Waitlisted" : "Booked";
}, 200);
});
});
</script>
</body> </body>
</html> </html>
+56 -1
View File
@@ -8,7 +8,7 @@ from pathlib import Path
import pytest import pytest
from snack_and_lift.domain.errors import LoginFailedError from snack_and_lift.domain.errors import GymError, LoginFailedError
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
@@ -55,3 +55,58 @@ def test_save_screenshot_creates_the_folder(gym: SeleniumGym, tmp_path: Path) ->
gym.save_screenshot(target) gym.save_screenshot(target)
assert target.stat().st_size > 0 assert target.stat().st_size > 0
FIXTURE_SCHEDULE = (Path(__file__).parent / "fixtures" / "schedule.html").as_uri()
@pytest.fixture
def schedule_gym(tmp_path: Path) -> Iterator[SeleniumGym]:
driver = create_driver(headless=True, profile_dir=tmp_path / "profile", keep_open=False)
try:
driver.get(FIXTURE_SCHEDULE)
yield SeleniumGym(driver, base_url=FIXTURE_SCHEDULE)
finally:
driver.quit()
def test_list_class_cards_reads_the_day_from_the_ancestor_group(schedule_gym: SeleniumGym) -> None:
cards = {card.card_id: card for card in schedule_gym.list_class_cards()}
assert list(cards) == ["101", "102", "201", "401", "301", "302"]
assert cards["102"].day_label == "Today (Wed, Aug 13)"
assert cards["201"].weekday == "Thu"
assert cards["302"].weekday == "Tue"
assert cards["302"].name == "Spin Class"
assert cards["302"].time_text == "Time: 6:00 PM"
assert cards["302"].button_text == "Book Class"
assert cards["101"].button_text == "Completed"
assert cards["401"].button_text == "Booked"
def test_press_class_button_confirms_booking(schedule_gym: SeleniumGym) -> None:
assert schedule_gym.press_class_button("102") == "Booked"
def test_press_class_button_confirms_waitlist(schedule_gym: SeleniumGym) -> None:
assert schedule_gym.press_class_button("201") == "Waitlisted"
def test_press_class_button_raises_with_the_error_text_then_succeeds(
schedule_gym: SeleniumGym,
) -> None:
with pytest.raises(GymError, match="Network error"):
schedule_gym.press_class_button("302")
assert schedule_gym.press_class_button("302") == "Booked"
def test_save_text_screenshot_writes_a_png_and_keeps_the_page(
schedule_gym: SeleniumGym, tmp_path: Path
) -> None:
target = tmp_path / "shots" / "step-4.png"
schedule_gym.save_text_screenshot("✓ Successfully booked: <Spin Class>", target)
assert target.read_bytes().startswith(b"\x89PNG")
assert schedule_gym.list_class_cards()
+39 -3
View File
@@ -4,17 +4,35 @@
from pathlib import Path from pathlib import Path
from snack_and_lift.domain.errors import LoginFailedError from snack_and_lift import constants
from snack_and_lift.domain.errors import GymError, LoginFailedError
from snack_and_lift.domain.models import ClassCard
_PRESS_RESULTS = {
constants.BUTTON_TEXT_BOOK: constants.BUTTON_TEXT_BOOKED,
constants.BUTTON_TEXT_JOIN_WAITLIST: constants.BUTTON_TEXT_WAITLISTED,
}
class FakeGym: class FakeGym:
"""@brief Records the calls; can be told to refuse the login.""" """@brief Records the calls; can refuse the login, serve cards and fail a click."""
def __init__(self, login_error: str | None = None) -> None: def __init__(
self,
login_error: str | None = None,
cards: list[ClassCard] | None = None,
press_result: str | None = None,
press_error: str | None = None,
) -> None:
self.login_error = login_error self.login_error = login_error
self.cards = cards or []
self.press_result = press_result
self.press_error = press_error
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] = []
self.text_screenshots: list[tuple[str, Path]] = []
self.pressed: list[str] = []
self.closed = False self.closed = False
def open_site(self) -> None: def open_site(self) -> None:
@@ -26,6 +44,24 @@ class FakeGym:
if self.login_error is not None: if self.login_error is not None:
raise LoginFailedError(self.login_error) raise LoginFailedError(self.login_error)
def list_class_cards(self) -> list[ClassCard]:
self.calls.append("list_class_cards")
return list(self.cards)
def press_class_button(self, card_id: str) -> str:
self.calls.append("press_class_button")
self.pressed.append(card_id)
if self.press_error is not None:
raise GymError(self.press_error)
if self.press_result is not None:
return self.press_result
card = next(card for card in self.cards if card.card_id == card_id)
return _PRESS_RESULTS[card.button_text]
def save_text_screenshot(self, text: str, path: Path) -> None:
self.calls.append("save_text_screenshot")
self.text_screenshots.append((text, path))
def save_screenshot(self, path: Path) -> None: def save_screenshot(self, path: Path) -> None:
self.calls.append("save_screenshot") self.calls.append("save_screenshot")
self.screenshots.append(path) self.screenshots.append(path)
+97
View File
@@ -0,0 +1,97 @@
"""@file test_booking.py
@brief Booking a class and booking the next Tuesday class, against a fake gym.
"""
import logging
import pytest
from tests.unit.fakes import FakeGym
from tests.unit.test_models import make_card
from snack_and_lift.application.booking import book_class, book_next_tuesday
from snack_and_lift.domain.errors import GymError, UnknownButtonError
from snack_and_lift.domain.models import Outcome
@pytest.mark.parametrize(
("button_text", "outcome", "message"),
[
("Book Class", Outcome.NEW_BOOKING, "✓ Successfully booked: Spin Class on"),
("Join Waitlist", Outcome.NEW_WAITLIST, "✓ Joined waitlist for: Spin Class on"),
],
)
def test_book_class_clicks_and_confirms(
button_text: str, outcome: Outcome, message: str, caplog: pytest.LogCaptureFixture
) -> None:
card = make_card(button_text=button_text)
gym = FakeGym(cards=[card])
with caplog.at_level(logging.INFO):
result = book_class(gym, card)
assert result.outcome is outcome
assert gym.pressed == ["123"]
assert message in caplog.text
@pytest.mark.parametrize(
("button_text", "outcome"),
[("Booked", Outcome.ALREADY_BOOKED), ("Waitlisted", Outcome.ALREADY_WAITLISTED)],
)
def test_book_class_does_not_click_when_already_done(button_text: str, outcome: Outcome) -> None:
gym = FakeGym()
result = book_class(gym, make_card(button_text=button_text))
assert result.outcome is outcome
assert gym.pressed == []
def test_book_class_raises_when_click_is_not_confirmed() -> None:
card = make_card()
gym = FakeGym(cards=[card], press_result="Book Class")
with pytest.raises(GymError, match="click not confirmed"):
book_class(gym, card)
def test_book_class_raises_when_waitlist_click_gives_booked() -> None:
card = make_card(button_text="Join Waitlist")
with pytest.raises(GymError, match="click not confirmed"):
book_class(FakeGym(cards=[card], press_result="Booked"), card)
def test_book_class_propagates_a_site_error() -> None:
with pytest.raises(GymError, match="Network down"):
book_class(FakeGym(cards=[make_card()], press_error="Network down"), make_card())
def test_book_class_raises_for_a_busy_button() -> None:
with pytest.raises(UnknownButtonError):
book_class(FakeGym(), make_card(button_text="Booking..."))
def test_book_next_tuesday_books_the_first_tuesday_class() -> None:
cards = [
make_card("Today (Wed, Aug 13)", card_id="1"),
make_card("Tue, Aug 19", card_id="2"),
make_card("Tue, Aug 26", card_id="3"),
]
gym = FakeGym(cards=cards)
result = book_next_tuesday(gym)
assert result is not None
assert result.card.card_id == "2"
assert gym.pressed == ["2"]
def test_book_next_tuesday_logs_when_there_is_no_class(caplog: pytest.LogCaptureFixture) -> None:
gym = FakeGym(cards=[make_card("Wed, Aug 13")])
with caplog.at_level(logging.INFO):
assert book_next_tuesday(gym) is None
assert "No Tue class at 18:00 found" in caplog.text
assert gym.pressed == []
+36 -2
View File
@@ -6,6 +6,7 @@ from pathlib import Path
import pytest import pytest
from tests.unit.fakes import FakeGym from tests.unit.fakes import FakeGym
from tests.unit.test_models import make_card
from snack_and_lift import constants from snack_and_lift import constants
from snack_and_lift.cli import main, parse_args from snack_and_lift.cli import main, parse_args
@@ -61,8 +62,41 @@ def test_main_saves_both_screenshots_on_success(tmp_path: Path) -> None:
assert code == 0 assert code == 0
assert gym.logins == [(DUMMY_EMAIL, DUMMY_PASSWORD)] assert gym.logins == [(DUMMY_EMAIL, DUMMY_PASSWORD)]
assert gym.screenshots == [tmp_path / "step-1.png", tmp_path / "step-2.png"] assert gym.screenshots == [
assert gym.calls == ["open_site", "save_screenshot", "log_in", "save_screenshot"] tmp_path / "step-1.png",
tmp_path / "step-2.png",
tmp_path / "step-3.png",
]
assert [path for _, path in gym.text_screenshots] == [tmp_path / "step-4.png"]
assert gym.calls == [
"open_site",
"save_screenshot",
"log_in",
"save_screenshot",
"list_class_cards",
"save_screenshot",
"save_text_screenshot",
]
@pytest.mark.usefixtures("credentials_env")
def test_main_books_the_class_and_draws_the_console_text(tmp_path: Path) -> None:
gym = FakeGym(cards=[make_card("Tue, Aug 19")])
assert main(["--screenshot-dir", str(tmp_path)], lambda *_: gym) == 0
assert gym.pressed == ["123"]
text, _ = gym.text_screenshots[0]
assert "✓ Logged in" in text
assert "✓ Successfully booked: Spin Class on Tue, Aug 19" in text
@pytest.mark.usefixtures("credentials_env")
def test_main_returns_1_when_the_booking_fails() -> None:
gym = FakeGym(cards=[make_card("Tue, Aug 19")], press_error="Network down")
assert main(["--headless"], lambda *_: gym) == 1
assert gym.closed is True
@pytest.mark.usefixtures("credentials_env") @pytest.mark.usefixtures("credentials_env")
+30
View File
@@ -0,0 +1,30 @@
"""@file test_console_recorder.py
@brief The recorder keeps the formatted lines of the log.
"""
import logging
from snack_and_lift.infrastructure.console_recorder import ConsoleRecorder
def test_recorder_keeps_formatted_messages() -> None:
logger = logging.getLogger("test_console_recorder")
logger.setLevel(logging.INFO)
recorder = ConsoleRecorder()
logger.addHandler(recorder)
try:
logger.info("✓ Logged %s", "in")
logger.warning("careful")
finally:
logger.removeHandler(recorder)
assert recorder.lines == ["✓ Logged in", "careful"]
assert recorder.text == "✓ Logged in\ncareful"
def test_recorder_lines_are_a_copy() -> None:
recorder = ConsoleRecorder()
recorder.lines.append("x")
assert recorder.lines == []
+57
View File
@@ -0,0 +1,57 @@
"""@file test_models.py
@brief Button states and the values derived from a class card.
"""
import dataclasses
import pytest
from snack_and_lift.domain.errors import PageFormatError
from snack_and_lift.domain.models import ButtonState, ClassCard
def make_card(
day_label: str = "Tomorrow (Thu, Aug 14)",
time_text: str = "Time: 6:00 PM",
button_text: str = "Book Class",
card_id: str = "123",
name: str = "Spin Class",
) -> ClassCard:
return ClassCard(card_id, name, day_label, time_text, button_text)
@pytest.mark.parametrize(
("text", "state"),
[
("Book Class", ButtonState.BOOK),
("Join Waitlist", ButtonState.JOIN_WAITLIST),
("Booked", ButtonState.BOOKED),
("Waitlisted", ButtonState.WAITLISTED),
("Booking...", ButtonState.BUSY),
("Joining...", ButtonState.BUSY),
("Completed", ButtonState.COMPLETED),
(" Booked ", ButtonState.BOOKED),
("", ButtonState.UNKNOWN),
("Cancel", ButtonState.UNKNOWN),
],
)
def test_button_state_is_read_from_text(text: str, state: ButtonState) -> None:
assert ButtonState.from_text(text) is state
def test_card_derives_state_weekday_and_date() -> None:
card = make_card()
assert card.state is ButtonState.BOOK
assert card.weekday == "Thu"
assert card.date_text == "Thu, Aug 14"
def test_card_with_bad_day_label_raises() -> None:
with pytest.raises(PageFormatError):
_ = make_card(day_label="Someday").weekday
def test_card_is_immutable() -> None:
with pytest.raises(dataclasses.FrozenInstanceError):
make_card().name = "Other" # type: ignore[misc]
+138
View File
@@ -0,0 +1,138 @@
"""@file test_rules.py
@brief Day and time parsing, finding the target class, the decision and the messages.
"""
import pytest
from tests.unit.test_models import make_card
from snack_and_lift.domain.errors import PageFormatError, UnknownButtonError
from snack_and_lift.domain.models import Outcome
from snack_and_lift.domain.rules import (
Action,
decide_outcome_action,
find_next_class,
format_outcome,
is_target_time,
parse_day_label,
parse_time,
)
@pytest.mark.parametrize(
("label", "expected"),
[
("Today (Tue, Aug 12)", ("Tue", "Tue, Aug 12")),
("Tomorrow (Wed, Aug 13)", ("Wed", "Wed, Aug 13")),
("Thu, Aug 14", ("Thu", "Thu, Aug 14")),
(" Fri, Sep 5 ", ("Fri", "Fri, Sep 5")),
],
)
def test_day_label_is_split(label: str, expected: tuple[str, str]) -> None:
assert parse_day_label(label) == expected
@pytest.mark.parametrize("label", ["", "Tuesday", "Today Tue, Aug 12", "Today (Tue, Aug 12", "x"])
def test_day_label_junk_raises(label: str) -> None:
with pytest.raises(PageFormatError):
parse_day_label(label)
@pytest.mark.parametrize(
("text", "expected"),
[
("6:00 PM", (18, 0)),
("Time: 6:00 PM", (18, 0)),
("12:00 AM", (0, 0)),
("12:30 AM", (0, 30)),
("12:00 PM", (12, 0)),
("11:59 PM", (23, 59)),
("1:05 am", (1, 5)),
],
)
def test_time_is_read(text: str, expected: tuple[int, int]) -> None:
assert parse_time(text) == expected
@pytest.mark.parametrize("text", ["", "noon", "6 PM", "0:00 AM", "13:00 PM", "6:60 PM", "18:00"])
def test_time_junk_raises(text: str) -> None:
with pytest.raises(PageFormatError):
parse_time(text)
def test_target_time_is_six_pm() -> None:
assert is_target_time(make_card(time_text="Time: 6:00 PM"))
assert not is_target_time(make_card(time_text="Time: 6:00 AM"))
assert not is_target_time(make_card(time_text="Time: 6:30 PM"))
def test_next_class_is_found_when_the_schedule_starts_on_a_wednesday() -> None:
cards = [
make_card("Today (Wed, Aug 13)", card_id="1"),
make_card("Thu, Aug 14", card_id="2"),
make_card("Tue, Aug 19", time_text="Time: 7:00 AM", card_id="3"),
make_card("Tue, Aug 19", card_id="4"),
make_card("Tue, Aug 26", card_id="5"),
]
found = find_next_class(cards, "Tue")
assert found is not None
assert found.card_id == "4"
def test_next_class_accepts_a_today_label() -> None:
cards = [make_card("Today (Tue, Aug 12)", card_id="9")]
found = find_next_class(cards, "Tue")
assert found is not None
assert found.card_id == "9"
def test_next_class_skips_completed_classes() -> None:
cards = [
make_card("Today (Tue, Aug 12)", button_text="Completed", card_id="1"),
make_card("Tue, Aug 19", card_id="2"),
]
found = find_next_class(cards, "Tue")
assert found is not None
assert found.card_id == "2"
def test_next_class_is_none_without_a_match() -> None:
assert find_next_class([make_card("Thu, Aug 14")], "Tue") is None
assert find_next_class([], "Tue") is None
@pytest.mark.parametrize(
("button_text", "action"),
[
("Book Class", Action.CLICK_BOOK),
("Join Waitlist", Action.CLICK_WAITLIST),
("Booked", Action.ALREADY_BOOKED),
("Waitlisted", Action.ALREADY_WAITLISTED),
],
)
def test_action_follows_the_button_state(button_text: str, action: Action) -> None:
assert decide_outcome_action(make_card(button_text=button_text)) is action
@pytest.mark.parametrize("button_text", ["Booking...", "Joining...", "Completed", "???"])
def test_action_for_busy_or_unknown_button_raises(button_text: str) -> None:
with pytest.raises(UnknownButtonError):
decide_outcome_action(make_card(button_text=button_text))
@pytest.mark.parametrize(
("outcome", "message"),
[
(Outcome.NEW_BOOKING, "✓ Successfully booked: Spin Class on Tomorrow (Thu, Aug 14)"),
(Outcome.NEW_WAITLIST, "✓ Joined waitlist for: Spin Class on Tomorrow (Thu, Aug 14)"),
(Outcome.ALREADY_BOOKED, "✓ Already booked: Spin Class on Tomorrow (Thu, Aug 14)"),
(Outcome.ALREADY_WAITLISTED, "✓ Already on waitlist: Spin Class on Tomorrow (Thu, Aug 14)"),
],
)
def test_outcome_message(outcome: Outcome, message: str) -> None:
assert format_outcome(outcome, make_card()) == message