spb/company-atlas

1from __future__ import annotations23import json45import pytest6from conftest import FakeFetcher, fixture_bytes, fixture_path, make_result78from companyatlas.fetch import file_result9from companyatlas.sdk import connector as C10from companyatlas.sdk.connector import ConnectorContext1112CASES = [13    ("greenhouse-v1", ("greenhouse", "stripe_jobs.json"), "https://boards-api.greenhouse.io/v1/boards/stripe/jobs?content=false", 20, "Abuse Investigator", "8172487"),14    ("lever-v1", ("lever", "palantir_postings.json"), "https://api.lever.co/v0/postings/palantir?mode=json", 20, "Administrative Business Partner", None),15    ("ashby-v1", ("ashby", "ashby_board.json"), "https://api.ashbyhq.com/posting-api/job-board/ashby", 20, "Engineering Manager - EU", None),16    ("smartrecruiters-v1", ("smartrecruiters", "smartrecruiters_postings.json"), "https://api.smartrecruiters.com/v1/companies/smartrecruiters/postings?limit=100", 1,17     "Data Operations Consultant", "744000148454651"),18    ("workable-v1", ("workable", "epignosis_widget.json"), "https://apply.workable.com/api/v1/widget/accounts/epignosis", 7, "Data Engineer", "E38DB16625"),19    ("workday-v1", ("workday", "nvidia_jobs_page1.json"), "https://nvidia.wd5.myworkdayjobs.com/wday/cxs/nvidia/NVIDIAExternalCareerSite/jobs", 20,20     "Senior Embedded Software Engineer, DPU - Networking", "JR2017846"),21    ("recruitee-v1", ("recruitee", "vandebron_offers.json"), "https://vandebron.recruitee.com/api/offers/", 11, "Manager Market Operations", None),22    ("personio-v1", ("personio", "personio_jobs.xml"), "https://personio.jobs.personio.de/xml", 1, "Staff Software Engineer, Data Platform", "1834171"),23    ("teamtailor-v1", ("teamtailor", "teamtailor_jobs_feed.json"), "https://career.teamtailor.com/jobs.json", 10, "Account Executive - UK Enterprise", "8021334"),24]252627@pytest.mark.parametrize("cid,fixture,url,count,first_title,first_ext", CASES)28def test_ats_connector_extracts_jobs(cid: str, fixture: tuple[str, str], url: str, count: int, first_title: str, first_ext: str | None) -> None:29    con = C.get(cid)30    assert C.for_surface("jobs_board", url).connector_id == cid31    ct = "application/xml" if fixture[1].endswith(".xml") else "application/json"32    ex = con.extract({"url": url, "surface": "jobs_board", "config": {}}, file_result(fixture_path(*fixture), url=url, content_type=ct))33    assert len(ex.jobs) == count34    assert ex.jobs[0].title == first_title35    if first_ext:36        assert ex.jobs[0].external_id == first_ext37    assert all(j.url for j in ex.jobs)38    assert len(ex.blocks) == count and all(b.kind == "job_listing" for b in ex.blocks)39    assert ex.meta["structured"] is True and ex.meta["vendor"] == con.vendor  # type: ignore[attr-defined]40    for j in ex.jobs:41        assert j.country is None or (len(j.country) == 2 and j.country.isupper())424344def test_greenhouse_fields() -> None:45    ex = C.get("greenhouse-v1").extract({"url": "https://boards-api.greenhouse.io/v1/boards/stripe/jobs?content=false", "config": {}},46                                        file_result(fixture_path("greenhouse", "stripe_jobs.json"), url="x", content_type="application/json"))47    j = ex.jobs[0]48    assert j.url.startswith("https://stripe.com/jobs/") and j.location_text == "Dublin" and j.posted_at is not None495051def test_workday_paginates_via_post(fake_fetcher: FakeFetcher) -> None:52    url = "https://nvidia.wd5.myworkdayjobs.com/wday/cxs/nvidia/NVIDIAExternalCareerSite/jobs"53    page = json.loads(fixture_bytes("workday", "nvidia_jobs_page1.json"))54    page["total"] = 2555    fake_fetcher.add(url, json.dumps(page), content_type="application/json")56    con = C.get("workday-v1")5758    async def go():  # type: ignore[no-untyped-def]59        return await con.fetch(ConnectorContext(company={}), {"url": url, "config": {}}, fake_fetcher)  # type: ignore[arg-type]6061    import asyncio6263    res = asyncio.run(go())64    assert res.headers["x-companyatlas-pages"] == "2"          # 25 total, 20 per page → 2 POSTs65    assert fake_fetcher.calls.count(f"POST {url}") == 266    ex = con.extract({"url": url, "config": {}}, res)67    assert ex.jobs and ex.jobs[0].external_id == "JR2017846"68    assert ex.jobs[0].url.endswith("_JR2017846")697071def test_smartrecruiters_pagination(fake_fetcher: FakeFetcher) -> None:72    base = "https://api.smartrecruiters.com/v1/companies/acme/postings?limit=100"73    p1 = {"offset": 0, "limit": 100, "totalFound": 2, "content": [{"id": "1", "name": "A", "location": {"city": "Berlin", "country": "de"}}]}74    p2 = {"offset": 1, "limit": 100, "totalFound": 2, "content": [{"id": "2", "name": "B", "location": {"city": "Paris", "country": "fr"}}]}75    fake_fetcher.add(base, json.dumps(p1), content_type="application/json")76    fake_fetcher.add(base + "&offset=1", json.dumps(p2), content_type="application/json")77    import asyncio7879    con = C.get("smartrecruiters-v1")80    res = asyncio.run(con.fetch(ConnectorContext(company={}), {"url": base, "config": {"token": "acme"}}, fake_fetcher))  # type: ignore[arg-type]81    ex = con.extract({"url": base, "config": {"token": "acme"}}, res)82    assert [j.title for j in ex.jobs] == ["A", "B"] and ex.jobs[1].country == "FR"83    assert ex.jobs[0].url == "https://jobs.smartrecruiters.com/acme/1"848586def test_statuspage_and_feed_and_sitemap() -> None:87    sp = C.get("statuspage-v1").extract({"url": "https://www.githubstatus.com/api/v2/summary.json", "config": {}},88                                        file_result(fixture_path("statuspage", "github_summary.json"), url="x", content_type="application/json"))89    assert sp.meta["indicator"] == "none" and sp.meta["component_count"] >= 5 and sp.blocks[0].kind == "hero"90    fd = C.get("feed-v1").extract({"url": "https://www.acme-cloud.example/blog/atom.xml", "surface": "feed", "config": {}},91                                  file_result(fixture_path("feed", "blog_atom.xml"), url="x", content_type="application/atom+xml"))92    assert fd.news[0].title == "Introducing Atlas AI" and fd.news[0].published_at.year == 2026 and fd.news[0].category == "blog"93    assert len(fd.blocks) == 394    sm = C.get("sitemap-v1").extract({"url": "https://www.acme-cloud.example/sitemap-pages.xml", "surface": "sitemap", "config": {"canonical_domain": "acme-cloud.example"}},95                                     file_result(fixture_path("sitemap", "sitemap_pages.xml"), url="x", content_type="application/xml"))96    assert sm.meta["url_count"] == 10                             # the .png is dropped97    surfaces = {d.surface.value for d in sm.discovered}98    assert {"pricing", "careers", "leadership", "locations", "newsroom", "changelog"} <= surfaces99    idx = C.get("sitemap-v1")100    from companyatlas.connectors.sitemap import parse_sitemap101102    pages, children = parse_sitemap(fixture_bytes("sitemap", "sitemap_index.xml").decode())103    assert not pages and len(children) == 2 and children[0][1].startswith("2026-09-10")104    assert idx.meta.supports_discovery105106107def test_feed_json_feed_variant() -> None:108    body = json.dumps({"version": "https://jsonfeed.org/version/1.1", "title": "T", "items": [{"id": "1", "title": "Hello world post", "url": "https://x.example/p/1",109                                                                                              "date_published": "2026-01-02T00:00:00Z", "summary": "s"}]})110    ex = C.get("feed-v1").extract({"url": "https://x.example/feed.json", "surface": "blog", "config": {}}, make_result("https://x.example/feed.json", body, content_type="application/feed+json"))111    assert ex.news[0].title == "Hello world post" and ex.news[0].category == "blog"112