free-claude-code/tests/api/test_app_lifespan_and_errors.py

import importlib
from collections.abc import MutableMapping
from types import SimpleNamespace
from typing import Any, cast
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

from config.settings import Settings
from providers.exceptions import ServiceUnavailableError
from providers.registry import ProviderRegistry

_RUNTIME_EXTRAS = {
    "voice_note_enabled": True,
    "whisper_model": "base",
    "whisper_device": "cpu",
    "hf_token": "",
    "nvidia_nim_api_key": "",
    "claude_cli_bin": "claude",
    "uses_process_anthropic_auth_token": lambda: False,
    "messaging_rate_limit": 1,
    "messaging_rate_window": 1.0,
    "max_message_log_entries_per_chat": None,
    "debug_platform_edits": False,
    "debug_subagent_stack": False,
    "log_api_error_tracebacks": False,
    "log_raw_messaging_content": False,
    "log_raw_cli_diagnostics": False,
    "log_messaging_error_details": False,
    "configured_chat_model_refs": lambda: (),
}


def _app_settings(**kwargs):
    """Minimal settings namespace for AppRuntime (matches typed :class:`Settings` fields used)."""
    data = {**_RUNTIME_EXTRAS, **kwargs}
    return SimpleNamespace(**data)


def test_warn_if_process_auth_token_logs_warning():
    api_runtime_mod = importlib.import_module("api.runtime")
    settings = cast(
        Settings, SimpleNamespace(uses_process_anthropic_auth_token=lambda: True)
    )

    with patch.object(api_runtime_mod.logger, "warning") as warning:
        api_runtime_mod.warn_if_process_auth_token(settings)

    warning.assert_called_once()
    assert "ANTHROPIC_AUTH_TOKEN" in warning.call_args.args[0]


def test_warn_if_process_auth_token_skips_explicit_dotenv_config():
    api_runtime_mod = importlib.import_module("api.runtime")
    settings = cast(
        Settings, SimpleNamespace(uses_process_anthropic_auth_token=lambda: False)
    )

    with patch.object(api_runtime_mod.logger, "warning") as warning:
        api_runtime_mod.warn_if_process_auth_token(settings)

    warning.assert_not_called()


def test_create_app_provider_error_handler_returns_anthropic_format():
    from api.app import create_app
    from providers.exceptions import AuthenticationError

    app = create_app()

    @app.get("/raise_provider")
    async def _raise_provider():
        raise AuthenticationError("bad key")

    api_app_mod = importlib.import_module("api.app")
    settings = _app_settings(
        messaging_platform="telegram",
        telegram_bot_token=None,
        allowed_telegram_user_id=None,
        discord_bot_token=None,
        allowed_discord_channels=None,
        allowed_dir="",
        claude_workspace="./agent_workspace",
        host="127.0.0.1",
        port=8082,
        log_file="server.log",
    )
    with (
        patch.object(api_app_mod, "get_settings", return_value=settings),
        patch.object(ProviderRegistry, "cleanup", new=AsyncMock()),
    ):
        with TestClient(app) as client:
            resp = client.get("/raise_provider")
        assert resp.status_code == 401
    body = resp.json()
    assert body["type"] == "error"
    assert body["error"]["type"] == "authentication_error"


def test_create_app_provider_error_default_logs_exclude_provider_message():
    """Provider errors must not log exc.message by default."""
    from api.app import create_app
    from providers.exceptions import AuthenticationError

    app = create_app()
    secret = "provider-upstream-secret-detail"

    @app.get("/raise_provider_secret")
    async def _raise():
        raise AuthenticationError(secret)

    api_app_mod = importlib.import_module("api.app")
    settings = _app_settings(
        messaging_platform="telegram",
        telegram_bot_token=None,
        allowed_telegram_user_id=None,
        discord_bot_token=None,
        allowed_discord_channels=None,
        allowed_dir="",
        claude_workspace="./agent_workspace",
        host="127.0.0.1",
        port=8082,
        log_file="server.log",
        log_api_error_tracebacks=False,
    )
    with (
        patch.object(api_app_mod, "get_settings", return_value=settings),
        patch.object(ProviderRegistry, "cleanup", new=AsyncMock()),
        patch.object(api_app_mod.logger, "error") as log_err,
    ):
        with TestClient(app) as client:
            resp = client.get("/raise_provider_secret")
        assert resp.status_code == 401

    blob = " ".join(str(a) for c in log_err.call_args_list for a in c.args)
    blob += repr([c.kwargs for c in log_err.call_args_list])
    assert secret not in blob
    assert "authentication_error" in blob


