"""The review area's reading service: the parts that hold without a database. The queries themselves are exercised against a real copy before deploy; what is pinned here is what silently rots otherwise: the platform names a person reads, and the fact that the route's accepted values and the service's own choices cannot drift apart. """ import re from pathlib import Path from app.services import collector_view as view from app.services.collectors.registry import COLLECTORS ROUTER = (Path(__file__).resolve().parents[1] / "app" / "routers" / "collectors.py").read_text() def _pattern(name: str) -> set[str]: """The alternatives a route parameter accepts, read from its regex.""" match = re.search(rf"{name}.*?pattern=\"\^\(([^)]+)\)\$\"", ROUTER, re.S) assert match, f"no pattern for {name}" return set(match.group(1).split("|")) class TestPlatforms: def test_every_collector_resolves_to_a_named_platform(self): # An unnamed platform reads as "shopify_beauty" on the page: a module name leaking # into something a person is meant to understand. unknown = [slug for slug in COLLECTORS if view.platform_of(slug) in ("unknown", "")] assert not unknown, f"no platform name for {unknown}" def test_the_avolta_airports_share_one_platform(self): # One collector per platform is the design; ten airports reading as ten platforms # would hide that a fix at one is a fix at all. avolta = {view.platform_of(s) for s in COLLECTORS if s.startswith("avolta-")} assert len(avolta) == 1 def test_an_unregistered_slug_is_not_invented(self): assert view.platform_of("no-such-collector") == "unknown" class TestRouteAndServiceAgree: def test_missing_choices(self): assert _pattern("missing") == set(view.MISSING_CHOICES) def test_sorts(self): assert _pattern("sort") == set(view.SORTS) def test_groups(self): assert _pattern("group") == set(view.GROUPS) def test_differs(self): assert _pattern("differs") == set(view.DIFFERS_CHOICES) class TestGrouping: def test_an_unknown_group_returns_nothing_rather_than_guessing(self): assert view.grouped(None, group="price") == {"group": "price", "rows": []} class TestTheStoryTellsTheTruthAboutARefusal: """Dubai's edge answers 403 to our declared identity, so no run exists after 22 Aug and the story read "nothing attempted for N days" while verification check 396 (11 Sep) said BLOCKED. The newest BLOCKED check wins when no run has started since (Stream L, LT8).""" def test_refused_since_wins_over_nothing_attempted(self): from datetime import UTC, datetime, timedelta from types import SimpleNamespace from app.services.collector_view import _story old = datetime.now(UTC) - timedelta(days=24) runs = [SimpleNamespace(status="ok", started_at=old)] assert _story(runs)["story"].startswith("nothing attempted for") blocked = old + timedelta(days=20) told = _story(runs, refused_since=blocked) assert told["story"] == f"refused since {blocked.date().isoformat()}" assert told["refused_since"] is not None # A run since the check speaks for itself. later = [SimpleNamespace(status="ok", started_at=blocked + timedelta(days=1))] assert _story(later, refused_since=blocked)["story"] != f"refused since {blocked.date().isoformat()}" assert _story([], refused_since=blocked)["story"] == f"refused since {blocked.date().isoformat()}" # --------------------------------------------------------------------------- the live read (Stream AW4.3) # # SQLite, no network, a fake /proc and cgroup on a temporary directory, a fake clock. What it # cost: the live tab drew the 18 Sep orphans as collections in progress; a stalled run had no # word; progress and an ETA needed a person with `psql`; the page could show 100 while a run was # still walking. from datetime import UTC, datetime, timedelta import pytest from sqlalchemy import create_engine, event, select from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import StaticPool from app.models import Account, AuditLog, Base, CollectionRun, Listing, ProductVariant, Retailer, Shop, Source from app.models.hours import AirportHours from app.models.quality import VerificationCheck from app.models.schemas import LiveOut from app.services import ingest, procinfo from app.services.collectors.fetch import FetchError, SourceBlocked from tests.test_collector_control import RATES, SLUG, TABLES, FakeClock, FakeCollector, _raw SLUGS = sorted(COLLECTORS) LIVE_TABLES = [*TABLES, AuditLog.__table__, AirportHours.__table__, VerificationCheck.__table__] @pytest.fixture def db(): engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) Base.metadata.create_all(engine, tables=LIVE_TABLES) ingest._table_cache.clear() factory = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False, future=True) with factory() as session: session.add(Account(id=1, username="rian", display_name="rian")) session.commit() yield session @pytest.fixture def registered(monkeypatch): """The fake collector in the registry for the read, so `live()` lists its source.""" monkeypatch.setitem(COLLECTORS, SLUG, FakeCollector([])) @pytest.fixture def roots(tmp_path): """A cgroup and a /proc the way the container writes them: 1.6 GiB used of 4 GiB, pid 77 a collector of the first registered source, pid 1 present so an absent pid reads as gone.""" cgroup = tmp_path / "cgroup" cgroup.mkdir() (cgroup / "memory.current").write_text(str(1_717_986_918)) (cgroup / "memory.max").write_text(str(4 * 1024 ** 3)) proc = tmp_path / "proc" (proc / "1").mkdir(parents=True) (proc / "77").mkdir() (proc / "77" / "cmdline").write_bytes(f"python\0-m\0app.cli\0collect\0--source\0{SLUGS[5]}\0".encode()) (proc / "77" / "status").write_text("Name:\tpython\nState:\tS (sleeping)\nVmRSS:\t 84000 kB\n") return cgroup, proc class TestProgressArithmetic: def test_no_percent_before_three_units_and_never_100_while_running(self): assert view.percent_done(2, 100, finished=False) is None assert view.percent_done(3, 100, finished=False) == 3 assert view.percent_done(100, 100, finished=False) == 99, "100 is a claim only a finished run may make" assert view.percent_done(150, 100, finished=False) == 99 assert view.percent_done(100, 100, finished=True) == 100 assert view.percent_done(150, 100, finished=True) == 100 assert view.percent_done(50, None, finished=False) is None assert view.percent_done(50, 0, finished=True) is None def test_the_eta_is_a_duration_from_the_rate_so_far(self): assert view.eta_seconds(50, 100, 600.0, finished=False) == 600.0 assert view.eta_seconds(2, 100, 600.0, finished=False) is None assert view.eta_seconds(50, None, 600.0, finished=False) is None assert view.eta_seconds(50, 100, 600.0, finished=True) is None assert view.eta_seconds(120, 100, 600.0, finished=False) == 0.0, "past the estimate there is nothing left to guess" def _source(db, n: int, **kw) -> Source: source = Source(id=n, slug=SLUGS[n], name=f"Source {n}", delay_seconds=kw.pop("delay_seconds", 1.0), **kw) db.add(source) return source def _run(db, n: int, **kw) -> CollectionRun: run = CollectionRun(source_id=n, **kw) db.add(run) return run class TestTheNineStateWords: """One seeded row per word, read back through `live()` with the fake /proc, so the words the page shows are the words the rows imply and nothing is marked on the way.""" def test_each_word_from_its_rows(self, db, roots, monkeypatch): cgroup, proc = roots now = datetime.now(UTC) monkeypatch.setattr(ingest, "process_boot", lambda *a, **k: now - timedelta(hours=2)) # 0 dead: began before this container (three hours ago, the container two hours old) _source(db, 0) _run(db, 0, started_at=now - timedelta(hours=3), status="running", heartbeat_at=now - timedelta(seconds=5), pid=77) # 1 stalled: the beat is twenty minutes old at a pace of 1 s (the ten-minute floor); no pid _source(db, 1) _run(db, 1, started_at=now - timedelta(minutes=40), status="running", heartbeat_at=now - timedelta(minutes=20)) # 2 stopping: a stop set after the run began _source(db, 2, control="stop", control_set_by="rian", control_set_at=now - timedelta(seconds=10)) _run(db, 2, started_at=now - timedelta(minutes=5), status="running", heartbeat_at=now - timedelta(seconds=3)) # 3 paused: the beat is newer than the pause _source(db, 3, control="pause", control_set_by="rian", control_set_at=now - timedelta(minutes=2)) _run(db, 3, started_at=now - timedelta(minutes=5), status="running", heartbeat_at=now - timedelta(seconds=2)) # 4 pausing: the pause is newer than the beat _source(db, 4, control="pause", control_set_by="rian", control_set_at=now - timedelta(seconds=1), delay_seconds=60) _run(db, 4, started_at=now - timedelta(minutes=5), status="running", heartbeat_at=now - timedelta(seconds=20)) # 5 running: pid 77 is its process in the fake /proc, with an estimate from the row _source(db, 5) _run(db, 5, started_at=now - timedelta(minutes=10), status="running", heartbeat_at=now - timedelta(seconds=3), pid=77, products_seen=300, prices_written=280, requests_made=25, expected_total=1000, mode="discover") # 6 off: the kill switch, no running row _source(db, 6, enabled=False) # 7 refused: the last run was a refusal _source(db, 7) _run(db, 7, started_at=now - timedelta(days=8), finished_at=now - timedelta(days=8), status="blocked", error="HTTP 403") # 8 idle: the last run ended ok _source(db, 8) _run(db, 8, started_at=now - timedelta(hours=5), finished_at=now - timedelta(hours=3), status="ok", prices_written=412, products_seen=420) db.commit() out = view.live(db, now=now, cgroup=cgroup, proc=proc) assert isinstance(out, LiveOut) by_slug = {c.slug: c for c in out.collectors} words = [by_slug[SLUGS[n]].state for n in range(9)] assert words == ["dead", "stalled", "stopping", "paused", "pausing", "running", "off", "refused", "idle"] assert by_slug[SLUGS[0]].state_detail == "dead: began before this container" assert by_slug[SLUGS[1]].state_detail == "stalled: no heartbeat for 20 min (pace 1 s, 10 min expected)" assert by_slug[SLUGS[1]].run.heartbeat_age_seconds == pytest.approx(20 * 60) assert by_slug[SLUGS[2]].state_detail.startswith("stopping by rian 10 s ago") assert by_slug[SLUGS[2]].run.stop_requested_at is not None assert by_slug[SLUGS[3]].state_detail == "paused 2 min by rian ยท holds approvals while paused" assert by_slug[SLUGS[4]].state_detail == "pausing: next request boundary in about 40 s" assert by_slug[SLUGS[6]].state_detail == "switched off (the kill switch)" assert by_slug[SLUGS[7]].refused is not None and "HTTP 403" in by_slug[SLUGS[7]].refused.why assert by_slug[SLUGS[7]].start_blocked.code == "SOURCE_REFUSED", "the same function the POST answers with" assert by_slug[SLUGS[8]].state_detail.startswith("last: ok ") and by_slug[SLUGS[8]].state_detail.endswith("412 prices in 2 h 00") assert by_slug[SLUGS[8]].start_blocked is None and by_slug[SLUGS[8]].last.prices == 412 assert by_slug[SLUGS[6]].start_blocked.code == "SOURCE_DISABLED" running = by_slug[SLUGS[5]] assert running.run.alive is True and running.run.rss_bytes == 84000 * 1024 assert running.run.percent == 30 and running.run.expected_kind == "estimate" and running.run.done == 300 assert running.run.eta_seconds == pytest.approx(600 * 700 / 300) assert running.run.requests_per_minute == 2.5 and running.holds_approvals is True assert by_slug[SLUGS[0]].holds_approvals is False, "a dead row does not hold the gate" assert by_slug[SLUGS[1]].run.alive is None, "a row with no pid is unknown to /proc, never gone" assert [c.slug for c in out.collectors[:6]] == [SLUGS[n] for n in range(6)], "live states first, then by slug" assert (out.totals.dead, out.totals.running, out.totals.idle) == (1, 1, len(SLUGS) - 8) assert out.memory.used_bytes == 1_717_986_918 and out.memory.limit_bytes == 4 * 1024 ** 3 and out.memory.readable assert out.memory.shed_at_bytes == procinfo.SHED_AT_BYTES and not out.memory.frozen assert out.totals.prices_24h == 280 + 412, "the running run and the run that ended three hours ago both count" def test_a_dead_row_says_why_and_a_stall_beats_every_control_word(self): from types import SimpleNamespace now = datetime.now(UTC) source = SimpleNamespace(enabled=True, delay_seconds=60, control="stop", control_set_by="rian", control_set_at=now - timedelta(seconds=5)) silent = SimpleNamespace(status="running", started_at=now - timedelta(hours=2), heartbeat_at=now - timedelta(minutes=72), prices_written=0, requests_made=0) assert view.state_of(source, silent, None, None, now, True, boot=None) == ("dead", "dead: no heartbeat for 1 h 12") old = SimpleNamespace(status="running", started_at=now - ingest.STUCK_AFTER - timedelta(hours=1), heartbeat_at=None, prices_written=0, requests_made=0) assert view.state_of(source, old, None, None, now, None, boot=None)[0] == "dead" gone = SimpleNamespace(status="running", started_at=now - timedelta(minutes=5), heartbeat_at=now - timedelta(seconds=3), prices_written=0, requests_made=0) assert view.state_of(source, gone, None, None, now, False, boot=None) == ("stalled", "stalled: process gone") # At a pace of 60 s the stall line is five paces, not the ten-minute floor. late = SimpleNamespace(status="running", started_at=now - timedelta(minutes=9), heartbeat_at=now - timedelta(minutes=6), prices_written=0, requests_made=0) assert view.state_of(source, late, None, None, now, True, boot=None)[0] == "stopping" source.delay_seconds = 1 assert view.state_of(source, late, None, None, now, True, boot=None)[0] == "stopping", "six minutes is under the floor" late.heartbeat_at = now - timedelta(minutes=11) assert view.state_of(source, late, None, None, now, True, boot=None)[0] == "stalled" # Off with a running row is stopping (the loop ends at its next poll); off without one is off. source.control, source.enabled = "run", False assert view.state_of(source, gone, None, None, now, True, boot=None)[0] == "stopping" assert view.state_of(source, None, None, None, now, None, boot=None)[0] == "off" # A verification refusal newer than the last run is a refusal; an older one is not. source.enabled = True last = SimpleNamespace(status="ok", started_at=now - timedelta(days=2), finished_at=now - timedelta(days=2), prices_written=5) assert view.state_of(source, None, last, now - timedelta(days=1), now, None, boot=None)[0] == "refused" assert view.state_of(source, None, last, now - timedelta(days=3), now, None, boot=None)[0] == "idle" assert view.state_of(source, None, None, None, now, None, boot=None) == ("idle", "never run") class TestTheCountersAfterResolution: """Two listings: one new variant, one price move; a listing the rules reject counts nothing. Cost: without these the page could say only "prices"; what a run found that was new, what it checked and what moved were a person's `psql` on the observation table. """ def test_new_then_checked_changed_and_missing(self, db, registered): clock = FakeClock() first = ingest.run_collector(db, FakeCollector([_raw(0, price=10.0), _raw(1, price=20.0)]), rates=RATES, sleeper=clock.sleep, clock=clock.clock) assert first.status == "ok" assert (first.new_found, first.new_variants, first.new_brands) == (2, 2, 1) assert first.new_lines >= 1 assert (first.existing_checked, first.existing_changed, first.existing_missing) == (0, 0, 0) moved = _raw(0, price=12.0) second = ingest.run_collector(db, FakeCollector([moved, _raw(2, price=30.0)]), rates=RATES, sleeper=clock.sleep, clock=clock.clock) assert second.status == "ok" assert (second.existing_checked, second.existing_changed) == (1, 1) assert (second.new_found, second.new_variants, second.new_brands) == (1, 1, 0) assert second.existing_missing == 1, "sku1 was held and never met by a full walk" limited = ingest.run_collector(db, FakeCollector([_raw(0, price=12.0)]), rates=RATES, limit=1, sleeper=clock.sleep, clock=clock.clock) assert (limited.existing_checked, limited.existing_changed, limited.existing_missing) == (1, 0, 0), \ "a limited walk measures no missing; the price did not move since the second run" out = view.live(db, now=datetime.now(UTC)) row = next(c for c in out.collectors if c.slug == SLUG) assert row.last is not None and row.last.existing_missing_measured is False assert row.held_listings == 3 and row.recheck_supported is True def test_a_listing_the_rules_reject_counts_nothing(self, db): clock = FakeClock() first = _raw(0, price=10.0) first.gtin = "5000000000001" ingest.run_collector(db, FakeCollector([first]), rates=RATES, sleeper=clock.sleep, clock=clock.clock) clash = _raw(9, price=10.0) clash.gtin, clash.quantity_ml, clash.name = "5000000000001", 50, "Fake Whisky 9 5cl" run = ingest.run_collector(db, FakeCollector([clash]), rates=RATES, sleeper=clock.sleep, clock=clock.clock) assert run.status == "ok" and run.skip_counts.get("gtin_size_veto") == 1 assert (run.new_found, run.new_variants, run.existing_checked, run.existing_changed) == (0, 0, 0, 0) def test_a_second_sighting_in_one_run_counts_once(self, db): clock = FakeClock() run = ingest.run_collector(db, FakeCollector([_raw(0), _raw(0)]), rates=RATES, sleeper=clock.sleep, clock=clock.clock) assert run.prices_written == 2 and run.new_found == 1 and run.new_variants == 1 class TestRecheckOutcomes: def test_missing_fetch_error_and_a_block(self, db, registered): clock = FakeClock() ingest.run_collector(db, FakeCollector([_raw(n) for n in range(3)]), rates=RATES, sleeper=clock.sleep, clock=clock.clock) results = {"sku0": _raw(0, price=11.0), "sku1": "missing", "sku2": FetchError("https://x failed: timeout")} run = ingest.run_collector(db, FakeCollector([], read_results=results), rates=RATES, mode="recheck", sleeper=clock.sleep, clock=clock.clock) assert (run.status, run.existing_checked, run.existing_changed, run.existing_missing) == ("ok", 1, 1, 1) assert run.skip_counts == {"recheck_fetch_error": 1} and run.expected_total == 3 out = view.live(db, now=datetime.now(UTC)) last = next(c for c in out.collectors if c.slug == SLUG).last assert last.mode == "recheck" and last.existing_missing_measured is True blocked = ingest.run_collector(db, FakeCollector([], read_results={"sku0": SourceBlocked("HTTP 403")}), rates=RATES, mode="recheck", sleeper=clock.sleep, clock=clock.clock) assert blocked.status == "blocked" and "403" in blocked.error row = next(c for c in view.live(db, now=datetime.now(UTC)).collectors if c.slug == SLUG) assert row.state == "refused" and row.start_blocked.code == "SOURCE_REFUSED" class TestTheLiveReadIsBounded: def test_four_hundred_runs_within_the_statement_budget(self, db, roots): cgroup, proc = roots now = datetime.now(UTC) retailer = Retailer(id=1, slug="r", name="R") db.add(retailer) db.flush() for n in range(9): _source(db, n) db.add(Shop(id=n + 1, retailer_id=1, code=f"S{n}", iata=f"A{n:02d}"[:3], name=f"Shop {n}", currency="EUR", visible=True)) db.flush() for i in range(400): n = i % 9 status = "running" if i >= 396 else ("ok", "error", "blocked", "stopped", "skipped")[i % 5] db.add(CollectionRun(source_id=n, started_at=now - timedelta(minutes=5 * (400 - i)), status=status, finished_at=None if status == "running" else now - timedelta(minutes=5 * (400 - i) - 4), heartbeat_at=now - timedelta(seconds=3) if status == "running" else None, prices_written=i)) for i in range(54): # six per collector, so the cap of five shows db.add(AuditLog(account_id=1, action="source.pause", entity_type="source", entity_key=SLUGS[i % 9], detail={"n": i})) db.add(AirportHours(shop_id=1, source_kind="collected", text="05:00-22:00", observed_at=now - timedelta(days=1), source_url="https://operator.example/hours")) db.add(AirportHours(shop_id=2, source_kind="hand", text="06:00-21:00", observed_at=now - timedelta(days=2), entered_by_id=1)) db.add(ProductVariant(id=1, name="Fake 1L", match_key="k1", vertical="liquor", attributes={})) db.flush() db.add(Listing(id=1, variant_id=1, shop_id=1, source_sku="x", url="https://x/1")) db.flush() db.add(VerificationCheck(run_id=1, listing_id=1, source_id=3, verdict="BLOCKED", checked_at=now - timedelta(days=1), detail="403", url="https://x")) db.commit() counted: list[str] = [] event.listen(db.get_bind(), "before_cursor_execute", lambda conn, cur, stmt, *a: counted.append(stmt)) view.live(db, now=now, cgroup=cgroup, proc=proc) # the first read learns which tables exist cold = len(counted) counted.clear() out = view.live(db, now=now, cgroup=cgroup, proc=proc) assert len(counted) <= view.LIVE_MAX_STATEMENTS, f"{len(counted)} statements warm ({cold} cold):\n" + "\n".join(counted) assert out.totals.running == 4 and len(out.recent) == 400 - 4 - sum(1 for i in range(396) if 5 * (400 - i) > 24 * 60) by_slug = {c.slug: c for c in out.collectors} assert len(by_slug[SLUGS[0]].events) == view.EVENTS_PER_COLLECTOR and by_slug[SLUGS[0]].events[0].by == "rian" assert by_slug[SLUGS[3]].state == "refused" or by_slug[SLUGS[3]].run is not None assert out.hours_by_hand and out.hours_by_hand[0].entered_by == "rian" assert [a.iata for a in out.hours_by_hand] == ["A01"] assert "A02" in out.hours_missing and "A00" not in out.hours_missing and "A01" not in out.hours_missing assert all(h.slug and h.operator and h.airports for h in out.hours) class TestTheStoryHasANeutralKind: def test_a_stop_and_a_skip_are_neither_ok_nor_failed(self): from types import SimpleNamespace now = datetime.now(UTC) runs = [SimpleNamespace(status="ok", started_at=now - timedelta(days=2)), SimpleNamespace(status="stopped", started_at=now - timedelta(days=1)), SimpleNamespace(status="skipped", started_at=now)] told = view._story(runs) assert told["recent"] == ["ok", "neutral", "neutral"] and told["neutral"] == 2 assert told["story"] == "clean every run", "a stop from the page is not trouble" assert view._story([SimpleNamespace(status="stopped", started_at=now)])["story"] == "nothing finished yet" assert view._story([SimpleNamespace(status="error", started_at=now - timedelta(days=1)), SimpleNamespace(status="stopped", started_at=now)])["story"] == "never once succeeded" def test_a_run_still_walking_is_not_a_refusal(self): """The 19 Sep 03:15 dump: avolta-ath, -bcn, -hkg, -mad and extime-paris were mid-sweep and each read "worked before, refused now" because a `running` row counted as failed. Cost: the page told a person a host closed its door while the collector was happily walking it.""" from types import SimpleNamespace now = datetime.now(UTC) runs = [SimpleNamespace(status="ok", started_at=now - timedelta(days=1)), SimpleNamespace(status="running", started_at=now)] told = view._story(runs) assert told["recent"] == ["ok", "neutral"] and told["failed"] == 0 assert told["story"] == "clean every run" assert view._story([SimpleNamespace(status="running", started_at=now)])["story"] == "nothing finished yet"