"""Tests for Task A55: Account Connection Defect Fixes (P0-1 through P0-5).""" from __future__ import annotations import json from pathlib import Path from unittest.mock import MagicMock, patch import pytest from fastapi.testclient import TestClient from antigravity_provider.version import __version__, VERSION_INFO from antigravity_provider.router.action_handler import ActionExecutor from antigravity_provider.router.profile_manager import ProfileAuthManager from antigravity_provider.router.adapters.ollama_adapter import OllamaAdapter, DEFAULT_OLLAMA_BASE_URL from antigravity_provider.router.connection_preflight import validate_connection from antigravity_provider.router.model_discovery_service import ModelDiscoveryService from antigravity_provider.router.web.server import app # ── P0-1: Antigravity & Browser Auth Tests ── def test_p0_1_add_account_unauthenticated_browser_auth_error_message(): """add_account should return Russian browser auth message for antigravity and claude.""" with patch.object(ProfileAuthManager, "get_profile_status", return_value={"authenticated": False}): # Antigravity without token/completed auth res_ag = ActionExecutor.execute("add_account", { "provider": "antigravity", "profile_id": "ag-1", "token": "", }) assert not res_ag["ok"] assert "Авторизация через браузер не завершена. Пожалуйста, откройте ссылку входа или вставьте адрес возврата." in res_ag["message"] # Claude without token/completed auth res_cl = ActionExecutor.execute("add_account", { "provider": "claude", "profile_id": "claude-1", "token": "", }) assert not res_cl["ok"] assert "Авторизация через браузер не завершена. Пожалуйста, откройте ссылку входа или вставьте адрес возврата." in res_cl["message"] def test_p0_1_profile_auth_manager_get_profile_status_antigravity(): """get_profile_status should return authenticated: False when auth_data is empty or invalid.""" # Unauthenticated/empty auth_data with patch.object(ProfileAuthManager, "load_profile_auth", return_value={}): st = ProfileAuthManager.get_profile_status("antigravity", "ag-1") assert st["authenticated"] is False assert st["status"] == "NOT_CONFIGURED" # Authenticated with access_token with patch.object(ProfileAuthManager, "load_profile_auth", return_value={"token": {"access_token": "ya29.valid-token"}}): st = ProfileAuthManager.get_profile_status("antigravity", "ag-1") assert st["authenticated"] is True assert st["status"] == "AUTHENTICATED" def test_p0_1_load_profile_auth_gemini_oauth_creds_fallback(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): """load_profile_auth should check .gemini/oauth_creds.json if auth.json is missing.""" monkeypatch.setattr("antigravity_provider.paths.get_hermes_home", lambda: tmp_path) pdir = tmp_path / "agy_profiles" / "ag-5" gemini_dir = pdir / ".gemini" gemini_dir.mkdir(parents=True, exist_ok=True) creds_file = gemini_dir / "oauth_creds.json" creds_file.write_text(json.dumps({"access_token": "token-123", "refresh_token": "refresh-456"}), encoding="utf-8") auth = ProfileAuthManager.load_profile_auth("antigravity", "ag-5") assert auth is not None assert auth.get("provider") == "antigravity" assert auth.get("token", {}).get("access_token") == "token-123" # ── P0-2: Ollama Base URL & Connection Error Tests ── def test_p0_2_ollama_default_base_url(): """DEFAULT_OLLAMA_BASE_URL must be http://127.0.0.1:11434 (without trailing /v1).""" assert DEFAULT_OLLAMA_BASE_URL == "http://127.0.0.1:11434" def test_p0_2_ollama_adapter_url_builders(): """OllamaAdapter URL methods should handle base_urls with or without /v1 cleanly.""" adapter = OllamaAdapter() # Standard base url without /v1 assert adapter._get_native_host("http://127.0.0.1:11434") == "http://127.0.0.1:11434" assert adapter._get_chat_url("http://127.0.0.1:11434") == "http://127.0.0.1:11434/v1/chat/completions" assert adapter._get_models_url("http://127.0.0.1:11434") == "http://127.0.0.1:11434/v1/models" # Base url with /v1 assert adapter._get_native_host("http://192.168.1.81:11434/v1") == "http://192.168.1.81:11434" assert adapter._get_chat_url("http://192.168.1.81:11434/v1") == "http://192.168.1.81:11434/v1/chat/completions" assert adapter._get_models_url("http://192.168.1.81:11434/v1") == "http://192.168.1.81:11434/v1/models" # Base url with trailing slash assert adapter._get_native_host("http://192.168.1.81:11434/") == "http://192.168.1.81:11434" assert adapter._get_chat_url("http://192.168.1.81:11434/") == "http://192.168.1.81:11434/v1/chat/completions" def test_p0_2_ollama_connection_refused_helpful_message(): """Ollama connection refused error should provide network IP guidance.""" # Test connection preflight when endpoint is down res = validate_connection("ollama", token="", base_url="http://127.0.0.1:11434") assert not res["ok"] assert "Не удалось подключиться к http://127.0.0.1:11434 (Connection refused)" in res["message"] assert "http://192.168.1.81:11434" in res["message"] # ── P0-3: Antigravity Dynamic Model Discovery Tests ── def test_p0_3_antigravity_dynamic_model_discovery(): """ModelDiscoveryService probe should find and query any authenticated Antigravity profile.""" discovery = ModelDiscoveryService.get() with patch("antigravity_provider.router.profile_manager.ProfileAuthManager.get_profile_status") as mock_st, \ patch("antigravity_provider.agy_subprocess.discover_models", return_value={"gemini-2.5-pro": "gemini-2.5-pro"}) as mock_disc: # Simulate ag-3 being authenticated def status_side_effect(prov, pid): if pid == "ag-3": return {"authenticated": True, "provider": prov, "profile_id": pid} return {"authenticated": False, "provider": prov, "profile_id": pid} mock_st.side_effect = status_side_effect models, err = discovery._probe_provider("antigravity") assert err is None assert models == ["gemini-2.5-pro"] mock_disc.assert_called_with(profile_id="ag-3") # ── P0-4: Version Info & API Propagation Tests ── def test_p0_4_version_single_source_of_truth(): """Номер версии и VERSION_INFO обязаны совпадать между собой. Сверять с записанным в тесте числом бессмысленно: при каждой сборке его пришлось бы править, и тест превращался бы в напоминание, а не в проверку. Смысл требования — единый источник, его и проверяем. """ assert VERSION_INFO == tuple(int(part) for part in __version__.split(".")) def test_p0_4_version_in_api_endpoints(): """Все точки API обязаны отдавать ту же версию, что и пакет.""" client = TestClient(app) # /api/health res_health = client.get("/api/health") assert res_health.status_code == 200 assert res_health.json().get("version") == __version__ # /api/settings res_settings = client.get("/api/settings") assert res_settings.status_code == 200 assert res_settings.json().get("version") == __version__ # /api/snapshot res_snap = client.get("/api/snapshot") assert res_snap.status_code == 200 snap = res_snap.json() assert snap.get("version") == __version__ assert (snap.get("metrics") or {}).get("version") == __version__ # ── P0-5: Settings View & System Paths Tests ── def test_p0_5_settings_api_returns_system_paths_and_hub_settings(): """GET /api/settings must return system_paths, server host/port, token status, and hub settings.""" client = TestClient(app) res = client.get("/api/settings") assert res.status_code == 200 data = res.json() # System paths dict assert "system_paths" in data sys_paths = data["system_paths"] assert "hermes_home" in sys_paths assert "config_dir" in sys_paths assert "log_file" in sys_paths # Server settings & tokens assert "web_api_host" in data assert "web_api_port" in data assert "web_api_token_configured" in data # Snapshot includes system_paths res_snap = client.get("/api/snapshot") assert res_snap.status_code == 200 snap = res_snap.json() assert "system_paths" in snap assert snap["system_paths"]["hermes_home"] == sys_paths["hermes_home"]