spb/company-atlas
Public
Python 66.3%
TypeScript 22.7%
JavaScript 8.6%
HTML 1.4%
CSS 0.7%
1"""Company profile enrichment: Wikidata mapping, Wikipedia/homepage precedence, provenance, grounded LLM text, people/relationship2persistence and idempotency. Fixtures in `fixtures/enrichment/` (trimmed Alphabet/Google entity, its English labels, a Wikipedia summary,3a homepage with a JSON-LD Organization). No live network: the fake fetcher answers the exact API URLs the client builds."""4from __future__ import annotations56import json7from pathlib import Path8from typing import Any9from urllib.parse import parse_qs, urlparse1011import pytest12from conftest import FakeFetcher13from factories import (14 cleanup,15 intel_db, # noqa: F401 — pytest fixture registered by import16 make_company,17)1819from companyatlas.db import execute, fetch_all, fetch_one, jsonb, transaction20from companyatlas.services import enrichment as en2122FIX = Path(__file__).resolve().parents[1] / "fixtures" / "enrichment"23ENTITY = json.loads((FIX / "wikidata-Q95.json").read_text(encoding="utf-8"))["entities"]["Q95"]24_LABELS = json.loads((FIX / "wikidata-labels.json").read_text(encoding="utf-8"))25LABELS = {q: v["label"] for q, v in _LABELS.items()}26DESCRIPTIONS = {q: v["description"] for q, v in _LABELS.items()}27SUMMARY = json.loads((FIX / "wikipedia-summary.json").read_text(encoding="utf-8"))28HOMEPAGE = (FIX / "homepage.html").read_text(encoding="utf-8")29EXTRA_LABELS: dict[str, str] = {} # test-local QIDs (never real ones, so DB tests cannot touch real companies)30CLAIMS = {("Q30", "P297"): ["US"], ("Q4917", "P498"): ["USD"]}31WD_URL = "https://www.wikidata.org/wiki/Q95"32WP_URL = "https://en.wikipedia.org/wiki/Google"333435class WikimediaFetcher(FakeFetcher):36 """Answers wbgetentities / wbgetclaims from the fixtures whatever the id order; other URLs use the FakeFetcher routes."""3738 async def get(self, url: str, **kw: Any) -> Any:39 p = urlparse(url)40 if p.netloc == "www.wikidata.org" and p.path == "/w/api.php":41 self.calls.append(url)42 assert kw.get("respect_robots") is False and kw.get("rate_per_min")43 q = {k: v[0] for k, v in parse_qs(p.query).items()}44 if q["action"] == "wbgetentities":45 ids = q["ids"].split("|")46 if "claims" in q["props"]:47 ents = {i: ENTITY for i in ids if i == "Q95"} | {i: {"id": i, "missing": ""} for i in ids if i != "Q95"}48 else:49 labels = LABELS | EXTRA_LABELS50 descs = DESCRIPTIONS | {q: "company" for q in EXTRA_LABELS}51 ents = {i: {"id": i, "labels": ({"en": {"language": "en", "value": labels[i]}} if labels.get(i) else {}),52 "descriptions": ({"en": {"language": "en", "value": descs[i]}} if descs.get(i) else {})} for i in ids}53 return self._json({"entities": ents})54 if q["action"] == "wbgetclaims":55 vals = CLAIMS.get((q["entity"], q["property"]), [])56 return self._json({"claims": {q["property"]: [{"mainsnak": {"snaktype": "value", "datavalue": {"value": v, "type": "string"}}, "rank": "normal"} for v in vals]}})57 return await super().get(url, **kw)5859 def _json(self, payload: dict[str, Any]) -> Any:60 from conftest import make_result6162 return make_result("https://www.wikidata.org/w/api.php", json.dumps(payload), content_type="application/json")636465def company_row(**over: Any) -> dict[str, Any]:66 base = {"id": "co_test", "slug": "google", "display_name": "Google", "canonical_domain": "google.com", "website": "https://www.google.com",67 "description": "American multinational technology company", "industries": [], "industry_primary": None, "country": None, "hq_city": None,68 "hq_region": None, "founded_year": None, "employees": None, "public_company": False, "ticker": None, "exchange": None, "legal_name": None,69 "lei": None, "sec_cik": None, "logo_url": None, "wikidata_id": "Q95", "source_meta": {"source": "wikidata", "industry_labels": ["technology company"]}}70 base.update(over)71 return base727374@pytest.fixture75def fetcher(monkeypatch: pytest.MonkeyPatch) -> WikimediaFetcher:76 async def no_dns(url: str) -> None: # SSRF validation resolves hosts — never in tests77 return None7879 monkeypatch.setattr(en, "validate_destination_async", no_dns)80 f = WikimediaFetcher()81 f.add("https://en.wikipedia.org/api/rest_v1/page/summary/Google", json.dumps(SUMMARY), content_type="application/json")82 f.add("https://www.example-robotics.test/", HOMEPAGE)83 return f848586# ================================================================================================================ Wikidata mapping878889async def test_wikidata_mapping_and_provenance(fetcher: WikimediaFetcher) -> None:90 res = await en.enrich_company(company_row(), fetcher=fetcher, sources=("wikidata",), use_db=False, llm=False)91 p = res.profile92 assert res.sources_used == ["wikidata"] and not res.errors93 assert p["legal_name"] == "Google LLC" and p["founded_year"] == 1998 and p["legal_form"] == "limited liability company"94 assert p["hq"]["city"] == "Mountain View" and p["hq"]["country"] == "US" and p["hq"]["region"] == "California"95 assert p["hq"]["lat"] == pytest.approx(37.42, abs=0.01) and p["hq"]["lon"] == pytest.approx(-122.08, abs=0.01)96 assert p["employees"] and p["employees_year"] and p["employees_year"] >= 2013 # latest P585 observation wins97 assert p["revenue"] == {"value": 305630000000.0, "currency": "USD", "year": 2023} # 2023 beats 202198 assert p["net_income"]["year"] == 2023 and p["total_assets"] is None99 assert "software" in p["industries"] and p["industry_labels"] and "Internet industry" in p["industry_labels"]100 assert "Google Search" in p["products"]101 assert p["exchange"] is None and p["ticker"] is None # both Nasdaq listings ended in 2016 (P582)102 assert p["isin"] == "US02079K3059" and p["lei"] == "7ZW8QJWVPR4P1J1KQY45" and p["sec_cik"] == "0001824723"103 assert p["official_website"] == "https://about.google/" and p["public_company"] is True # ISIN present104 assert p["logo_url"].startswith("https://commons.wikimedia.org/wiki/Special:FilePath/Google_2026_logo.svg")105 assert p["socials"]["linkedin"] == "https://www.linkedin.com/company/google" and p["socials"]["x"] == "https://x.com/Google"106 assert p["socials"]["github"] == "https://github.com/google" and p["socials"]["crunchbase"] == "https://www.crunchbase.com/organization/google"107 assert p["wikipedia_url"] == WP_URL and p["wikidata_url"] == WD_URL108 assert p["description"] == "American multinational technology company" and p["description_source"] == "wikidata"109 by_field = {s["field"]: s for s in p["sources"]}110 assert by_field["employees"] == {"field": "employees", "source": "wikidata", "url": WD_URL, "retrieved_at": by_field["employees"]["retrieved_at"]}111 assert by_field["industry_labels"]["source"] == "wikidata" and by_field["revenue"]["url"] == WD_URL112 assert p["enriched_at"] and p["version"] == "profile-v1"113 # people: CEO (current) listed, founders listed, an ended CEO tenure → no_longer_listed (never "left")114 people = {x.name: x for x in res.people}115 assert people["Sundar Pichai"].status == "listed" and people["Sundar Pichai"].role_category == "ceo" and people["Sundar Pichai"].is_executive116 assert people["Larry Page"].role_category == "founder" and people["Sergey Brin"].source_url == WD_URL117 assert people["Eric Schmidt"].status == "no_longer_listed" and people["Eric Schmidt"].valid_to is not None118 assert all(x.title in ("Chief Executive Officer", "Founder") for x in res.people)119 # relationships: parent → SUBSIDIARY_OF, subsidiaries → PARENT_OF, owner of → OWNER_OF, capped per property120 kinds = {(r.kind, r.to_qid): r for r in res.relationships}121 assert kinds[("SUBSIDIARY_OF", "Q20800404")].to_name == "Alphabet Inc." and kinds[("SUBSIDIARY_OF", "Q20800404")].valid_from is not None122 assert kinds[("PARENT_OF", "Q1318441")].to_name == "AdMob" and kinds[("PARENT_OF", "Q1318441")].property == "P355"123 assert not any(k == "OWNER_OF" for k, _ in kinds) # Google's "owner of" items in the fixture are products → filtered124 assert ("PARENT_OF", "Q1053674") not in kinds # deprecated-rank statement (DoubleClick) ignored125 assert en.looks_like_organisation("top-level domain", default=True) is False and en.looks_like_organisation("American advertising company", default=False)126 assert en.looks_like_organisation("provides Internet ad serving services", default=False) is True127 assert en.looks_like_organisation("note-taking service developed by Google", default=False) is False128 assert en.looks_like_organisation(None, default=False) is False and en.looks_like_organisation("something unusual", default=True) is True129 # column back-fills: null columns filled, registry description NOT replaced by the one-line Wikidata description (same rank)130 assert res.column_updates["founded_year"] == 1998 and res.column_updates["hq_city"] == "Mountain View" and res.column_updates["country"] == "US"131 assert res.column_updates["legal_name"] == "Google LLC" and res.column_updates["public_company"] is True132 assert "description" not in res.column_updates and res.industries and "software" in res.industries133 # label lookups are batched (≤ 50 ids per call) and country/currency codes resolved through wbgetclaims134 label_calls = [c for c in fetcher.calls if "props=labels%7Cdescriptions" in c]135 assert label_calls and all(len(parse_qs(urlparse(c).query)["ids"][0].split("|")) <= 50 for c in label_calls)136 assert any("wbgetclaims" in c and "P297" in c for c in fetcher.calls)137138139def _item(qid: str) -> dict[str, Any]:140 return {"snaktype": "value", "datavalue": {"value": {"entity-type": "item", "id": qid}, "type": "wikibase-entityid"}}141142143def _string(v: str) -> dict[str, Any]:144 return {"snaktype": "value", "datavalue": {"value": v, "type": "string"}}145146147async def test_current_listing_and_ticker_qualifier(fetcher: WikimediaFetcher) -> None:148 """A current P414 statement gives the exchange; the ticker comes from its P249 qualifier when there is no top-level P249."""149 entity = {"id": "Q95", "labels": {}, "descriptions": {}, "sitelinks": {},150 "claims": {"P414": [{"mainsnak": {**_item("Q82059"), "property": "P414"}, "rank": "normal", "qualifiers": {"P249": [{**_string("GOOGL"), "property": "P249"}]}},151 {"mainsnak": {**_item("Q82059"), "property": "P414"}, "rank": "normal",152 "qualifiers": {"P249": [{**_string("OLD"), "property": "P249"}],153 "P582": [{"snaktype": "value", "property": "P582", "datavalue": {"value": {"time": "+2010-01-01T00:00:00Z", "precision": 11}, "type": "time"}}]}}],154 "P1128": [{"mainsnak": {"snaktype": "value", "property": "P1128", "datavalue": {"value": {"amount": "+10", "unit": "1"}, "type": "quantity"}}, "rank": "normal"},155 {"mainsnak": {"snaktype": "value", "property": "P1128", "datavalue": {"value": {"amount": "+12", "unit": "1"}, "type": "quantity"}}, "rank": "normal",156 "qualifiers": {"P585": [{"snaktype": "value", "property": "P585", "datavalue": {"value": {"time": "+2020-01-01T00:00:00Z", "precision": 9}, "type": "time"}}]}}]}}157 res = await en.enrich_company(company_row(description=None), fetcher=fetcher, entity=entity, sources=("wikidata",), use_db=False, llm=False)158 assert res.profile["exchange"] == "Nasdaq" and res.profile["ticker"] == "GOOGL" and res.profile["public_company"] is True159 assert res.profile["employees"] == 12 and res.profile["employees_year"] == 2020 # dated observation beats an undated one160 assert res.column_updates["ticker"] == "GOOGL" and res.column_updates["exchange"] == "Nasdaq"161162163async def test_wikidata_helpers() -> None:164 assert en.wd_time({"time": "+1998-09-04T00:00:00Z", "precision": 11})[0] == 1998165 assert en.wd_time({"time": "+2015-00-00T00:00:00Z", "precision": 9})[1].isoformat() == "2015-01-01"166 assert en.wd_time({"time": "-0050-00-00T00:00:00Z", "precision": 9}) == (None, None, 9)167 assert en.wd_quantity({"amount": "+47756", "unit": "1"}) == (47756.0, None)168 assert en.wd_quantity({"amount": "-3.5", "unit": "http://www.wikidata.org/entity/Q4917"}) == (-3.5, "Q4917")169 assert en.commons_url("Google 2026 logo.svg") == "https://commons.wikimedia.org/wiki/Special:FilePath/Google_2026_logo.svg"170 assert en.commons_url("Éclair (1).png") == "https://commons.wikimedia.org/wiki/Special:FilePath/%C3%89clair_%281%29.png"171 q = en.latest_quantity(ENTITY, "P2139")172 assert q and q[2] == 2023173 url = en.WikidataClient.entities_url(["Q95", "Q3884"])174 assert url.startswith("https://www.wikidata.org/w/api.php?format=json&action=wbgetentities&ids=Q95%7CQ3884&props=labels%7Cdescriptions%7Cclaims%7Csitelinks")175 assert "sitefilter=" in url and "enwiki" in url176 assert en.rank_of("description", "wikipedia") > en.rank_of("description", "llm") > en.rank_of("description", "homepage") > en.rank_of("description", "wikidata")177 assert en.rank_of("employees", "wikidata") > en.rank_of("employees", "homepage") > en.rank_of("employees", "registry")178179180async def test_wikidata_client_tolerates_failures() -> None:181 f = FakeFetcher() # every URL → 404182 wd = en.WikidataClient(f)183 assert await wd.entities(["Q95"]) == {} and await wd.labels(["Q30"]) == {} and await wd.country_iso("Q30") is None184 assert await wd.currency_code("Q4917") == "USD" # static table, no request185 res = await en.enrich_company(company_row(), fetcher=f, sources=("wikidata", "wikipedia"), use_db=False, llm=False)186 assert res.errors == ["wikidata: entity unavailable"] and res.profile["description"] == "American multinational technology company"187 assert res.column_updates == {} and res.people == [] and res.relationships == []188189190# ================================================================================================================ Wikipedia / homepage / precedence191192193async def test_wikipedia_description_with_attribution(fetcher: WikimediaFetcher) -> None:194 res = await en.enrich_company(company_row(), fetcher=fetcher, sources=("wikidata", "wikipedia"), use_db=False, llm=False)195 p = res.profile196 assert res.sources_used == ["wikidata", "wikipedia"]197 assert p["description"].startswith("Google LLC is an American multinational technology corporation") and len(p["description"]) >= 200198 assert p["description_source"] == "wikipedia" and p["description_url"] == WP_URL and p["description_license"] == "CC BY-SA 4.0"199 assert p["description_attribution"] == "Text from Wikipedia (en), CC BY-SA 4.0"200 assert p["logo_url"].startswith("https://commons.wikimedia.org/") # Wikidata logo outranks the Wikipedia thumbnail201 assert res.column_updates["description"] == p["description"] # wikipedia outranks the registry one-liner202 src = {s["field"]: s["source"] for s in p["sources"]}203 assert src["description"] == "wikipedia" and src["wikipedia_url"] in ("wikidata", "wikipedia")204205206def test_clean_extract() -> None:207 raw = "Acme Corp (pronounced /ˈækmi/) is a company.[1] It makes things.[citation needed]\n\nSecond paragraph here.\nThird.\nFourth is dropped."208 assert en.clean_extract(raw) == "Acme Corp is a company. It makes things. Second paragraph here. Third."209 long = " ".join([f"Sentence number {i} is here." for i in range(200)])210 out = en.clean_extract(long, max_chars=300)211 assert out and len(out) <= 300 and out.endswith(".")212 assert en.clean_extract("") is None213214215async def test_homepage_facts_and_icon(fetcher: WikimediaFetcher) -> None:216 company = company_row(id="co_home", slug="example-robotics", display_name="Example Robotics", canonical_domain="example-robotics.test",217 website="https://www.example-robotics.test/", wikidata_id=None, description=None, source_meta={"source": "manual"})218 res = await en.enrich_company(company, fetcher=fetcher, sources=("homepage",), use_db=False, llm=False)219 p = res.profile220 assert res.sources_used == ["homepage"] and not res.errors221 assert p["description"].startswith("Example Robotics designs and builds autonomous mobile robots") and p["description_source"] == "homepage"222 assert p["description_attribution"] == en.HOMEPAGE_ATTRIBUTION and p["description_license"] is None223 assert p["legal_name"] == "Example Robotics Inc." and p["founded_year"] == 2014 and p["employees"] == 420 and p["phone"] == "+1 514-555-0100"224 assert p["hq"] == {"city": "Montréal", "region": "Quebec", "country": "CA", "address": "1200 Rue Example, H2X 1Y4, Montréal, Quebec, CA", "lat": None, "lon": None}225 assert p["logo_url"] == "https://www.example-robotics.test/static/logo.svg"226 assert p["icon_url"] == "https://www.example-robotics.test/static/apple-touch-icon.png" # apple-touch-icon beats favicon and og:image227 assert p["socials"] == {"linkedin": "https://www.linkedin.com/company/example-robotics", "x": "https://twitter.com/examplerobotics",228 "github": "https://github.com/example-robotics"}229 assert res.column_updates["description"] == p["description"] and res.column_updates["country"] == "CA" and res.column_updates["employees"] == 420230 facts = en.profile_facts(p)231 keys = {f["key"]: f for f in facts}232 assert keys["headquarters"]["value"] == "Montréal, Quebec, CA" and keys["employees"]["value"] == "420" and keys["founded"]["source"] == "homepage"233 assert keys["founded"]["url"] == "https://www.example-robotics.test/"234235236def test_parse_homepage_ignores_relative_junk() -> None:237 html = '<html><head><meta name="description" content="short"><link rel="icon" href="javascript:alert(1)"><meta property="og:image" content="//cdn.example.test/og.png">' \238 '<script type="application/ld+json">{"@type":"Organization","name":"X","numberOfEmployees":{"minValue":10,"maxValue":50},"foundingDate":"not a date","address":"12 Main St"}</script></head></html>'239 f = en.parse_homepage(html, "https://www.example.test/")240 assert f.description is None and f.icon == "https://cdn.example.test/og.png" and f.employees is None and f.founded_year is None and f.address == "12 Main St"241242243async def test_better_source_is_never_overwritten(fetcher: WikimediaFetcher) -> None:244 """A company whose description column already came from Wikipedia keeps it when only the homepage runs; a homepage logo does not245 replace a Wikidata logo, but does replace a registry one."""246 company = company_row(id="co_keep", slug="example-robotics", canonical_domain="example-robotics.test", website="https://www.example-robotics.test/",247 wikidata_id=None, description="Long encyclopedic text from Wikipedia about the company.", logo_url="https://commons.wikimedia.org/x.svg",248 employees=400, source_meta={"source": "wikidata", "provenance": {"description": {"source": "wikipedia", "url": WP_URL},249 "logo_url": {"source": "wikidata", "url": WD_URL}}})250 res = await en.enrich_company(company, fetcher=fetcher, sources=("homepage",), use_db=False, llm=False)251 assert res.profile["description"] == "Long encyclopedic text from Wikipedia about the company." and res.profile["description_source"] == "wikipedia"252 assert res.profile["logo_url"] == "https://commons.wikimedia.org/x.svg"253 assert res.profile["employees"] == 420 # homepage JSON-LD outranks the registry value…254 assert res.column_updates == {"employees": 420, "hq_city": "Montréal", "hq_region": "Quebec", "country": "CA", "founded_year": 2014,255 "legal_name": "Example Robotics Inc."} # …and description / logo_url are left alone256 # registry logo (no provenance) is replaced by the homepage JSON-LD logo257 company2 = company_row(id="co_keep2", slug="example-robotics", canonical_domain="example-robotics.test", website="https://www.example-robotics.test/",258 wikidata_id=None, logo_url="https://seed.example/logo.png", source_meta={"source": "wikidata"})259 res2 = await en.enrich_company(company2, fetcher=fetcher, sources=("homepage",), use_db=False, llm=False)260 assert res2.column_updates["logo_url"] == "https://www.example-robotics.test/static/logo.svg"261262263# ================================================================================================================ LLM guard rails264265266def test_numbers_grounded() -> None:267 src = "Founded in Montréal in 2014, the company employs 420 people and operates 2 sites."268 assert en.numbers_grounded("The company was founded in 2014 and has 420 employees.", src)269 assert not en.numbers_grounded("The company has 1,200 employees.", src)270 assert not en.numbers_grounded("Revenue reached $3.5 billion in 2014.", src)271 assert en.numbers_grounded("No figures here.", src)272273274async def test_llm_only_without_wikipedia_and_with_enough_text(fetcher: WikimediaFetcher, monkeypatch: pytest.MonkeyPatch) -> None:275 calls: list[dict[str, Any]] = []276277 async def fake_llm(company: dict[str, Any], text: str, *, source_url: str) -> tuple[str | None, str | None, dict[str, Any]]:278 calls.append({"company": company["id"], "chars": len(text), "url": source_url})279 return "Example Robotics builds autonomous mobile robots for warehouses and serves retailers in Canada and Europe.", "llm_x", {"status": "done"}280281 monkeypatch.setattr(en, "llm_profile_text", fake_llm)282 monkeypatch.setattr(en.settings, "llm_enabled", True)283 monkeypatch.setattr(en.settings, "llm_base_url", "http://llm.test/v1")284 company = company_row(id="co_llm", slug="example-robotics", canonical_domain="example-robotics.test", website="https://www.example-robotics.test/",285 wikidata_id=None, description=None, source_meta={"source": "manual"})286 res = await en.enrich_company(company, fetcher=fetcher, sources=("homepage", "llm"), use_db=False, llm=True)287 assert calls and calls[0]["chars"] >= en.settings.enrich_llm_min_text_chars and calls[0]["url"] == "https://www.example-robotics.test/"288 assert res.profile["description_source"] == "llm" and res.profile["description_attribution"] == en.LLM_ATTRIBUTION and res.llm_job_id == "llm_x"289 assert res.profile["description"].startswith("Example Robotics builds") and "llm" in res.sources_used290 # with a Wikipedia extract the LLM is never called291 calls.clear()292 res = await en.enrich_company(company_row(), fetcher=fetcher, sources=("wikidata", "wikipedia", "llm"), use_db=False, llm=True)293 assert not calls and res.profile["description_source"] == "wikipedia"294295296# ================================================================================================================ persistence (Postgres)297298299def test_entity() -> dict[str, Any]:300 """The Q95 fixture with every company-like target renamed to a test-only QID (real companies in the local DB must never be linked)."""301 import copy302303 ent = copy.deepcopy(ENTITY)304 ent["id"] = "QZTEST95"305 for prop in ("P749", "P355", "P127", "P1830"):306 for st in ent["claims"].get(prop) or []:307 v = st["mainsnak"]["datavalue"]["value"]308 old = v["id"]309 v["id"] = "QZTEST" + old[1:]310 EXTRA_LABELS[v["id"]] = LABELS.get(old) or f"ZTest {old}"311 EXTRA_LABELS["QZTEST20800404"] = "ZTest Alphabet"312 return ent313314315@pytest.mark.usefixtures("intel_db")316async def test_persist_people_relationships_idempotent(fetcher: WikimediaFetcher) -> None:317 entity = test_entity()318 wd_url = "https://www.wikidata.org/wiki/QZTEST95"319 try:320 async with transaction() as conn:321 google = await make_company(conn, name="ZTest Google", country="US")322 alphabet = await make_company(conn, name="ZTest Alphabet", country="US")323 await execute(conn, "update companies set wikidata_id = 'QZTEST95', description = 'American multinational technology company', source_meta = cast(:m as jsonb) where id = :id",324 id=google["id"], m=jsonb({"source": "wikidata"}))325 await execute(conn, "update companies set wikidata_id = 'QZTEST20800404' where id = :id", id=alphabet["id"])326 # a page-sourced person that Wikidata also knows: must keep its page provenance and title327 await execute(conn, """insert into people (id, company_id, name, name_norm, title, role_category, is_executive, status, source_url)328 values ('person_ztest_pichai', :c, 'Sundar Pichai', :nn, 'CEO, Google and Alphabet', 'ceo', true, 'listed', :url)""",329 c=google["id"], nn=en.norm_name("Sundar Pichai"), url="https://about.google/leadership/")330 row = await fetch_one(conn, "select * from companies where id = :id", id=google["id"])331 res = await en.enrich_company(row, fetcher=fetcher, entity=entity, sources=("wikidata", "wikipedia"), use_db=True, llm=False)332 async with transaction() as conn:333 stored = await en.persist(conn, row, res)334 assert stored["people"] >= 4 and stored["relationships_new"] >= 4335 async with transaction() as conn:336 c = await fetch_one(conn, "select * from companies where id = :id", id=google["id"])337 people = await fetch_all(conn, "select * from people where company_id = :c order by name", c=google["id"])338 rels = await fetch_all(conn, "select * from company_relationships where from_company_id = :c or to_company_id = :c order by kind", c=google["id"])339 meta = c["source_meta"]340 assert meta["profile"]["description_source"] == "wikipedia" and meta["enriched_at"] and meta["enrichment"]["sources"] == ["wikidata", "wikipedia"]341 assert c["description"].startswith("Google LLC is an American") and meta["provenance"]["description"]["source"] == "wikipedia"342 assert meta["provenance"]["description"]["previous"] == "American multinational technology company"343 assert c["founded_year"] == 1998 and c["hq_city"] == "Mountain View" and c["legal_name"] == "Google LLC" and c["public_company"] is True344 assert "software" in c["industries"] and c["industry_primary"] == c["industries"][0]345 by_name = {p["name"]: p for p in people}346 pichai = by_name["Sundar Pichai"]347 assert pichai["id"] == "person_ztest_pichai" and pichai["title"] == "CEO, Google and Alphabet" and pichai["source_url"] == "https://about.google/leadership/"348 assert by_name["Larry Page"]["source_url"] == wd_url and by_name["Larry Page"]["title"] == "Founder" and by_name["Larry Page"]["status"] == "listed"349 assert by_name["Eric Schmidt"]["status"] == "no_longer_listed" and by_name["Eric Schmidt"]["removed_at"] is not None350 sub = [r for r in rels if r["kind"] == "SUBSIDIARY_OF" and r["from_company_id"] == google["id"]]351 assert len(sub) == 1 and sub[0]["to_company_id"] == alphabet["id"] and sub[0]["provenance"]["property"] == "P749" and float(sub[0]["confidence"]) == pytest.approx(0.85)352 inverse = [r for r in rels if r["kind"] == "PARENT_OF" and r["from_company_id"] == alphabet["id"] and r["to_company_id"] == google["id"]]353 assert len(inverse) == 1354 named = [r for r in rels if r["kind"] == "PARENT_OF" and r["from_company_id"] == google["id"]]355 assert named and all(r["to_company_id"] is None and r["to_name"] for r in named)356 # second run: no new people / relationships, profile refreshed, columns unchanged357 res2 = await en.enrich_company(c, fetcher=fetcher, entity=entity, sources=("wikidata", "wikipedia"), use_db=True, llm=False)358 assert res2.column_updates == {} and res2.industries == []359 async with transaction() as conn:360 stored2 = await en.persist(conn, c, res2)361 n_people = await fetch_one(conn, "select count(*) as n from people where company_id = :c", c=google["id"])362 n_rel = await fetch_one(conn, "select count(*) as n from company_relationships where from_company_id = :c or to_company_id = :c", c=google["id"])363 c2 = await fetch_one(conn, "select * from companies where id = :id", id=google["id"])364 assert stored2["relationships_new"] == 0 and stored2["relationships_seen"] == len(rels)365 assert n_people["n"] == len(people) and n_rel["n"] == len(rels)366 assert c2["source_meta"]["profile"]["enriched_at"] >= meta["profile"]["enriched_at"] and c2["description"] == c["description"]367 # pending queue: freshly enriched companies are no longer due368 async with transaction() as conn:369 due = {r["id"] for r in await en.pending_companies(conn, 100000)}370 assert google["id"] not in due and alphabet["id"] in due371 # batch runner end-to-end (fake fetcher: the entity lookup for a test QID is "missing" → recorded, still persisted, never raises)372 stats = await en.enrich_pending(company_keys=[alphabet["slug"]], fetcher=fetcher, sources=("wikidata",), llm=False)373 assert stats["companies"] == 1 and stats["ok"] == 1 and stats["failed"] == 0 and stats["requests"] >= 1374 async with transaction() as conn:375 a = await fetch_one(conn, "select source_meta from companies where id = :id", id=alphabet["id"])376 assert a["source_meta"]["enrichment"]["errors"] == ["wikidata: entity unavailable"] and a["source_meta"]["profile"]["version"] == "profile-v1"377 finally:378 await cleanup()379