deer-flow/backend/tests/test_client_live_policy.py

"""Regression tests for the explicit opt-in policy of live client tests."""

from __future__ import annotations

import os
import subprocess
import sys
from pathlib import Path

import pytest

BACKEND_ROOT = Path(__file__).resolve().parents[1]
REPO_ROOT = BACKEND_ROOT.parent
LIVE_TEST_PATH = BACKEND_ROOT / "tests" / "test_client_live.py"
LIVE_OPT_IN = "DEER_FLOW_RUN_LIVE_TESTS"
REPRESENTATIVE_LIVE_NODE_IDS = (
    "::TestLiveBasicChat::test_chat_returns_nonempty_string",
    "::TestLiveStreaming::test_stream_yields_messages_tuple_and_end",
)


def _collect_live_tests(
    tmp_path: Path,
    *,
    config_exists: bool,
    opt_in: bool,
    ci: bool,
) -> subprocess.CompletedProcess[str]:
    """Collect a temporary copy without inheriting credentials or a real .env."""
    temp_repo = tmp_path / "repo"
    temp_tests = temp_repo / "backend" / "tests"
    temp_tests.mkdir(parents=True)
    (temp_tests / "test_client_live.py").write_text(
        LIVE_TEST_PATH.read_text(encoding="utf-8"),
        encoding="utf-8",
    )
    if config_exists:
        (temp_repo / "config.yaml").write_text("models: []\n", encoding="utf-8")

    env = {
        "PATH": os.environ.get("PATH", ""),
        "PYTHONDONTWRITEBYTECODE": "1",
        "PYTHONIOENCODING": "utf-8",
        "PYTHONUTF8": "1",
        "PYTHONPATH": os.pathsep.join(
            [
                str(BACKEND_ROOT),
                str(BACKEND_ROOT / "packages" / "harness"),
            ]
        ),
        "PYTEST_DISABLE_PLUGIN_AUTOLOAD": "1",
    }
    if opt_in:
        env[LIVE_OPT_IN] = "1"
    if ci:
        env["CI"] = "1"

    return subprocess.run(
        [
            sys.executable,
            "-m",
            "pytest",
            "-c",
            str(BACKEND_ROOT / "pyproject.toml"),
            "--collect-only",
            "-q",
            "-rs",
            "-p",
            "no:cacheprovider",
            str(temp_tests / "test_client_live.py"),
        ],
        cwd=temp_repo / "backend",
        env=env,
        capture_output=True,
        text=True,
        check=False,
    )


def _assert_collection_skipped(result: subprocess.CompletedProcess[str], reason: str) -> None:
    output = result.stdout + result.stderr
    assert result.returncode in {0, pytest.ExitCode.NO_TESTS_COLLECTED}, output
    assert reason in output
    assert "no tests collected" in output
    assert all(node_id not in output for node_id in REPRESENTATIVE_LIVE_NODE_IDS)


def _dry_run_make_target(target: str) -> str:
    result = subprocess.run(
        ["make", "-n", target],
        cwd=BACKEND_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    output = result.stdout + result.stderr
    assert result.returncode == 0, output
    return output


def test_config_alone_does_not_enable_live_collection(tmp_path: Path) -> None:
    result = _collect_live_tests(tmp_path, config_exists=True, opt_in=False, ci=False)

    _assert_collection_skipped(result, f"Set {LIVE_OPT_IN}=1")


def test_explicit_opt_in_collects_live_tests(tmp_path: Path) -> None:
    result = _collect_live_tests(tmp_path, config_exists=True, opt_in=True, ci=False)

    output = result.stdout + result.stderr
    assert result.returncode == 0, output
    assert all(node_id in output for node_id in REPRESENTATIVE_LIVE_NODE_IDS)


def test_ci_blocks_live_collection_even_with_opt_in(tmp_path: Path) -> None:
    result = _collect_live_tests(tmp_path, config_exists=True, opt_in=True, ci=True)

    _assert_collection_skipped(result, "Live tests skipped in CI")


def test_opt_in_without_config_reports_missing_config(tmp_path: Path) -> None:
    result = _collect_live_tests(tmp_path, config_exists=False, opt_in=True, ci=False)

    _assert_collection_skipped(result, "No config.yaml found")


def test_make_targets_keep_default_tests_offline_and_support_live_opt_in() -> None:
    default_command = _dry_run_make_target("test")
    live_command = _dry_run_make_target("test-live")

    assert 'pytest -m "not live"' in default_command
    assert "tests/" in default_command
    assert LIVE_OPT_IN not in default_command

    assert f"{LIVE_OPT_IN}=1" in live_command
    assert "pytest -m live" in live_command
    assert "tests/" in live_command
    assert "tests/test_client_live.py" not in live_command


def test_live_marker_is_registered() -> None:
    pyproject = (BACKEND_ROOT / "pyproject.toml").read_text(encoding="utf-8")

    assert '"live: tests that call real external APIs and require explicit opt-in"' in pyproject


def test_documentation_matches_live_test_commands() -> None:
    contributor_docs = (REPO_ROOT / "CONTRIBUTING.md").read_text(encoding="utf-8")
    backend_agent_docs = (BACKEND_ROOT / "AGENTS.md").read_text(encoding="utf-8")

    for docs in (contributor_docs, backend_agent_docs):
        assert "make test-live" in docs
        assert LIVE_OPT_IN in docs
        assert "API" in docs


def test_default_ci_workflow_does_not_opt_in_to_live_tests() -> None:
    workflow = (REPO_ROOT / ".github" / "workflows" / "backend-unit-tests.yml").read_text(encoding="utf-8")

    assert "make test" in workflow
    assert LIVE_OPT_IN not in workflow