python/tests/conftest.py

"""Simple test configuration for Archon - Essential tests only."""

import os
import uuid
from unittest.mock import MagicMock, patch

import httpx
import pytest
import respx
from fastapi.testclient import TestClient

# Set test environment - always override to ensure test isolation
os.environ["TEST_MODE"] = "true"
os.environ["TESTING"] = "true"
# Set fake database credentials to prevent connection attempts
os.environ["SUPABASE_URL"] = "https://test.supabase.co"
os.environ["SUPABASE_SERVICE_KEY"] = "test-key"
# Set required port environment variables for ServiceDiscovery
os.environ["ARCHON_SERVER_PORT"] = "8181"
os.environ["ARCHON_MCP_PORT"] = "8051"
os.environ["ARCHON_AGENTS_PORT"] = "8052"

# Global patches that need to be active during module imports and app initialization
from unittest.mock import AsyncMock

# --- START: GLOBAL LLM MOCKING DEFINITIONS ---
# Mock GenAI Response
mock_genai_response = MagicMock()
mock_genai_response.text = "Mocked GenAI Response in Traditional Chinese (繁體中文)"

# Mock GenAI Client
mock_genai_client_instance = MagicMock()
mock_genai_client_instance.models.generate_content.return_value = mock_genai_response
mock_genai_client_instance.aio.models.generate_content = AsyncMock(return_value=mock_genai_response)

# Mock OpenAI Response
mock_openai_response = MagicMock()
mock_openai_choice = MagicMock()
mock_openai_message = MagicMock()
mock_openai_message.content = "Mocked OpenAI Response in Traditional Chinese (繁體中文)"
mock_openai_choice.message = mock_openai_message
mock_openai_response.choices = [mock_openai_choice]

# Mock OpenAI Embeddings Response
mock_embeddings_response = MagicMock()
mock_embedding_data = MagicMock()
mock_embedding_data.embedding = [0.1] * 384
mock_embeddings_response.data = [mock_embedding_data]

# Mock OpenAI Client Instances
mock_openai_client_instance = MagicMock()
mock_openai_client_instance.chat.completions.create = AsyncMock(return_value=mock_openai_response)
mock_openai_client_instance.embeddings.create = AsyncMock(return_value=mock_embeddings_response)

mock_openai_sync_client_instance = MagicMock()
mock_openai_sync_client_instance.chat.completions.create.return_value = mock_openai_response
mock_openai_sync_client_instance.embeddings.create.return_value = mock_embeddings_response

# Mock LiteLLM Response
mock_litellm_response = {
    "choices": [
        {
            "message": {
                "content": "Mocked LiteLLM Summary Response"
            }
        }
    ]
}

# Mock PydanticAI RunResult
mock_run_result = MagicMock()
mock_run_result.data = "Mocked Agent Response"
mock_run_result.output = "Mocked Agent Response"
mock_run_result.usage = MagicMock()
mock_run_result.usage.input_tokens = 10
mock_run_result.usage.output_tokens = 10
mock_run_result.model_used = "mock-model"

# --- END: GLOBAL LLM MOCKING DEFINITIONS ---

# --- START: PYTEST NETWORK & KEY BOUNDARY CHECKS ---
def check_internet_connection():
    import socket
    try:
        socket.setdefaulttimeout(1)
        socket.socket(socket.AF_INET, socket.SOCK_STREAM).connect(("8.8.8.8", 53))
        return True
    except OSError:
        return False

def pytest_configure(config):
    config.addinivalue_line("markers", "integration: mark test as integration test")

def pytest_runtest_setup(item):
    if "integration" in item.keywords:
        api_key = os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY")
        if not api_key or not check_internet_connection():
            pytest.skip("Skipping integration test: No internet or API keys found.")
# --- END: PYTEST BOUNDARY CHECKS ---

mock_client = MagicMock()
mock_table = MagicMock()
mock_select = MagicMock()
mock_execute = MagicMock()
mock_execute.data = []
mock_select.execute.return_value = mock_execute
mock_select.eq.return_value = mock_select
mock_select.order.return_value = mock_select
mock_table.select.return_value = mock_select
mock_client.table.return_value = mock_table

import sys  # noqa: E402
from pathlib import Path  # noqa: E402

# --- START: CRITICAL PATH FIX FOR PYTEST SRC-LAYOUT ---
# This ensures Python can find 'server' as a top-level module when running tests
src_path = str(Path(__file__).parent.parent / "src")
if src_path not in sys.path:
    sys.path.insert(0, src_path)
