Igloo MCP

"""Tests for HealthCheckTool covering success and failure branches.""" from __future__ import annotations import sys import types from dataclasses import dataclass from pathlib import Path from typing import Any import pytest from igloo_mcp.config import Config from igloo_mcp.mcp.tools.health import HealthCheckTool from igloo_mcp.mcp_health import HealthStatus from igloo_mcp.profile_utils import ProfileValidationError class _StubCursor: def __init__(self) -> None: self._current: dict[str, Any] = {} def execute(self, query: str) -> None: key = query.split("CURRENT_", 1)[-1].split("()")[0].lower() self._current = {key: key.upper()} def fetchone(self) -> dict[str, Any]: return self._current class _StubConnection: def __init__(self, cursor: _StubCursor) -> None: self.cursor = cursor def __enter__(self) -> tuple[None, _StubCursor]: return None, self.cursor def __exit__(self, exc_type, exc, tb) -> bool: return False class StubSnowflakeService: def __init__(self, cursor: _StubCursor | None = None) -> None: self._cursor = cursor or _StubCursor() def get_query_tag_param(self) -> dict[str, Any]: return {} def get_connection(self, **kwargs: Any) -> _StubConnection: return _StubConnection(self._cursor) @dataclass class StubProfileHealth: is_valid: bool profile_name: str = "DEFAULT" config_path: Path = Path("/tmp/config.toml") available_profiles: list[str] = None # type: ignore[assignment] config_exists: bool = True validation_error: str | None = None def __post_init__(self) -> None: if self.available_profiles is None: self.available_profiles = ["DEFAULT"] @dataclass class StubSystemStatus: status: str = "ok" is_healthy: bool = True error_count: int = 0 warning_count: int = 0 metrics: dict[str, int] = None # type: ignore[assignment] recent_errors: list[str] = None # type: ignore[assignment] def __post_init__(self) -> None: if self.metrics is None: self.metrics = { "total_queries": 1, "successful_queries": 1, "failed_queries": 0, } if self.recent_errors is None: self.recent_errors = [] class StubHealthMonitor: def __init__(self, profile_health: StubProfileHealth, system_status: StubSystemStatus): self._profile_health = profile_health self._system_status = system_status self.connection_status = HealthStatus.HEALTHY def get_profile_health(self) -> StubProfileHealth: return self._profile_health def check_connection_health(self, _service=None) -> HealthStatus: return self.connection_status def get_health_status(self) -> StubSystemStatus: return self._system_status class StubResourceManager: def __init__(self, resources: list[str]): self._resources = resources def list_resources(self) -> list[str]: return list(self._resources) class SummaryStub: def __init__(self) -> None: self.config_path = Path("/tmp/config.toml") self.config_exists = True self.available_profiles = ["DEFAULT"] self.default_profile = "DEFAULT" self.current_profile = "DEFAULT" self.profile_count = 1 self.current_profile_authenticator = "externalbrowser" @pytest.mark.asyncio async def test_health_check_tool_success(monkeypatch: pytest.MonkeyPatch) -> None: config = Config.from_env() service = StubSnowflakeService() monitor = StubHealthMonitor( profile_health=StubProfileHealth(is_valid=True), system_status=StubSystemStatus(), ) resource_manager = StubResourceManager(["catalog"]) module = types.ModuleType("mcp_server_snowflake.cortex_services.tools") async def fake_complete_cortex(**kwargs: Any) -> None: # pragma: no cover - stub return None module.complete_cortex = fake_complete_cortex monkeypatch.setitem( sys.modules, "mcp_server_snowflake.cortex_services.tools", module, ) monkeypatch.setattr( "igloo_mcp.mcp.tools.health.validate_and_resolve_profile", lambda: "DEFAULT", ) monkeypatch.setattr( "igloo_mcp.mcp.tools.health.get_profile_summary", lambda: SummaryStub(), ) tool = HealthCheckTool( config=config, snowflake_service=service, health_monitor=monitor, resource_manager=resource_manager, ) result = await tool.execute(include_cortex=True, include_catalog=True, response_mode="full") assert result["checks"]["connection"]["status"] == "connected" assert result["checks"]["profile"]["status"] == "valid" assert result["checks"]["cortex"]["available"] is True assert result["checks"]["catalog"]["status"] == "available" assert result["checks"]["system"]["healthy"] is True assert result["status"] == "healthy" @pytest.mark.asyncio async def test_health_check_handles_failures(monkeypatch: pytest.MonkeyPatch) -> None: config = Config.from_env() class FailingService(StubSnowflakeService): def get_connection(self, **kwargs: Any) -> _StubConnection: raise RuntimeError("connection refused") failing_monitor = StubHealthMonitor( profile_health=StubProfileHealth( is_valid=False, validation_error="invalid profile", ), system_status=StubSystemStatus(is_healthy=False, status="error"), ) failing_monitor.connection_status = HealthStatus.UNHEALTHY def raise_validation() -> None: raise ProfileValidationError("profile error") monkeypatch.setattr( "igloo_mcp.mcp.tools.health.validate_and_resolve_profile", raise_validation, ) monkeypatch.setattr( "igloo_mcp.mcp.tools.health.get_profile_summary", lambda: SummaryStub(), ) monkeypatch.delitem( sys.modules, "mcp_server_snowflake.cortex_services.tools", raising=False, ) class FailingResourceManager: def list_resources(self) -> list[str]: raise RuntimeError("catalog offline") tool = HealthCheckTool( config=config, snowflake_service=FailingService(), health_monitor=failing_monitor, resource_manager=FailingResourceManager(), ) result = await tool.execute(include_cortex=True, include_catalog=True, response_mode="full") assert result["checks"]["connection"]["connected"] is False assert result["checks"]["profile"]["status"] in {"invalid", "error"} assert result["checks"]["catalog"]["status"] == "error" assert result["checks"]["cortex"]["available"] is False assert result["checks"]["system"]["status"] == "error" assert result["status"] == "unhealthy" @pytest.mark.asyncio async def test_health_check_full_mode_includes_storage_paths(monkeypatch: pytest.MonkeyPatch) -> None: """Test that full mode includes storage paths diagnostics.""" config = Config.from_env() service = StubSnowflakeService() monitor = StubHealthMonitor( profile_health=StubProfileHealth(is_valid=True), system_status=StubSystemStatus(), ) # Mock profile validation to avoid dependency on real profile monkeypatch.setattr( "igloo_mcp.mcp.tools.health.validate_and_resolve_profile", lambda: "DEFAULT", ) monkeypatch.setattr( "igloo_mcp.mcp.tools.health.get_profile_summary", lambda: SummaryStub(), ) tool = HealthCheckTool( config=config, snowflake_service=service, health_monitor=monitor, ) result = await tool.execute(response_mode="full", include_profile=True, include_cortex=False, include_catalog=False) # Verify storage_paths is present in diagnostics assert "diagnostics" in result assert "storage_paths" in result["diagnostics"] storage = result["diagnostics"]["storage_paths"] # Verify required fields assert "scope" in storage assert "base_directory" in storage assert "query_history" in storage assert "artifacts" in storage assert "cache" in storage assert "reports" in storage assert "catalogs" in storage assert "namespaced" in storage # Verify scope value is valid assert storage["scope"] in ["global", "repo"] # Verify all paths are strings and non-empty assert isinstance(storage["base_directory"], str) assert storage["base_directory"] assert isinstance(storage["query_history"], str) assert storage["query_history"] assert isinstance(storage["artifacts"], str) assert storage["artifacts"] assert isinstance(storage["cache"], str) assert storage["cache"] assert isinstance(storage["reports"], str) assert storage["reports"] assert isinstance(storage["catalogs"], str) assert storage["catalogs"] # Verify namespaced is boolean assert isinstance(storage["namespaced"], bool) @pytest.mark.asyncio async def test_health_check_minimal_mode_excludes_storage_paths(monkeypatch: pytest.MonkeyPatch) -> None: """Test that minimal mode does NOT include storage paths (token efficiency).""" config = Config.from_env() service = StubSnowflakeService() tool = HealthCheckTool( config=config, snowflake_service=service, ) result = await tool.execute(response_mode="minimal", include_profile=False, include_cortex=False) # Minimal mode should not have diagnostics at all assert "diagnostics" not in result # Double-check storage_paths is not in the result anywhere assert "storage_paths" not in result.get("diagnostics", {}) @pytest.mark.asyncio async def test_storage_paths_reflect_repo_scope(monkeypatch: pytest.MonkeyPatch) -> None: """Test storage paths correctly reflect repo scope configuration.""" config = Config.from_env() service = StubSnowflakeService() monitor = StubHealthMonitor( profile_health=StubProfileHealth(is_valid=True), system_status=StubSystemStatus(), ) # Mock profile validation monkeypatch.setattr( "igloo_mcp.mcp.tools.health.validate_and_resolve_profile", lambda: "DEFAULT", ) monkeypatch.setattr( "igloo_mcp.mcp.tools.health.get_profile_summary", lambda: SummaryStub(), ) # Mock repo scope environment monkeypatch.setenv("IGLOO_MCP_LOG_SCOPE", "repo") tool = HealthCheckTool( config=config, snowflake_service=service, health_monitor=monitor, ) result = await tool.execute(response_mode="full", include_profile=True, include_cortex=False, include_catalog=False) storage = result["diagnostics"]["storage_paths"] # Verify scope is repo assert storage["scope"] == "repo" # Repo scope paths should NOT start with ~/.igloo_mcp home_igloo = str(Path.home() / ".igloo_mcp") assert not storage["base_directory"].startswith(home_igloo), ( f"Repo scope should not use global base: {storage['base_directory']}" )