conftest.py

import os
import subprocess
import sys
import tempfile
from pathlib import Path

import pytest
from dotenv import load_dotenv

load_dotenv()  # Load .env if present, so OPENROUTER_API_KEY is available in tests


def pytest_addoption(parser):
    """Cursor-audit Item 2 + Phase 1.9 (v4):

    - ``--snapshot-update`` gates visual snapshot regeneration. Running the
      suite with ``--snapshot-update`` rewrites baseline screenshots under
      ``platform/tests/snapshots/v4/dashboard/`` instead of asserting
      pixel equality.

    - ``--run-slow`` opts in to the long-running tests gated by
      ``@pytest.mark.slow``. Default CI (PRs) runs without this flag and
      skips them; nightly / local full-suite runs pass ``--run-slow``.
    """
    parser.addoption(
        "--snapshot-update",
        action="store_true",
        default=False,
        help="Regenerate dashboard visual snapshot baselines (Phase 1.9).",
    )
    parser.addoption(
        "--run-slow",
        action="store_true",
        default=False,
        help=(
            "Include tests marked @pytest.mark.slow. Default-off so PR "
            "CI stays under the ~4 minute budget; nightly / local full "
            "runs pass --run-slow to exercise the full suite."
        ),
    )
    parser.addoption(
        "--run-live",
        action="store_true",
        default=False,
        help="Include tests marked @pytest.mark.live (require real network/API access).",
    )


def _docker_available() -> bool:
    """Return True if Docker daemon is reachable."""
    try:
        result = subprocess.run(
            ["docker", "info"],
            capture_output=True,
            timeout=5,
        )
        return result.returncode == 0
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return False


_DOCKER_OK: bool | None = None


def pytest_collection_modifyitems(config, items):
    """Auto-skip tests based on markers and available infrastructure.

    - ``@pytest.mark.slow``: skipped unless ``--run-slow``
    - ``@pytest.mark.live``: skipped unless ``--run-live``
    - ``@pytest.mark.docker``: skipped unless Docker daemon is reachable
    """
    global _DOCKER_OK  # noqa: PLW0603
    run_slow = config.getoption("--run-slow")
    run_live = config.getoption("--run-live")

    if _DOCKER_OK is None:
        _DOCKER_OK = _docker_available()

    skip_slow = pytest.mark.skip(reason="need --run-slow option to run")
    skip_live = pytest.mark.skip(reason="need --run-live option to run")
    skip_docker = pytest.mark.skip(reason="Docker daemon not available")

    for item in items:
        if not run_slow and "slow" in item.keywords:
            item.add_marker(skip_slow)
        if not run_live and "live" in item.keywords:
            item.add_marker(skip_live)
        if not _DOCKER_OK and "docker" in item.keywords:
            item.add_marker(skip_docker)


# ---------------------------------------------------------------------------
# Hypothesis CI profile
#
# Cleanup-audit Item 2: property-based tests can balloon default CI
# wall-time. We register two profiles and load ``ci`` by default so
# property tests that do NOT override @settings(max_examples=...) run
# with a compact budget on PRs. Nightly (``HYPOTHESIS_PROFILE=nightly``)
# restores full coverage. Tests that *explicitly* pin ``max_examples``
# via @settings keep their per-test setting; the profile only moves the
# default floor.
# ---------------------------------------------------------------------------
try:
    from hypothesis import HealthCheck
    from hypothesis import settings as _hyp_settings

    _hyp_settings.register_profile(
        "ci",
        max_examples=10,
        deadline=None,
        suppress_health_check=[HealthCheck.too_slow],
    )
    _hyp_settings.register_profile(
        "nightly",
        max_examples=40,
        deadline=None,
        suppress_health_check=[HealthCheck.too_slow],
    )
    _hyp_settings.load_profile(os.environ.get("HYPOTHESIS_PROFILE", "ci"))
except ImportError:
    pass

# Add forge root so `studio.*` packages are importable alongside `forge.*`.
# Also add ``<root>/platform`` so ``from forge.X import Y`` resolves --
# previously every test file under ``platform/tests/**`` repeated a
# 4-liner ``sys.path.insert(0, str(<repo>/platform))``; centralising
# both insertions here is the Round-5 F2 cleanup.
_forge_root = Path(__file__).resolve().parent.parent.parent
_platform_root = _forge_root / "platform"
for _p in (_forge_root, _platform_root):
    _ps = str(_p)
    if _ps not in sys.path:
        sys.path.insert(0, _ps)