# --- END: CRITICAL PATH FIX ---

# Apply global patches immediately
_global_patches = [
    patch("supabase.create_client", return_value=mock_client),
    patch("server.services.client_manager.get_supabase_client", return_value=mock_client),
    patch("server.utils.get_supabase_client", return_value=mock_client),
    patch("google.genai.Client", return_value=mock_genai_client_instance),
    patch("openai.OpenAI", return_value=mock_openai_sync_client_instance),
    patch("openai.AsyncOpenAI", return_value=mock_openai_client_instance),
    patch("litellm.completion", AsyncMock(return_value=mock_litellm_response)),
    patch("pydantic_ai.Agent.run", AsyncMock(return_value=mock_run_result)),
    patch("pydantic_ai.Agent.run_sync", MagicMock(return_value=mock_run_result)),
]

print("\n--- DEBUG CONTEST.PY IMPORT ENVIRONMENT ---")
print(f"sys.path: {sys.path}")
print(f"sys.modules keys (first 20): {list(sys.modules.keys())[:20]}")
print(f"sys.modules['server']: {sys.modules.get('server')}")
print(f"sys.modules['server.services']: {sys.modules.get('server.services')}")
print("--- END DEBUG ---")

for p in _global_patches:
    p.start()


# --- START: STATEFUL MOCK SUPABASE CLIENT DEFINITION ---



class MockQueryBuilder:
    def __init__(self, data: list, table_name: str, client: 'StatefulMockSupabaseClient', action: str, action_data: any = None):
        self.data = data
        self.table_name = table_name
        self.client = client
        self.action = action
        self.action_data = action_data
        self.filters = []
        self.orders = []
        self.limit_val = None

    def eq(self, field: str, value: any):
        self.filters.append(lambda x: x.get(field) == value)
        return self

    def neq(self, field: str, value: any):
        self.filters.append(lambda x: x.get(field) != value)
        return self

    def ilike(self, field: str, pattern: str):
        clean_pat = pattern.replace("%", "").lower()
        self.filters.append(lambda x: clean_pat in str(x.get(field, "")).lower())
        return self

    def contains(self, field: str, value: any):
        def _check_contains(x):
            val = x.get(field)
            if isinstance(val, list):
                if isinstance(value, list):
                    return all(item in val for item in value)
                return value in val
            return False
        self.filters.append(_check_contains)
        return self

    def or_(self, filter_str: str):
        # Allow pass-through for test simplicity
        return self

    def order(self, field: str, desc: bool = False):
        self.orders.append((field, desc))
        return self

    def limit(self, limit_val: int):
        self.limit_val = limit_val
        return self

    def execute(self):
        target_records = self.data

        filtered_records = []
        for r in target_records:
            match = True
            for f in self.filters:
                try:
                    if not f(r):
                        match = False
                        break
                except Exception:
                    match = False
                    break
            if match:
                filtered_records.append(r)

        for field, desc in self.orders:
            try:
                filtered_records.sort(key=lambda x: x.get(field), reverse=desc)
            except Exception:
                pass

        if self.limit_val is not None:
            filtered_records = filtered_records[:self.limit_val]

        response = MagicMock()

        if self.action == 'select':
            response.data = [dict(r) for r in filtered_records]

        elif self.action == 'insert':
            inserted = []
            if isinstance(self.action_data, list):
                for item in self.action_data:
                    new_item = dict(item)
                    if "id" not in new_item:
                        new_item["id"] = str(uuid.uuid4())
                    self.data.append(new_item)
                    inserted.append(new_item)
            else:
                new_item = dict(self.action_data)
                if "id" not in new_item:
                    new_item["id"] = str(uuid.uuid4())
                self.data.append(new_item)
                inserted.append(new_item)
            response.data = inserted

        elif self.action == 'update':
            updated = []
            for r in filtered_records:
                r.update(self.action_data)
                updated.append(dict(r))
            response.data = updated

        elif self.action == 'delete':
            for r in filtered_records:
                if r in self.data:
                    self.data.remove(r)
            response.data = [dict(r) for r in filtered_records]

        return response


