import json import pytest from edu_crawler.errors import SourceStructureChanged from edu_crawler.models import RawDocument from edu_crawler.parsing import discover_job_links, parse_document from edu_crawler.policies import source_policy def test_parse_schema_org_job_posting() -> None: body = b""" """ document = RawDocument.create( "BRZP", "https://brzp.chinasme.cn/job/BRZP-1", "text/html", body ) records = parse_document(document, allow_html=True) assert records[0]["@type"] == "JobPosting" def test_discover_links_uses_versioned_patterns_and_deduplicates() -> None: body = b""" one duplicate company """ document = RawDocument.create( "BRZP", "https://brzp.chinasme.cn/", "text/html", body ) assert discover_job_links(document, source_policy("BRZP")) == [ "https://brzp.chinasme.cn/job/1" ] def test_invalid_ndjson_reports_line_without_payload() -> None: document = RawDocument.create( "BOSS", "file:///authorized.ndjson", "application/x-ndjson", b'{"id":"one"}\nnot-json\n', ) with pytest.raises(SourceStructureChanged, match="line 2") as caught: parse_document(document, allow_html=False) assert "not-json" not in str(caught.value) def test_json_collection_wrappers() -> None: document = RawDocument.create( "BOSS", "file:///authorized.json", "application/json", json.dumps({"data": {"items": [{"id": "one"}, {"id": "two"}]}}).encode(), ) assert [record["id"] for record in parse_document(document, allow_html=False)] == [ "one", "two", ]