"""The collector control plane (Stream AW4.2): the hook a running collector reads between requests, the cooperative end, the advisory lock, what a Start must satisfy, the pace floor, and the six routes. SQLite, no network, a fake sleeper and clock: seconds of pace pass in milliseconds. What it cost: on 19 Sep the guard shed four collectors with SIGTERM and the four `running` rows they left behind held the review's approval gate for the 24 hours the timer had left; a pace change or a stop needed `docker exec`; a double Start would have run one source twice. """ from __future__ import annotations import os import signal from datetime import UTC, datetime, timedelta from types import SimpleNamespace import pytest from sqlalchemy import create_engine, func, select, update from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import StaticPool from app.models import ( Account, AttributeAlias, Base, Brand, CollectionRun, Listing, Merge, PriceObservation, ProductLine, ProductVariant, RawRecord, Retailer, Shop, Source, Suggestion, ) from app.models.decisions import LEDGER_TABLES from app.models.places import Place, ShopPlace from app.services import ingest, procinfo from app.services.collectors import control from app.services.collectors.base import ListingRef, RawListing, ShopSpec from app.services.collectors.fetch import FetchError from app.services.fx import FxRates TABLES = [ Account.__table__, Source.__table__, CollectionRun.__table__, Brand.__table__, ProductLine.__table__, ProductVariant.__table__, Retailer.__table__, Shop.__table__, RawRecord.__table__, Listing.__table__, PriceObservation.__table__, Merge.__table__, Suggestion.__table__, AttributeAlias.__table__, *LEDGER_TABLES, Place.__table__, ShopPlace.__table__, ] RATES = FxRates({"USD": 1.0, "EUR": 0.9}, datetime.now(UTC), False) SLUG = "fake-src" def _raw(n: int, price: float = 10.0) -> RawListing: return RawListing(source_sku=f"sku{n}", name=f"Fake Whisky {n} 1L", price=price, currency="USD", shop_code="FAKE1", brand="Fake", quantity_ml=1000, url=f"https://shop.example.com/p/{n}", vertical="liquor") class FakeCollector: """A collector that calls the hook where `fetch()` would, once per listing.""" slug = SLUG retailer_name = "Fake Retailer" retailer_slug = "fake" operator = "Fake" homepage = "https://shop.example.com/" parser_version = "fake/1" def __init__(self, listings, on_each=None, read_results=None): self.listings = listings self.on_each = on_each self.read_results = read_results or {} self.reads: list[str] = [] def shops(self): return [ShopSpec(code="FAKE1", name="Fake shop", currency="USD", iata="FK1")] def collect(self, *, limit=None, delay=1.0): for i, raw in enumerate(self.listings[:limit] if limit else self.listings): control.wait(delay) if self.on_each: self.on_each(i) yield raw def read_one(self, ref: ListingRef): control.wait(1.0) self.reads.append(ref.source_sku) result = self.read_results.get(ref.source_sku, "missing") if isinstance(result, Exception): raise result return None if result == "missing" else result class FakeClock: """A clock the sleeper advances, so a 60 s pace costs nothing.""" def __init__(self): self.now = 1000.0 self.slept: list[float] = [] self.on_sleep = None def clock(self): return self.now def sleep(self, seconds): self.slept.append(seconds) self.now += seconds if self.on_sleep: self.on_sleep(len(self.slept)) @pytest.fixture def db(): engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) Base.metadata.create_all(engine, tables=TABLES) 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.add(Source(id=1, slug=SLUG, name="Fake Retailer", delay_seconds=1.0)) session.commit() yield session def _running(db, **kw) -> CollectionRun: run = CollectionRun(source_id=1, started_at=kw.pop("started_at", datetime.now(UTC)), status="running", **kw) db.add(run) db.commit() return run def _hook(db, run, clock: FakeClock) -> control.RunControl: return control.RunControl(run_id=run.id, source_id=1, slug=SLUG, db=db, run_started_at=run.started_at, sleeper=clock.sleep, clock=clock.clock) # --------------------------------------------------------------------------- the hook class TestTheHookUnbound: def test_it_sleeps_the_requested_delay_as_the_fetch_port_always_did(self, monkeypatch): slept = [] monkeypatch.setattr(control.time, "sleep", slept.append) control.wait(2.5) control.wait(0) assert slept == [2.5] assert control.current() is None class TestTheHookBound: def test_a_boundary_commits_the_batch_heartbeats_and_counts_the_request(self, db): run = _running(db) run.products_seen = 7 # a pending ORM change: the batch the boundary commits hook = _hook(db, run, FakeClock()) hook.wait(0) fresh = db.execute(select(CollectionRun.heartbeat_at, CollectionRun.requests_made, CollectionRun.products_seen) .where(CollectionRun.id == run.id)).one() assert fresh.heartbeat_at is not None and fresh.requests_made == 1 and fresh.products_seen == 7 assert (hook.polls, hook.boundaries) == (1, 1) def test_zero_reads_the_control_once_and_never_sleeps(self, db): run = _running(db) clock = FakeClock() _hook(db, run, clock).wait(0) assert clock.slept == [] def test_the_wait_is_the_slowest_of_requested_robots_and_the_live_pace_in_two_second_slices(self, db): run = _running(db) db.execute(update(Source).where(Source.id == 1).values(delay_seconds=3.0)) db.commit() clock = FakeClock() hook = _hook(db, run, clock) hook.note_robots(5.0) hook.wait(2.0) assert sum(clock.slept) == 5.0 and max(clock.slept) <= control.CONTROL_POLL assert hook.boundaries == 1 and hook.polls == 1 + len(clock.slept) # A poll of the wait heartbeats without counting a request. assert db.scalar(select(CollectionRun.requests_made).where(CollectionRun.id == run.id)) == 1 def test_the_hosts_crawl_delay_is_published_on_the_source_at_the_next_poll(self, db): run = _running(db) hook = _hook(db, run, FakeClock()) hook.note_robots(60.0) hook.wait(0) row = db.execute(select(Source.robots_crawl_delay, Source.robots_read_at).where(Source.id == 1)).one() assert float(row.robots_crawl_delay) == 60.0 and row.robots_read_at is not None assert hook.robots_pending is None def test_a_pace_change_mid_wait_lengthens_the_wait_and_never_shortens_it(self, db): run = _running(db) db.execute(update(Source).where(Source.id == 1).values(delay_seconds=4.0)) db.commit() clock = FakeClock() def slower_after_first_slice(n): if n == 1: db.execute(update(Source).where(Source.id == 1).values(delay_seconds=8.0)) db.commit() clock.on_sleep = slower_after_first_slice _hook(db, run, clock).wait(1.0) assert sum(clock.slept) == 8.0 def test_the_kill_switch_ends_the_run_at_the_next_poll(self, db): run = _running(db) db.execute(update(Source).where(Source.id == 1).values(enabled=False)) db.commit() with pytest.raises(control.RunStopped) as stopped: _hook(db, run, FakeClock()).wait(1.0) assert (stopped.value.reason, stopped.value.by) == ("kill switch", "kill-switch") def test_a_stop_from_the_page_ends_it_with_who_pressed_it(self, db): run = _running(db) db.execute(update(Source).where(Source.id == 1).values(control="stop", control_set_by="rian", control_set_at=datetime.now(UTC) + timedelta(seconds=1))) db.commit() with pytest.raises(control.RunStopped) as stopped: _hook(db, run, FakeClock()).wait(1.0) assert (stopped.value.reason, stopped.value.by) == ("stopped from the page", "rian") def test_a_stop_pressed_before_the_run_began_never_applies(self, db): """`effective_control`: a stop pressed yesterday must not kill tonight's shell run, and nothing resets the field.""" run = _running(db) db.execute(update(Source).where(Source.id == 1).values(control="stop", control_set_by="rian", control_set_at=run.started_at - timedelta(days=1))) db.commit() clock = FakeClock() _hook(db, run, clock).wait(1.0) assert sum(clock.slept) == 1.0 assert control.effective_control("stop", run.started_at - timedelta(days=1), run.started_at) == "run" assert control.effective_control("pause", run.started_at + timedelta(seconds=1), run.started_at) == "pause" assert control.effective_control("run", None, run.started_at) == "run" def test_a_pause_mid_wait_heartbeats_every_poll_and_holds_the_deadline(self, db): run = _running(db) db.execute(update(Source).where(Source.id == 1).values(delay_seconds=4.0)) db.commit() clock = FakeClock() beats = [] def pause_then_resume(n): beats.append(db.scalar(select(CollectionRun.heartbeat_at).where(CollectionRun.id == run.id))) if n == 1: db.execute(update(Source).where(Source.id == 1).values(control="pause", control_set_by="rian", control_set_at=datetime.now(UTC) + timedelta(seconds=1))) if n == 3: db.execute(update(Source).where(Source.id == 1).values(control="run", control_set_at=datetime.now(UTC) + timedelta(seconds=2))) db.commit() clock.on_sleep = pause_then_resume _hook(db, run, clock).wait(1.0) # One slice of the 4 s wait, two slices paused; the deadline was held, so on resume it has # already passed and no further slice is slept: two requests never closer than the pace. assert clock.slept == [control.CONTROL_POLL] * 3 assert all(beats) and beats[2] > beats[0], "a paused run is never silent" def test_a_pause_at_the_boundary_is_followed_by_the_full_pace(self, db): run = _running(db) db.execute(update(Source).where(Source.id == 1).values(delay_seconds=4.0, control="pause", control_set_by="rian", control_set_at=datetime.now(UTC) + timedelta(seconds=1))) db.commit() clock = FakeClock() def resume_on_second(n): if n == 2: db.execute(update(Source).where(Source.id == 1).values(control="run", control_set_at=datetime.now(UTC) + timedelta(seconds=2))) db.commit() clock.on_sleep = resume_on_second _hook(db, run, clock).wait(1.0) assert sum(clock.slept) == 2 * control.CONTROL_POLL + 4.0, "slower after a pause, never faster" def test_the_stop_flag_lands_inside_a_sixty_second_wait_after_three_slices(self, db): run = _running(db) db.execute(update(Source).where(Source.id == 1).values(delay_seconds=60.0)) db.commit() clock = FakeClock() hook = _hook(db, run, clock) clock.on_sleep = lambda n: hook.stop_flag.set() if n == 3 else None with pytest.raises(control.RunStopped) as stopped: hook.wait(60.0) assert stopped.value.by == "signal" and sum(clock.slept) == 6.0 def test_a_row_someone_else_closed_ends_the_run_as_superseded(self, db): """The stale-row check: the heartbeat UPDATE is guarded on status='running', so a row `backfill stuck_runs` or Mark as ended closed is no longer this process's to write.""" run = _running(db) db.execute(update(CollectionRun).where(CollectionRun.id == run.id).values(status="error")) db.commit() with pytest.raises(control.RunStopped) as stopped: _hook(db, run, FakeClock()).wait(0) assert (stopped.value.reason, stopped.value.by) == ("superseded", "rules") def test_run_stopped_survives_a_collectors_broad_except(self): """dubai.py wraps an enrichment in `except Exception`; a stop must pass through it.""" with pytest.raises(control.RunStopped): try: raise control.RunStopped("stopped from the page", "rian") except Exception: # noqa: BLE001 - the stub of the broad except under test pytest.fail("a broad except swallowed the stop") # --------------------------------------------------------------------------- run_collector def _prices(db) -> int: return db.scalar(select(func.count(PriceObservation.id))) or 0 class TestTheCooperativeEnd: def test_a_sigterm_mid_run_leaves_stopped_by_signal_with_the_batch_committed(self, db): """The four `running` rows the 19 Sep OOM shed left behind held the approval gate for the 24 hours the timer had left, and the prices in their open batches were lost. The handler only sets a flag; the hook raises at the next request boundary, after the batch commit. """ clock = FakeClock() previous = signal.getsignal(signal.SIGTERM) def term_on_second(i): if i == 1: os.kill(os.getpid(), signal.SIGTERM) run = ingest.run_collector(db, FakeCollector([_raw(n) for n in range(4)], on_each=term_on_second), rates=RATES, sleeper=clock.sleep, clock=clock.clock, by="rian") assert (run.status, run.stopped_by) == ("stopped", "signal") assert run.prices_written == 2 and _prices(db) == 2, "the batch is committed before the row closes" assert run.finished_at is not None assert signal.getsignal(signal.SIGTERM) is previous, "the handler is restored" def test_a_stop_from_the_page_records_who_pressed_it(self, db): def stop_on_second(i): if i == 1: db.execute(update(Source).where(Source.id == 1).values(control="stop", control_set_by="rian", control_set_at=datetime.now(UTC) + timedelta(seconds=1))) clock = FakeClock() run = ingest.run_collector(db, FakeCollector([_raw(n) for n in range(4)], on_each=stop_on_second), rates=RATES, sleeper=clock.sleep, clock=clock.clock) assert (run.status, run.stopped_by, run.prices_written) == ("stopped", "rian", 2) def test_the_kill_switch_mid_run_ends_it_stopped_by_the_switch(self, db): def off_on_first(i): if i == 0: db.execute(update(Source).where(Source.id == 1).values(enabled=False)) clock = FakeClock() run = ingest.run_collector(db, FakeCollector([_raw(n) for n in range(3)], on_each=off_on_first), rates=RATES, sleeper=clock.sleep, clock=clock.clock) assert (run.status, run.stopped_by, run.prices_written) == ("stopped", "kill-switch", 1) def test_the_run_row_carries_pid_mode_limit_and_who_started_it(self, db): clock = FakeClock() run = ingest.run_collector(db, FakeCollector([_raw(n) for n in range(3)]), rates=RATES, limit=2, sleeper=clock.sleep, clock=clock.clock, by="rian") assert (run.status, run.pid, run.mode, run.limit_n, run.started_by) == ("ok", os.getpid(), "discover", 2, "rian") assert run.requests_made == 2 and run.heartbeat_at is not None assert db.scalar(select(Source.control).where(Source.id == 1)) == "run", "the loop never writes the control" def test_the_shell_is_the_default_starter(self, db): run = ingest.run_collector(db, FakeCollector([]), rates=RATES, sleeper=FakeClock().sleep) assert run.started_by == "shell" and run.status == "ok" class TestTheLock: def test_a_held_lock_refuses_the_second_start_before_any_run_row(self, db, monkeypatch): def held(engine, slug): raise control.CollectorLocked(f"{slug}: another collector process holds the lock") monkeypatch.setattr(ingest, "_try_lock", held) with pytest.raises(control.CollectorLocked): ingest.run_collector(db, FakeCollector([_raw(1)]), rates=RATES) assert db.scalar(select(func.count(CollectionRun.id))) == 0 def test_the_lock_is_an_advisory_lock_on_its_own_autocommit_connection(self): class Conn: def __init__(self, got): self.got, self.closed, self.statements, self.options = got, False, [], None def execution_options(self, **kw): self.options = kw return self def execute(self, statement, params=None): self.statements.append((str(statement), params)) return SimpleNamespace(scalar=lambda: self.got) def close(self): self.closed = True class Engine: dialect = SimpleNamespace(name="postgresql") def __init__(self, got): self.conn = Conn(got) def connect(self): return self.conn winner = Engine(True) conn = ingest._try_lock(winner, "avolta-ath") assert conn is winner.conn and not conn.closed and conn.options == {"isolation_level": "AUTOCOMMIT"} assert "pg_try_advisory_lock" in conn.statements[0][0] and conn.statements[0][1] == {"key": "dfp.collect.avolta-ath"} ingest._release_lock(conn, "avolta-ath") assert conn.closed and "pg_advisory_unlock" in conn.statements[1][0] loser = Engine(False) with pytest.raises(control.CollectorLocked): ingest._try_lock(loser, "avolta-ath") assert loser.conn.closed assert ingest._try_lock(SimpleNamespace(dialect=SimpleNamespace(name="sqlite")), "x") is None class TestModesAndTheEstimate: def test_the_estimate_skips_a_limited_ok_run(self, db): source = db.get(Source, 1) now = datetime.now(UTC) db.add(CollectionRun(source_id=1, started_at=now - timedelta(days=2), status="ok", products_seen=1240)) db.add(CollectionRun(source_id=1, started_at=now - timedelta(hours=1), status="ok", products_seen=30, limit_n=30)) db.commit() assert ingest.expected_total_for(db, source, "discover", None, None) == 1240 assert ingest.expected_total_for(db, source, "discover", 100, None) == 100 assert ingest.expected_total_for(db, source, "recheck", None, 412) == 412 assert ingest.expected_total_for(db, source, "recheck", 30, 412) == 30 def test_no_unlimited_ok_run_means_no_estimate(self, db): db.add(CollectionRun(source_id=1, started_at=datetime.now(UTC), status="ok", products_seen=30, limit_n=30)) db.commit() assert ingest.expected_total_for(db, db.get(Source, 1), "discover", None, None) is None def test_a_recheck_walks_the_held_listings_through_read_one(self, db): clock = FakeClock() first = ingest.run_collector(db, FakeCollector([_raw(n) for n in range(4)]), rates=RATES, sleeper=clock.sleep, clock=clock.clock) assert first.status == "ok" and first.prices_written == 4 results = {"sku0": _raw(0, price=12.0), "sku1": "missing", "sku2": FetchError("https://x returned HTTP 404"), "sku3": FetchError("https://x failed: timeout")} collector = FakeCollector([], read_results=results) run = ingest.run_collector(db, collector, rates=RATES, mode="recheck", sleeper=clock.sleep, clock=clock.clock) assert collector.reads == ["sku0", "sku1", "sku2", "sku3"] assert (run.status, run.mode, run.expected_total) == ("ok", "recheck", 4) assert run.existing_missing == 2 and run.skip_counts == {"recheck_fetch_error": 1} assert run.prices_written == 1 and run.requests_made == 4 def test_a_recheck_with_nothing_held_ends_skipped(self, db): run = ingest.run_collector(db, FakeCollector([]), rates=RATES, mode="recheck", sleeper=FakeClock().sleep) assert run.status == "skipped" and "nothing to recheck" in (run.error or "") # --------------------------------------------------------------------------- what a Start needs LOW = procinfo.ContainerMemory(used_bytes=1024 ** 3, limit_bytes=4 * 1024 ** 3, readable=True) HIGH = procinfo.ContainerMemory(used_bytes=procinfo.START_HEADROOM_BYTES + 1, limit_bytes=4 * 1024 ** 3, readable=True) UNREADABLE = procinfo.ContainerMemory(None, None, False) @pytest.fixture def uploads(tmp_path, monkeypatch): from app.config import settings monkeypatch.setattr(settings, "uploads_dir", str(tmp_path)) return tmp_path def _refusal(db, mode="discover", memory=LOW, collector=None, **kw): now = datetime.now(UTC) return control.start_refusal(db, db.get(Source, 1), collector or FakeCollector([]), mode, None, memory, now, **kw) class TestStartRefusal: def test_nothing_refuses_a_plain_start(self, db, uploads): assert _refusal(db) is None def test_the_kill_switch_first(self, db, uploads): db.execute(update(Source).where(Source.id == 1).values(enabled=False)) db.commit() assert _refusal(db).code == "SOURCE_DISABLED" def test_a_refusal_is_final_and_names_the_host_and_the_date(self, db, uploads): db.add(CollectionRun(source_id=1, started_at=datetime(2026, 9, 11, 3, 0, tzinfo=UTC), status="blocked", error="shop.example.com robots.txt answered HTTP 403")) db.commit() refusal = _refusal(db) assert refusal.code == "SOURCE_REFUSED" assert "shop.example.com" in refusal.summary and "11 Sep" in refusal.summary and "rian" in refusal.summary def test_a_verification_refusal_newer_than_the_last_run_counts(self, db, uploads): now = datetime.now(UTC) db.add(CollectionRun(source_id=1, started_at=now - timedelta(days=9), status="ok")) db.commit() assert _refusal(db, prefetched=True, last=db.scalar(select(CollectionRun)), running=None, refused_since=now - timedelta(days=1)).code == "SOURCE_REFUSED" assert _refusal(db, prefetched=True, last=db.scalar(select(CollectionRun)), running=None, refused_since=now - timedelta(days=20)) is None def test_source_refused_comes_before_memory_low(self, db, uploads): db.add(CollectionRun(source_id=1, started_at=datetime.now(UTC), status="blocked", error="403")) db.commit() assert _refusal(db, memory=HIGH).code == "SOURCE_REFUSED" def test_a_live_run_refuses_and_a_dead_one_does_not(self, db, uploads): _running(db) assert _refusal(db).code == "SOURCE_RUNNING" db.execute(update(CollectionRun).values(started_at=datetime.now(UTC) - ingest.STUCK_AFTER - timedelta(minutes=1))) db.commit() assert _refusal(db) is None def test_recheck_is_refused_on_a_rendered_collector_and_a_platform_with_no_read_path(self, db, uploads): from app.services.collectors.registry import COLLECTORS rendered = next(c for c in COLLECTORS.values() if control.is_rendered(c)) assert _refusal(db, mode="recheck", collector=rendered).code == "MODE_UNSUPPORTED" no_path = next(c for c in COLLECTORS.values() if not control.has_read_path(c)) assert _refusal(db, mode="recheck", collector=no_path).code == "MODE_UNSUPPORTED" assert control.has_read_path(FakeCollector([])) def test_recheck_with_nothing_held_is_refused(self, db, uploads): assert _refusal(db, mode="recheck").code == "NOTHING_TO_RECHECK" ingest.run_collector(db, FakeCollector([_raw(1)]), rates=RATES, sleeper=FakeClock().sleep) assert _refusal(db, mode="recheck") is None def test_the_freeze_marker_refuses_with_its_stamp(self, db, uploads): (uploads / ".collect-freeze").write_text("D3 2026-09-19T21:00:00Z by rian\n") refusal = _refusal(db) assert refusal.code == "COLLECT_FROZEN" and "D3 2026-09-19T21:00:00Z by rian" in refusal.summary def test_memory_over_the_headroom_refuses_and_unreadable_is_skipped(self, db, uploads): refusal = _refusal(db, memory=HIGH) assert refusal.code == "MEMORY_LOW" and "3.0 GiB" in refusal.summary and "3.3 GiB" in refusal.summary assert _refusal(db, memory=UNREADABLE) is None class TestPaceRefusal: def _source(self, **kw): base = dict(slug="s", delay_seconds=1.0, robots_crawl_delay=None, robots_read_at=None, delay_set_by=None, delay_set_at=None) return SimpleNamespace(**{**base, **kw}) def test_the_hosts_crawl_delay_is_the_floor_and_the_sentence_names_it(self): source = self._source(robots_crawl_delay=60, robots_read_at=datetime(2026, 9, 19, 13, 24, tzinfo=UTC)) refusal = control.pace_refusal(source, FakeCollector([]), 30) assert refusal.code == "PACE_BELOW_FLOOR" and "at least 60 s" in refusal.summary and "19 Sep 13:24" in refusal.summary assert control.pace_refusal(source, FakeCollector([]), 60) is None pace = control.pace_of(source, FakeCollector([])) assert (pace.floor, pace.floor_reason, pace.recommended) == (60.0, "robots", 60.0) def test_our_own_second_when_the_host_asks_nothing(self): refusal = control.pace_refusal(self._source(), FakeCollector([]), 0.5) assert refusal.code == "PACE_BELOW_FLOOR" and "our own floor is 1 s" in refusal.summary def test_a_rendered_source_never_goes_under_the_render_floor(self): from app.services.collectors.registry import COLLECTORS rendered = next(c for c in COLLECTORS.values() if control.is_rendered(c)) refusal = control.pace_refusal(self._source(), rendered, 5) assert refusal.code == "PACE_BELOW_FLOOR" and "rendered source waits at least" in refusal.summary assert control.pace_of(self._source(), rendered).floor_reason == "render" def test_the_cap(self): assert control.pace_refusal(self._source(), FakeCollector([]), 601).code == "PACE_ABOVE_CAP" assert control.pace_refusal(self._source(), FakeCollector([]), 600) is None class TestTheSpawn: def test_the_argv_is_a_list_with_the_session_user(self): argv = control.collect_argv("shopify-yul", "recheck", 30, "rian") assert argv[1:] == ["-m", "app.cli", "collect", "--source", "shopify-yul", "--mode", "recheck", "--by", "rian", "--limit", "30"] assert "--limit" not in control.collect_argv("shopify-yul", "discover", None, "rian") def test_spawn_starts_its_own_session_and_reaps(self, monkeypatch): calls = {} class Proc: pid = 4242 def wait(self): return 0 def popen(argv, **kw): calls["argv"], calls["kw"] = argv, kw return Proc() monkeypatch.setattr(control.subprocess, "Popen", popen) proc = control.spawn_collector("shopify-yul", "discover", None, "rian") assert proc.pid == 4242 and calls["kw"] == {"start_new_session": True, "close_fds": True} assert calls["argv"][3:] == ["collect", "--source", "shopify-yul", "--mode", "discover", "--by", "rian"] class TestProcinfo: def test_memory_and_a_process_read_from_a_fake_root(self, tmp_path): (tmp_path / "memory.current").write_text("1717986918\n") (tmp_path / "memory.max").write_text("max\n") memory = procinfo.container_memory(tmp_path) assert (memory.used_bytes, memory.limit_bytes, memory.readable) == (1717986918, None, True) assert procinfo.container_memory(tmp_path / "nowhere") == procinfo.ContainerMemory(None, None, False) proc = tmp_path / "proc" (proc / "1").mkdir(parents=True) (proc / "77").mkdir() (proc / "77" / "cmdline").write_bytes(b"python\0-m\0app.cli\0collect\0--source\0avolta-ath\0") (proc / "77" / "status").write_text("Name:\tpython\nState:\tS (sleeping)\nVmRSS:\t 84000 kB\n") assert procinfo.alive(77, "avolta-ath", proc) is True assert procinfo.alive(77, "avolta-lhr", proc) is False, "another source's process is not this run" assert procinfo.alive(78, "avolta-ath", proc) is False, "a pid that is gone" assert procinfo.alive(1, "avolta-ath", proc) is False and procinfo.alive(None, "x", proc) is False assert procinfo.alive(77, "avolta-ath", tmp_path / "no-proc") is None, "unreadable is unknown, never false" assert procinfo.rss_of(77, proc) == 84000 * 1024 (proc / "77" / "status").write_text("State:\tZ (zombie)\n") assert procinfo.alive(77, "avolta-ath", proc) is False def test_a_collect_all_process_and_a_source_equals_form_are_this_runs_collector(self, tmp_path): """`alive` once demanded the literal argv items `--source` and ``. A run started the documented default way, `python -m app.cli collect` (every enabled collector), or as `--source=`, read `alive is False`; `state_of` then called it "stalled: process gone" ahead of its fresh heartbeat, and Stop now on that row would have `mark_stuck`ed a live run, which the hook ends as superseded at its next poll: a healthy sweep killed from the page by a false reading. Match the process, not the flag shape.""" proc = tmp_path / "proc" (proc / "1").mkdir(parents=True) status = "Name:\tpython\nState:\tS (sleeping)\nVmRSS:\t 84000 kB\n" (proc / "80").mkdir() (proc / "80" / "cmdline").write_bytes(b"python\0-m\0app.cli\0collect\0--limit\030\0") (proc / "80" / "status").write_text(status) assert procinfo.alive(80, "avolta-ath", proc) is True, "a collect-all run collects this source too" assert procinfo.alive(80, "shopify-yul", proc) is True (proc / "81").mkdir() (proc / "81" / "cmdline").write_bytes(b"python\0-m\0app.cli\0collect\0--source=avolta-ath\0--by\0rian\0") (proc / "81" / "status").write_text(status) assert procinfo.alive(81, "avolta-ath", proc) is True assert procinfo.alive(81, "avolta-lhr", proc) is False (proc / "82").mkdir() (proc / "82" / "cmdline").write_bytes(b"python\0-m\0app.cli\0hours\0collect\0") (proc / "82" / "status").write_text(status) assert procinfo.alive(82, "avolta-ath", proc) is False, "the hours collectors are another command" assert procinfo.collects(["python", "-m", "app.cli", "collect", "--source"], "avolta-ath") is False def test_the_freeze_marker(self, tmp_path): marker = tmp_path / ".collect-freeze" assert procinfo.freeze_state(marker).frozen is False marker.write_text("D3 2026-09-19T21:00:00Z by rian\n") state = procinfo.freeze_state(marker) assert state.frozen and state.stamp == "D3 2026-09-19T21:00:00Z by rian" and not state.stale assert procinfo.freeze_state(marker, datetime.now(UTC) + procinfo.FREEZE_STALE + timedelta(minutes=1)).stale # --------------------------------------------------------------------------- the six routes @pytest.fixture def world(monkeypatch, tmp_path): from tests import _accounts as T from tests.kit import _env catalogue = [Retailer.__table__, Shop.__table__, ProductLine.__table__, ProductVariant.__table__, Listing.__table__, Source.__table__, CollectionRun.__table__, PriceObservation.__table__, RawRecord.__table__, Merge.__table__, Suggestion.__table__, AttributeAlias.__table__] T.fresh(monkeypatch) Base.metadata.drop_all(_env.engine, tables=catalogue) Base.metadata.create_all(_env.engine, tables=catalogue) T.person("rian") T.person("adam", level="admin") from app.config import settings from app.services.collectors.registry import COLLECTORS monkeypatch.setattr(settings, "uploads_dir", str(tmp_path)) monkeypatch.setattr(procinfo, "container_memory", lambda *a, **k: LOW) slug = next(s for s, c in COLLECTORS.items() if type(c).__name__ == "ShopifyCollector") with _env.TestSessionLocal() as db: db.add(Source(id=1, slug=slug, name="Shopify shop", delay_seconds=2.0)) db.commit() yield SimpleNamespace(slug=slug, T=T, env=_env) Base.metadata.drop_all(_env.engine, tables=catalogue) def _owner(world): c = world.T.client() world.T.as_user(c, "rian") return c def _seed_running(world, **kw) -> int: with world.env.TestSessionLocal() as db: run = CollectionRun(source_id=1, started_at=kw.pop("started_at", datetime.now(UTC)), status="running", **kw) db.add(run) db.commit() return run.id def _run(world, run_id): with world.env.TestSessionLocal() as db: return db.get(CollectionRun, run_id) def _source(world): with world.env.TestSessionLocal() as db: return db.get(Source, 1) class TestTheRoutesRefuseTheWrongPerson: def test_401_signed_out_and_403_for_a_developer_on_all_six(self, world): c = world.T.client() bodies = {"start": {}, "pause": None, "resume": None, "stop": {}, "pace": {"delay_seconds": 5}, "mode": {"mode": "discover"}} for verb, body in bodies.items(): assert c.post(f"/api/collectors/{world.slug}/{verb}", json=body).status_code == 401, verb world.T.as_user(c, "adam") for verb, body in bodies.items(): assert c.post(f"/api/collectors/{world.slug}/{verb}", json=body).status_code == 403, verb assert world.T.audit_actions() == [] class TestStart: def test_the_owner_starts_a_run_and_the_audit_row_carries_mode_limit_and_pid(self, world, monkeypatch): spawned = {} def fake_spawn(slug, mode, limit, by): spawned.update(slug=slug, mode=mode, limit=limit, by=by) return SimpleNamespace(pid=4242) monkeypatch.setattr(control, "spawn_collector", fake_spawn) r = _owner(world).post(f"/api/collectors/{world.slug}/start", json={"mode": "discover", "limit": 30}) assert r.status_code == 202, r.text assert r.json()["started"] and r.json()["pid"] == 4242 and r.json()["control"] == "run" assert spawned == {"slug": world.slug, "mode": "discover", "limit": 30, "by": "rian"} assert world.T.audit_actions() == ["source.start"] assert _source(world).control_set_by == "rian" def test_a_running_source_refuses_and_a_refused_one_is_final(self, world, monkeypatch): monkeypatch.setattr(control, "spawn_collector", lambda *a: pytest.fail("spawned despite a refusal")) _seed_running(world) r = _owner(world).post(f"/api/collectors/{world.slug}/start", json={}) assert r.status_code == 409 and r.json()["detail"]["error_code"] == "SOURCE_RUNNING" with world.env.TestSessionLocal() as db: db.execute(update(CollectionRun).values(status="blocked", error="403", finished_at=datetime.now(UTC))) db.commit() r = _owner(world).post(f"/api/collectors/{world.slug}/start", json={}) assert r.status_code == 409 and r.json()["detail"]["error_code"] == "SOURCE_REFUSED" assert "rian" in r.json()["detail"]["summary"] assert world.T.audit_actions() == [] def test_a_refusal_behind_a_dead_running_row_is_still_final(self, world, monkeypatch): """The route once took `last` as the newest row of any status while `live()` takes the newest FINISHED row. A `blocked` run followed by a dead `running` row (the shape the 19 Sep OOM left behind) made the page say `start_blocked: SOURCE_REFUSED` while POST /start saw only the dead row, skipped it, and let the Start through: a refused host recollected from the page, against the brief's "SOURCE_REFUSED is final" and "start_blocked is exactly what the POST answers". One definition of `last` for both.""" monkeypatch.setattr(control, "spawn_collector", lambda *a: pytest.fail("spawned on a refused source")) now = datetime.now(UTC) with world.env.TestSessionLocal() as db: db.add(CollectionRun(source_id=1, started_at=now - timedelta(days=2), finished_at=now - timedelta(days=2), status="blocked", error="403")) db.commit() _seed_running(world, started_at=now - ingest.STUCK_AFTER - timedelta(hours=1)) r = _owner(world).post(f"/api/collectors/{world.slug}/start", json={}) assert r.status_code == 409, r.text assert r.json()["detail"]["error_code"] == "SOURCE_REFUSED" with world.env.TestSessionLocal() as db: source, collector = db.get(Source, 1), FakeCollector([]) last = db.scalar(select(CollectionRun).where(CollectionRun.status != "running")) running = db.scalar(select(CollectionRun).where(CollectionRun.status == "running")) page = control.start_refusal(db, source, collector, "discover", None, LOW, now, last=last, running=running, refused_since=None, prefetched=True) assert page.code == "SOURCE_REFUSED", "the page's start_blocked and the POST answer the same" assert world.T.audit_actions() == [] def test_an_unknown_collector_is_404_and_a_bad_body_422(self, world): c = _owner(world) assert c.post("/api/collectors/no-such/start", json={}).status_code == 404 assert c.post(f"/api/collectors/{world.slug}/start", json={"mode": "bogus"}).status_code == 422 assert c.post(f"/api/collectors/{world.slug}/start", json={"limit": 0}).status_code == 422 class TestPauseResumeStop: def test_pause_then_resume_with_an_audit_row_each_and_idempotent_pause(self, world): c = _owner(world) run_id = _seed_running(world) r = c.post(f"/api/collectors/{world.slug}/pause") assert r.status_code == 200 and r.json()["control"] == "pause" and r.json()["effective_control"] == "pause" again = c.post(f"/api/collectors/{world.slug}/pause") assert again.status_code == 200 and again.json()["changed"] is False r = c.post(f"/api/collectors/{world.slug}/resume") assert r.status_code == 200 and r.json()["control"] == "run" assert c.post(f"/api/collectors/{world.slug}/resume").json()["detail"]["error_code"] == "NOT_PAUSED" assert world.T.audit_actions() == ["source.pause", "source.resume"] assert _run(world, run_id).status == "running", "a route never writes the run's status" def test_pause_and_resume_refuse_without_a_live_run(self, world): c = _owner(world) assert c.post(f"/api/collectors/{world.slug}/pause").json()["detail"]["error_code"] == "NOT_RUNNING" assert c.post(f"/api/collectors/{world.slug}/resume").json()["detail"]["error_code"] == "NOT_PAUSED" _seed_running(world, started_at=datetime.now(UTC) - ingest.STUCK_AFTER - timedelta(minutes=1)) assert c.post(f"/api/collectors/{world.slug}/pause").json()["detail"]["error_code"] == "NOT_RUNNING" def test_a_stop_is_cooperative_and_stop_now_needs_a_prior_stop(self, world, monkeypatch): c = _owner(world) run_id = _seed_running(world, pid=777) assert c.post(f"/api/collectors/{world.slug}/stop", json={"force": True}).json()["detail"]["error_code"] == "STOP_NOT_REQUESTED" r = c.post(f"/api/collectors/{world.slug}/stop", json={}) assert r.status_code == 200 and r.json()["effective_control"] == "stop" assert c.post(f"/api/collectors/{world.slug}/pause").json()["detail"]["error_code"] == "ALREADY_STOPPING" assert c.post(f"/api/collectors/{world.slug}/resume").json()["detail"]["error_code"] == "ALREADY_STOPPING" killed = [] monkeypatch.setattr(procinfo, "alive", lambda pid, slug, proc=None: True) monkeypatch.setattr(os, "kill", lambda pid, sig: killed.append((pid, sig))) r = c.post(f"/api/collectors/{world.slug}/stop", json={"force": True}) assert r.status_code == 200 and killed == [(777, signal.SIGTERM)] assert _run(world, run_id).status == "running", "SIGTERM is the loop's to answer; the row is not closed here" assert world.T.audit_actions() == ["source.stop", "source.stop"] def test_stop_now_on_a_stalled_run_whose_process_is_gone_marks_it_ended(self, world, monkeypatch): c = _owner(world) run_id = _seed_running(world, pid=777) c.post(f"/api/collectors/{world.slug}/stop", json={}) monkeypatch.setattr(procinfo, "alive", lambda pid, slug, proc=None: False) monkeypatch.setattr(os, "kill", lambda pid, sig: pytest.fail("signalled a pid that is gone")) assert c.post(f"/api/collectors/{world.slug}/stop", json={"force": True}).status_code == 200 run = _run(world, run_id) assert run.status == "error" and "process gone" in run.error and "rian" in run.error def test_stop_now_without_a_pid_is_refused(self, world): c = _owner(world) _seed_running(world) c.post(f"/api/collectors/{world.slug}/stop", json={}) assert c.post(f"/api/collectors/{world.slug}/stop", json={"force": True}).json()["detail"]["error_code"] == "PID_UNKNOWN" def test_stop_on_a_dead_run_marks_it_ended_by_the_actor(self, world): c = _owner(world) run_id = _seed_running(world, started_at=datetime.now(UTC) - ingest.STUCK_AFTER - timedelta(minutes=1)) r = c.post(f"/api/collectors/{world.slug}/stop", json={}) assert r.status_code == 200 and r.json()["run"] is None run = _run(world, run_id) assert run.status == "error" and run.error == "stuck: ended by rian from the page" assert c.post(f"/api/collectors/{world.slug}/stop", json={}).json()["detail"]["error_code"] == "NOT_RUNNING" class TestPaceAndMode: def test_the_pace_is_floored_and_capped_and_audited(self, world): c = _owner(world) assert c.post(f"/api/collectors/{world.slug}/pace", json={"delay_seconds": 0.5}).json()["detail"]["error_code"] == "PACE_BELOW_FLOOR" assert c.post(f"/api/collectors/{world.slug}/pace", json={"delay_seconds": 601}).json()["detail"]["error_code"] == "PACE_ABOVE_CAP" assert c.post(f"/api/collectors/{world.slug}/pace", json={"delay_seconds": 0}).status_code == 422 r = c.post(f"/api/collectors/{world.slug}/pace", json={"delay_seconds": 3}) assert r.status_code == 200 and r.json()["pace"]["delay_seconds"] == 3.0 and r.json()["pace"]["floor"] == 1.0 source = _source(world) assert float(source.delay_seconds) == 3.0 and source.delay_set_by == "rian" assert world.T.audit_actions() == ["source.pace"] with world.env.TestSessionLocal() as db: db.execute(update(Source).values(robots_crawl_delay=60, robots_read_at=datetime.now(UTC))) db.commit() r = c.post(f"/api/collectors/{world.slug}/pace", json={"delay_seconds": 30}) assert r.json()["detail"]["error_code"] == "PACE_BELOW_FLOOR" and "at least 60 s" in r.json()["detail"]["summary"] def test_the_mode_is_a_source_default_and_recheck_is_refused_where_unsupported(self, world): from app.services.collectors.registry import COLLECTORS c = _owner(world) r = c.post(f"/api/collectors/{world.slug}/mode", json={"mode": "recheck"}) assert r.status_code == 200 and r.json()["mode_default"] == "recheck" and _source(world).mode == "recheck" assert c.post(f"/api/collectors/{world.slug}/mode", json={"mode": "bogus"}).status_code == 422 no_path = next(s for s, col in COLLECTORS.items() if not control.has_read_path(col)) with world.env.TestSessionLocal() as db: db.add(Source(id=2, slug=no_path, name="No read path")) db.commit() r = c.post(f"/api/collectors/{no_path}/mode", json={"mode": "recheck"}) assert r.status_code == 409 and r.json()["detail"]["error_code"] == "MODE_UNSUPPORTED" assert world.T.audit_actions() == ["source.mode"]