class MockTable:
    def __init__(self, data: list, table_name: str, client: 'StatefulMockSupabaseClient'):
        self.data = data
        self.table_name = table_name
        self.client = client

    def select(self, fields: str = "*"):
        return MockQueryBuilder(self.data, self.table_name, self.client, 'select')

    def insert(self, data: dict | list):
        return MockQueryBuilder(self.data, self.table_name, self.client, 'insert', data)

    def update(self, data: dict):
        return MockQueryBuilder(self.data, self.table_name, self.client, 'update', data)

    def delete(self):
        return MockQueryBuilder(self.data, self.table_name, self.client, 'delete')


class StatefulMockSupabaseClient:
    def __init__(self):
        self._db = {}
        self.auth = MagicMock()
        self.auth.get_user.return_value = None
        self.storage = MagicMock()

    def table(self, table_name: str):
        if table_name not in self._db:
            self._db[table_name] = []
        return MockTable(self._db[table_name], table_name, self)

# --- END: STATEFUL MOCK SUPABASE CLIENT DEFINITION ---


@pytest.fixture(autouse=True)
def ensure_test_environment():
    """Ensure test environment is properly set for each test."""
    # Force test environment settings - this runs before each test
    os.environ["TEST_MODE"] = "true"
    os.environ["TESTING"] = "true"
    os.environ["SUPABASE_URL"] = "https://test.supabase.co"
    os.environ["SUPABASE_SERVICE_KEY"] = "test-key"
    os.environ["ARCHON_SERVER_PORT"] = "8181"
    os.environ["ARCHON_MCP_PORT"] = "8051"
    os.environ["ARCHON_AGENTS_PORT"] = "8052"
    yield


@pytest.fixture(scope="function", autouse=True)
def mock_supabase_db():
    """Shared stateful Supabase database emulator that intercepts network requests."""
    client = StatefulMockSupabaseClient()
    with respx.mock(assert_all_called=False) as respx_mock:
        # Allow pass-through for live AI requests in integration tests
        respx_mock.route(host__regex=r".*googleapis\.com").pass_through()

        respx_mock.route(host__regex=r".*supabase\.co").mock(
            side_effect=httpx.ConnectError("Blocked connection to Supabase during hermetic tests.")
        )
        yield client


@pytest.fixture(autouse=True)
def prevent_real_db_calls(mock_supabase_db):
    """Automatically prevent any real database calls in all tests."""
    with patch("supabase.create_client", return_value=mock_supabase_db):
        with patch("server.services.client_manager.get_supabase_client", return_value=mock_supabase_db):
            with patch("server.utils.get_supabase_client", return_value=mock_supabase_db):
                yield


@pytest.fixture
def mock_supabase_client(mock_supabase_db):
    """Mock Supabase client for testing."""
    return mock_supabase_db


@pytest.fixture
def client(mock_supabase_client):
    """FastAPI test client with mocked database."""
    # Patch all the ways Supabase client can be created
    with patch(
        "server.services.client_manager.get_supabase_client",
        return_value=mock_supabase_client,
    ):
        with patch(
            "server.utils.get_supabase_client",
            return_value=mock_supabase_client,
        ):
            with patch(
                "src.server.utils.get_supabase_client",
                return_value=mock_supabase_client,
            ):
                with patch(
                    "server.services.credential_service.create_client",
                    return_value=mock_supabase_client,
                    create=True,
                ):
                    with patch("supabase.create_client", return_value=mock_supabase_client):
                        from unittest.mock import AsyncMock

                        import server.main as server_main
                        from server.auth.dependencies import get_current_user

                    # Mark initialization as complete for testing (before accessing app)
                    server_main._initialization_complete = True
                    app = server_main.app

                    # Mock the schema check to always return valid
                    mock_schema_check = AsyncMock(return_value={"valid": True, "message": "Schema is up to date"})

                    # Global Auth Mock for tests
                    app.dependency_overrides[get_current_user] = lambda: {
                        "id": "test-admin",
                        "role": "admin",
                        "email": "admin@test.com",
                    }

                    with patch("server.main._check_database_schema", new=mock_schema_check, create=True):
                        return TestClient(app)


@pytest.fixture
def test_project():
    """Simple test project data."""
    return {"title": "Test Project", "description": "A test project for essential tests"}


@pytest.fixture
def test_task():
    """Simple test task data."""
    return {
        "title": "Test Task",
        "description": "A test task for essential tests",
        "status": "todo",
        "assignee": "User",
    }


@pytest.fixture
def test_knowledge_item():
    """Simple test knowledge item data."""
    return {
        "url": "https://example.com/test",
        "title": "Test Knowledge Item",
        "content": "This is test content for knowledge base",
        "source_id": "test-source",
    }