opencode-config/tests/test_memory_tools_e2e.py

import json
import os
import shutil
import subprocess
import time
import uuid
from pathlib import Path

import pytest

pytestmark = pytest.mark.skipif(
    not os.environ.get("RUN_LIVE"),
    reason="requires real OpenRouter API",
)

REPO_ROOT = Path(__file__).resolve().parent.parent
UNIQUE_TERM = "zzuniqtestterm42"
SYNONYM = "unique test concept"


def _rg_bin() -> str:
    rg = shutil.which("rg") or "/root/.cache/opencode/bin/rg"
    if not Path(rg).exists():
        pytest.skip("ripgrep not available")
    return rg


def _rg_search(pattern: str, memory_dir: Path) -> list[str]:
    rg = _rg_bin()
    result = subprocess.run(
        [
            rg,
            "-il",
            "--glob",
            "*.md",
            "--glob",
            "!.git",
            "--glob",
            "!.rag",
            "-e",
            pattern,
            str(memory_dir),
        ],
        capture_output=True,
        text=True,
        timeout=30,
        check=False,
    )
    if result.returncode != 0:
        return []
    return [line.strip() for line in result.stdout.strip().splitlines() if line.strip()]


def _semantic_search(
    query: str, index_dir: Path, env: dict[str, str] | None = None, timeout: int = 120
) -> list[dict]:
    run_env = env if env is not None else dict(os.environ)
    result = subprocess.run(
        [
            "python3",
            "-m",
            "src.memory",
            "search",
            query,
            "-i",
            str(index_dir),
            "-k",
            "15",
            "--json",
        ],
        capture_output=True,
        text=True,
        timeout=timeout,
        cwd=str(REPO_ROOT),
        env=run_env,
        check=False,
    )
    if result.returncode != 0:
        return []
    try:
        parsed = json.loads(result.stdout)
    except json.JSONDecodeError:
        return []
    return parsed if isinstance(parsed, list) else []


def _reindex(
    memory_dir: Path, index_dir: Path, env: dict[str, str] | None = None, timeout: int = 180
) -> None:
    index_dir.mkdir(parents=True, exist_ok=True)
    run_env = env if env is not None else dict(os.environ)
    result = subprocess.run(
        [
            "python3",
            "-m",
            "src.memory",
            "index",
            str(memory_dir),
            "-o",
            str(index_dir),
        ],
        capture_output=True,
        text=True,
        timeout=timeout,
        cwd=str(REPO_ROOT),
        env=run_env,
        check=False,
    )
    assert result.returncode == 0, (
        f"reindex failed:\nstdout: {result.stdout}\nstderr: {result.stderr}"
    )
    log_path = index_dir / "reindex.log"
    deadline = time.monotonic() + 30
    while time.monotonic() < deadline:
        if log_path.exists():
            content = log_path.read_text(encoding="utf-8")
            if "done:" in content or "failed:" in content:
                return
        time.sleep(1)
    pytest.fail("reindex did not write done/failed to log in time")


def _make_memory_file(memory_dir: Path, term: str, synonym: str = "") -> Path:
    uid = uuid.uuid4().hex[:8]
    tech_dir = memory_dir / "technical"
    tech_dir.mkdir(parents=True, exist_ok=True)
    fname = tech_dir / f"e2e-{uid}.md"
    body = (
        f"---\ntitle: E2E test {uid}\ntags: [e2e, test]\n"
        f"summary: E2E coverage probe\n---\n\n# E2E test\n\n"
        f"Contains {term} as exact marker. {synonym}\n"
    )
    fname.write_text(body, encoding="utf-8")
    return fname


def _make_dummy_file(memory_dir: Path) -> Path:
    tech_dir = memory_dir / "technical"
    tech_dir.mkdir(parents=True, exist_ok=True)
    fname = tech_dir / "dummy.md"
    fname.write_text(
        "---\ntitle: dummy\n---\n\n# Dummy\n\nfiller content for reindex stability\n",
        encoding="utf-8",
    )
    return fname


