"""Integration: the pipeline against the local Postgres with a temporary `ztest-` company (skipped when the DB is unavailable).""" from __future__ import annotations import asyncio from typing import Any import pytest from conftest import FakeFetcher, fixture_path from companyatlas.fetch import FetchError, NotModified, file_result from companyatlas.taxonomy import FailureClass BASE = "https://www.acme-cloud.example" async def _setup() -> tuple[str, dict[str, str]]: from companyatlas.db import execute, jsonb, transaction from companyatlas.ids import new_id cid = new_id("company") slug = f"ztest-pipe-{cid[-6:].lower()}" dom = f"{slug}.example" sensors: dict[str, str] = {} async with transaction() as conn: await execute(conn, """insert into companies (id, slug, display_name, canonical_domain, website, tier, importance, onboarding_status) values (:id, :slug, 'Pipeline test', :dom, :web, 3, 0.4, 'active')""", id=cid, slug=slug, dom=dom, web=f"https://{dom}/") for surface, url, con in [("pricing", BASE + "/pricing", "generic-html-v1"), ("careers", BASE + "/careers", "generic-html-v1"), ("jobs_board", "https://boards-api.greenhouse.io/v1/boards/ztest/jobs?content=false", "greenhouse-v1"), ("leadership", BASE + "/about/leadership", "generic-html-v1"), ("locations", BASE + "/company/locations", "generic-html-v1"), ("newsroom", BASE + "/news", "generic-html-v1"), ("legal_terms", BASE + "/legal/terms", "generic-html-v1")]: sid = new_id("sensor") await execute(conn, """insert into sensors (id, company_id, surface, connector_id, url, canonical_url, domain, base_interval_s, current_interval_s, status, config) values (:id, :c, :s, :con, :u, :u, :d, 43200, 43200, 'pending', cast(:cfg as jsonb))""", id=sid, c=cid, s=surface, con=con, u=url, d=dom, cfg=jsonb({"canonical_domain": "acme-cloud.example", "token": "ztest"})) sensors[surface] = sid return cid, sensors async def _teardown(cid: str) -> None: from companyatlas.db import dispose, execute, transaction async with transaction() as conn: await execute(conn, "delete from companies where id = :c", c=cid) await execute(conn, "delete from domain_budgets where domain like 'ztest-%'") await dispose() async def _sensor(sid: str) -> dict[str, Any]: from companyatlas.db import fetch_one, transaction async with transaction() as conn: row = await fetch_one(conn, "select * from sensors where id = :id", id=sid) assert row is not None return row async def _run(sid: str, path: str, ct: str = "text/html", fetcher: Any = None): # type: ignore[no-untyped-def] from companyatlas.services.pipeline import run_sensor row = await _sensor(sid) return await run_sensor(row, fetcher=fetcher or FakeFetcher(), worker="pytest", result=file_result(path, url=row["url"], content_type=ct)) async def _q(sql: str, **params: Any) -> list[dict[str, Any]]: from companyatlas.db import fetch_all, transaction async with transaction() as conn: return await fetch_all(conn, sql, **params) @pytest.mark.usefixtures("db") def test_pipeline_end_to_end() -> None: async def scenario() -> None: cid, s = await _setup() try: gh = fixture_path("greenhouse", "stripe_jobs.json") # ---- baseline runs (version 1, no change rows, entities inserted) o = await _run(s["pricing"], fixture_path("generic_html", "pricing.html")) assert o.status == "ok" and o.snapshot_id and not o.change_id and o.delta_counts == {"plans_added": 3} # ---- same content again → unchanged, no snapshot, interval grows ×1.25 o = await _run(s["pricing"], fixture_path("generic_html", "pricing.html")) assert o.status == "unchanged" and o.snapshot_id is None and o.interval_s == int(43200 * 1.25) # ---- price change → new snapshot version, change row pending, plan superseded, burst interval o = await _run(s["pricing"], fixture_path("generic_html", "pricing_v2.html")) assert o.status == "changed" and o.kind in ("major", "critical") and o.change_id and o.delta_counts.get("plans_price_changed") == 1 assert o.interval_s == 900 plans = await _q("select plan_name, price, status, version_no from pricing_plans where company_id = :c order by plan_name, version_no", c=cid) starter = [p for p in plans if p["plan_name"] == "Starter"] assert [(int(p["price"]), p["status"], p["version_no"]) for p in starter] == [(29, "superseded", 1), (39, "current", 2)] chg = (await _q("select * from changes where company_id = :c and surface = 'pricing'", c=cid))[0] assert chg["status"] == "pending" and chg["structured_delta"]["plans"]["price_changed"][0]["after"] == 39 and chg["diff"]["counts"]["modified"] == 1 snaps = await _q("select version_no, previous_snapshot_id from snapshots where sensor_id = :s order by version_no", s=s["pricing"]) assert [x["version_no"] for x in snaps] == [1, 2] and snaps[1]["previous_snapshot_id"] == (await _q("select id from snapshots where sensor_id = :s and version_no = 1", s=s["pricing"]))[0]["id"] # ---- careers: jobs added / removed with no_longer_listed await _run(s["careers"], fixture_path("generic_html", "careers.html")) o = await _run(s["careers"], fixture_path("generic_html", "careers_v2.html")) assert o.delta_counts == {"jobs_added": 2, "jobs_removed": 1} jobs = await _q("select title, status, removed_at, is_ai from jobs where sensor_id = :s order by title", s=s["careers"]) by = {j["title"]: j for j in jobs} assert by["People Operations Lead"]["status"] == "no_longer_listed" and by["People Operations Lead"]["removed_at"] is not None assert by["AI Research Scientist"]["is_ai"] and by["AI Research Scientist"]["status"] == "open" assert len(jobs) == 7 # ---- untrusted extraction never mass-removes: a page with no job listing leaves the 6 open jobs alone o = await _run(s["careers"], fixture_path("generic_html", "legal_terms.html")) open_after = await _q("select count(*) n from jobs where sensor_id = :s and status = 'open'", s=s["careers"]) assert open_after[0]["n"] == 6 # ---- structured board: baseline + unchanged o = await _run(s["jobs_board"], gh, "application/json") assert o.status == "ok" and o.delta_counts == {"jobs_added": 20} o = await _run(s["jobs_board"], gh, "application/json") assert o.status == "unchanged" # ---- other surfaces baseline for surf, fn in (("leadership", "leadership.html"), ("locations", "locations.html"), ("newsroom", "newsroom.html"), ("legal_terms", "legal_terms.html")): o = await _run(s[surf], fixture_path("generic_html", fn)) assert o.status == "ok", (surf, o.error) assert (await _q("select count(*) n from people where company_id = :c and is_executive", c=cid))[0]["n"] == 4 assert (await _q("select count(*) n from locations where company_id = :c and country = 'US'", c=cid))[0]["n"] == 2 assert (await _q("select count(*) n from news_items where company_id = :c", c=cid))[0]["n"] == 4 # ---- failure path: fetch error → observation + failures row + backoff; repeated → failing from companyatlas.services.pipeline import run_sensor ff = FakeFetcher() ff.add_error(BASE + "/legal/terms", FetchError("http 404", status=404, url=BASE + "/legal/terms", failure=FailureClass.PAGE_REMOVED)) row = await _sensor(s["legal_terms"]) o = await run_sensor(row, fetcher=ff, worker="pytest") # type: ignore[arg-type] assert o.status == "failed" and o.failure_class == "PAGE_REMOVED" and o.interval_s == min(7 * 86400, int(row["current_interval_s"] * 4.0)) o = await run_sensor(await _sensor(s["legal_terms"]), fetcher=ff, worker="pytest") # type: ignore[arg-type] assert o.sensor_status == "failing" fails = await _q("select failure_class from failures where sensor_id = :s", s=s["legal_terms"]) assert len(fails) == 2 # ---- not modified path ff2 = FakeFetcher() ff2.add_error(BASE + "/news", NotModified(5)) o = await run_sensor(await _sensor(s["newsroom"]), fetcher=ff2, worker="pytest") # type: ignore[arg-type] assert o.status == "not_modified" obs = await _q("select not_modified from observations where sensor_id = :s order by fetched_at desc limit 1", s=s["newsroom"]) assert obs[0]["not_modified"] is True # ---- robots block → review queue item ff3 = FakeFetcher() from companyatlas.fetch import BlockedError ff3.add_error(BASE + "/company/locations", BlockedError("robots", url=BASE + "/company/locations", failure=FailureClass.ROBOTS)) o = await run_sensor(await _sensor(s["locations"]), fetcher=ff3, worker="pytest") # type: ignore[arg-type] assert o.sensor_status == "blocked" assert (await _q("select count(*) n from review_queue where ref_id = :s and kind = 'blocked_source'", s=s["locations"]))[0]["n"] == 1 # ---- company + ledger touched comp = (await _q("select stats, first_observed_at, last_change_at from companies where id = :c", c=cid))[0] assert comp["first_observed_at"] is not None and comp["last_change_at"] is not None and comp["stats"]["meaningful_changes"] >= 2 led = await _q("select units from cost_ledger where day = current_date and dimension = 'fetch' and key = :c", c=cid) assert led and led[0]["units"] >= 10 finally: await _teardown(cid) asyncio.run(scenario()) @pytest.mark.usefixtures("db") def test_scheduler_claims_and_releases() -> None: async def scenario() -> None: from companyatlas.services.scheduler import claim_due_sensors, release_claims cid, _sensors = await _setup() try: rows = await claim_due_sensors("pytest-worker", 100) mine = [r for r in rows if r["company_id"] == cid] assert len(mine) == 7 and all(r["claimed_by"] == "pytest-worker" for r in mine) again = await claim_due_sensors("pytest-worker-2", 100) assert not [r for r in again if r["company_id"] == cid] # already claimed await release_claims("pytest-worker") assert (await _q("select count(*) n from sensors where company_id = :c and claimed_by is null", c=cid))[0]["n"] == 7 finally: await _teardown(cid) asyncio.run(scenario()) @pytest.mark.usefixtures("db") def test_connectors_table_sync() -> None: async def scenario() -> None: from companyatlas.db import dispose, transaction from companyatlas.sdk.connector import all_connectors, sync_connectors_table async with transaction() as conn: n = await sync_connectors_table(conn) assert n == len(all_connectors()) >= 14 rows = await _q("select id from connectors where id = 'generic-html-v1'") assert rows await dispose() asyncio.run(scenario())