@pytest.fixture(autouse=True)
def _resolve_config_path(monkeypatch):
    """Ensure DEFAULT_CONFIG_PATH resolves regardless of pytest's cwd.

    DEFAULT_CONFIG_PATH is relative ('studio/config.yaml') which only works
    when cwd is the repo root. Pytest runs from platform/, so patch it at
    every import site so tests that trigger config loading don't fail with
    ConfigError.
    """
    abs_config = str(_forge_root / "studio" / "config.yaml")
    monkeypatch.setattr("forge.defaults.DEFAULT_CONFIG_PATH", abs_config)
    try:
        import forge.agents.registry
        monkeypatch.setattr(forge.agents.registry, "DEFAULT_CONFIG_PATH", abs_config)
    except (ImportError, AttributeError):
        pass


@pytest.fixture(autouse=True)
def _default_lsp_disabled_in_tests(monkeypatch):
    """R9.C Phase A.0 (safety): default ``FORGE_LSP_ENABLED=0`` in tests.

    Phase A wires the MCP server's ``_get_lsp_bridge()`` factory to
    auto-spawn ``pyright-langserver`` when it's on PATH. On dev boxes
    where pyright happens to be installed, that would silently turn
    every existing ``_get_references`` / ``_get_definition`` /
    ``_get_type`` MCP test into a subprocess-spawning test -- making
    them slow and flaky.

    We pin the env var to ``"0"`` (kill-switch) for the entire test
    suite, so the factory short-circuits to ``None`` and the MCP
    handlers fall back to tree-sitter unconditionally. Tests that
    *want* to exercise the on-path (Phase A.5) explicitly
    ``monkeypatch.delenv("FORGE_LSP_ENABLED", raising=False)`` or
    set it to ``"1"`` after this fixture runs.
    """
    monkeypatch.setenv("FORGE_LSP_ENABLED", "0")
    yield


