obsidian-wiki/tests/test_mcp_ingest_verification.py

import hashlib
import json
import sqlite3
import tempfile
import unittest
from pathlib import Path

from obsidian_wiki.mcp_ingest_verification import verify_ingest_outputs


class McpIngestVerificationTests(unittest.TestCase):
    def setUp(self):
        self.tempdir = tempfile.TemporaryDirectory()
        self.root = Path(self.tempdir.name)
        self.vault = self.root / "vault"
        self.runtime = self.root / "runtime"
        self.vault.mkdir()
        self.runtime.mkdir()
        (self.vault / "_meta").mkdir()
        self.payload = b"verified source"
        self.digest = hashlib.sha256(self.payload).hexdigest()
        self.upload_path = self.runtime / "source.pdf"
        self.upload_path.write_bytes(self.payload)
        self.upload = {
            "upload_id": "upload-1",
            "path": str(self.upload_path),
            "sha256": f"sha256:{self.digest}",
        }
        self.started_at = "2026-07-16T10:00:00Z"
        self.archive_path = "_sources/archive/project/2026-07/source.pdf"
        self.registry_record = {
            "source_id": f"sha256:{self.digest}",
            "archive_path": self.archive_path,
            "related_pages": ["references/source.md"],
            "catalog_paths": ["_sources/catalog/source.md"],
        }
        self.manifest_record = {
            "ingested_at": "2026-07-16T10:01:00Z",
            "content_hash": f"sha256:{self.digest}",
            "source_id": f"sha256:{self.digest}",
            "original_path": str(self.upload_path),
            "archive_path": self.archive_path,
            "pages_created": ["references/source.md"],
            "pages_updated": [],
            "catalog_paths": ["_sources/catalog/source.md"],
        }
        self._write_complete_outputs()

    def tearDown(self):
        self.tempdir.cleanup()

    def _write_complete_outputs(self):
        archive = self.vault / self.archive_path
        archive.parent.mkdir(parents=True, exist_ok=True)
        archive.write_bytes(self.payload)
        for relative_path in ("references/source.md", "_sources/catalog/source.md"):
            path = self.vault / relative_path
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_text("verified", encoding="utf-8")
        registry = self.vault / "_meta" / "source-registry.jsonl"
        registry.write_text(json.dumps(self.registry_record) + "\n", encoding="utf-8")
        manifest = {"version": 1, "sources": {str(self.upload_path): self.manifest_record}}
        (self.vault / ".manifest.json").write_text(
            json.dumps(manifest),
            encoding="utf-8",
        )
        db_path = self.vault / "_meta" / "file-assets.sqlite"
        with sqlite3.connect(db_path) as conn:
            conn.execute("CREATE TABLE file_assets (sha256 TEXT NOT NULL)")
            conn.execute("INSERT INTO file_assets (sha256) VALUES (?)", (self.digest,))

    def _verify(self):
        return verify_ingest_outputs(
            vault_path=self.vault,
            uploads=[self.upload],
            job_started_at=self.started_at,
            archive_sources=True,
        )

    def test_passes_when_all_deterministic_outputs_exist(self):
        result = self._verify()

        self.assertEqual(result["status"], "passed")
        self.assertEqual(result["failures"], [])
        self.assertEqual(
            result["outputs"],
            {
                "source_count": 1,
                "registry_count": 1,
                "archive_count": 1,
                "manifest_count": 1,
                "indexed_count": 1,
            },
        )
        self.assertTrue(result["checked_at"])

    def test_fails_when_registry_record_is_missing(self):
        (self.vault / "_meta" / "source-registry.jsonl").write_text("", encoding="utf-8")

        result = self._verify()

        self.assertEqual(result["status"], "failed")
        self.assertIn("registry_missing", {item["code"] for item in result["failures"]})

    def test_fails_when_required_archive_is_missing(self):
        (self.vault / self.archive_path).unlink()

        result = self._verify()

        self.assertEqual(result["status"], "failed")
        self.assertIn("archive_missing", {item["code"] for item in result["failures"]})

    def test_fails_when_manifest_record_is_missing_or_stale(self):
        self.manifest_record["ingested_at"] = "2026-07-16T09:59:59Z"
        manifest = {"version": 1, "sources": {str(self.upload_path): self.manifest_record}}
        (self.vault / ".manifest.json").write_text(json.dumps(manifest), encoding="utf-8")

        result = self._verify()

        self.assertEqual(result["status"], "failed")
        self.assertIn("manifest_missing", {item["code"] for item in result["failures"]})

    def test_fails_when_file_asset_index_row_is_missing(self):
        with sqlite3.connect(self.vault / "_meta" / "file-assets.sqlite") as conn:
            conn.execute("DELETE FROM file_assets")

        result = self._verify()

        self.assertEqual(result["status"], "failed")
        self.assertIn("index_missing", {item["code"] for item in result["failures"]})

    def test_fails_when_declared_output_path_is_missing(self):
        (self.vault / "references" / "source.md").unlink()

        result = self._verify()

        self.assertEqual(result["status"], "failed")
        self.assertIn("declared_path_missing", {item["code"] for item in result["failures"]})

    def test_archive_check_is_skipped_when_archiving_is_disabled(self):
        (self.vault / self.archive_path).unlink()

        result = verify_ingest_outputs(
            vault_path=self.vault,
            uploads=[self.upload],
            job_started_at=self.started_at,
            archive_sources=False,
        )

        self.assertEqual(result["status"], "passed")
        self.assertEqual(result["outputs"]["archive_count"], 0)


if __name__ == "__main__":
    unittest.main()