GEWU/tests/application/test_registry.py

import hashlib
import json
import subprocess
import sys
from pathlib import Path

import portalocker
import pytest
import yaml

from gewu.adapters.git_repository import GitCanonicalRepository
from gewu.application.registry import (
    MAX_CONTEXT_RECORD_BYTES,
    LocalContextProvider,
    LocalProjectRegistry,
    RegistryError,
)
from gewu.domain.contracts import AutoAcceptPolicy, ExtractionBlock, ProjectRecord, ProjectStatus
from gewu.domain.validation import extraction_digest
from tests.domain.proposal_fixtures import EXTRACTION, NODE, POLICIES, PROJECTS


def test_context_loads_large_normalization_maps_with_a_separate_bound(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    text = "a" * 100_000
    block = ExtractionBlock(
        block_id="long-block",
        ordinal=0,
        original_text=text,
        normalized_text=text,
        normalization_map=tuple((index, index + 1) for index in range(len(text))),
    )
    candidate = EXTRACTION.model_copy(update={"blocks": (block,)})
    extraction = candidate.model_copy(update={"extraction_id": extraction_digest(candidate)})
    raw = extraction.model_dump_json().encode()
    assert len(raw) > MAX_CONTEXT_RECORD_BYTES
    repository = GitCanonicalRepository(tmp_path / "knowledge.git")
    repository.publish_files(
        expected_head=repository.head(),
        files={f"extractions/{extraction.extraction_id}.json": raw},
        message="Publish long normalized extraction",
        index_directory=tmp_path / "publishing",
    )
    provider = LocalContextProvider(repository, tmp_path / "registry.json")

    assert provider().extractions[extraction.extraction_id] == extraction

    monkeypatch.setattr("gewu.application.registry.MAX_CONTEXT_EXTRACTION_BYTES", len(raw) - 1)
    with pytest.raises(RegistryError, match="Canonical extraction is unavailable or too large"):
        provider()


def test_project_registry_create_and_archive_are_atomic_and_idempotent(tmp_path: Path) -> None:
    registry_file = tmp_path / "control" / "registry.json"
    registry = LocalProjectRegistry(
        registry_file,
        repository=GitCanonicalRepository(tmp_path / "knowledge.git"),
        index_directory=tmp_path / "publishing",
        authority_lock_path=tmp_path / "workflow" / "locks" / "canonical-writer.lock",
    )
    project = ProjectRecord(
        project_id="new-research",
        title="New research",
        status=ProjectStatus.ACTIVE,
        owners=("owner-one",),
        north_star="Investigate a new research area.",
        key_questions=("What is known?",),
        autonomy_policy_id="new-policy",
        default_includes=("shared",),
    )
    policy = AutoAcceptPolicy(
        policy_id="new-policy",
        policy_version=1,
        project_id="new-research",
        require_supported_assessment=True,
    )

    assert registry.create(project, policy) == project
    assert registry.repository is not None
    revision_after_create = registry.repository.head()
    assert (
        registry.repository.read_file(revision_after_create, "projects/new-research/project.md")
        is not None
    )
    assert registry.repository.list_files(revision_after_create, "policies")
    canonical_context = LocalContextProvider(registry.repository, registry_file)()
    assert canonical_context.projects["new-research"] == project
    assert canonical_context.policies["new-policy"] == policy
    assert registry.records() == (project,)
    assert "new-research" in registry()
    lock_path = tmp_path / "workflow" / "locks" / "canonical-writer.lock"
    with portalocker.Lock(lock_path, mode="a", timeout=0):
        with pytest.raises(portalocker.exceptions.LockException):
            registry.archive("new-research")
    assert registry.archive("new-research").status == ProjectStatus.ARCHIVED
    assert registry.repository.head() != revision_after_create
    assert registry.archive("new-research").status == ProjectStatus.ARCHIVED
    assert registry.records()[0].status == ProjectStatus.ARCHIVED
    assert (
        LocalContextProvider(registry.repository, registry_file)().projects["new-research"].status
        == ProjectStatus.ARCHIVED
    )


def test_archive_migrates_legacy_project_to_canonical_registry(tmp_path: Path) -> None:
    registry_file = tmp_path / "control" / "registry.json"
    registry_file.parent.mkdir(parents=True)
    registry_file.write_text(
        json.dumps(
            {
                "schema_version": 1,
                "projects": [project.model_dump(mode="json") for project in PROJECTS.values()],
                "policies": [policy.model_dump(mode="json") for policy in POLICIES.values()],
            }
        ),
        encoding="utf-8",
    )
    repository = GitCanonicalRepository(tmp_path / "knowledge.git")
    registry = LocalProjectRegistry(
        registry_file,
        repository=repository,
        index_directory=tmp_path / "publishing",
        authority_lock_path=tmp_path / "workflow" / "locks" / "canonical-writer.lock",
    )

    archived = registry.archive("dome-performance")
    context = LocalContextProvider(repository, registry_file)()

    assert archived.status == ProjectStatus.ARCHIVED
    assert context.projects["dome-performance"].status == ProjectStatus.ARCHIVED
    assert context.projects["interactive-video"] == PROJECTS["interactive-video"]
    assert context.policies == POLICIES


def test_canonical_registry_retains_prior_extractions_after_parser_replacement(
    tmp_path: Path,
) -> None:
    repository = GitCanonicalRepository(tmp_path / "knowledge.git")
    prior = EXTRACTION
    replacement_candidate = prior.model_copy(
        update={"extraction_id": "0" * 64, "parser_version": "replacement-parser-1"}
    )
    replacement = replacement_candidate.model_copy(
        update={"extraction_id": extraction_digest(replacement_candidate)}
    )

    node_path = (
        f"projects/{hashlib.sha256(NODE.project_id.encode()).hexdigest()}/nodes/"
        f"{hashlib.sha256(NODE.node_id.encode()).hexdigest()}.yaml"
    )
    first_revision = repository.publish_files(
        expected_head=repository.head(),
        files={
            f"extractions/{prior.extraction_id}.json": prior.model_dump_json().encode(),
            node_path: yaml.safe_dump(NODE.model_dump(mode="json")).encode(),
        },
        message="Publish a cited node using the original parser extraction",
        index_directory=tmp_path / "publishing",
    )
    replacement_revision = repository.publish_files(
        expected_head=repository.head(),
        files={
            f"extractions/{replacement.extraction_id}.json": replacement.model_dump_json().encode()
        },
        message="Add replacement parser extraction without deleting history",
        index_directory=tmp_path / "publishing",
    )

    context = LocalContextProvider(repository, tmp_path / "control/registry.json")()
    assert context.extractions[prior.extraction_id] == prior
    assert context.extractions[replacement.extraction_id] == replacement
    assert context.existing_nodes[NODE.node_id] == NODE
    assert (
        context.existing_nodes[NODE.node_id].claims[0].evidence[0].extraction_id
        == prior.extraction_id
    )
    assert (
        repository.read_file(first_revision, f"extractions/{prior.extraction_id}.json") is not None
    )
    assert (
        repository.read_file(replacement_revision, f"extractions/{prior.extraction_id}.json")
        is not None
    )


def test_disposable_registry_seed_loads_with_versioned_contracts(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setenv("GEWU_DATA_ROOT", str(tmp_path))
    result = subprocess.run(
        [sys.executable, "scripts/seed_test_registry.py"],
        check=False,
        capture_output=True,
        text=True,
    )
    assert result.returncode == 0, result.stderr
    repository = GitCanonicalRepository(tmp_path / "knowledge.git")
    context = LocalContextProvider(repository, tmp_path / "control/registry.json")()
    assert set(context.projects) == {"dome-performance", "interactive-video"}
    assert set(context.policies) == {"dome-pilot-policy", "video-pilot-policy", "shared"}