content-explorer/tests/test_wikidata_connector.py

"""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()