ATOCore/tests/test_project_state.py

"""Tests for Trusted Project State."""

import pytest

from atocore.context.project_state import (
    CATEGORIES,
    ensure_project,
    format_project_state,
    get_state,
    init_project_state_schema,
    invalidate_state,
    set_state,
)
from atocore.models.database import init_db


@pytest.fixture(autouse=True)
def setup_db(tmp_data_dir):
    """Initialize DB and project state schema for every test."""
    init_db()
    init_project_state_schema()


def test_ensure_project_creates():
    """Test creating a new project."""
    pid = ensure_project("test-project", "A test project")
    assert pid
    # Second call returns same ID
    pid2 = ensure_project("test-project")
    assert pid == pid2


def test_set_state_creates_entry():
    """Test creating a project state entry."""
    entry = set_state("myproject", "status", "phase", "Phase 0.5 — PoC complete")
    assert entry.category == "status"
    assert entry.key == "phase"
    assert entry.value == "Phase 0.5 — PoC complete"
    assert entry.status == "active"


def test_set_state_upserts():
    """Test that setting same key updates the value."""
    set_state("myproject", "status", "phase", "Phase 0")
    entry = set_state("myproject", "status", "phase", "Phase 1")
    assert entry.value == "Phase 1"

    # Only one entry should exist
    entries = get_state("myproject", category="status")
    assert len(entries) == 1
    assert entries[0].value == "Phase 1"


def test_set_state_invalid_category():
    """Test that invalid category raises ValueError."""
    with pytest.raises(ValueError, match="Invalid category"):
        set_state("myproject", "invalid_category", "key", "value")


def test_set_state_validates_confidence():
    """Project-state confidence should stay within the documented range."""
    with pytest.raises(ValueError, match="Confidence must be between 0.0 and 1.0"):
        set_state("myproject", "status", "phase", "Phase 1", confidence=1.2)


def test_get_state_all():
    """Test getting all state entries for a project."""
    set_state("proj", "status", "phase", "Phase 1")
    set_state("proj", "decision", "database", "SQLite for v1")
    set_state("proj", "requirement", "latency", "<2 seconds")

    entries = get_state("proj")
    assert len(entries) == 3
    categories = {e.category for e in entries}
    assert categories == {"status", "decision", "requirement"}


def test_get_state_by_category():
    """Test filtering by category."""
    set_state("proj", "status", "phase", "Phase 1")
    set_state("proj", "decision", "database", "SQLite")
    set_state("proj", "decision", "vectordb", "ChromaDB")

    entries = get_state("proj", category="decision")
    assert len(entries) == 2
    assert all(e.category == "decision" for e in entries)


def test_get_state_nonexistent_project():
    """Test getting state for a project that doesn't exist."""
    entries = get_state("nonexistent")
    assert entries == []


def test_invalidate_state():
    """Test marking a state entry as superseded."""
    set_state("invalidate-test", "decision", "approach", "monolith")
    success = invalidate_state("invalidate-test", "decision", "approach")
    assert success

    # Active entries should be empty
    entries = get_state("invalidate-test", active_only=True)
    assert len(entries) == 0

    # But entry still exists if we include inactive
    entries = get_state("invalidate-test", active_only=False)
    assert len(entries) == 1
    assert entries[0].status == "superseded"


def test_invalidate_nonexistent():
    """Test invalidating a nonexistent entry."""
    success = invalidate_state("proj", "decision", "nonexistent")
    assert not success


def test_format_project_state():
    """Test formatting state entries for context injection."""
    set_state("proj", "status", "phase", "Phase 1")
    set_state("proj", "decision", "database", "SQLite", source="Build Spec V1")
    entries = get_state("proj")

    formatted = format_project_state(entries)
    assert "--- Trusted Project State ---" in formatted
    assert "--- End Project State ---" in formatted
    assert "phase: Phase 1" in formatted
    assert "database: SQLite" in formatted
    assert "(source: Build Spec V1)" in formatted


def test_format_empty():
    """Test formatting empty state."""
    assert format_project_state([]) == ""


# --- Alias canonicalization regression tests --------------------------------


def test_set_state_canonicalizes_alias(project_registry):
    """Writing state via an alias should land under the canonical project id.

    Regression for codex's P1 finding: previously /project/state with
    project="p05" created a separate alias row that later context builds
    (which canonicalize the hint) would never see.
    """
    project_registry(("p05-interferometer", ["p05", "interferometer"]))

    set_state("p05", "status", "next_focus", "Wave 2 ingestion")

    # The state must be reachable via every alias AND the canonical id
    via_alias = get_state("p05")
    via_canonical = get_state("p05-interferometer")
    via_other_alias = get_state("interferometer")

    assert len(via_alias) == 1
    assert len(via_canonical) == 1
    assert len(via_other_alias) == 1
    # All three reads return the same row id (no fragmented duplicates)
    assert via_alias[0].id == via_canonical[0].id == via_other_alias[0].id
    assert via_canonical[0].value == "Wave 2 ingestion"


def test_get_state_canonicalizes_alias_after_canonical_write(project_registry):
    """Reading via an alias should find state written under the canonical id."""
    project_registry(("p04-gigabit", ["p04", "gigabit"]))

    set_state("p04-gigabit", "status", "phase", "Phase 1 baseline")
    via_alias = get_state("gigabit")

    assert len(via_alias) == 1
    assert via_alias[0].value == "Phase 1 baseline"


def test_invalidate_state_canonicalizes_alias(project_registry):
    """Invalidating via an alias should hit the canonical row."""
    project_registry(("p06-polisher", ["p06", "polisher"]))

    set_state("p06-polisher", "decision", "frame", "kinematic mounts")
    success = invalidate_state("polisher", "decision", "frame")

    assert success is True
    active = get_state("p06-polisher")
    assert len(active) == 0


def test_unregistered_project_state_still_works(project_registry):
    """Hand-curated state for an unregistered project must still round-trip.

    Backwards compatibility with state created before the project
    registry existed: resolve_project_name returns the input unchanged
    when the registry has no record, so the raw name is used as-is.
    """
    project_registry()  # empty registry

    set_state("orphan-project", "status", "phase", "Standalone")
    entries = get_state("orphan-project")
    assert len(entries) == 1
    assert entries[0].value == "Standalone"