"""Certification for the Wikidata export connector.""" from __future__ import annotations import ast import json import sys import tempfile import unittest from pathlib import Path ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT)) from content_explorer.contract import validate_documents # noqa: E402 from content_explorer.ingest import ALLOWED_LICENSES, load_json_records # noqa: E402 from connectors.wikidata_export import CONNECTOR_VERSION, run # noqa: E402 PIN = ROOT / "examples" / "sources" / "wikidata" / "computing-basics.json" PUBLISHED = ROOT / "examples" / "dataspaces" / "computing-basics" KIND_NAMES = ("programming-language", "language-designer", "formalism") UI_ROOTS = (ROOT / "ui" / "app", ROOT / "ui" / "components") def _write(document: dict, directory: Path) -> Path: directory.mkdir(parents=True, exist_ok=True) path = directory / "export.json" path.write_text(json.dumps(document), encoding="utf-8") return path class WikidataConnector(unittest.TestCase): def test_pinned_export_publishes_and_replays(self) -> None: with self.assertRaises(ValueError): load_json_records(PIN) first = run(PIN) second = run(PIN) snapshot = first["snapshot"] report = first["report"] self.assertTrue(report["published"]) self.assertEqual(report["connectorVersion"], f"wikidata-export/{CONNECTOR_VERSION}") digest = __import__("hashlib").sha256(PIN.read_bytes()).hexdigest() self.assertEqual(report["sourceDigest"], digest) self.assertEqual(report["inputDigest"], digest) self.assertEqual(report["counts"]["candidates"], 16) self.assertEqual(report["counts"]["accepted"], len(snapshot["entities"])) self.assertEqual( [entity["id"] for entity in snapshot["entities"]], [entity["id"] for entity in second["snapshot"]["entities"]], ) self.assertEqual(report["contentHash"], second["report"]["contentHash"]) python = next(entity for entity in snapshot["entities"] if entity["id"].endswith(":python")) self.assertTrue(python["id"].startswith("urn:ce:computing-basics:")) self.assertIn("Q28865", python["aliases"]) source = snapshot["sources"][0] self.assertIn(source["license"], ALLOWED_LICENSES) for claim in snapshot["claims"]: self.assertEqual(claim["evidenceRefs"], [source["id"]]) self.assertEqual(validate_documents(snapshot, first["manifest"], require_size=True), []) committed = json.loads((PUBLISHED / "snapshot.json").read_text(encoding="utf-8")) committed_manifest = json.loads((PUBLISHED / "manifest.json").read_text(encoding="utf-8")) self.assertEqual(snapshot, committed) self.assertEqual(first["manifest"], committed_manifest) human = (PUBLISHED / "import-report.md").read_text(encoding="utf-8") self.assertIn(f"Connector version: wikidata-export/{CONNECTOR_VERSION}", human) self.assertIn(f"Source digest: {digest}", human) def test_forbidden_rights_unresolved_reference_and_identity_conflict_do_not_publish(self) -> None: document = json.loads(PIN.read_text(encoding="utf-8")) with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) rights = json.loads(json.dumps(document)) rights["source"]["license"] = "CC-BY-NC-4.0" rights_dest = root / "rights" rights_result = run(_write(rights, root / "rights-src"), rights_dest) self.assertFalse(rights_result["report"]["published"]) self.assertIsNone(rights_result["snapshot"]) self.assertEqual(rights_result["report"]["quarantined"][0]["reason"], "rights") self.assertFalse((rights_dest / "snapshot.json").exists()) unresolved = json.loads(json.dumps(document)) unresolved["entities"]["Q28865"]["claims"]["P287"].append({ "mainsnak": { "snaktype": "value", "property": "P287", "datavalue": {"type": "wikibase-entityid", "value": {"id": "Q1"}}, }, "type": "statement", }) unresolved_dest = root / "unresolved" unresolved_result = run(_write(unresolved, root / "unresolved-src"), unresolved_dest) self.assertFalse(unresolved_result["report"]["published"]) self.assertIsNone(unresolved_result["snapshot"]) self.assertTrue(unresolved_result["report"]["dangling"]) self.assertFalse((unresolved_dest / "snapshot.json").exists()) conflict = json.loads(json.dumps(document)) conflict["kindByQid"]["Q2005"]["key"] = conflict["kindByQid"]["Q28865"]["key"] conflict["kindByQid"]["Q2005"]["kind"] = conflict["kindByQid"]["Q28865"]["kind"] conflict_dest = root / "conflict" conflict_result = run(_write(conflict, root / "conflict-src"), conflict_dest) self.assertFalse(conflict_result["report"]["published"]) self.assertIsNone(conflict_result["snapshot"]) self.assertEqual(conflict_result["report"]["conflicts"][0]["reason"], "duplicate-identity") self.assertFalse((conflict_dest / "snapshot.json").exists()) def test_changed_export_refreshes_the_snapshot(self) -> None: document = json.loads(PIN.read_text(encoding="utf-8")) with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) dest = root / "out" first = run(PIN, dest) document["entities"]["Q28865"]["labels"]["en"]["value"] = "Python language" changed_path = root / "changed.json" changed_path.write_text(json.dumps(document), encoding="utf-8") changed = run(changed_path, dest) self.assertTrue(changed["report"]["published"]) self.assertNotEqual(changed["report"]["sourceDigest"], first["report"]["sourceDigest"]) self.assertEqual(changed["report"]["connectorVersion"], first["report"]["connectorVersion"]) self.assertGreater(changed["report"]["counts"]["accepted"], 0) published = json.loads((dest / "snapshot.json").read_text(encoding="utf-8")) python = next(entity for entity in published["entities"] if entity["id"].endswith(":python")) self.assertEqual(python["title"], "Python language") self.assertNotEqual( [entity["title"] for entity in first["snapshot"]["entities"]], [entity["title"] for entity in published["entities"]], ) def test_connector_does_not_import_ui_and_components_omit_kind_names(self) -> None: source = (ROOT / "connectors" / "wikidata_export.py").read_text(encoding="utf-8") tree = ast.parse(source) imported: list[str] = [] for node in ast.walk(tree): if isinstance(node, ast.Import): imported.extend(alias.name for alias in node.names) elif isinstance(node, ast.ImportFrom): imported.append(node.module or "") self.assertFalse(any(name == "react" or name.startswith("ui") for name in imported)) blob = "\n".join( path.read_text(encoding="utf-8") for root in UI_ROOTS for path in root.rglob("*") if path.suffix in {".ts", ".tsx"} ) for name in KIND_NAMES: self.assertNotIn(name, blob) if __name__ == "__main__": unittest.main()