@pytest.fixture(autouse=True)
def _isolate_forge_state(tmp_path_factory, monkeypatch):
    """v4.5.2 Phase 3 - redirect every relative ``.forge/*.db`` default
    into the per-test ``tmp_path`` so unit tests never create rogue
    ``platform/.forge/`` artefacts or pollute the shipped production
    databases.

    Target classes and their defaults (all verified via ``grep``):

    - ``CostTracker`` (``forge.dashboard.costs``) - ``.forge/costs.db``
    - ``StructuredStore`` (``forge.memory.store``) - ``.forge/metrics.db``
    - ``MetricsTracker`` (``forge.meta.metrics``) - ``.forge/metrics.db``
    - ``TrustRatchet`` (``forge.trust.ratchet``) - ``.forge/trust.db``
    - ``compute_slo_snapshot`` default param - ``.forge/metrics.db``
    - ``calibration.default_samples_path`` / ``default_multiplier_path``

    Plus ``monkeypatch.chdir(tmp_path)`` so any remaining bare
    ``.forge/*.db`` literal resolves inside the tmp dir instead of
    the repo root.
    """
    iso_root = tmp_path_factory.mktemp("forge_state_iso")
    (iso_root / ".forge").mkdir(parents=True, exist_ok=True)

    # Default-arg monkeypatching via functools.partial-style wrappers
    # on each constructor.
    def _redirect(db_path: str, db_name: str) -> str:
        """Rewrite any relative ``.forge/<db>`` path into ``iso_root``.

        Absolute paths are left alone so tests that explicitly point
        at a tmp path still work. This handles the case where a
        caller passes ``db_path=".forge/trust.db"`` explicitly
        (``studio.workflows.orchestrator.merge_node`` does this),
        which would otherwise bypass our default-arg monkeypatch.
        """
        from pathlib import Path as _P

        p = str(db_path)
        if p.startswith(str(iso_root)):
            return p
        pp = _P(p)
        if pp.is_absolute():
            return p
        if p.startswith(".forge/") or p.endswith(f"/{db_name}") or p == f".forge/{db_name}":
            return str(iso_root / ".forge" / db_name)
        return p

    try:
        from forge.dashboard import costs as _costs

        _orig_cost_init = _costs.CostTracker.__init__

        def _cost_init(self, db_path: str = ".forge/costs.db"):
            return _orig_cost_init(self, db_path=_redirect(db_path, "costs.db"))

        monkeypatch.setattr(_costs.CostTracker, "__init__", _cost_init)
    except Exception:
        pass

    try:
        from forge.memory import store as _store_mod

        _orig_store_init = _store_mod.StructuredStore.__init__

        def _store_init(self, db_path: str = ".forge/metrics.db"):
            return _orig_store_init(self, db_path=_redirect(db_path, "metrics.db"))

        monkeypatch.setattr(_store_mod.StructuredStore, "__init__", _store_init)
    except Exception:
        pass

    try:
        from forge.meta import metrics as _metrics_mod

        _orig_metrics_init = _metrics_mod.MetricsTracker.__init__

        def _metrics_init(self, db_path: str = ".forge/metrics.db"):
            return _orig_metrics_init(self, db_path=_redirect(db_path, "metrics.db"))

        monkeypatch.setattr(_metrics_mod.MetricsTracker, "__init__", _metrics_init)
    except Exception:
        pass

    try:
        from forge.trust import ratchet as _trust_mod

        _orig_trust_init = _trust_mod.TrustRatchet.__init__

        def _trust_init(
            self,
            db_path: str = ".forge/trust.db",
            promote_threshold: int = 20,
        ):
            return _orig_trust_init(
                self,
                db_path=_redirect(db_path, "trust.db"),
                promote_threshold=promote_threshold,
            )

        monkeypatch.setattr(_trust_mod.TrustRatchet, "__init__", _trust_init)
    except Exception:
        pass

    try:
        from forge.meta import slo as _slo_mod

        _orig_slo = _slo_mod.compute_slo_snapshot

        def _slo(db_path=str(iso_root / ".forge" / "metrics.db"), *a, **kw):
            return _orig_slo(db_path, *a, **kw)

        monkeypatch.setattr(_slo_mod, "compute_slo_snapshot", _slo)
    except Exception:
        pass

    try:
        from forge.engine import calibration as _calib

        monkeypatch.setattr(
            _calib,
            "default_samples_path",
            lambda root=None: iso_root / ".forge" / "calibration.jsonl",
        )
        monkeypatch.setattr(
            _calib,
            "default_multiplier_path",
            lambda root=None: iso_root / ".forge" / "calibration.json",
        )
    except Exception:
        pass

    try:
        from forge.engine import decomposer_memory as _dm

        def _iso_default_lessons_read_path(root=None):
            """R10 B.11a: honor ``FORGE_DECOMPOSER_LESSONS_PATH`` even
            inside the iso-state fixture so cassette E2E tests can
            point the orchestrator at a captured-snapshot sidecar
            (without that override the autouse iso fixture would
            redirect every read to a fresh empty tmp file, drifting
            the cassette prompt hash). When the env var is unset,
            fall back to the prior iso-root behavior so nothing else
            in the test suite changes.
            """
            import os as _os

            env = _os.environ.get("FORGE_DECOMPOSER_LESSONS_PATH", "")
            if env.strip():
                return Path(env)
            return iso_root / ".forge" / "decomposer_lessons.jsonl"

        def _iso_default_lessons_write_path(root=None):
            """R10 P0.3: writes always go to iso-root, never to the
            env-pointed sidecar. This isolates test-time writes from
            the captured-snapshot sidecars that ``FORGE_DECOMPOSER_LESSONS_PATH``
            points at, preserving the read/write split that the
            production code now enforces.
            """
            return iso_root / ".forge" / "decomposer_lessons.jsonl"

        monkeypatch.setattr(
            _dm,
            "default_lessons_read_path",
            _iso_default_lessons_read_path,
        )
        monkeypatch.setattr(
            _dm,
            "default_lessons_write_path",
            _iso_default_lessons_write_path,
        )
        # Backward-compat alias kept for legacy callers / tests that
        # still reference the un-directional name. It now mirrors the
        # READ path semantics (env-aware) since pre-P0.3 code expected
        # exactly that.
        monkeypatch.setattr(
            _dm,
            "default_lessons_path",
            _iso_default_lessons_read_path,
        )
    except Exception:
        pass

    try:
        import studio.workflows.orchestrator as _orch_mod
        import studio.workflows._helpers as _helpers_mod

        (iso_root / "docs").mkdir(parents=True, exist_ok=True)
        fake_taskboard = iso_root / "docs" / "TASKBOARD.md"
        if not fake_taskboard.exists():
            fake_taskboard.write_text(
                "# Test TASKBOARD (isolated)\n\n## Active\n\n## Completed\n\n"
                "| ID | Description |\n|------|-------------|\n",
                encoding="utf-8",
            )

        monkeypatch.setattr(_orch_mod, "_REPO_ROOT", iso_root, raising=False)
        monkeypatch.setattr(_helpers_mod, "REPO_ROOT", iso_root, raising=False)
    except Exception:
        pass

    # NOTE: we deliberately do NOT ``monkeypatch.chdir`` here. Several
    # tests (MCP stdio subprocess, E2E pipeline) rely on the starting
    # cwd to locate ``studio/config.yaml`` or to launch subprocesses
    # with the repo-root PYTHONPATH. Defence-in-depth for stray bare
    # ``.forge/*.db`` literals comes from the per-class ``__init__``
    # monkeypatches above rather than a global chdir.

    yield iso_root