def test_create_app_general_exception_handler_returns_500():
    from api.app import create_app

    app = create_app()

    @app.get("/raise_general")
    async def _raise_general():
        raise RuntimeError("boom")

    api_app_mod = importlib.import_module("api.app")
    settings = _app_settings(
        messaging_platform="telegram",
        telegram_bot_token=None,
        allowed_telegram_user_id=None,
        discord_bot_token=None,
        allowed_discord_channels=None,
        allowed_dir="",
        claude_workspace="./agent_workspace",
        host="127.0.0.1",
        port=8082,
        log_file="server.log",
    )
    with (
        patch.object(api_app_mod, "get_settings", return_value=settings),
        patch.object(ProviderRegistry, "cleanup", new=AsyncMock()),
    ):
        with TestClient(app, raise_server_exceptions=False) as client:
            resp = client.get("/raise_general")
        assert resp.status_code == 500
        body = resp.json()
        assert body["type"] == "error"
        assert body["error"]["type"] == "api_error"


def test_create_app_general_exception_default_logs_exclude_exception_message():
    """Unhandled errors must not log exception text by default (may echo user content)."""
    from api.app import create_app

    app = create_app()

    secret = "user-provided-secret-token-xyzzy"

    @app.get("/raise_secret")
    async def _raise_secret():
        raise ValueError(secret)

    api_app_mod = importlib.import_module("api.app")
    settings = _app_settings(
        messaging_platform="telegram",
        telegram_bot_token=None,
        allowed_telegram_user_id=None,
        discord_bot_token=None,
        allowed_discord_channels=None,
        allowed_dir="",
        claude_workspace="./agent_workspace",
        host="127.0.0.1",
        port=8082,
        log_file="server.log",
        log_api_error_tracebacks=False,
    )
    with (
        patch.object(api_app_mod, "get_settings", return_value=settings),
        patch.object(ProviderRegistry, "cleanup", new=AsyncMock()),
        patch.object(api_app_mod.logger, "error") as log_err,
    ):
        with TestClient(app, raise_server_exceptions=False) as client:
            resp = client.get("/raise_secret")
        assert resp.status_code == 500

    flattened: list[str] = []
    for call in log_err.call_args_list:
        flattened.extend(str(arg) for arg in call.args)
        flattened.append(repr(call.kwargs))
    blob = " ".join(flattened)
    assert secret not in blob
    assert "ValueError" in blob


@pytest.mark.parametrize(
    "messaging_enabled", [True, False], ids=["with_platform", "no_platform"]
)
def test_app_lifespan_sets_state_and_cleans_up(tmp_path, messaging_enabled):
    from api.app import create_app

    app = create_app()

    settings = _app_settings(
        messaging_platform="telegram",
        telegram_bot_token="token" if messaging_enabled else None,
        allowed_telegram_user_id="123",
        discord_bot_token=None,
        allowed_discord_channels=None,
        allowed_dir=str(tmp_path / "workspace"),
        claude_workspace=str(tmp_path / "data"),
        host="127.0.0.1",
        port=8082,
        log_file=str(tmp_path / "server.log"),
    )

    fake_platform = MagicMock()
    fake_platform.name = "fake"
    fake_platform.on_message = MagicMock()
    fake_platform.start = AsyncMock()
    fake_platform.stop = AsyncMock()

    session_store = MagicMock()
    session_store.get_all_trees.return_value = [{"t": 1}] if messaging_enabled else []
    session_store.get_node_mapping.return_value = {"n": "t"}
    session_store.sync_from_tree_data = MagicMock()

    fake_queue = MagicMock()
    fake_queue.cleanup_stale_nodes.return_value = 1
    fake_queue.to_dict.return_value = {
        "trees": [{"t": 1}],
        "node_to_tree": {"n": "t"},
    }

    cli_manager = MagicMock()
    cli_manager.stop_all = AsyncMock()

    api_app_mod = importlib.import_module("api.app")

    registry_cleanup = AsyncMock()
    with (
        patch.object(api_app_mod, "get_settings", return_value=settings),
        patch.object(ProviderRegistry, "cleanup", new=registry_cleanup),
        patch(
            "messaging.platforms.factory.create_messaging_platform",
            return_value=fake_platform if messaging_enabled else None,
        ) as create_platform,
        patch("messaging.session.SessionStore", return_value=session_store),
        patch("cli.manager.CLISessionManager", return_value=cli_manager),
        patch(
            "messaging.trees.queue_manager.TreeQueueManager.from_dict",
            return_value=fake_queue,
        ),
        TestClient(app),
    ):
        pass

    if messaging_enabled:
        create_platform.assert_called_once()
        fake_platform.on_message.assert_called_once()
        fake_platform.start.assert_awaited_once()
        fake_platform.stop.assert_awaited_once()
        cli_manager.stop_all.assert_awaited_once()
        assert getattr(app.state, "message_handler", None) is not None
        session_store.sync_from_tree_data.assert_called_once_with(
            [{"t": 1}],
            {"n": "t"},
        )
    else:
        fake_platform.start.assert_not_awaited()
        fake_platform.stop.assert_not_awaited()
        cli_manager.stop_all.assert_not_awaited()
        assert getattr(app.state, "messaging_platform", "missing") is None

    registry_cleanup.assert_awaited_once()


