test_memory_scenarios.py

from __future__ import annotations

import json
import os
import subprocess
import sys
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from dev_harness import config, safety
from dev_harness import memory_scenarios

REPO_ROOT = Path(__file__).resolve().parents[3]


def _env(tmp_path: Path) -> dict[str, str]:
    env = os.environ.copy()
    env["PROVIDER_MODE"] = "offline"
    env["OMI_LOCAL_STATE_ROOT"] = str(tmp_path / "state")
    pythonpath = [str(REPO_ROOT / "scripts" / "dev-harness")]
    if existing := env.get("PYTHONPATH"):
        pythonpath.append(existing)
    env["PYTHONPATH"] = os.pathsep.join(pythonpath)
    return env


def test_all_memory_scenarios_import_and_validate() -> None:
    memory_scenarios.validate_all_scenarios()
    names = {scenario.scenario_id for scenario in memory_scenarios.list_scenarios()}
    assert {
        "happy_path",
        "default_off",
        "kill_switch",
        "malformed_cursor",
        "stale_short_exclusion",
        "archive_default_exclusion",
        "cross_user_isolation",
    }.issubset(names)

    happy = memory_scenarios.get_scenario("happy_path")
    assert {user.uid for user in happy.users} >= {
        "local_default_user",
        "alice",
        "bob",
        "omi-local-emulator-chat-first-enabled-v1",
        "omi-local-emulator-chat-first-disabled-v1",
    }
    assert happy.selected_user == "alice"
    assert happy.report_metadata.evidence_class == "LOCAL_EMULATOR_DEV"
    assert happy.report_metadata.activation_eligible is False
    assert all(case.expected_no_write for case in happy.request_cases)


def test_memory_scenario_cli_listing_json(tmp_path: Path) -> None:
    env = _env(tmp_path)
    result = subprocess.run(
        [sys.executable, "-m", "dev_harness.memory_scenarios", "list", "--json"],
        cwd=REPO_ROOT,
        env=env,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        timeout=30,
    )
    assert result.returncode == 0, result.stdout
    payload = json.loads(result.stdout)
    assert any(item["scenario_id"] == "happy_path" for item in payload)
    assert all("LOCAL_EMULATOR_DEV" not in item.get("description", "") for item in payload)


def test_seed_manifest_generation_is_dry_run_without_emulators(tmp_path: Path) -> None:
    env = _env(tmp_path)
    cfg = config.load_config(REPO_ROOT, env=env, create_layout=True)

    manifest = memory_scenarios.seed_scenario("happy_path", cfg, dry_run=True)

    assert manifest.scenario_id == "happy_path"
    assert manifest.dry_run is True
    assert manifest.applied is False
    assert manifest.report_metadata.evidence_class == "LOCAL_EMULATOR_DEV"
    assert manifest.report_metadata.activation_eligible is False
    assert any(op.kind == "firestore" and op.target == "memory_control/global_read_gate" for op in manifest.operations)
    manifest_path = cfg.layout.process_manifest.parent / "memory-scenario-happy_path-seed.json"
    assert manifest_path.is_file()
    saved = json.loads(manifest_path.read_text(encoding="utf-8"))
    assert saved["report_metadata"]["watermark"] == "NOT_ACTIVATION_EVIDENCE"


def test_reset_manifest_is_idempotent_in_temp_state(tmp_path: Path) -> None:
    env = _env(tmp_path)
    cfg = config.load_config(REPO_ROOT, env=env, create_layout=True)

    first = memory_scenarios.reset_scenario("happy_path", cfg, dry_run=True)
    second = memory_scenarios.reset_scenario("happy_path", cfg, dry_run=True)

    assert first.dry_run is True
    assert second.dry_run is True
    assert [op.target for op in first.operations] == [op.target for op in second.operations]
    safety.read_and_validate_sentinel(cfg.layout.state_root, repo_root=REPO_ROOT, instance="default")


def test_fixtures_cannot_choose_evidence_labels() -> None:
    for scenario in memory_scenarios.list_scenarios():
        assert scenario.report_metadata.evidence_class == "LOCAL_EMULATOR_DEV"
        assert scenario.report_metadata.activation_eligible is False
        for seed in (*scenario.profile_seed, *scenario.firestore_seed):
            assert "evidence_class" not in seed.data
            assert "activation_eligible" not in seed.data
        assert scenario.local_flags["LOCAL_EMULATOR_DEV"] is True
        assert scenario.local_flags["activation_eligible"] is False


