"""Hours by hand and the reader's rule (Stream G, G2), on an in-memory SQLite. What it pins: a hand row outranks a collected one, a collected row written after a hand row is stored and not shown, the newest row within a kind wins, an airport with two shop rows reads hours written against either, `guide_for(iata, db)` fills the guide from the table (and makes a bare guide for an airport nobody has written), `hours set --check` writes nothing, a real `hours set` names the account, and `backfill hours_seed` seeds once and never twice. No network, no Postgres; the four tables come from the models. """ import argparse from datetime import UTC, datetime, timedelta import pytest from sqlalchemy import create_engine, func, select from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import StaticPool from app import cli_hours from app.models import Account, Base, Location, Retailer from app.models.hours import AirportHours from app.models.hubs import AirportGuide from app.services import airport_guides from app.services.hours import store TABLES = [Account.__table__, Retailer.__table__, Location.__table__, AirportHours.__table__] T0 = datetime(2026, 9, 11, 20, 0, tzinfo=UTC) @pytest.fixture def factory(): engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) Base.metadata.create_all(engine, tables=TABLES) make = sessionmaker(bind=engine, expire_on_commit=False) with make() as db: avolta = Retailer(slug="avolta", name="Avolta") ari = Retailer(slug="ari", name="ARI") db.add_all([avolta, ari]) db.flush() db.add_all([ Location(retailer_id=avolta.id, code="LHR", iata="LHR", name="London Heathrow", currency="GBP"), Location(retailer_id=ari.id, code="LHR-2", iata="LHR", name="London Heathrow", currency="GBP"), Location(retailer_id=ari.id, code="DUB", iata="DUB", name="Dublin", currency="EUR"), Account(display_name="rian", username="rian"), Account(display_name="Adam", username="adam"), ]) db.commit() return make @pytest.fixture def db(factory): with factory() as session: yield session def ids(db, iata): return store.location_ids_for(db, iata) class TestReader: def test_hand_beats_collected_and_a_later_collect_is_stored_not_shown(self, db): lhr = ids(db, "LHR") store.record(db, location_id=lhr[0], source_kind="collected", text="collected first", observed_at=T0) assert store.current(db, lhr).text == "collected first" store.record(db, location_id=lhr[0], source_kind="hand", text="typed by a person", observed_at=T0 + timedelta(days=1), entered_by_id=1) assert store.current(db, lhr).text == "typed by a person" later = store.record(db, location_id=lhr[0], source_kind="collected", text="collected later", observed_at=T0 + timedelta(days=2)) shown = store.current(db, lhr) assert shown.text == "typed by a person" and shown.source_kind == "hand" # Stored, complete history, nothing deleted. assert db.scalar(select(func.count(AirportHours.id))) == 3 and later.id == 3 def test_the_newest_row_within_a_kind_wins(self, db): dub = ids(db, "DUB") store.record(db, location_id=dub[0], source_kind="collected", text="old", observed_at=T0) store.record(db, location_id=dub[0], source_kind="collected", text="new", observed_at=T0 + timedelta(hours=1)) assert store.current(db, dub).text == "new" assert store.current(db, ids(db, "LHR")) is None def test_an_airport_with_two_shop_rows_reads_hours_written_against_either(self, db): lhr = ids(db, "LHR") assert len(lhr) == 2 store.record(db, location_id=lhr[1], source_kind="collected", text="against the second shop", observed_at=T0) assert store.current_for_airport(db, "lhr").text == "against the second shop" def test_record_refuses_an_unknown_kind_and_an_empty_line(self, db): with pytest.raises(ValueError): store.record(db, location_id=ids(db, "DUB")[0], source_kind="guessed", text="x", observed_at=T0) with pytest.raises(ValueError): store.record(db, location_id=ids(db, "DUB")[0], source_kind="hand", text=" ", observed_at=T0) def test_the_account_behind_a_hand_row(self, db): row = store.record(db, location_id=ids(db, "DUB")[0], source_kind="hand", text="x", observed_at=T0, entered_by_id=2) assert store.entered_by_username(db, row) == "adam" collected = store.record(db, location_id=ids(db, "DUB")[0], source_kind="collected", text="y", observed_at=T0) assert store.entered_by_username(db, collected) is None assert store.account_by_name(db, "Rian").username == "rian" and store.account_by_name(db, "Adam").username == "adam" assert store.account_by_name(db, "nobody") is None and store.account_by_name(db, "") is None class TestGuide: def test_the_written_guide_gains_the_tables_hours(self, db): assert airport_guides.guide_for("LHR").hours is None # the constant carries none store.record(db, location_id=ids(db, "LHR")[0], source_kind="collected", text="World Duty Free, 14 stores", observed_at=T0) guide = airport_guides.guide_for("LHR", db) assert guide.hours == "World Duty Free, 14 stores" and len(guide.terminals) == 4 # The constant itself is untouched. assert airport_guides.GUIDES["LHR"].hours is None and airport_guides.guide_for("LHR").hours is None def test_an_airport_nobody_has_written_gets_a_guide_of_hours_alone(self, db): assert airport_guides.guide_for("DUB", db) is None store.record(db, location_id=ids(db, "DUB")[0], source_kind="hand", text="T1 04:00-22:00", observed_at=T0, entered_by_id=1) guide = airport_guides.guide_for("DUB", db) assert guide.model_dump(exclude={"hours_provenance"}) == AirportGuide(hours="T1 04:00-22:00").model_dump(exclude={"hours_provenance"}) assert guide.hours_provenance.kind == "hand" and guide.hours_provenance.entered_by_username == "rian" assert guide.hours_provenance.source_host is None assert airport_guides.guide_for("QQQ", db) is None and airport_guides.guide_for(None, db) is None class TestHandCommand: def test_check_prints_and_writes_nothing(self, factory, monkeypatch, tmp_path, capsys): monkeypatch.setattr(cli_hours, "SessionLocal", factory) path = tmp_path / "dub.txt" path.write_text("Dublin Airport Duty Free,\n T1 04:00-22:00 and T2 04:00-21:00\n") code = cli_hours.cmd_hours_set(argparse.Namespace(airport="dub", file=str(path), by="rian", source_url=None, check=True)) assert code == 0 assert "[check, nothing written] DUB hand, by rian: Dublin Airport Duty Free, T1 04:00-22:00 and T2 04:00-21:00" in capsys.readouterr().out with factory() as db: assert db.scalar(select(func.count(AirportHours.id))) == 0 def test_a_hand_row_is_dated_and_named_to_the_account(self, factory, monkeypatch, tmp_path, capsys): monkeypatch.setattr(cli_hours, "SessionLocal", factory) path = tmp_path / "dub.txt" path.write_text("T1 04:00-22:00") assert cli_hours.cmd_hours_set(argparse.Namespace(airport="DUB", file=str(path), by="Adam", source_url="https://www.dublinairport.com/x", check=False)) == 0 with factory() as db: row = store.current_for_airport(db, "DUB") assert row.source_kind == "hand" and row.text == "T1 04:00-22:00" and row.entered_by_id == 2 assert row.source_url == "https://www.dublinairport.com/x" and row.detail == {"entered_by": "adam", "file": "dub.txt"} assert row.observed_at.replace(tzinfo=UTC) >= T0 assert "DUB hand row 1, by adam" in capsys.readouterr().out def test_an_unknown_account_or_airport_or_empty_file_writes_nothing(self, factory, monkeypatch, tmp_path, capsys): monkeypatch.setattr(cli_hours, "SessionLocal", factory) path = tmp_path / "x.txt" path.write_text("T1 04:00-22:00") assert cli_hours.cmd_hours_set(argparse.Namespace(airport="DUB", file=str(path), by="nobody", source_url=None, check=False)) == 2 assert cli_hours.cmd_hours_set(argparse.Namespace(airport="QQQ", file=str(path), by="rian", source_url=None, check=False)) == 2 path.write_text(" \n") assert cli_hours.cmd_hours_set(argparse.Namespace(airport="DUB", file=str(path), by="rian", source_url=None, check=False)) == 2 with factory() as db: assert db.scalar(select(func.count(AirportHours.id))) == 0 class TestSeed: def test_the_seed_writes_once_and_never_twice(self, db, monkeypatch): monkeypatch.setattr(airport_guides, "GUIDES", { "LHR": AirportGuide(hours="World Duty Free opens with the first flight"), "DUB": AirportGuide(operator="ARI"), # no hours constant: nothing to seed }) monkeypatch.setattr(cli_hours.settings, "account_owner", "rian") first = cli_hours.backfill_hours_seed(db) assert first.startswith("hours_seed: seeded 1 (LHR); 0 already") and "1 guides carry no hours constant (DUB)" in first row = store.current_for_airport(db, "LHR") assert row.source_kind == "hand" and row.entered_by_id == 1 and row.detail["seed"] == "airport_guides" second = cli_hours.backfill_hours_seed(db) assert second.startswith("hours_seed: seeded 0 (none); 1 already had a hand row") assert db.scalar(select(func.count(AirportHours.id))) == 1 def test_the_seed_today_seeds_nothing_because_no_guide_carries_hours(self, db): assert all(g.hours is None for g in airport_guides.GUIDES.values()) assert cli_hours.backfill_hours_seed(db).startswith("hours_seed: seeded 0 (none)")