def test_zero_config_keyword_only(tmp_path: Path) -> None:
    """Scenario 1: keyword works without OPENAI_BASE_URL, semantic returns []."""
    memory_dir = tmp_path / "memory"
    index_dir = memory_dir / ".rag"
    fname = _make_memory_file(memory_dir, UNIQUE_TERM)
    rel_path = fname.relative_to(memory_dir).as_posix()

    env_no_base = {k: v for k, v in os.environ.items() if k != "OPENAI_BASE_URL"}

    semantic = _semantic_search(UNIQUE_TERM, index_dir, env=env_no_base)
    keyword = _rg_search(UNIQUE_TERM, memory_dir)

    assert semantic == [], f"semantic should return [] without OPENAI_BASE_URL, got {semantic}"
    assert any(rel_path in p for p in keyword), f"keyword should find {rel_path}, got {keyword}"


@pytest.mark.timeout(180)
def test_full_hybrid_semantic_and_keyword(tmp_path: Path) -> None:
    """Scenario 2: semantic finds by synonym, keyword finds by exact term."""
    memory_dir = tmp_path / "memory"
    index_dir = memory_dir / ".rag"
    fname = _make_memory_file(memory_dir, UNIQUE_TERM, SYNONYM)
    _make_dummy_file(memory_dir)
    rel_path = fname.relative_to(memory_dir).as_posix()

    _reindex(memory_dir, index_dir)

    semantic_term = _semantic_search(UNIQUE_TERM, index_dir)
    semantic_synonym = _semantic_search(SYNONYM, index_dir)
    keyword = _rg_search(UNIQUE_TERM, memory_dir)

    sem_sources_term = {r["source"] for r in semantic_term}
    sem_sources_syn = {r["source"] for r in semantic_synonym}
    assert rel_path in sem_sources_term, (
        f"semantic should find by exact term, got {sem_sources_term}"
    )
    assert rel_path in sem_sources_syn, f"semantic should find by synonym, got {sem_sources_syn}"
    assert any(rel_path in p for p in keyword), f"keyword should find {rel_path}, got {keyword}"


@pytest.mark.timeout(300)
def test_fallback_openrouter_down(tmp_path: Path) -> None:
    """Scenario 3: keyword works when OpenRouter returns 401."""
    memory_dir = tmp_path / "memory"
    index_dir = memory_dir / ".rag"
    fname = _make_memory_file(memory_dir, UNIQUE_TERM, SYNONYM)
    _make_dummy_file(memory_dir)
    rel_path = fname.relative_to(memory_dir).as_posix()

    _reindex(memory_dir, index_dir)

    env_invalid_key = dict(os.environ)
    env_invalid_key["OPENAI_API_KEY"] = "invalid_key"

    semantic = _semantic_search(UNIQUE_TERM, index_dir, env=env_invalid_key)
    keyword = _rg_search(UNIQUE_TERM, memory_dir)

    assert semantic == [], f"semantic should return [] with invalid key, got {semantic}"
    assert any(rel_path in p for p in keyword), (
        f"keyword fallback should find {rel_path}, got {keyword}"
    )


@pytest.mark.timeout(180)
def test_file_deletion(tmp_path: Path) -> None:
    """Scenario 4: after deletion + reindex, search returns []."""
    memory_dir = tmp_path / "memory"
    index_dir = memory_dir / ".rag"
    fname = _make_memory_file(memory_dir, UNIQUE_TERM, SYNONYM)
    _make_dummy_file(memory_dir)
    rel_path = fname.relative_to(memory_dir).as_posix()

    _reindex(memory_dir, index_dir)

    fname.unlink()
    _reindex(memory_dir, index_dir)

    semantic = _semantic_search(UNIQUE_TERM, index_dir)
    keyword = _rg_search(UNIQUE_TERM, memory_dir)

    sem_sources = {r["source"] for r in semantic}
    assert rel_path not in sem_sources, (
        f"deleted file should not be in semantic results, got {sem_sources}"
    )
    assert not any(rel_path in p for p in keyword), (
        f"deleted file should not be in keyword results, got {keyword}"
    )