049-snack-and-lift/tests/unit/test_resilience.py

"""@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)