def test_app_lifespan_cleanup_continues_if_platform_stop_raises(tmp_path):
    from api.app import create_app

    app = create_app()

    settings = _app_settings(
        messaging_platform="telegram",
        telegram_bot_token="token",
        allowed_telegram_user_id="123",
        discord_bot_token=None,
        allowed_discord_channels=None,
        allowed_dir=str(tmp_path / "workspace"),
        claude_workspace=str(tmp_path / "data"),
        host="127.0.0.1",
        port=8082,
        log_file=str(tmp_path / "server.log"),
    )

    fake_platform = MagicMock()
    fake_platform.name = "fake"
    fake_platform.on_message = MagicMock()
    fake_platform.start = AsyncMock()
    fake_platform.stop = AsyncMock(side_effect=RuntimeError("stop failed"))

    session_store = MagicMock()
    session_store.get_all_trees.return_value = []
    session_store.get_node_mapping.return_value = {}
    session_store.sync_from_tree_data = MagicMock()

    cli_manager = MagicMock()
    cli_manager.stop_all = AsyncMock()

    api_app_mod = importlib.import_module("api.app")
    registry_cleanup = AsyncMock()
    with (
        patch.object(api_app_mod, "get_settings", return_value=settings),
        patch.object(ProviderRegistry, "cleanup", new=registry_cleanup),
        patch(
            "messaging.platforms.factory.create_messaging_platform",
            return_value=fake_platform,
        ),
        patch("messaging.session.SessionStore", return_value=session_store),
        patch("cli.manager.CLISessionManager", return_value=cli_manager),
        TestClient(app),
    ):
        pass

    fake_platform.stop.assert_awaited_once()
    cli_manager.stop_all.assert_awaited_once()
    registry_cleanup.assert_awaited_once()


@pytest.mark.asyncio
async def test_runtime_startup_validation_blocks_messaging_and_cleans_up(tmp_path):
    import api.runtime as api_runtime_mod

    settings = _app_settings(
        messaging_platform="telegram",
        telegram_bot_token="token",
        allowed_telegram_user_id="123",
        discord_bot_token=None,
        allowed_discord_channels=None,
        allowed_dir=str(tmp_path / "workspace"),
        claude_workspace=str(tmp_path / "data"),
        host="127.0.0.1",
        port=8082,
        log_file=str(tmp_path / "server.log"),
    )
    app = FastAPI()
    runtime = api_runtime_mod.AppRuntime(
        app=app,
        settings=cast(Settings, settings),
    )

    validation = AsyncMock(side_effect=ServiceUnavailableError("bad model"))
    cleanup = AsyncMock()
    with (
        patch.object(ProviderRegistry, "validate_configured_models", new=validation),
        patch.object(ProviderRegistry, "cleanup", new=cleanup),
        patch.object(api_runtime_mod.logger, "error") as log_error,
        patch(
            "messaging.platforms.factory.create_messaging_platform"
        ) as create_platform,
        pytest.raises(ServiceUnavailableError, match="bad model"),
    ):
        await runtime.startup()

    validation.assert_awaited_once_with(settings)
    cleanup.assert_awaited_once()
    create_platform.assert_not_called()
    logged = " ".join(
        str(arg) for call in log_error.call_args_list for arg in call.args
    )
    assert "Startup failed" in logged
    assert "bad model" in logged
    assert "Traceback" not in logged


@pytest.mark.asyncio
async def test_graceful_asgi_lifespan_failure_sends_no_traceback(tmp_path):
    import api.app as api_app_mod

    settings = _app_settings(
        messaging_platform="none",
        telegram_bot_token=None,
        allowed_telegram_user_id=None,
        discord_bot_token=None,
        allowed_discord_channels=None,
        allowed_dir=str(tmp_path / "workspace"),
        claude_workspace=str(tmp_path / "data"),
        host="127.0.0.1",
        port=8082,
        log_file=str(tmp_path / "server.log"),
    )
    app = api_app_mod.GracefulLifespanApp(FastAPI())
    sent: list[MutableMapping[str, Any]] = []

    async def receive() -> MutableMapping[str, Any]:
        return {"type": "lifespan.startup"}

    async def send(message: MutableMapping[str, Any]) -> None:
        sent.append(message)

    validation = AsyncMock(side_effect=ServiceUnavailableError("bad model"))
    cleanup = AsyncMock()
    with (
        patch.object(api_app_mod, "get_settings", return_value=settings),
        patch.object(ProviderRegistry, "validate_configured_models", new=validation),
        patch.object(ProviderRegistry, "cleanup", new=cleanup),
    ):
        await app({"type": "lifespan"}, receive, send)

    assert sent == [{"type": "lifespan.startup.failed", "message": "bad model"}]