@pytest.fixture(autouse=True)
def _isolate_forge_logging(tmp_path):
    """Redirect log_event to a temp dir so tests don't pollute production logs.

    Cleanup-audit Item 1: the async sink (QueueHandler + background
    QueueListener) must be torn down between tests so (a) the next
    test's ``LOG_DIR`` override is respected and (b) we don't leak
    listener threads across the suite. ``_reset_logging_for_tests``
    handles both.
    """
    import forge.logging as _flog

    original_log_dir = _flog.LOG_DIR
    _flog.LOG_DIR = tmp_path / "test-logs"
    _flog.LOG_DIR.mkdir(parents=True, exist_ok=True)
    _flog._reset_logging_for_tests()

    yield

    _flog._reset_logging_for_tests()
    _flog.LOG_DIR = original_log_dir


@pytest.fixture(autouse=True)
def _reset_orchestrator_caches():
    """Reset module-level caches in _helpers, tdd, and registry between tests."""
    try:
        from forge.agents.registry import (
            _reset_config_cache,
            _reset_scope_check,
        )

        _reset_config_cache()
        _reset_scope_check()
    except (ImportError, AttributeError):
        pass
    try:
        from forge.runtime.mcp_config import _reset_mcp_tools_cache

        _reset_mcp_tools_cache()
    except (ImportError, AttributeError):
        pass
    yield
    try:
        from forge.agents.registry import (
            _reset_config_cache,
            _reset_scope_check,
        )

        _reset_config_cache()
        _reset_scope_check()
    except (ImportError, AttributeError):
        pass
    try:
        from studio.workflows._helpers import reset_caches as _rc

        _rc()
    except (ImportError, AttributeError):
        pass
    try:
        import studio.workflows.tdd as tdd

        tdd._get_runtime.reset()
        tdd._code_graph_cache = None
        tdd._get_build_knowledge.reset()
    except (ImportError, AttributeError):
        pass
    # R9.E: reset the memoized ``features.mcp_tools`` flag on
    # forge.runtime.mcp_config so tests that monkeypatch ``load_config``
    # to flip the feature value don't observe stale cached values from
    # earlier tests.
    try:
        from forge.runtime.mcp_config import _reset_mcp_tools_cache

        _reset_mcp_tools_cache()
    except (ImportError, AttributeError):
        pass
    try:
        import studio.workflows._utils as _wu

        _wu._shutdown_requested = False
    except (ImportError, AttributeError):
        pass
    try:
        from studio.workflows.orchestrator import _file_lock_manager

        _file_lock_manager.clear_all()
    except (ImportError, AttributeError):
        pass


