shd.build/tools/game-data-capture/tests/unit/test_cli.py

from __future__ import annotations

import io
import logging
import subprocess
import sys
from contextlib import redirect_stderr, redirect_stdout
from pathlib import Path

import pytest

from game_data_capture.cli import EXIT_CONFIG_ERROR, EXIT_UNIMPLEMENTED, main
from game_data_capture.logging import RedactingFormatter, configure_logging, redact_log_data


def test_doctor_reports_required_diagnostics_without_openai_value(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    monkeypatch.setenv("HOME", str(tmp_path / "home"))
    monkeypatch.setenv("OPENAI_API_KEY", "sk-this-must-not-appear")
    output = io.StringIO()

    with redirect_stdout(output):
        result = main(["doctor", "--json"])

    report = output.getvalue()
    assert result == 0
    for name in ("niri", "wl-paste", "ffmpeg", "gpu_provider", "storage_permissions", "openai"):
        assert f'"name":"{name}"' in report
    assert "sk-this-must-not-appear" not in report


def test_serve_rejects_public_binding() -> None:
    error = io.StringIO()

    with redirect_stderr(error):
        result = main(["serve", "--host", "0.0.0.0"])

    assert result == EXIT_CONFIG_ERROR
    assert "loopback IP literal" in error.getvalue()


def test_serve_rejects_invalid_port_without_echoing_value() -> None:
    error = io.StringIO()

    with redirect_stderr(error), pytest.raises(SystemExit) as raised:
        main(["serve", "--port", "99999-private-value"])

    assert raised.value.code == EXIT_CONFIG_ERROR
    assert "port must be an integer from 1024 to 65535" in error.getvalue()
    assert "99999-private-value" not in error.getvalue()


def test_subsystem_commands_are_explicit_stubs() -> None:
    error = io.StringIO()

    with redirect_stderr(error):
        result = main(["watch"])

    assert result == EXIT_UNIMPLEMENTED
    assert "requires gdc-runtime-integration" in error.getvalue()


def test_unimplemented_command_validates_config_before_returning_stub_status(
    tmp_path: Path,
) -> None:
    config_path = tmp_path / "invalid.toml"
    config_path.write_text("[server]\nport = 80\n", encoding="utf-8")
    error = io.StringIO()

    with redirect_stderr(error):
        result = main(["watch", "--config", str(config_path)])

    assert result == EXIT_CONFIG_ERROR
    assert "server.port" in error.getvalue()
    assert "subsystem is not implemented" not in error.getvalue()


def test_logging_redacts_sensitive_fields_and_messages() -> None:
    redacted = redact_log_data(
        {
            "authorization": "Bearer secret-value",
            "image_payload": "data:image/png;base64,abc",
            "ocr_text": "unrestricted source text",
            "status": "ok",
        }
    )
    record = logging.LogRecord(
        "test", logging.INFO, __file__, 1, "Authorization: Bearer hidden sk-visible123", (), None
    )

    assert redacted == {
        "authorization": "[REDACTED]",
        "image_payload": "[REDACTED]",
        "ocr_text": "[REDACTED]",
        "status": "ok",
    }
    assert "hidden" not in RedactingFormatter("%(message)s").format(record)
    assert "sk-visible123" not in RedactingFormatter("%(message)s").format(record)


def test_configure_logging_replaces_unredacted_handlers_and_redacts_exceptions() -> None:
    package_logger = logging.getLogger("game_data_capture")
    package_handlers = package_logger.handlers[:]
    package_level = package_logger.level
    package_propagate = package_logger.propagate
    root_logger = logging.getLogger()
    root_handler = logging.NullHandler()
    root_logger.addHandler(root_handler)
    leaked_output = io.StringIO()
    existing_handler = logging.StreamHandler(leaked_output)
    package_logger.handlers = [existing_handler]
    captured_output = io.StringIO()

    try:
        with redirect_stderr(captured_output):
            configured_logger = configure_logging()
            configure_logging()
            try:
                raise RuntimeError("Authorization: Bearer exception-secret; ocr_text=private text")
            except RuntimeError:
                configured_logger.exception("capture failed")

        output = captured_output.getvalue()
        assert package_logger.handlers != [existing_handler]
        assert len(package_logger.handlers) == 1
        assert leaked_output.getvalue() == ""
        assert root_handler in root_logger.handlers
        assert "exception-secret" not in output
        assert "private text" not in output
        assert "[REDACTED EXCEPTION]" in output
    finally:
        for handler in package_logger.handlers[:]:
            package_logger.removeHandler(handler)
        package_logger.handlers = package_handlers
        package_logger.setLevel(package_level)
        package_logger.propagate = package_propagate
        root_logger.removeHandler(root_handler)


def test_package_import_has_no_runtime_side_effects(tmp_path: Path) -> None:
    root = tmp_path / "fresh-xdg"
    source_root = Path(__file__).parents[2] / "src"
    environment = {
        "GDC_IMPORT_ROOT": str(root),
        "PYTHONDONTWRITEBYTECODE": "1",
        "PYTHONPATH": str(source_root),
    }
    script = """
import os
import sys
from pathlib import Path

root = Path(os.environ["GDC_IMPORT_ROOT"])
targets = [root / name for name in ("config", "data", "cache", "runtime")]
events = []

def audit(event, arguments):
    if event in {"os.mkdir", "os.chmod", "subprocess.Popen", "socket.connect", "sqlite3.connect"}:
        events.append(event)
    if event == "open":
        mode = arguments[1] if len(arguments) > 1 else None
        flags = arguments[2] if len(arguments) > 2 else 0
        if isinstance(mode, str) and any(marker in mode for marker in ("w", "a", "x", "+")):
            events.append("open-write")
        if isinstance(flags, int) and flags & (os.O_WRONLY | os.O_RDWR | os.O_CREAT | os.O_TRUNC):
            events.append("open-write")

sys.addaudithook(audit)
os.environ.update({
    "HOME": str(root / "home"),
    "XDG_CONFIG_HOME": str(targets[0]),
    "XDG_DATA_HOME": str(targets[1]),
    "XDG_CACHE_HOME": str(targets[2]),
    "XDG_RUNTIME_DIR": str(targets[3]),
})
import game_data_capture

if events or any(path.exists() for path in targets):
    raise SystemExit(1)
"""

    completed = subprocess.run(
        [sys.executable, "-B", "-c", script],
        capture_output=True,
        check=False,
        env=environment,
        text=True,
        timeout=5,
    )

    assert completed.returncode == 0, completed.stderr