spb/company-atlas
Public
Python 66.3%
TypeScript 22.7%
JavaScript 8.6%
HTML 1.4%
CSS 0.7%
1"""Deterministic event rules: subtypes, wording, importance/confidence, idempotency and cross-surface clustering."""2from __future__ import annotations34from datetime import UTC, datetime, timedelta56import pytest7from factories import intel_db # noqa: F401 — registers the fixture89from companyatlas.services.events import derive_events, safe_wording, scale_importance10from companyatlas.taxonomy import FORBIDDEN_WORDING, ChangeKind1112COMPANY = {"id": "co_test", "slug": "ztest-acme", "display_name": "Acme", "country": "CA", "industries": []}131415def _change(surface: str, *, significance: float = 0.6, delta: dict | None = None, diff: dict | None = None, kind: str | None = None) -> dict:16 from companyatlas.taxonomy import change_kind1718 return {"id": "chg_test", "sensor_id": "sen_test", "company_id": "co_test", "surface": surface, "significance": significance,19 "kind": kind or str(change_kind(significance)), "structured_delta": delta or {}, "diff": diff or {}, "detected_at": datetime(2026, 9, 12, 12, tzinfo=UTC),20 "snapshot_before": "snap_a", "snapshot_after": "snap_b", "blocks_added": 0, "blocks_removed": 0, "blocks_modified": 0, "text_delta_ratio": 0.2}212223def _sensor(surface: str, connector: str = "generic-html-v1") -> dict:24 return {"id": "sen_test", "surface": surface, "connector_id": connector, "url": f"https://acme.example/{surface}"}252627def _by_subtype(derived): # type: ignore[no-untyped-def]28 out: dict[str, list] = {}29 for d in derived.events:30 out.setdefault(d.subtype, []).append(d)31 return out323334def _assert_clean(derived) -> None: # type: ignore[no-untyped-def]35 for d in derived.events:36 low = (d.title + " " + (d.summary or "")).lower()37 assert not any(bad in low for bad in FORBIDDEN_WORDING), d.title383940# ------------------------------------------------------------------------------------------------------------ hiring414243def test_job_count_increase_aggregate_and_ai():44 added = [{"title": f"Engineer {i}", "country": "CA", "department": "Engineering"} for i in range(12)]45 added[0]["title"] = "Senior Machine Learning Engineer"46 added[1]["title"] = "AI Product Manager"47 delta = {"jobs": {"added": added, "removed": [], "open_before": 40, "open_after": 52}}48 d = derive_events(_change("careers", delta=delta), COMPANY, _sensor("careers"))49 by = _by_subtype(d)50 assert "JOB_COUNT_INCREASE" in by and "NEW_JOB" not in by # > 5 added → aggregate only51 ev = by["JOB_COUNT_INCREASE"][0]52 assert ev.title == "12 new positions detected on careers page"53 assert ev.old_value == "40" and ev.new_value == "52"54 assert len(ev.entities["jobs"]) == 1255 assert "AI_HIRING" in by and by["AI_HIRING"][0].title == "2 AI-related positions detected on careers page"56 assert ev.confidence == 0.8 # HTML extraction57 assert 0 < ev.importance <= 158 assert d.needs_classification is False59 _assert_clean(d)606162def test_per_job_events_when_few_added_and_ats_confidence():63 delta = {"jobs": {"added": [{"title": "Data Scientist", "location_text": "Toronto, CA"}, {"title": "Account Executive", "remote": True}], "removed": [],64 "open_before": 10, "open_after": 12}}65 d = derive_events(_change("jobs_board", delta=delta), COMPANY, _sensor("jobs_board", "greenhouse-v1"))66 by = _by_subtype(d)67 assert len(by["NEW_JOB"]) == 268 assert by["NEW_JOB"][0].title == "New position listed: Data Scientist (Toronto, CA)"69 assert by["NEW_JOB"][1].title == "New position listed: Account Executive (Remote)"70 assert all(e.confidence == 0.95 for e in d.events) # structured ATS JSON717273def test_job_count_decrease_wording_and_freeze_signal():74 removed = [{"title": f"Role {i}"} for i in range(30)]75 delta = {"jobs": {"added": [], "removed": removed, "open_before": 40, "open_after": 10}}76 d = derive_events(_change("careers", significance=0.7, delta=delta), COMPANY, _sensor("careers"))77 by = _by_subtype(d)78 dec = by["JOB_COUNT_DECREASE"][0]79 assert dec.title == "30 monitored job listings no longer visible on careers page"80 assert "HIRING_FREEZE_SIGNAL" in by81 assert "no longer visible" in by["HIRING_FREEZE_SIGNAL"][0].title82 assert by["HIRING_FREEZE_SIGNAL"][0].review == "unexpected_activity"83 _assert_clean(d)848586def test_hiring_surge_uses_company_baseline():87 added = [{"title": f"Role {i}"} for i in range(9)]88 delta = {"jobs": {"added": added, "removed": [], "open_before": 100, "open_after": 109}}89 no_baseline = derive_events(_change("careers", delta=delta), COMPANY, _sensor("careers"))90 assert "HIRING_SURGE" not in _by_subtype(no_baseline) # 9 < fallback threshold of 1091 baseline = {"jobs_new_weekly": {"mean": 1.0, "stddev": 1.0, "samples": 8}}92 with_baseline = derive_events(_change("careers", delta=delta), COMPANY, _sensor("careers"), baseline=baseline)93 surge = _by_subtype(with_baseline)["HIRING_SURGE"][0]94 assert "baseline ≈ 1.0 new/week" in surge.title959697# ------------------------------------------------------------------------------------------------------------ pricing9899100def test_price_increase_and_tier_changes():101 delta = {"plans": {"price_changed": [{"plan_name": "Pro", "before": 49, "after": 59, "currency": "USD", "billing_period": "month"}],102 "added": [{"plan_name": "Enterprise", "contact_sales": True}], "removed": [{"plan_name": "Starter", "price": 9, "currency": "USD"}]}}103 d = derive_events(_change("pricing", significance=0.7, delta=delta), COMPANY, _sensor("pricing"))104 by = _by_subtype(d)105 inc = by["PRICE_INCREASE"][0]106 assert inc.title == "Pro plan price observed at $59 (was $49)"107 assert inc.old_value == "$49" and inc.new_value == "$59"108 assert inc.payload["pct"] == pytest.approx(20.4, abs=0.1)109 assert by["NEW_PRICING_TIER"][0].title == "New pricing tier listed: Enterprise (contact sales)"110 assert "enterprise" in by["NEW_PRICING_TIER"][0].tags111 assert by["PRICING_TIER_REMOVED"][0].title == "Pricing tier no longer listed: Starter"112 assert inc.importance > by["PRICING_TIER_REMOVED"][0].importance113114115def test_price_decrease_eur():116 delta = {"plans": {"price_changed": [{"plan_name": "Team", "before": 30, "after": 24, "currency": "EUR", "billing_period": "month", "pct": -20}]}}117 d = derive_events(_change("pricing", delta=delta), COMPANY, _sensor("pricing"))118 ev = _by_subtype(d)["PRICE_DECREASE"][0]119 assert ev.title == "Team plan price observed at €24 (was €30)"120 assert ev.summary == "Decrease of 20.0%, billed per month."121122123def test_generic_pricing_change_from_text_diff():124 diff = {"modified": [{"path": "Pricing > Pro", "before": "a", "after": "b"}, {"path": "Pricing > FAQ", "before": "c", "after": "d"}], "added": [], "removed": [],125 "counts": {"added": 0, "removed": 0, "modified": 2}, "text_delta_ratio": 0.3}126 d = derive_events(_change("pricing", diff=diff), COMPANY, _sensor("pricing"))127 ev = _by_subtype(d)["PRICING_CHANGE"][0]128 assert ev.title == "Pricing page materially updated (2 blocks changed)"129 assert ev.confidence == 0.7 # text-diff only130 assert ev.payload["sections"] == ["Pro", "FAQ"]131132133# ------------------------------------------------------------------------------------------------------------ leadership134135136def test_leadership_events_wording():137 delta = {"people": {"added": [{"name": "Jane Doe", "title": "Chief Financial Officer", "role_category": "cfo", "is_executive": True}],138 "removed": [{"name": "John Roe", "title": "Chief Technology Officer", "role_category": "cto", "is_executive": True}],139 "title_changed": [{"name": "Ann Lee", "before": "VP Operations", "after": "COO"}]}}140 d = derive_events(_change("leadership", significance=0.7, delta=delta), COMPANY, _sensor("leadership"))141 by = _by_subtype(d)142 assert by["NEW_EXECUTIVE"][0].title == "Jane Doe listed as Chief Financial Officer on leadership page"143 assert by["EXECUTIVE_NO_LONGER_LISTED"][0].title == "John Roe no longer listed on leadership page"144 assert by["EXECUTIVE_TITLE_CHANGE"][0].title == "Ann Lee now listed as COO (was VP Operations)"145 assert by["LEADERSHIP_CHANGE"][0].title == "Leadership page updated: 1 added, 1 no longer listed, 1 title change"146 _assert_clean(d)147148149# ------------------------------------------------------------------------------------------------------------ products / locations150151152def test_product_and_location_rules():153 delta = {"products": {"added": [{"name": "Atlas Copilot", "url": "https://acme.example/copilot"}], "removed": [{"name": "Atlas Lite"}]},154 "locations": {"added": [{"name": "Toronto office", "city": "Toronto", "country": "CA", "kind": "office"}, {"name": "Tokyo", "city": "Tokyo", "country": "JP", "kind": "office"}],155 "removed": [{"name": "Berlin", "city": "Berlin", "country": "DE", "kind": "office"}], "new_countries": ["JP"]}}156 d = derive_events(_change("locations", delta=delta), COMPANY, _sensor("locations"), country_names={"JP": "Japan"})157 by = _by_subtype(d)158 assert by["NEW_PRODUCT"][0].title == "New product listed: Atlas Copilot"159 assert "ai" in by["NEW_PRODUCT"][0].tags160 assert by["PRODUCT_REMOVED"][0].title == "Product no longer listed: Atlas Lite"161 assert by["NEW_LOCATION"][0].title == "New office listed: Toronto, CA"162 assert by["OFFICE_REMOVED"][0].title == "Office no longer listed: Berlin, DE"163 assert by["COUNTRY_EXPANSION"][0].title == "New country presence listed: Japan (Tokyo)"164 assert by["COUNTRY_EXPANSION"][0].importance > by["NEW_LOCATION"][0].importance165166167# ------------------------------------------------------------------------------------------------------------ news / legal / homepage168169170def test_news_subtype_heuristics():171 delta = {"news": {"added": [{"title": "Acme announces Q3 2026 earnings results", "url": "https://acme.example/ir/q3", "category": "press"},172 {"title": "How we built our new AI search", "url": "https://acme.example/blog/ai-search", "category": "blog"},173 {"title": "v2.4 — webhooks and SDK updates", "url": "https://acme.example/changelog/2-4", "category": "changelog"}]}}174 diff = {"added": [{"path": "Feed", "before": None, "after": "Acme announces Q3 2026 earnings results — revenue up…"}], "modified": [], "removed": [], "counts": {"added": 1}}175 d = derive_events(_change("feed", delta=delta, diff=diff), COMPANY, _sensor("feed", "rss-feed-v1"))176 by = _by_subtype(d)177 assert "EARNINGS_RELEASE" in by and by["EARNINGS_RELEASE"][0].title.startswith("Earnings release: ")178 assert "BLOG_POST" in by and "ai" in by["BLOG_POST"][0].tags179 assert "CHANGELOG_ENTRY" in by180 assert all(e.confidence == 0.9 for e in d.events) # feed / JSON-LD grade evidence181 assert "EARNINGS_RELEASE" in d.summarize182183184def test_terms_change_sections_and_review():185 diff = {"modified": [{"path": "Terms > 7. Termination", "before": "x", "after": "y"}, {"path": "Terms > 12. Governing law", "before": "x", "after": "y"}],186 "added": [{"path": "Terms > 14. Arbitration", "before": None, "after": "New section"}], "removed": [], "counts": {"added": 1, "removed": 0, "modified": 2},187 "text_delta_ratio": 0.12}188 d = derive_events(_change("legal_terms", significance=0.55, diff=diff), COMPANY, _sensor("legal_terms"))189 ev = _by_subtype(d)["TERMS_CHANGE"][0]190 assert ev.title == "Terms of service page materially updated (3 sections changed)"191 assert ev.payload["sections"] == ["7. Termination", "12. Governing law", "14. Arbitration"]192 assert ev.review == "legal_sensitive" and ev.confidence == 0.7193 assert "TERMS_CHANGE" in d.summarize194195196def test_homepage_redesign_vs_change_and_messaging():197 diff = {"modified": [{"path": "Hero", "before": "Old", "after": "New"}], "added": [], "removed": [], "counts": {"added": 0, "removed": 0, "modified": 1}, "text_delta_ratio": 0.6}198 meta = {"meta": {"title_changed": {"before": "Acme — Payments", "after": "Acme — AI Payments Platform"}}}199 major = derive_events(_change("homepage", significance=0.7, diff=diff, delta=meta), COMPANY, _sensor("homepage"))200 by = _by_subtype(major)201 assert "HOMEPAGE_REDESIGN" in by and "MESSAGING_CHANGE" in by202 assert by["MESSAGING_CHANGE"][0].old_value == "Acme — Payments"203 meaningful = derive_events(_change("homepage", significance=0.5, diff=diff), COMPANY, _sensor("homepage"))204 assert list(_by_subtype(meaningful)) == ["WEBSITE_CHANGE"]205 assert meaningful.needs_classification is True and meaningful.classification_reason == "ambiguous_surface"206207208def test_docs_api_changelog_text_rules():209 diff = {"added": [{"path": "Changelog", "before": None, "after": "2026-09-10: Added bulk export endpoint\nDetails…"}], "modified": [], "removed": [],210 "counts": {"added": 1, "removed": 0, "modified": 0}, "text_delta_ratio": 0.15}211 assert _by_subtype(derive_events(_change("changelog", diff=diff), COMPANY, _sensor("changelog")))["CHANGELOG_ENTRY"][0].title == \212 "Changelog entry detected: 2026-09-10: Added bulk export endpoint"213 assert "API_CHANGE" in _by_subtype(derive_events(_change("api", diff=diff), COMPANY, _sensor("api")))214 assert "DOC_CHANGE" in _by_subtype(derive_events(_change("docs", diff=diff), COMPANY, _sensor("docs")))215216217# ------------------------------------------------------------------------------------------------------------ guards218219220@pytest.mark.parametrize("significance", [0.05, 0.3])221def test_noise_and_minor_never_emit(significance):222 delta = {"jobs": {"added": [{"title": "X"}] * 20, "removed": [], "open_before": 1, "open_after": 21}}223 d = derive_events(_change("careers", significance=significance, delta=delta), COMPANY, _sensor("careers"))224 assert d.events == [] and d.needs_classification is False225226227def test_unknown_surface_without_structure_requests_classification():228 diff = {"modified": [{"path": "Partners", "before": "a", "after": "b"}], "added": [], "removed": [], "counts": {"modified": 1}}229 d = derive_events(_change("partners", diff=diff), COMPANY, _sensor("partners"))230 assert d.events == [] and d.needs_classification and d.classification_reason == "no_deterministic_event"231232233def test_critical_changes_go_to_review_and_importance_scales():234 delta = {"plans": {"price_changed": [{"plan_name": "Pro", "before": 10, "after": 12, "currency": "USD"}]}}235 crit = derive_events(_change("pricing", significance=0.9, delta=delta), COMPANY, _sensor("pricing"))236 mid = derive_events(_change("pricing", significance=0.5, delta=delta), COMPANY, _sensor("pricing"))237 assert crit.events[0].review == "major_event" and crit.events[0].importance > mid.events[0].importance238 assert scale_importance(0.8, 0.5) == pytest.approx(0.8, abs=1e-6)239 assert scale_importance(0.8, 1.0, 1.0) == 1.0240241242def test_safe_wording_rewrites_forbidden_phrases():243 assert "no longer listed" in safe_wording("72 employees laid off")244 assert "shut down" not in safe_wording("office shut down").lower()245246247# ============================================================================================================ database flow248249250@pytest.mark.usefixtures("intel_db")251async def test_process_pending_is_idempotent_and_clusters_across_surfaces():252 from factories import cleanup, make_change, make_company, make_sensor253254 from companyatlas.db import fetch_all, fetch_one, fetch_val, transaction255 from companyatlas.services.events import process_pending_changes, reprocess_events256257 try:258 async with transaction() as conn:259 co = await make_company(conn)260 newsroom = await make_sensor(conn, co, "newsroom")261 feed = await make_sensor(conn, co, "feed", path="feed.xml")262 careers = await make_sensor(conn, co, "careers")263 item = {"title": "Acme launches Atlas Copilot for enterprises", "url": f"https://{co['canonical_domain']}/news/copilot", "category": "press"}264 await make_change(conn, newsroom, significance=0.6, structured_delta={"news": {"added": [item]}})265 await make_change(conn, feed, significance=0.6, structured_delta={"news": {"added": [{**item, "url": item["url"] + "?utm=rss"}]}})266 await make_change(conn, careers, significance=0.6, structured_delta={"jobs": {"added": [{"title": "ML Engineer", "country": "CA"}], "removed": [],267 "open_before": 5, "open_after": 6}})268 await make_change(conn, careers, significance=0.1, structured_delta={}) # noise → archived, no event269 stats = await process_pending_changes(limit=50)270 assert stats["changes"] == 3 and stats["archived"] == 1271 assert stats["events"] == 5 # NEWS×2 (one duplicate) + JOB_COUNT_INCREASE + AI_HIRING + NEW_JOB272 assert stats["duplicates"] == 1273 async with transaction() as conn:274 events = await fetch_all(conn, "select * from events where company_id = :c order by created_at", c=co["id"])275 news = [e for e in events if e["event_subtype"] == "NEWS_RELEASE"]276 assert len(news) == 2 and {e["status"] for e in news} == {"active", "duplicate"}277 canonical = next(e for e in news if e["status"] == "active")278 dup = next(e for e in news if e["status"] == "duplicate")279 assert canonical["cluster_id"] == dup["cluster_id"]280 cluster = await fetch_one(conn, "select * from event_clusters where id = :id", id=canonical["cluster_id"])281 assert cluster["source_count"] == 2 and set(cluster["surfaces"]) == {"newsroom", "feed"}282 assert canonical["confidence"] == pytest.approx(0.93, abs=0.011) # max(0.8 html, 0.9 feed) + 0.03 corroboration283 assert canonical["payload"]["corroborations"] == 1284 sources = await fetch_all(conn, "select * from event_sources where event_id = :e", e=canonical["id"])285 assert {s["kind"] for s in sources} == {"primary", "corroboration"}286 assert await fetch_val(conn, "select count(*) from changes where company_id = :c and status = 'pending'", c=co["id"]) == 0287 assert await fetch_val(conn, "select event_count from sensors where id = :s", s=careers["id"]) == 3288 assert await fetch_val(conn, "select last_event_at from companies where id = :c", c=co["id"]) is not None289 assert await fetch_val(conn, "select count(*) from llm_jobs where company_id = :c", c=co["id"]) >= 0290 # re-run: nothing pending, nothing new291 again = await process_pending_changes(limit=50)292 assert again["events"] == 0 and again["changes"] == 0293 # reprocess over processed changes: dedupe keys make it a no-op294 rep = await reprocess_events(datetime.now(UTC) - timedelta(days=1), company_id=co["id"])295 assert rep["changes"] == 3 and rep["events"] == 0296 async with transaction() as conn:297 assert await fetch_val(conn, "select count(*) from events where company_id = :c", c=co["id"]) == 5298 finally:299 await cleanup()300301302@pytest.mark.usefixtures("intel_db")303async def test_review_queue_and_llm_jobs_for_legal_change(monkeypatch):304 from factories import cleanup, make_change, make_company, make_sensor305306 from companyatlas.config import settings307 from companyatlas.db import fetch_all, transaction308 from companyatlas.services.events import process_pending_changes309310 monkeypatch.setattr(settings, "llm_base_url", "https://llm.example/v1")311 monkeypatch.setattr(settings, "llm_enabled", True)312 try:313 async with transaction() as conn:314 co = await make_company(conn)315 legal = await make_sensor(conn, co, "legal_terms")316 other = await make_sensor(conn, co, "partners")317 diff = {"modified": [{"path": "Terms > 7. Termination", "before": "old text", "after": "new text"}], "added": [], "removed": [], "counts": {"modified": 1}, "text_delta_ratio": 0.2}318 await make_change(conn, legal, significance=0.55, diff=diff)319 await make_change(conn, other, significance=0.5, diff=diff)320 stats = await process_pending_changes()321 assert stats["events"] == 1 and stats["llm_jobs"] == 2322 async with transaction() as conn:323 reviews = await fetch_all(conn, "select kind from review_queue where company_id = :c", c=co["id"])324 assert {r["kind"] for r in reviews} == {"legal_sensitive"}325 jobs = await fetch_all(conn, "select kind from llm_jobs where company_id = :c", c=co["id"])326 assert sorted(j["kind"] for j in jobs) == ["classify_change", "summarize_event"]327 finally:328 await cleanup()329330331def test_kind_enum_values_used_by_rules():332 assert ChangeKind.NOISE == "noise" and ChangeKind.CRITICAL == "critical"333334335def test_news_bursts_become_one_aggregate_event() -> None:336 from companyatlas.services.events import derive_events337338 items = [{"title": f"Release {i}: quarterly update", "url": f"https://x.com/news/{i}", "published_at": None, "category": "press"} for i in range(9)]339 change = {"id": "chg_x", "sensor_id": "sen_x", "surface": "newsroom", "significance": 0.55, "kind": "meaningful", "diff": {}, "structured_delta": {"news": {"added": items}}}340 drafts = derive_events(change, {"id": "co_x", "display_name": "X"}, {"id": "sen_x", "surface": "newsroom", "connector_id": "generic-html-v1", "url": "https://x.com/news"}).events341 news = [d for d in drafts if d.subtype == "NEWS_RELEASE"]342 assert len(news) == 1 and news[0].title.startswith("9 new news releases published on") and len(news[0].entities["news"]) == 9343 small = dict(change, structured_delta={"news": {"added": items[:2]}})344 drafts = derive_events(small, {"id": "co_x", "display_name": "X"}, {"id": "sen_x", "surface": "newsroom", "connector_id": "generic-html-v1", "url": "https://x.com/news"}).events345 assert len([d for d in drafts if d.subtype == "NEWS_RELEASE"]) == 2346