def test_auth_live_seed_prefers_the_admin_api_for_deterministic_uids(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    env = _env(tmp_path)
    cfg = config.load_config(REPO_ROOT, env=env, create_layout=True)
    calls: list[tuple[str, str, dict[str, object] | None]] = []

    def fake_request(method: str, url: str, payload: dict[str, object] | None = None) -> tuple[int, str]:
        calls.append((method, url, dict(payload) if payload is not None else None))
        return 200, '{}'

    monkeypatch.setattr(memory_scenarios, '_request_json', fake_request)
    admin_calls = []
    monkeypatch.setattr(
        memory_scenarios,
        '_apply_auth_admin_sdk',
        lambda received_cfg, op: admin_calls.append((received_cfg, op)) or True,
    )
    op = memory_scenarios.SeedOperation(
        kind='auth',
        action='upsert',
        target='alice',
        payload={'localId': 'alice', 'email': 'alice@local.omi.invalid', 'password': 'local-test'},
    )

    memory_scenarios._apply_operation(cfg, op)

    assert calls == []
    assert admin_calls == [(cfg, op)]


def test_happy_path_has_rich_default_memory_fixture_set() -> None:
    ctx = memory_scenarios._clock()
    default_ids = memory_scenarios._alice_default_memory_ids(ctx)
    assert len(default_ids) == 24
    assert ctx.ids["alice_short_active"] in default_ids
    assert ctx.ids["alice_long_edu"] in default_ids
    assert ctx.ids["alice_short_stale"] not in default_ids
    assert ctx.ids["alice_archive"] not in default_ids


def test_happy_path_sourced_seeds_include_synthetic_local_qa_evidence() -> None:
    """Every sourced happy_path memory has matching synthetic local-QA evidence docs."""
    happy = memory_scenarios.get_scenario("happy_path")
    uid = memory_scenarios.ALICE_USER_ID
    evidence_by_path = {
        seed.path: seed for seed in happy.firestore_seed if seed.path.startswith(f"users/{uid}/memory_evidence/")
    }

    memory_seeds = [
        seed
        for seed in happy.firestore_seed
        if seed.path.startswith(f"users/{uid}/memory_items/") and seed.data.get("evidence")
    ]
    assert memory_seeds, "expected sourced memory fixtures with embedded evidence"

    short_term_ids: list[str] = []
    long_term_ids: list[str] = []
    for memory_seed in memory_seeds:
        tier = memory_seed.data.get("tier")
        embedded = memory_seed.data["evidence"]
        assert isinstance(embedded, list) and embedded, f"missing embedded evidence on {memory_seed.path}"
        ev = embedded[0]
        evidence_id = ev["evidence_id"]
        source_id = ev["source_id"]
        content = memory_seed.data["content"]

        evidence_path = f"users/{uid}/memory_evidence/{evidence_id}"
        evidence_seed = evidence_by_path.get(evidence_path)
        assert evidence_seed is not None, f"missing memory_evidence doc for {memory_seed.path}"
        assert evidence_seed.data["evidence_id"] == evidence_id
        assert evidence_seed.data["source_id"] == source_id
        assert evidence_seed.data["quote_refs"][0]["quote"] == content
        assert evidence_seed.data["quote_refs"][0]["source_id"] == source_id

        memory_id = memory_seed.path.rsplit("/", 1)[-1]
        if tier == "short_term":
            short_term_ids.append(memory_id)
        elif tier == "long_term":
            long_term_ids.append(memory_id)

    assert "mem_alice_short_active_030" in short_term_ids
    assert "mem_alice_short_demo_030" in short_term_ids
    assert "mem_alice_long_030" in long_term_ids
    assert "mem_alice_long_edu_030" in long_term_ids
    assert len(short_term_ids) >= 6
    assert len(long_term_ids) >= 15


def test_remap_firestore_seed_to_auth_uid() -> None:
    seed = memory_scenarios.FirestoreSeed(
        path="users/alice/memory_items/mem_alice_long_030",
        protected=True,
        data={"uid": "alice", "memory_id": "mem_alice_long_030"},
    )
    remapped = memory_scenarios._remap_firestore_seed(seed, {"alice": "auth-uid-123"})
    assert remapped.path == "users/auth-uid-123/memory_items/mem_alice_long_030"
    assert remapped.data["uid"] == "auth-uid-123"


def test_entrypoint_seed_dry_run_outputs_manifest(tmp_path: Path) -> None:
    env = _env(tmp_path)
    result = subprocess.run(
        [sys.executable, "scripts/dev-harness/seed-memory-scenario.py", "happy_path", "--dry-run"],
        cwd=REPO_ROOT,
        env=env,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        timeout=30,
    )
    assert result.returncode == 0, result.stdout
    payload = json.loads(result.stdout)
    assert payload["scenario_id"] == "happy_path"
    assert payload["dry_run"] is True
    assert payload["report_metadata"]["evidence_class"] == "LOCAL_EMULATOR_DEV"