Gan, Jimmy
574 lines
20 KiB
Python
574 lines
20 KiB
Python
"""
|
|
Shared pytest fixtures for backend tests.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
from datetime import datetime, timedelta
|
|
from unittest.mock import AsyncMock, Mock, patch
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
# Set up environment variables before any imports during pytest collection
|
|
os.environ.setdefault("SECRET_KEY", "test-secret-key-at-least-32-chars-long")
|
|
os.environ.setdefault("ADMIN_USER", "testadmin")
|
|
os.environ.setdefault("ADMIN_PASSWORD_HASH", "$pbkdf2-sha256$29000$N2YMIQQgBAAgxBgjhPD.vw$1234567890abcdef")
|
|
os.environ.setdefault("VOLUME_ROOT", "/tmp/test-volume")
|
|
os.environ.setdefault("DOCKER_HOST", "tcp://mock-docker:2375")
|
|
os.environ.setdefault("GITEA_URL", "http://mock-gitea:3000")
|
|
os.environ.setdefault("GITEA_TOKEN", "mock-token")
|
|
os.environ.setdefault("TRANSMISSION_USER", "test-tx-user")
|
|
os.environ.setdefault("TRANSMISSION_PASS", "test-tx-pass")
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_volume_root(tmp_path):
|
|
"""Create temporary volume root for testing."""
|
|
volume_root = tmp_path / "volume1"
|
|
volume_root.mkdir()
|
|
(volume_root / "docker" / "nas-dashboard").mkdir(parents=True)
|
|
return str(volume_root)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_config(temp_volume_root, monkeypatch):
|
|
"""Mock config module with test values."""
|
|
monkeypatch.setenv("SECRET_KEY", "test-secret-key-at-least-32-chars-long")
|
|
monkeypatch.setenv("ADMIN_USER", "testadmin")
|
|
monkeypatch.setenv("ADMIN_PASSWORD_HASH", "$pbkdf2-sha256$29000$N2YMIQQgBAAgxBgjhPD.vw$1234567890abcdef")
|
|
monkeypatch.setenv("VOLUME_ROOT", temp_volume_root)
|
|
monkeypatch.setenv("DOCKER_HOST", "tcp://mock-docker:2375")
|
|
monkeypatch.setenv("GITEA_URL", "http://mock-gitea:3000")
|
|
monkeypatch.setenv("GITEA_TOKEN", "mock-token")
|
|
monkeypatch.setenv("TRANSMISSION_USER", "test-tx-user")
|
|
monkeypatch.setenv("TRANSMISSION_PASS", "test-tx-pass")
|
|
|
|
# Reload config module to pick up new env vars
|
|
import importlib
|
|
|
|
import config
|
|
|
|
importlib.reload(config)
|
|
|
|
return config
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_auth_file(temp_volume_root):
|
|
"""Create temporary auth.json file."""
|
|
auth_file = os.path.join(temp_volume_root, "docker/nas-dashboard/auth.json")
|
|
data = {"totp_secret": "", "passkey_credentials": [], "token_version": 0, "last_totp_ts": 0}
|
|
with open(auth_file, "w") as f:
|
|
json.dump(data, f)
|
|
return auth_file
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_rbac_file(temp_volume_root):
|
|
"""Create temporary rbac.json file."""
|
|
rbac_file = os.path.join(temp_volume_root, "docker/nas-dashboard/rbac.json")
|
|
data = {
|
|
"role_defaults": {
|
|
"admin": {"pages": "*", "sidebar_links": "*"},
|
|
"member": {"pages": ["dashboard"], "sidebar_links": "*"},
|
|
"viewer": {"pages": ["dashboard"], "sidebar_links": ["dashboard"]},
|
|
},
|
|
"user_overrides": {},
|
|
}
|
|
with open(rbac_file, "w") as f:
|
|
json.dump(data, f)
|
|
return rbac_file
|
|
|
|
|
|
@pytest.fixture
|
|
def valid_access_token(mock_config):
|
|
"""Generate a valid access token for testing."""
|
|
from auth_service import create_access_token
|
|
|
|
return create_access_token(data={"sub": "testuser", "role": "admin"}, expires_delta=timedelta(minutes=30))
|
|
|
|
|
|
@pytest.fixture
|
|
def valid_refresh_token(mock_config, temp_auth_file):
|
|
"""Generate a valid refresh token for testing."""
|
|
from auth_service import create_refresh_token
|
|
|
|
return create_refresh_token(data={"sub": "testuser", "role": "admin"})
|
|
|
|
|
|
@pytest.fixture
|
|
def expired_access_token(mock_config):
|
|
"""Generate an expired access token for testing."""
|
|
from auth_service import create_access_token
|
|
|
|
return create_access_token(data={"sub": "testuser", "role": "admin"}, expires_delta=timedelta(seconds=-1))
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_docker_client():
|
|
"""Mock Docker client for testing.
|
|
|
|
Returns a mock client with a single running container.
|
|
For error scenarios, use mock_docker_client_with_errors.
|
|
"""
|
|
client = Mock()
|
|
|
|
# Mock container image
|
|
mock_image = Mock()
|
|
mock_image.tags = ["test-image:latest"]
|
|
mock_image.id = "sha256:abc123def456"
|
|
|
|
# Mock container
|
|
container = Mock()
|
|
container.id = "abc123"
|
|
container.short_id = "abc123"
|
|
container.name = "test-container"
|
|
container.status = "running"
|
|
container.image = mock_image
|
|
container.ports = {"80/tcp": [{"HostPort": "8080"}]}
|
|
container.attrs = {"State": {"Health": {"Status": "healthy"}}, "Config": {"Image": "test-image:latest"}}
|
|
container.logs.return_value = b"test log line 1\ntest log line 2\n"
|
|
|
|
client.containers.list.return_value = [container]
|
|
client.containers.get.return_value = container
|
|
|
|
return client
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_docker_client_with_errors():
|
|
"""Mock Docker client that simulates error scenarios.
|
|
|
|
Use this fixture to test error handling:
|
|
- Container not found (NotFound exception)
|
|
- Connection errors (APIError)
|
|
- Permission denied (APIError with 403)
|
|
|
|
Example:
|
|
def test_container_not_found(test_app, admin_headers, mock_docker_client_with_errors):
|
|
# Configure the mock to raise NotFound
|
|
import docker.errors
|
|
mock_docker_client_with_errors.containers.get.side_effect = docker.errors.NotFound("Container not found")
|
|
"""
|
|
import docker.errors
|
|
|
|
client = Mock()
|
|
|
|
# By default, raise NotFound for get operations
|
|
client.containers.get.side_effect = docker.errors.NotFound("Container not found")
|
|
client.containers.list.return_value = []
|
|
|
|
return client
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_docker_client_factory():
|
|
"""Factory for creating customizable Docker client mocks.
|
|
|
|
Returns a function that creates mock clients with specific configurations.
|
|
|
|
Example:
|
|
def test_custom_container(mock_docker_client_factory):
|
|
client = mock_docker_client_factory(
|
|
container_name="my-container",
|
|
container_status="stopped",
|
|
image_tag="my-image:v1.0"
|
|
)
|
|
"""
|
|
|
|
def _create_mock(
|
|
container_name="test-container",
|
|
container_status="running",
|
|
image_tag="test-image:latest",
|
|
container_id="abc123",
|
|
):
|
|
client = Mock()
|
|
|
|
# Mock container image
|
|
mock_image = Mock()
|
|
mock_image.tags = [image_tag]
|
|
mock_image.id = f"sha256:{container_id}def456"
|
|
|
|
# Mock container
|
|
container = Mock()
|
|
container.id = container_id
|
|
container.short_id = container_id[:12]
|
|
container.name = container_name
|
|
container.status = container_status
|
|
container.image = mock_image
|
|
container.ports = {"80/tcp": [{"HostPort": "8080"}]}
|
|
container.attrs = {"State": {"Health": {"Status": "healthy"}}, "Config": {"Image": image_tag}}
|
|
container.logs.return_value = b"test log line 1\ntest log line 2\n"
|
|
|
|
client.containers.list.return_value = [container]
|
|
client.containers.get.return_value = container
|
|
|
|
return client
|
|
|
|
return _create_mock
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_ssh_connection():
|
|
"""Mock asyncssh connection for testing."""
|
|
conn = AsyncMock()
|
|
process = AsyncMock()
|
|
process.stdout = AsyncMock()
|
|
process.stdin = AsyncMock()
|
|
process.stderr = AsyncMock()
|
|
conn.create_process.return_value = process
|
|
return conn
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_httpx_client():
|
|
"""Mock httpx AsyncClient for testing."""
|
|
client = AsyncMock()
|
|
response = AsyncMock()
|
|
response.status_code = 200
|
|
response.json.return_value = {"status": "ok"}
|
|
client.get.return_value = response
|
|
client.post.return_value = response
|
|
return client
|
|
|
|
|
|
class _DictRow:
|
|
"""Mock for asyncpg Record that supports both dict() conversion and positional access."""
|
|
def __init__(self, data: dict):
|
|
self._data = data
|
|
|
|
def __getitem__(self, key):
|
|
if isinstance(key, int):
|
|
return list(self._data.values())[key]
|
|
return self._data[key]
|
|
|
|
def keys(self):
|
|
return self._data.keys()
|
|
|
|
def values(self):
|
|
return self._data.values()
|
|
|
|
def __iter__(self):
|
|
return iter(self._data)
|
|
|
|
def get(self, key, default=None):
|
|
return self._data.get(key, default)
|
|
|
|
|
|
def _patch_opc_db():
|
|
"""Mock OPC database to avoid requiring PostgreSQL in tests.
|
|
|
|
Returns a context manager that patches db.opc_db.get_pool to return
|
|
an in-memory mock pool. Tasks/executions are stored in lists keyed by ID.
|
|
"""
|
|
import db.opc_db as _opc_db
|
|
|
|
_task_counter = [0]
|
|
_exec_counter = [0]
|
|
_tasks: dict[int, dict] = {}
|
|
_executions: dict[int, dict] = {}
|
|
_history: list[dict] = []
|
|
|
|
def _next_id(counter):
|
|
counter[0] += 1
|
|
return counter[0]
|
|
|
|
def _make_task(title, description, status, priority, project_id,
|
|
assigned_to, assigned_type, tags, due_date):
|
|
tid = _next_id(_task_counter)
|
|
now = datetime.utcnow()
|
|
task = {
|
|
"id": tid,
|
|
"title": title,
|
|
"description": description,
|
|
"status": status,
|
|
"priority": priority,
|
|
"project_id": project_id,
|
|
"assigned_to": assigned_to,
|
|
"assigned_type": assigned_type,
|
|
"tags": json.dumps(tags or []),
|
|
"metadata": json.dumps({"created_by": "admin"}),
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
"completed_at": None,
|
|
"due_date": due_date,
|
|
}
|
|
_tasks[tid] = task
|
|
return task
|
|
|
|
class _MockConnection:
|
|
async def fetchrow(self, query, *params):
|
|
query_lower = query.strip().lower()
|
|
# INSERT ... RETURNING
|
|
if query_lower.startswith("insert into tasks"):
|
|
title = params[0] if len(params) > 0 else ""
|
|
desc = params[1] if len(params) > 1 else None
|
|
status = params[2] if len(params) > 2 else "parking_lot"
|
|
priority = params[3] if len(params) > 3 else "medium"
|
|
pid = params[4] if len(params) > 4 else None
|
|
assigned_to = params[5] if len(params) > 5 else None
|
|
assigned_type = params[6] if len(params) > 6 else "human"
|
|
tags = params[7] if len(params) > 7 else None
|
|
due_date = params[8] if len(params) > 8 else None
|
|
task = _make_task(title, desc, status, priority, pid,
|
|
assigned_to, assigned_type, tags, due_date)
|
|
return _DictRow(task)
|
|
elif query_lower.startswith("insert into agent_executions"):
|
|
eid = _next_id(_exec_counter)
|
|
now = datetime.utcnow()
|
|
exec_data = {
|
|
"id": eid,
|
|
"task_id": params[0] if len(params) > 0 else None,
|
|
"agent_id": params[1] if len(params) > 1 else None,
|
|
"status": params[2] if len(params) > 2 else "pending",
|
|
"input_context": params[3] if len(params) > 3 else "{}",
|
|
"requires_approval": params[4] if len(params) > 4 else False,
|
|
"started_at": now,
|
|
"completed_at": None,
|
|
"output_result": None,
|
|
"actions_proposed": None,
|
|
"actions_executed": None,
|
|
"error_message": None,
|
|
"approved_by": None,
|
|
"approved_at": None,
|
|
}
|
|
_executions[eid] = exec_data
|
|
return _DictRow(exec_data)
|
|
elif query_lower.startswith("insert into task_history"):
|
|
_history.append({"task_id": params[0], "action": params[1]})
|
|
return None
|
|
elif query_lower.startswith("select * from tasks where id ="):
|
|
tid = params[0] if params else None
|
|
if tid in _tasks:
|
|
return _DictRow(_tasks[tid])
|
|
return None
|
|
elif query_lower.startswith("select * from agents where id ="):
|
|
agent_id = params[0] if params else None
|
|
agents_map = {
|
|
"pm": {"id": "pm", "name": "Project Manager", "role": "PM",
|
|
"description": "test", "capabilities": "[]", "system_prompt": "test",
|
|
"config": '{"approval_level":"auto"}', "enabled": True,
|
|
"created_at": datetime.utcnow().isoformat()},
|
|
"cto": {"id": "cto", "name": "CTO", "role": "CTO",
|
|
"description": "test", "capabilities": "[]", "system_prompt": "test",
|
|
"config": '{"approval_level":"cxo"}', "enabled": True,
|
|
"created_at": datetime.utcnow().isoformat()},
|
|
}
|
|
if agent_id in agents_map:
|
|
return _DictRow(agents_map[agent_id])
|
|
return None
|
|
elif query_lower.startswith("select * from agent_executions where id ="):
|
|
eid = params[0] if params else None
|
|
if eid in _executions:
|
|
return _DictRow(_executions[eid])
|
|
return None
|
|
elif "update tasks set" in query_lower:
|
|
tid = params[-1] if params else None
|
|
if tid in _tasks:
|
|
return _DictRow(_tasks[tid])
|
|
return None
|
|
elif "update agent_executions" in query_lower:
|
|
eid = params[-1] if params else None
|
|
if eid in _executions:
|
|
if "approved_by" in query_lower:
|
|
_executions[eid]["approved_by"] = params[0] if len(params) > 0 else None
|
|
_executions[eid]["approved_at"] = datetime.utcnow().isoformat()
|
|
_executions[eid]["status"] = "approved" if (len(params) > 1 and params[1]) else "rejected"
|
|
return _DictRow(_executions[eid])
|
|
return None
|
|
return None
|
|
|
|
async def fetch(self, query, *params):
|
|
query_lower = query.strip().lower()
|
|
if "from agent_executions" in query_lower:
|
|
return [_DictRow(e) for e in _executions.values()]
|
|
if "from tasks" in query_lower or "select * from tasks" in query_lower:
|
|
return [_DictRow(t) for t in _tasks.values()]
|
|
if "from task_history" in query_lower:
|
|
return [_DictRow(h) for h in _history]
|
|
if "from agents" in query_lower:
|
|
return [_DictRow({
|
|
"id": "pm", "name": "Project Manager", "role": "PM",
|
|
"description": "test", "capabilities": "[]", "system_prompt": "test",
|
|
"config": '{"approval_level":"auto"}', "enabled": True,
|
|
"created_at": datetime.utcnow().isoformat(),
|
|
}), _DictRow({
|
|
"id": "cto", "name": "CTO", "role": "CTO",
|
|
"description": "test", "capabilities": "[]", "system_prompt": "test",
|
|
"config": '{"approval_level":"cxo"}', "enabled": True,
|
|
"created_at": datetime.utcnow().isoformat(),
|
|
})]
|
|
if "from time_entries" in query_lower:
|
|
return []
|
|
return []
|
|
|
|
async def execute(self, query, *params):
|
|
# no-op for DDL/DML
|
|
return "OK"
|
|
|
|
class _MockAcquireContext:
|
|
async def __aenter__(self):
|
|
return _MockConnection()
|
|
|
|
async def __aexit__(self, *args):
|
|
pass
|
|
|
|
class _MockPool:
|
|
def acquire(self):
|
|
return _MockAcquireContext()
|
|
|
|
async def _mock_get_pool():
|
|
return _MockPool()
|
|
|
|
return patch.object(_opc_db, "get_pool", side_effect=_mock_get_pool)
|
|
|
|
|
|
@pytest.fixture
|
|
def test_app(mock_config, temp_auth_file, temp_rbac_file, mock_docker_client, mock_conversation_db):
|
|
"""Create FastAPI test client with mocked dependencies."""
|
|
# Reload routers to pick up new config values
|
|
import importlib
|
|
|
|
import routers.files
|
|
import routers.conversation_tracker
|
|
import routers.security
|
|
import routers.system
|
|
|
|
importlib.reload(routers.files)
|
|
importlib.reload(routers.conversation_tracker)
|
|
importlib.reload(routers.security)
|
|
importlib.reload(routers.system)
|
|
|
|
# Reset cached Docker client in security router
|
|
routers.security._client = None
|
|
|
|
# Patch Docker client before importing main
|
|
with patch("docker.DockerClient", return_value=mock_docker_client):
|
|
# Mock the limiter to disable rate limiting during tests
|
|
mock_limiter = Mock()
|
|
mock_limiter.limit = lambda *args, **kwargs: lambda func: func
|
|
|
|
with patch("slowapi.Limiter", return_value=mock_limiter):
|
|
# Mock OPC database to avoid requiring PostgreSQL in tests
|
|
with _patch_opc_db():
|
|
from main import app
|
|
|
|
client = TestClient(app)
|
|
yield client
|
|
|
|
|
|
@pytest.fixture
|
|
def admin_headers(valid_access_token):
|
|
"""Headers with admin access token."""
|
|
return {"Authorization": f"Bearer {valid_access_token}"}
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_rbac_data():
|
|
"""Sample RBAC configuration for testing."""
|
|
return {
|
|
"role_defaults": {
|
|
"admin": {"pages": "*", "sidebar_links": "*"},
|
|
"member": {"pages": ["dashboard", "files"], "sidebar_links": "*"},
|
|
"viewer": {"pages": ["dashboard"], "sidebar_links": ["dashboard"]},
|
|
},
|
|
"user_overrides": {
|
|
"customuser": {"pages": ["dashboard"], "sidebar_links": ["dashboard", "files"]}
|
|
},
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_totp_secret():
|
|
"""Sample TOTP secret for testing."""
|
|
return "JBSWY3DPEHPK3PXP" # Base32 encoded secret
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_request():
|
|
"""Mock FastAPI Request object."""
|
|
request = Mock()
|
|
request.client = Mock()
|
|
request.client.host = "192.168.31.100"
|
|
request.headers = {}
|
|
return request
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_conversation_db(tmp_path, monkeypatch):
|
|
"""Mock conversation tracker database"""
|
|
import sqlite3
|
|
|
|
db_path = tmp_path / "conversations.db"
|
|
|
|
# Create minimal SQLite schema
|
|
conn = sqlite3.connect(db_path)
|
|
conn.execute("""
|
|
CREATE TABLE conversations (
|
|
session_id TEXT PRIMARY KEY,
|
|
project_path TEXT,
|
|
git_branch TEXT,
|
|
slug TEXT,
|
|
started_at DATETIME,
|
|
last_updated_at DATETIME,
|
|
message_count INTEGER,
|
|
total_input_tokens INTEGER DEFAULT 0,
|
|
total_output_tokens INTEGER DEFAULT 0,
|
|
model TEXT
|
|
)
|
|
""")
|
|
conn.execute("""
|
|
CREATE TABLE messages (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
session_id TEXT NOT NULL,
|
|
message_uuid TEXT UNIQUE NOT NULL,
|
|
role TEXT NOT NULL,
|
|
content_text TEXT,
|
|
timestamp DATETIME NOT NULL,
|
|
model TEXT,
|
|
input_tokens INTEGER DEFAULT 0,
|
|
output_tokens INTEGER DEFAULT 0,
|
|
has_tool_use INTEGER DEFAULT 0,
|
|
has_thinking INTEGER DEFAULT 0
|
|
)
|
|
""")
|
|
conn.execute("""
|
|
CREATE TABLE tool_calls (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
message_uuid TEXT NOT NULL,
|
|
tool_name TEXT NOT NULL,
|
|
tool_input TEXT,
|
|
tool_result TEXT,
|
|
is_error INTEGER DEFAULT 0,
|
|
timestamp DATETIME NOT NULL
|
|
)
|
|
""")
|
|
conn.execute("""
|
|
CREATE TABLE daily_summaries (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
date TEXT UNIQUE NOT NULL,
|
|
summary_content TEXT NOT NULL,
|
|
conversation_count INTEGER DEFAULT 0,
|
|
total_messages INTEGER DEFAULT 0,
|
|
total_tokens INTEGER DEFAULT 0,
|
|
created_at DATETIME,
|
|
project_path TEXT
|
|
)
|
|
""")
|
|
|
|
# Insert test data
|
|
conn.execute("""
|
|
INSERT INTO conversations VALUES
|
|
('test-session-id', '/test/project', 'main', 'test-slug', '2026-04-19 10:00:00', '2026-04-19 11:00:00', 10, 1000, 500, 'claude-sonnet-4')
|
|
""")
|
|
conn.execute("""
|
|
INSERT INTO messages VALUES
|
|
(1, 'test-session-id', 'msg-1', 'user', 'test message', '2026-04-19 10:00:00', 'claude-sonnet-4', 100, 50, 0, 0)
|
|
""")
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
monkeypatch.setenv("CONVERSATION_TRACKER_DB", str(db_path))
|
|
return db_path
|