From 559196b3dac71ba09ef8873454d40145d4e4a048 Mon Sep 17 00:00:00 2001 From: Darren Zal Date: Wed, 5 Aug 2026 00:57:41 -0700 Subject: [PATCH] test: cover functions regen-prod already ships untested MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Recovered from the NUC runtime snapshot (nuc-runtime@316f2e2). These files existed ONLY in the NUC's .git — no MacBook clone had them, and the newest tarball backup predated them by 2.5 months. Additive: no existing file is modified. --- tests/test_project_bridge_notes.py | 236 +++++++++++++++++++++++++++ tests/test_research_author_sensor.py | 150 +++++++++++++++++ tests/test_vault_note_exists.py | 79 +++++++++ 3 files changed, 465 insertions(+) create mode 100644 tests/test_project_bridge_notes.py create mode 100644 tests/test_research_author_sensor.py create mode 100644 tests/test_vault_note_exists.py diff --git a/tests/test_project_bridge_notes.py b/tests/test_project_bridge_notes.py new file mode 100644 index 00000000..3f048a78 --- /dev/null +++ b/tests/test_project_bridge_notes.py @@ -0,0 +1,236 @@ +import asyncio +import importlib.util +import subprocess +import sys +from pathlib import Path + +import pytest + + +ROOT = Path(__file__).resolve().parents[1] +SCRIPT = ROOT / "scripts" / "project_bridge_notes.py" + + +def load_projector(): + spec = importlib.util.spec_from_file_location("project_bridge_notes", SCRIPT) + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def write_note(tmp_path, body): + path = tmp_path / "bridge-note.md" + path.write_text( + """--- +doc_id: test.connection.note +doc_kind: research +status: draft +research_subkind: bridge_note +disposition: implementation hypothesis +concepts: + - memory-governance +depends_on: + - ic.intelligence-primitives +relates_to: [] +--- + +""" + + body + ) + return path + + +def test_parse_v2_review_claims_and_stop_c_claim_body(tmp_path): + projector = load_projector() + path = write_note( + tmp_path, + """# Test + +## Claim Register + +**C1** [confidence: high] [anchor: section 1] +The source claim should not absorb the following review claim. + +**C2** [confidence: medium] [anchor: section 2] +The second source claim supports the same review target. + +- **R1**: Add a transformation-edge audit lane. [target: ic.project-learning-membrane] [concept: memory-governance] TODO: slug-deferred + supported_by: C1, C2. +""", + ) + + note = projector.parse_bridge_note(path, "ic") + report = projector.build_parse_report(note, path.read_text()) + + assert [claim.c_id for claim in note.claims] == ["C1", "C2"] + assert "**R1**" not in note.claims[-1].statement + assert len(note.review_directives) == 1 + directive = note.review_directives[0] + assert directive.r_id == "R1" + assert directive.target_doc == "ic.project-learning-membrane" + assert directive.concept == "memory-governance" + assert directive.supported_by == ["C1", "C2"] + assert directive.statement == "Add a transformation-edge audit lane." + assert report.issues == [] + + +def test_parse_legacy_review_claims(tmp_path): + projector = load_projector() + path = write_note( + tmp_path, + """# Test + +## Claim Register + +**C1** [confidence: high] [anchor: theorem] +Legacy source claim. + +**R1** [review claim] [target: ic.intelligence-primitives] [concept: memory-governance] +The Memory primitive should mention freshness budgets. +*R1 is supported by C1.* +""", + ) + + note = projector.parse_bridge_note(path, "ic") + report = projector.build_parse_report(note, path.read_text()) + + assert len(note.claims) == 1 + assert len(note.review_directives) == 1 + directive = note.review_directives[0] + assert directive.r_id == "R1" + assert directive.target_doc == "ic.intelligence-primitives" + assert directive.concept == "memory-governance" + assert directive.supported_by == ["C1"] + assert directive.statement == "The Memory primitive should mention freshness budgets." + assert report.issues == [] + + +def test_parse_report_flags_missing_concept(tmp_path): + projector = load_projector() + path = write_note( + tmp_path, + """# Test + +## Claim Register + +**C1** [confidence: high] [anchor: section] +Source claim. + +- **R1**: Add an audit lane. [target: ic.project-learning-membrane] + supported_by: C1. +""", + ) + + note = projector.parse_bridge_note(path, "ic") + report = projector.build_parse_report(note, path.read_text()) + + assert note.review_directives == [] + assert any(issue.code == "missing_concept" for issue in report.issues) + assert any(issue.severity == "error" for issue in report.issues) + + +def test_parse_report_flags_unknown_support_ref(tmp_path): + projector = load_projector() + path = write_note( + tmp_path, + """# Test + +## Claim Register + +**C1** [confidence: high] [anchor: section] +Source claim. + +- **R1**: Add an audit lane. [target: ic.project-learning-membrane] [concept: memory-governance] + supported_by: C9. +""", + ) + + note = projector.parse_bridge_note(path, "ic") + report = projector.build_parse_report(note, path.read_text()) + + assert len(note.review_directives) == 1 + assert any(issue.code == "unknown_support_ref" for issue in report.issues) + + +def test_parse_report_cli_does_not_require_db(tmp_path): + path = write_note( + tmp_path, + """# Test + +## Claim Register + +**C1** [confidence: high] [anchor: section] +Source claim. + +- **R1**: Add an audit lane. [target: ic.project-learning-membrane] [concept: memory-governance] + supported_by: C1. +""", + ) + + result = subprocess.run( + [sys.executable, str(SCRIPT), "--parse-report", "--note", str(path)], + cwd=ROOT, + text=True, + capture_output=True, + check=False, + ) + + assert result.returncode == 0 + assert "C-claims parsed: 1" in result.stdout + assert "R-claims parsed: 1" in result.stdout + assert "Issues: none" in result.stdout + + +def test_claims_service_token_prefers_env(monkeypatch, tmp_path): + projector = load_projector() + monkeypatch.setenv("KOI_CLAIMS_SERVICE_TOKEN", "env-token") + monkeypatch.setenv("HOME", str(tmp_path)) + + assert projector._claims_service_token() == "env-token" + + +def test_claims_service_token_reads_state_file(monkeypatch, tmp_path): + projector = load_projector() + monkeypatch.delenv("KOI_CLAIMS_SERVICE_TOKEN", raising=False) + monkeypatch.setenv("HOME", str(tmp_path)) + token_path = tmp_path / ".config/personal-koi/koi-state/claims_service_token" + token_path.parent.mkdir(parents=True) + token_path.write_text("state-token\n") + + assert projector._claims_service_token() == "state-token" + + +def test_projector_knows_current_intake_dispositions(): + projector = load_projector() + + assert projector.DISPOSITION_SLUG["candidate protocol"] == "propose-protocol" + assert projector.DISPOSITION_SLUG["novel synthesis"] == "synthesize" + + +def test_create_review_claim_raises_on_auth_failure(): + projector = load_projector() + + class Response: + status_code = 401 + text = "auth required" + + class Client: + async def post(self, *args, **kwargs): + return Response() + + async def run(): + with pytest.raises(RuntimeError, match="/claims auth failed"): + await projector.create_review_claim( + Client(), + conn=object(), + claimant_uri="org:ic-learning-field", + concept_name="memory-governance", + about_uri="concept:memory-governance", + target_spec_doc="ic.memory-layers", + disposition_slug="implementation-hypothesis", + project_uri="project:intelligence-commons", + projection_batch="test", + ) + + asyncio.run(run()) diff --git a/tests/test_research_author_sensor.py b/tests/test_research_author_sensor.py new file mode 100644 index 00000000..bd1b71dc --- /dev/null +++ b/tests/test_research_author_sensor.py @@ -0,0 +1,150 @@ +import importlib.util +import sys +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +SCRIPT = ROOT / "scripts" / "research_author_sensor.py" + + +def load_sensor(): + spec = importlib.util.spec_from_file_location("research_author_sensor", SCRIPT) + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def make_author(tmp_path): + sensor = load_sensor() + return sensor.AuthorConfig( + author_id="ghrist-robert", + canonical_name="Robert Ghrist", + aliases=["R. Ghrist"], + corpus_root=tmp_path / "Papers", + author_dir=tmp_path / "Papers/authors/ghrist-robert", + official_preprints="https://www2.math.upenn.edu/~ghrist/preprints.html", + arxiv_queries=['au:"Robert Ghrist"'], + project_tags=["sheaf-explorer", "spore"], + corpus_tags=["sheaf-theory", "applied-topology"], + direct_patterns=["sheaf", "cohomology", "discourse"], + project_patterns=["network", "robot", "persistence"], + ) + + +def test_parse_official_preprints(tmp_path): + sensor = load_sensor() + author = make_author(tmp_path) + html = """ + + """ + + records = sensor.parse_official_preprints(html, author) + + assert len(records) == 2 + assert records[0].year == 2026 + assert records[0].source_url == "https://www2.math.upenn.edu/~ghrist/preprints/localglobal.pdf" + assert records[1].title == "Network Sheaf Models for Social Information Systems" + + +def test_parse_arxiv_xml(tmp_path): + sensor = load_sensor() + author = make_author(tmp_path) + xml = """ + + + http://arxiv.org/abs/2603.14831v3 + 2026-03-20T00:00:00Z + 2026-03-18T00:00:00Z + Neural Networks as Local-to-Global Computations + We study sheaf-style local-to-global computation. + Victor Bosca + Robert Ghrist + + + """ + + records = sensor.parse_arxiv_xml(xml, author) + + assert len(records) == 1 + record = records[0] + assert record.year == 2026 + assert record.arxiv_id == "2603.14831" + assert record.arxiv_version == "v3" + assert record.pdf_url == "https://arxiv.org/pdf/2603.14831" + assert record.authors == ["Victor Bosca", "Robert Ghrist"] + + +def test_merge_scores_and_keeps_official_source(tmp_path): + sensor = load_sensor() + author = make_author(tmp_path) + official = sensor.PaperRecord( + author_id=author.author_id, + canonical_author=author.canonical_name, + title="Neural Networks as Local-to-Global Computations", + year=2026, + source_url="https://www2.math.upenn.edu/~ghrist/preprints/localglobal.pdf", + source_kinds=["official_preprints"], + ) + arxiv = sensor.PaperRecord( + author_id=author.author_id, + canonical_author=author.canonical_name, + title="Neural Networks as Local-to-Global Computations", + year=2026, + abstract="This paper uses sheaf cohomology language for neural network computation.", + source_url="https://arxiv.org/abs/2603.14831v3", + pdf_url="https://arxiv.org/pdf/2603.14831", + arxiv_id="2603.14831", + source_kinds=["arxiv"], + ) + + merged = sensor.merge_records([official, arxiv]) + sensor.apply_record_ids(merged) + score, matches, decision = sensor.score_record(merged[0], author) + + assert len(merged) == 1 + assert merged[0].official_pdf_or_page == "https://www2.math.upenn.edu/~ghrist/preprints/localglobal.pdf" + assert merged[0].source_url == "https://arxiv.org/abs/2603.14831v3" + assert merged[0].paper_id == "ghrist-robert/2026-neural-networks-as-local-to-global-computations" + assert score >= 6 + assert "sheaf" in matches + assert decision == "download_now" + + +def test_existing_index_uses_manifest_and_metadata_titles(tmp_path): + sensor = load_sensor() + author = make_author(tmp_path) + author.author_dir.mkdir(parents=True) + paper_dir = author.author_dir / "2026-existing-paper" + paper_dir.mkdir() + (paper_dir / "metadata.yaml").write_text('title: "[2026] Existing Paper Title"\n') + manifest = author.corpus_root / "manifest.jsonl" + manifest.write_text( + '{"paper_id":"ghrist-robert/2025-manifest-paper","title":"Manifest Paper"}\n' + ) + records = [ + sensor.PaperRecord( + author_id=author.author_id, + canonical_author=author.canonical_name, + title="Manifest Paper", + year=2025, + ), + sensor.PaperRecord( + author_id=author.author_id, + canonical_author=author.canonical_name, + title="Existing Paper Title", + year=2026, + ), + ] + sensor.apply_record_ids(records) + + ids, titles = sensor.load_existing_index(author) + sensor.mark_existing(records, ids, titles) + + assert records[0].existing is True + assert records[0].existing_reason == "paper_id" + assert records[1].existing is True + assert records[1].existing_reason == "title" diff --git a/tests/test_vault_note_exists.py b/tests/test_vault_note_exists.py new file mode 100644 index 00000000..fe351dff --- /dev/null +++ b/tests/test_vault_note_exists.py @@ -0,0 +1,79 @@ +"""Unit tests for ``_vault_note_exists`` — the phantom-path guard primitive. + +This pure-filesystem helper is the foundation of the phantom-path work on the +``feat/document-ingest-phase1`` branch: it is called by +``resolve_canonical_to_vault`` (to skip orphaned mappings) and +``/entities/mentioned-in`` (to report ``vault_note_exists``) so the vault never +gets a dangling wikilink for a registry entity whose note was deleted, relocated, +or never materialized (the ~113-orphan class observed 2026-06-02; also the +Jacob/MOVE37/Cascadia phantom resolutions observed 2026-06-10). + +The two behaviours these tests lock in: + + 1. **Existence is checked against the real FS** (with or without ``.md``), so a + stored ``vault_path`` pointing at a missing note returns ``False``. + 2. **Basename drift is NOT silently "repaired" here** — ``Organizations/MOVE37`` + returns ``False`` even when ``Organizations/MOVE37XR.md`` exists. Relocation + repair is the job of ``scripts/audit_orphan_mappings.py`` (which UPDATEs the + stored path); this guard's contract is strictly "does THIS path exist". + 3. **Headless safety**: if the vault root is not mounted on this host, the guard + returns ``True`` so a deploy that can't see the vault never strips legit links. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +import api.personal_ingest_api as api # noqa: E402 + + +@pytest.fixture +def temp_vault(tmp_path, monkeypatch): + """Point ``_VAULT_ROOT`` at a temp vault with a couple of notes.""" + (tmp_path / "People").mkdir() + (tmp_path / "Organizations").mkdir() + (tmp_path / "People" / "Jacob Sayles.md").write_text("# Jacob Sayles\n") + (tmp_path / "Organizations" / "MOVE37XR.md").write_text("# MOVE37XR\n") + monkeypatch.setattr(api, "_VAULT_ROOT", tmp_path) + return tmp_path + + +def test_empty_path_is_false(temp_vault): + assert api._vault_note_exists("") is False + assert api._vault_note_exists(None) is False + + +def test_existing_note_with_md_suffix(temp_vault): + assert api._vault_note_exists("People/Jacob Sayles.md") is True + + +def test_existing_note_without_md_suffix(temp_vault): + # The stored vault_path / wikilink form usually has no extension. + assert api._vault_note_exists("People/Jacob Sayles") is True + + +def test_missing_note_is_phantom(temp_vault): + # The phantom case: registry resolved "Jacob" but People/Jacob.md doesn't exist. + assert api._vault_note_exists("People/Jacob.md") is False + assert api._vault_note_exists("People/Jacob") is False + + +def test_basename_drift_is_not_auto_repaired(temp_vault): + # Stored phantom "Organizations/MOVE37" must report False even though the real + # note lives at "Organizations/MOVE37XR.md" under a different basename. + # Relocation repair belongs to audit_orphan_mappings.py, not this guard. + assert api._vault_note_exists("Organizations/MOVE37") is False + assert api._vault_note_exists("Organizations/MOVE37XR") is True + + +def test_vault_not_mounted_returns_true(tmp_path, monkeypatch): + # Headless deploy: vault root isn't a directory → don't second-guess paths. + monkeypatch.setattr(api, "_VAULT_ROOT", tmp_path / "does-not-exist") + assert api._vault_note_exists("People/Anyone") is True