def test_app_lifespan_messaging_import_error_no_crash(tmp_path, caplog):
    """Messaging import failure logs warning and continues without crash."""
    from api.app import create_app

    app = create_app()

    settings = _app_settings(
        messaging_platform="telegram",
        telegram_bot_token="token",
        allowed_telegram_user_id="123",
        discord_bot_token=None,
        allowed_discord_channels=None,
        allowed_dir=str(tmp_path / "workspace"),
        claude_workspace=str(tmp_path / "data"),
        host="127.0.0.1",
        port=8082,
        log_file=str(tmp_path / "server.log"),
    )

    api_app_mod = importlib.import_module("api.app")
    registry_cleanup = AsyncMock()
    with (
        patch.object(api_app_mod, "get_settings", return_value=settings),
        patch.object(ProviderRegistry, "cleanup", new=registry_cleanup),
        patch(
            "messaging.platforms.factory.create_messaging_platform",
            side_effect=ImportError("discord not installed"),
        ),
        TestClient(app),
    ):
        pass

    assert getattr(app.state, "messaging_platform", None) is None
    registry_cleanup.assert_awaited_once()


def test_app_lifespan_platform_start_exception_cleanup_still_runs(tmp_path):
    """Exception during platform.start() logs error, cleanup still runs."""
    from api.app import create_app

    app = create_app()

    settings = _app_settings(
        messaging_platform="telegram",
        telegram_bot_token="token",
        allowed_telegram_user_id="123",
        discord_bot_token=None,
        allowed_discord_channels=None,
        allowed_dir=str(tmp_path / "workspace"),
        claude_workspace=str(tmp_path / "data"),
        host="127.0.0.1",
        port=8082,
        log_file=str(tmp_path / "server.log"),
    )

    fake_platform = MagicMock()
    fake_platform.name = "fake"
    fake_platform.on_message = MagicMock()
    fake_platform.start = AsyncMock(side_effect=RuntimeError("start failed"))
    fake_platform.stop = AsyncMock()

    session_store = MagicMock()
    session_store.get_all_trees.return_value = []
    session_store.get_node_mapping.return_value = {}
    session_store.sync_from_tree_data = MagicMock()

    cli_manager = MagicMock()
    cli_manager.stop_all = AsyncMock()

    api_app_mod = importlib.import_module("api.app")
    registry_cleanup = AsyncMock()
    with (
        patch.object(api_app_mod, "get_settings", return_value=settings),
        patch.object(ProviderRegistry, "cleanup", new=registry_cleanup),
        patch(
            "messaging.platforms.factory.create_messaging_platform",
            return_value=fake_platform,
        ),
        patch("messaging.session.SessionStore", return_value=session_store),
        patch("cli.manager.CLISessionManager", return_value=cli_manager),
        TestClient(app),
    ):
        pass

    registry_cleanup.assert_awaited_once()


def test_app_lifespan_flush_pending_save_exception_warning_only(tmp_path):
    """Session store flush exception on shutdown is logged as warning, no crash."""
    from api.app import create_app

    app = create_app()

    settings = _app_settings(
        messaging_platform="telegram",
        telegram_bot_token="token",
        allowed_telegram_user_id="123",
        discord_bot_token=None,
        allowed_discord_channels=None,
        allowed_dir=str(tmp_path / "workspace"),
        claude_workspace=str(tmp_path / "data"),
        host="127.0.0.1",
        port=8082,
        log_file=str(tmp_path / "server.log"),
    )

    fake_platform = MagicMock()
    fake_platform.name = "fake"
    fake_platform.on_message = MagicMock()
    fake_platform.start = AsyncMock()
    fake_platform.stop = AsyncMock()

    session_store = MagicMock()
    session_store.get_all_trees.return_value = []
    session_store.get_node_mapping.return_value = {}
    session_store.sync_from_tree_data = MagicMock()
    session_store.flush_pending_save = MagicMock(side_effect=OSError("disk full"))

    cli_manager = MagicMock()
    cli_manager.stop_all = AsyncMock()

    api_app_mod = importlib.import_module("api.app")
    registry_cleanup = AsyncMock()
    with (
        patch.object(api_app_mod, "get_settings", return_value=settings),
        patch.object(ProviderRegistry, "cleanup", new=registry_cleanup),
        patch(
            "messaging.platforms.factory.create_messaging_platform",
            return_value=fake_platform,
        ),
        patch("messaging.session.SessionStore", return_value=session_store),
        patch("cli.manager.CLISessionManager", return_value=cli_manager),
        TestClient(app),
    ):
        pass

    session_store.flush_pending_save.assert_called_once()
    registry_cleanup.assert_awaited_once()