orchestrator/tests/test_fs_normalize_startup.py

"""ORCH-057 D3: startup-hook observability + the clear pre-launch outcome.

TC-10 / TC-11 (04-test-plan.yaml):
  * TC-10 — the lifespan startup hook, on a detected mismatch, emits a WARNING and a
    Telegram message; a detect error never crashes the start (never-fatal).
  * TC-11 — the "clear, early" outcome on a permission failure is delivered by the
    actionable ensure_worktree error (ADR-001 D3: claim is NOT blocked), i.e. the
    launch surfaces an actionable diagnosis, never a raw git-fatal.

Background daemons are disabled via env so the lifespan is cheap and deterministic.
"""
import os
import tempfile

import pytest

_test_db = os.path.join(tempfile.gettempdir(), "test_orchestrator_fsn_startup.db")
os.environ["ORCH_DB_PATH"] = _test_db
os.environ["ORCH_REPOS_DIR"] = tempfile.gettempdir()
os.environ["ORCH_GITEA_TOKEN"] = "test-token"
os.environ["ORCH_PLANE_API_TOKEN"] = "test-token"
os.environ["ORCH_PLANE_WEBHOOK_SECRET"] = ""
os.environ["ORCH_GITEA_WEBHOOK_SECRET"] = ""
# Keep the lifespan light: no background daemons during the test.
os.environ["ORCH_RECONCILE_ENABLED"] = "false"
os.environ["ORCH_REAPER_ENABLED"] = "false"
os.environ["ORCH_DISK_MONITOR_ENABLED"] = "false"
os.environ["ORCH_BUILD_CACHE_PRUNE_ENABLED"] = "false"
os.environ["ORCH_FS_NORMALIZE_ENABLED"] = "true"

from fastapi.testclient import TestClient

from src import fs_normalize, git_worktree
from src.main import app
from src.db import init_db


@pytest.fixture(autouse=True)
def _db():
    if os.path.exists(_test_db):
        os.unlink(_test_db)
    init_db()
    fs_normalize.reset_cache()
    yield
    if os.path.exists(_test_db):
        os.unlink(_test_db)


# ---------------------------------------------------------------------------
# TC-10 — startup observability
# ---------------------------------------------------------------------------
def test_tc10_startup_mismatch_warns_and_telegrams(monkeypatch, caplog):
    """TC-10: on a detected mismatch the startup hook logs a WARNING and sends a
    Telegram message (mocked)."""
    sent = []
    monkeypatch.setattr(
        "src.notifications.send_telegram", lambda *a, **k: sent.append(a[0] if a else "")
    )
    scan = fs_normalize.OwnershipScan(
        mismatch=True, target_uid=1000, roots_checked=["/repos/_wt"],
        roots_mismatch=["/repos/_wt"], sample_path="/repos/_wt/x", checked_at=1.0,
    )
    monkeypatch.setattr("src.fs_normalize.scan_ownership", lambda *a, **k: scan)

    with caplog.at_level("WARNING"):
        with TestClient(app):
            pass

    assert any("FS-ownership mismatch" in r.message for r in caplog.records)
    # Filter for the fs-ownership message (the shared startup may emit other,
    # unrelated Telegram traffic — e.g. a leftover task's tracker card).
    fs_msgs = [m for m in sent if "legacy root-owned" in m.lower() or "chown" in m.lower()]
    assert fs_msgs, "expected a Telegram message on mismatch"


def test_tc10_startup_detect_error_never_fatal(monkeypatch):
    """TC-10: a detect error must NOT crash the start (never-fatal)."""
    def boom(*a, **k):
        raise RuntimeError("simulated detect failure")

    monkeypatch.setattr("src.fs_normalize.scan_ownership", boom)
    # Entering/exiting the lifespan must not raise.
    with TestClient(app):
        pass


def test_tc10_startup_clean_no_telegram(monkeypatch):
    """A clean environment (no mismatch) sends no Telegram and does not warn."""
    sent = []
    monkeypatch.setattr(
        "src.notifications.send_telegram", lambda *a, **k: sent.append(a[0] if a else "")
    )
    clean = fs_normalize.OwnershipScan(mismatch=False, target_uid=1000, checked_at=1.0)
    monkeypatch.setattr("src.fs_normalize.scan_ownership", lambda *a, **k: clean)
    with TestClient(app):
        pass
    # No fs-ownership message on a clean environment (unrelated startup Telegram
    # traffic from a shared-DB leftover task is ignored).
    fs_msgs = [m for m in sent if "legacy root-owned" in m.lower() or "обнаружены legacy" in m.lower()]
    assert fs_msgs == []


# ---------------------------------------------------------------------------
# TC-11 — clear pre-launch outcome (D1, not a claim gate)
# ---------------------------------------------------------------------------
def test_tc11_launch_permission_failure_is_actionable_not_raw(tmp_path, monkeypatch):
    """TC-11: the launch-time worktree creation surfaces an actionable error (clear,
    before the agent spends a token), not a raw git-fatal — the ADR-001 D3 "внятно и
    заранее" outcome that replaces a blocking claim gate."""
    repo = "orchestrator"
    repos_dir = tmp_path / "repos"
    (repos_dir / repo).mkdir(parents=True)
    monkeypatch.setattr(git_worktree.settings, "repos_dir", str(repos_dir))
    monkeypatch.setattr(git_worktree.settings, "worktrees_dir", str(repos_dir / "_wt"))
    monkeypatch.setattr(git_worktree.settings, "fs_normalize_enabled", True)

    class _R:
        def __init__(self, rc, err=""):
            self.returncode = rc
            self.stderr = err
            self.stdout = ""

    def fake_run(cmd, *a, **k):
        if "fetch" in cmd:
            return _R(0)
        if "worktree" in cmd and "add" in cmd:
            return _R(128, "fatal: ...: Permission denied")
        return _R(0)

    monkeypatch.setattr(git_worktree.subprocess, "run", fake_run)

    with pytest.raises(RuntimeError) as ei:
        git_worktree.ensure_worktree(repo, "feature/x")
    msg = str(ei.value)
    assert "INFRA.md" in msg and "chown" in msg.lower()
    assert "git worktree add failed" not in msg  # not the raw passthrough