@pytest.fixture(autouse=True)
def _reset_process_sandbox():
    """Commit 5: clear the sandbox installed state between tests.

    ``forge.tools.context.install_sandbox_from_env()`` installs the
    sandbox as both a ContextVar and a module-level process-global
    (the latter is needed because ``ThreadPoolExecutor`` worker threads
    don't inherit ContextVar state). Without a reset, a test that
    exercises the subprocess bridge would leak a sandbox into every
    downstream test's ``get_current_sandbox()`` call -- breaking the
    ``host-fallback'' assumption that most tool-surface tests rely on.

    We clear both the ContextVar and the process-global, before and
    after each test, defensively.
    """
    from forge.tools import context as _ctx_mod

    def _clear_both() -> None:
        try:
            _ctx_mod.current_sandbox.set(None)
        except Exception:  # pragma: no cover - defensive
            pass
        if hasattr(_ctx_mod, "_clear_process_sandbox"):
            _ctx_mod._clear_process_sandbox()

    _clear_both()
    yield
    _clear_both()


@pytest.fixture
def temp_dir():
    """Create a temporary directory for test isolation."""
    with tempfile.TemporaryDirectory() as d:
        yield Path(d)


@pytest.fixture
def git_repo(tmp_path):
    """A git repo with an initial commit, suitable for git tool tests."""
    subprocess.run(["git", "init", str(tmp_path)], check=True, capture_output=True)
    subprocess.run(
        ["git", "config", "user.email", "test@forge.dev"],
        check=True,
        capture_output=True,
        cwd=str(tmp_path),
    )
    subprocess.run(
        ["git", "config", "user.name", "Forge Test"],
        check=True,
        capture_output=True,
        cwd=str(tmp_path),
    )
    readme = tmp_path / "README.md"
    readme.write_text("# Test\n")
    subprocess.run(["git", "add", "."], check=True, capture_output=True, cwd=str(tmp_path))
    subprocess.run(
        ["git", "commit", "-m", "init"],
        check=True,
        capture_output=True,
        cwd=str(tmp_path),
    )
    return tmp_path


# ---------------------------------------------------------------------------
# Audit-Round-4 Q1: shared git_repo_main / git_repo_master fixtures.
#
# Several tests (default-branch detection, worker_diff regression
# tests, ...) need a fresh git repo whose initial branch is
# specifically ``main`` or ``master`` and which already has one
# commit so refs resolve. Pre-Q1, the same setup was duplicated in
# multiple test files (most prominently
# ``platform/tests/branching/test_default_branch.py``).
#
# The fixtures live in repo-level ``conftest.py`` so any sibling
# test under ``platform/tests/**`` can request them by name. They
# are idempotent: each invocation gets its own ``tmp_path``, and
# they only mutate the supplied directory (no global config, no
# environment variables).
# ---------------------------------------------------------------------------


def _init_git_repo_with_branch(repo_dir: Path, branch: str) -> Path:
    """Initialise ``repo_dir`` as a git repo on ``branch`` with one commit.

    Pure helper used by both ``git_repo_main`` and ``git_repo_master`` so
    the two fixtures share their setup verbatim — only the initial
    branch name differs.
    """
    subprocess.run(
        ["git", "init", f"--initial-branch={branch}", str(repo_dir)],
        capture_output=True,
        check=True,
    )
    subprocess.run(
        ["git", "config", "user.email", "test@test.com"],
        cwd=str(repo_dir),
        capture_output=True,
        check=True,
    )
    subprocess.run(
        ["git", "config", "user.name", "Test"],
        cwd=str(repo_dir),
        capture_output=True,
        check=True,
    )
    (repo_dir / "README.md").write_text("# test")
    subprocess.run(
        ["git", "add", "."],
        cwd=str(repo_dir),
        capture_output=True,
        check=True,
    )
    subprocess.run(
        ["git", "commit", "-m", "init"],
        cwd=str(repo_dir),
        capture_output=True,
        check=True,
    )
    return repo_dir


@pytest.fixture
def git_repo_main(tmp_path):
    """Git repo whose default branch is ``main`` with one initial commit."""
    return _init_git_repo_with_branch(tmp_path, "main")


@pytest.fixture
def git_repo_master(tmp_path):
    """Git repo whose default branch is ``master`` with one initial commit."""
    return _init_git_repo_with_branch(tmp_path, "master")


