"""Web dashboard test fixtures.""" from typing import Any, Generator from unittest.mock import MagicMock, patch import pytest from fastapi.testclient import TestClient from httpx import Response from meshcore_hub.web.app import create_app # Explicit all-enabled features dict so tests are not affected by the user's # local .env file (pydantic-settings loads .env by default). ALL_FEATURES_ENABLED = { "dashboard": True, "nodes": True, "advertisements": True, "messages": True, "map": True, "members": True, "pages": True, } class MockHttpClient: """Mock HTTP client for testing web routes.""" def __init__(self) -> None: """Initialize mock client with default responses.""" self._responses: dict[str, dict[str, Any]] = {} self._default_responses() def _default_responses(self) -> None: """Set up default mock API responses.""" # Default stats response self._responses["GET:/api/v1/dashboard/stats"] = { "status_code": 200, "json": { "total_nodes": 10, "active_nodes": 5, "total_messages": 100, "messages_today": 15, "total_advertisements": 50, "channel_message_counts": {"0": 30, "1": 20}, }, } # Default nodes list response self._responses["GET:/api/v1/nodes"] = { "status_code": 200, "json": { "items": [ { "id": "node-1", "public_key": "abc123def456abc123def456abc123def456abc123def456abc123def456abc1", "name": "Node One", "adv_type": "REPEATER", "last_seen": "2024-01-01T12:00:00Z", "tags": [], }, { "id": "node-2", "public_key": "def456abc123def456abc123def456abc123def456abc123def456abc123def4", "name": "Node Two", "adv_type": "CLIENT", "last_seen": "2024-01-01T11:00:00Z", "tags": [ {"key": "lat", "value": "40.7128"}, {"key": "lon", "value": "-74.0060"}, ], }, ], "total": 2, }, } # Default single node response (exact match) self._responses[ "GET:/api/v1/nodes/abc123def456abc123def456abc123def456abc123def456abc123def456abc1" ] = { "status_code": 200, "json": { "id": "node-1", "public_key": "abc123def456abc123def456abc123def456abc123def456abc123def456abc1", "name": "Node One", "adv_type": "REPEATER", "last_seen": "2024-01-01T12:00:00Z", "tags": [], }, } # Default messages response self._responses["GET:/api/v1/messages"] = { "status_code": 200, "json": { "items": [ { "id": "msg-1", "message_type": "direct", "pubkey_prefix": "abc123", "text": "Hello World", "received_at": "2024-01-01T12:00:00Z", "snr": -5.5, "hops": 2, }, { "id": "msg-2", "message_type": "channel", "channel_idx": 0, "text": "Channel message", "received_at": "2024-01-01T11:00:00Z", "snr": None, "hops": None, }, ], "total": 2, }, } # Default advertisements response self._responses["GET:/api/v1/advertisements"] = { "status_code": 200, "json": { "items": [ { "id": "adv-1", "public_key": "abc123def456abc123def456abc123def456abc123def456abc123def456abc1", "name": "Node One", "adv_type": "REPEATER", "received_at": "2024-01-01T12:00:00Z", }, ], "total": 1, }, } # Default telemetry response self._responses["GET:/api/v1/telemetry"] = { "status_code": 200, "json": { "items": [ { "id": "tel-1", "node_public_key": "abc123def456abc123def456abc123def456abc123def456abc123def456abc1", "parsed_data": {"battery_level": 85.5}, "received_at": "2024-01-01T12:00:00Z", }, ], "total": 1, }, } # Default members response (empty) self._responses["GET:/api/v1/members"] = { "status_code": 200, "json": { "items": [], "total": 0, "limit": 100, "offset": 0, }, } # Default activity response (for home page chart) self._responses["GET:/api/v1/dashboard/activity"] = { "status_code": 200, "json": { "days": 7, "data": [ {"date": "2024-01-01", "count": 10}, {"date": "2024-01-02", "count": 15}, {"date": "2024-01-03", "count": 8}, {"date": "2024-01-04", "count": 12}, {"date": "2024-01-05", "count": 20}, {"date": "2024-01-06", "count": 5}, {"date": "2024-01-07", "count": 18}, ], }, } # Default message activity response (for network page chart) self._responses["GET:/api/v1/dashboard/message-activity"] = { "status_code": 200, "json": { "days": 7, "data": [ {"date": "2024-01-01", "count": 5}, {"date": "2024-01-02", "count": 8}, {"date": "2024-01-03", "count": 3}, {"date": "2024-01-04", "count": 10}, {"date": "2024-01-05", "count": 7}, {"date": "2024-01-06", "count": 2}, {"date": "2024-01-07", "count": 9}, ], }, } # Default node count response (for network page chart) self._responses["GET:/api/v1/dashboard/node-count"] = { "status_code": 200, "json": { "days": 7, "data": [ {"date": "2024-01-01", "count": 5}, {"date": "2024-01-02", "count": 6}, {"date": "2024-01-03", "count": 7}, {"date": "2024-01-04", "count": 8}, {"date": "2024-01-05", "count": 9}, {"date": "2024-01-06", "count": 9}, {"date": "2024-01-07", "count": 10}, ], }, } # Health check response self._responses["GET:/health"] = { "status_code": 200, "json": {"status": "healthy"}, } def set_response( self, method: str, path: str, status_code: int = 200, json_data: Any = None ) -> None: """Set a custom response for a specific endpoint. Args: method: HTTP method (GET, POST, etc.) path: URL path status_code: Response status code json_data: JSON response body """ key = f"{method}:{path}" self._responses[key] = { "status_code": status_code, "json": json_data, } def _create_response(self, key: str) -> Response: """Create a mock response for a given key.""" import json as _json response_data = self._responses.get(key) if response_data is None: # Return 404 for unknown endpoints response = MagicMock(spec=Response) response.status_code = 404 response.json.return_value = {"detail": "Not found"} response.content = b'{"detail": "Not found"}' response.headers = {"content-type": "application/json"} return response response = MagicMock(spec=Response) response.status_code = response_data["status_code"] response.json.return_value = response_data["json"] response.content = _json.dumps(response_data["json"]).encode() response.headers = {"content-type": "application/json"} return response async def request( self, method: str, url: str, params: dict | None = None, content: bytes | None = None, headers: dict | None = None, ) -> Response: """Mock generic request (used by API proxy).""" key = f"{method.upper()}:{url}" if key in self._responses: return self._create_response(key) # Try base path without query params base_path = url.split("?")[0] key = f"{method.upper()}:{base_path}" return self._create_response(key) async def get(self, path: str, params: dict | None = None) -> Response: """Mock GET request.""" # Try exact match first key = f"GET:{path}" if key in self._responses: return self._create_response(key) # Try without query params for list endpoints base_path = path.split("?")[0] key = f"GET:{base_path}" return self._create_response(key) async def post( self, path: str, json: dict | None = None, params: dict | None = None ) -> Response: """Mock POST request.""" key = f"POST:{path}" return self._create_response(key) async def put( self, path: str, json: dict | None = None, params: dict | None = None ) -> Response: """Mock PUT request.""" key = f"PUT:{path}" return self._create_response(key) async def delete(self, path: str, params: dict | None = None) -> Response: """Mock DELETE request.""" key = f"DELETE:{path}" return self._create_response(key) async def aclose(self) -> None: """Mock close method.""" pass @pytest.fixture def mock_http_client() -> MockHttpClient: """Create a mock HTTP client.""" return MockHttpClient() @pytest.fixture def web_app(mock_http_client: MockHttpClient, monkeypatch: pytest.MonkeyPatch) -> Any: """Create a web app with mocked HTTP client.""" # Ensure tests use a consistent locale regardless of local .env monkeypatch.setenv("WEB_DATETIME_LOCALE", "en-US") monkeypatch.setenv("COLLECTOR_INCLUDE_TEST_CHANNEL", "true") monkeypatch.setenv("OIDC_ENABLED", "false") app = create_app( api_url="http://localhost:8000", api_key="test-api-key", network_name="Test Network", network_city="Test City", network_country="Test Country", network_radio_config="Test Radio Config", network_contact_email="test@example.com", network_contact_discord="https://discord.gg/test", features=ALL_FEATURES_ENABLED, ) # Override the lifespan to use our mock client app.state.http_client = mock_http_client return app @pytest.fixture def client(web_app: Any, mock_http_client: MockHttpClient) -> TestClient: """Create a test client for the web app. Note: We don't use the context manager to skip lifespan events since we've already set up the mock client. """ # Ensure the mock client is attached web_app.state.http_client = mock_http_client return TestClient(web_app, raise_server_exceptions=True) @pytest.fixture def web_app_with_oidc( mock_http_client: MockHttpClient, monkeypatch: pytest.MonkeyPatch ) -> Any: """Create a web app with OIDC enabled and session middleware.""" monkeypatch.setenv("OIDC_ENABLED", "true") monkeypatch.setenv("OIDC_CLIENT_ID", "test-client-id") monkeypatch.setenv("OIDC_CLIENT_SECRET", "test-client-secret") monkeypatch.setenv( "OIDC_DISCOVERY_URL", "https://idp.example.com/.well-known/openid-configuration" ) monkeypatch.setenv("OIDC_SESSION_SECRET", "test-session-secret") monkeypatch.setenv("WEB_DATETIME_LOCALE", "en-US") monkeypatch.setenv("COLLECTOR_INCLUDE_TEST_CHANNEL", "true") app = create_app( api_url="http://localhost:8000", api_key="test-api-key", network_name="Test Network", features=ALL_FEATURES_ENABLED, ) app.state.http_client = mock_http_client return app ADMIN_USER = { "sub": "admin-1", "name": "Admin User", "email": "admin@example.com", "picture": None, "roles": ["admin", "member"], } MEMBER_USER = { "sub": "member-1", "name": "Member User", "email": "member@example.com", "picture": None, "roles": ["member"], } @pytest.fixture def client_with_oidc( web_app_with_oidc: Any, mock_http_client: MockHttpClient ) -> TestClient: """Create a test client with OIDC enabled (no session).""" web_app_with_oidc.state.http_client = mock_http_client return TestClient(web_app_with_oidc, raise_server_exceptions=True) @pytest.fixture def client_with_oidc_admin_session( web_app_with_oidc: Any, mock_http_client: MockHttpClient ) -> Generator[TestClient, None, None]: """Create a test client with OIDC enabled and admin session via mock.""" web_app_with_oidc.state.http_client = mock_http_client with ( patch("meshcore_hub.web.app.get_session_user", return_value=ADMIN_USER), patch("meshcore_hub.web.oidc.get_session_user", return_value=ADMIN_USER), ): yield TestClient(web_app_with_oidc, raise_server_exceptions=True) @pytest.fixture def client_with_oidc_member_session( web_app_with_oidc: Any, mock_http_client: MockHttpClient ) -> Generator[TestClient, None, None]: """Create a test client with OIDC enabled and member session via mock.""" web_app_with_oidc.state.http_client = mock_http_client with ( patch("meshcore_hub.web.app.get_session_user", return_value=MEMBER_USER), patch("meshcore_hub.web.oidc.get_session_user", return_value=MEMBER_USER), ): yield TestClient(web_app_with_oidc, raise_server_exceptions=True) @pytest.fixture def web_app_no_features(mock_http_client: MockHttpClient) -> Any: """Create a web app with all features disabled.""" app = create_app( api_url="http://localhost:8000", api_key="test-api-key", network_name="Test Network", network_city="Test City", network_country="Test Country", features={ "dashboard": False, "nodes": False, "advertisements": False, "messages": False, "map": False, "members": False, "pages": False, }, ) app.state.http_client = mock_http_client return app @pytest.fixture def client_no_features( web_app_no_features: Any, mock_http_client: MockHttpClient ) -> TestClient: """Create a test client with all features disabled.""" web_app_no_features.state.http_client = mock_http_client return TestClient(web_app_no_features, raise_server_exceptions=True)