# ---------------------------------------------------------------------------
# Cleanup-audit follow-up C2: shared mock_studio_hooks fixture.
#
# Many node/orchestrator tests need to swap the real
# ``studio.workflows._helpers.get_hooks`` return value for a no-op
# ``MagicMock`` so the test exercises pure node logic without firing
# actual ``forge.pipeline_hooks`` side effects (recorders, training
# stores, evolution proposals, etc.). Pre-C2, ~13 test files repeated
# the same ``patch("studio.workflows._helpers.get_hooks", return_value=
# MagicMock())`` shape -- the densest single file did it 11 times. The
# fixture below collapses that boilerplate.
#
# Usage:
#
#     def test_x(self, mock_studio_hooks):
#         mock_studio_hooks.flags.red_team_review = True   # tweak as needed
#         red_team_node(state)
#
# The fixture is opt-in (NOT autouse): a test file gets the patch
# only when it requests the fixture, so legacy ``with patch(...)``
# blocks keep working unchanged. Each test gets its own fresh
# ``MagicMock`` to avoid cross-test pollution.
# ---------------------------------------------------------------------------
@pytest.fixture
def mock_studio_hooks():
    """Patch ``studio.workflows._helpers.get_hooks`` for the test
    duration and yield the underlying ``MagicMock`` hook object so
    the test can configure flags or assert call counts.
    """
    from unittest.mock import MagicMock, patch

    hooks = MagicMock()
    with patch("studio.workflows._helpers.get_hooks", return_value=hooks):
        yield hooks


@pytest.fixture
def go_project(tmp_path):
    """A minimal valid Go project for compiler and playtest tests."""
    module_name = "testmodule"
    go_mod = tmp_path / "go.mod"
    go_mod.write_text(f"module {module_name}\n\ngo 1.21\n")
    main_go = tmp_path / "main.go"
    main_go.write_text("package main\n\nfunc main() {}\n")
    main_test_go = tmp_path / "main_test.go"
    main_test_go.write_text('package main\n\nimport "testing"\n\nfunc TestAlwaysPasses(t *testing.T) {}\n')
    return tmp_path


@pytest.fixture
def initial_forge_state():
    """Canonical ForgeState dict for graph/node unit tests.

    Intended for future graph/playtest tests that need a shared baseline state;
    not all tests use this fixture yet.
    """
    return {
        "current_phase": "",
        "current_task": {"id": "DH-001", "description": "Test task"},
        "task_history": [],
        "test_result": None,
        "build_result": None,
        "failures": 0,
        "max_failures": 3,
        "messages": [],
        "memory_context": "",
    }


@pytest.fixture
def sample_config(temp_dir):
    """Create a minimal studio config for testing.

    Round 12 A2/B2/C2 (2026-05-02 cleanup): no ``provider_profile:``
    top-level key (the provider-profile system was deleted in favour
    of model-id-prefix routing); no per-agent ``provider:`` literal
    (derived from model id at ``create_agent`` time).  See
    docs/MODELS.md.
    """
    config_content = """
providers:
  openrouter:
    base_url: "https://openrouter.ai/api/v1"
    api_key_env: "OPENROUTER_API_KEY"
  lmstudio:
    base_url: "http://localhost:1234/v1"
    api_key: "lm-studio"
  bedrock:
    region: "us-east-1"
    api_key_env: "BEDROCK_API_KEY"

models:
  flash: "z-ai/glm-4.7-flash"
  coding_agent: "qwen/qwen3-coder-next"
  allrounder: "deepseek/deepseek-v3.2"
  swe_coder: "minimax/minimax-m2.7"
  reasoning: "qwen/qwen3.5-397b-a17b"
  deep_reasoning: "moonshotai/kimi-k2-thinking"
  embedding: "text-embedding-nomic-embed-text-v1.5"

agents:
  director:
    model: reasoning
    temperature: 0.7
    max_tokens: 4096
  red:
    model: coding_agent
    temperature: 0.3
    max_tokens: 4096
"""
    config_path = temp_dir / "config.yaml"
    config_path.write_text(config_content)
    return str(config_path)