alex/headquarter
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

merge: fix/stale-backend-test-cleanup

Browse Source
This commit is contained in:
Developer
2026-06-12 20:23:21 +00:00
parent 79be4eb525 81b9a66ef5
commit c6073fe9d2
35 changed files with 268 additions and 625 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/src/api/tool/tool_types_validation.py
+8 -2
View File
@@ -60,8 +60,14 @@ def check_port_exposed(parsed: dict, port: int) -> bool:
for service_config in parsed["services"].values():
if isinstance(service_config, dict) and "ports" in service_config:
for port_mapping in service_config["ports"]:
if isinstance(port_mapping, str) and port_str in port_mapping:
return True
if isinstance(port_mapping, str):
# Port mappings can be "host:container", "ip:host:container",
# "container", or ".../protocol". The container port is the
# last numeric segment before any protocol suffix.
mapping = port_mapping.split("/")[0]
parts = mapping.split(":")
if parts[-1] == port_str:
return True
elif isinstance(port_mapping, int) and port_mapping == port:
return True
apps/api/src/config.py
+6 -1
View File
@@ -47,6 +47,11 @@ class Settings(BaseSettings):
session_secret: str = "change-me-session-secret"
session_ttl_hours: int = 24
# Internal JWT configuration (used for service-to-service tokens)
jwt_algorithm: str = "HS256"
access_token_ttl_minutes: int = 15
refresh_token_ttl_days: int = 7
# Repository storage
repo_base_path: str = "/data/repos"
@@ -122,7 +127,7 @@ class Settings(BaseSettings):
@property
def cookie_samesite(self) -> str:
if self.app_env == "production":
return "none"
return "strict"
return "lax"
apps/api/src/database.py
+3 -2
View File
@@ -1,5 +1,6 @@
import asyncio
import logging
import os
import subprocess
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
@@ -54,10 +55,10 @@ async def init_database(
result = await asyncio.get_event_loop().run_in_executor(
None,
lambda: subprocess.run(
["alembic", "upgrade", "head"],
["python3", "-m", "alembic", "upgrade", "head"],
capture_output=True,
text=True,
cwd="/app",
cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
),
)
apps/api/src/schemas/tool/tool_type.py
+3 -3
View File
@@ -82,8 +82,8 @@ class ToolTypeCreate(BaseModel):
@classmethod
def validate_default_port(cls, v: int, info) -> int:
data = info.data
requires_port = data.get("requires_port", True)
if not requires_port:
requires_port = data.get("requires_port")
if requires_port is False:
return v
if v <= 0 or v > 65535:
raise ValueError("Port must be between 1 and 65535")
@@ -140,7 +140,7 @@ class ToolTypeCreate(BaseModel):
if not check_port_exposed(parsed, self.default_port):
raise ValueError(
f"Port {self.default_port} is not exposed in the compose template. Add it to the 'ports' section."
f"default_port {self.default_port} is not exposed in the compose template. Add it to the 'ports' section."
)
return self
apps/api/tests/conftest.py
+4 -4
View File
@@ -87,7 +87,7 @@ def authenticated_client(test_client) -> Generator[TestClient, None, None]:
"""Provide an authenticated test client with a test user."""
import uuid
from src.auth.session import create_session_cookie
from src.models.user import User
from src.models.user.user import User
user_id = str(uuid.uuid4())
settings = Settings()
@@ -135,8 +135,8 @@ def authenticated_client(test_client) -> Generator[TestClient, None, None]:
def test_project_and_repo(authenticated_client) -> tuple[str, str]:
"""Create a project and repository directly in the database."""
import uuid
from src.models.project import Project
from src.models.git_repository import GitRepository
from src.models.project.project import Project
from src.models.project.git_repository import GitRepository
project_id = uuid.uuid4()
repo_id = uuid.uuid4()
@@ -195,7 +195,7 @@ def admin_client(test_client) -> Generator[TestClient, None, None]:
"""Provide an authenticated test client with an admin user."""
import uuid
from src.auth.session import create_session_cookie
from src.models.user import User
from src.models.user.user import User
user_id = str(uuid.uuid4())
settings = Settings()
apps/api/tests/integration/test_auth_api.py
+25 -2
View File
@@ -10,7 +10,30 @@ from sqlalchemy.ext.asyncio import create_async_engine
from src.auth.session import create_session_cookie
from src.config import Settings, build_database_url
from src.models import Base
from src.models.user import User
from src.models.user.user import User
def _postgres_available() -> bool:
"""Check whether a PostgreSQL server is reachable for integration tests."""
import asyncpg
async def _check() -> bool:
try:
conn = await asyncpg.connect(
host="localhost", port=5432, user="headquarter", password="headquarter", database="headquarter"
)
await conn.close()
return True
except Exception:
return False
return asyncio.run(_check())
pytestmark = pytest.mark.skipif(
not _postgres_available(),
reason="PostgreSQL not available on localhost:5432",
)
def _prepare_auth_test_db() -> None:
@@ -34,7 +57,7 @@ def _prepare_auth_test_db() -> None:
def _load_app():
import src.database as database_module
import src.api.auth as auth_module
import src.api.user.auth as auth_module
import src.main as main_module
importlib.reload(database_module)
apps/api/tests/integration/test_events.py
+10 -10
View File
@@ -10,15 +10,15 @@ from fastapi.testclient import TestClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from src.api import events as events_module
from src.api.system import events as events_module
from src.auth.session import decode_session_cookie
from src.config import Settings
from src.models.git_repository import GitRepository
from src.models.instance_event import InstanceEvent
from src.models.project import Project
from src.models.tool_instance import ToolInstance
from src.models.tool_type import ToolType
from src.services.event_bus import InstanceEventBus, InstanceEventPayload
from src.models.project.git_repository import GitRepository
from src.models.system.instance_event import InstanceEvent
from src.models.project.project import Project
from src.models.tool.tool_instance import ToolInstance
from src.models.tool.tool_type import ToolType
from src.services.instance.event_bus import InstanceEventBus, InstanceEventPayload
@pytest.fixture
@@ -75,7 +75,7 @@ def test_sse_enforces_connection_limit(authenticated_client: TestClient) -> None
@pytest.mark.integration
def test_sse_event_generator_format() -> None:
"""Test the SSE endpoint is registered."""
from src.api.events import router
from src.api.system.events import router
route_paths = [getattr(r, "path", "") for r in router.routes]
assert any("/stream" in str(p) for p in route_paths)
@@ -142,7 +142,7 @@ async def test_lifecycle_hook_publishes_event_and_persists(
event_bus.subscribe("instance.created", subscriber)
from src.services.lifecycle_hooks import publish_lifecycle_event
from src.services.instance.lifecycle_hooks import publish_lifecycle_event
await publish_lifecycle_event(
event_bus=event_bus,
@@ -220,7 +220,7 @@ async def test_lifecycle_event_persists_audit_row(
db_session.add(instance)
await db_session.commit()
from src.services.lifecycle_hooks import publish_lifecycle_event
from src.services.instance.lifecycle_hooks import publish_lifecycle_event
await publish_lifecycle_event(
event_bus=event_bus,
apps/api/tests/integration/test_models.py
+6 -5
View File
@@ -3,11 +3,11 @@ from sqlalchemy.ext.asyncio import AsyncSession
from src.models import Base
from src.models.base import TimestampMixin, UUIDPrimaryKeyMixin
from src.models.git_repository import GitRepository
from src.models.project import Project
from src.models.ssh_key import SSHKey
from src.models.user import User
from src.models.user_config import UserConfig
from src.models.project.git_repository import GitRepository
from src.models.project.project import Project
from src.models.user.ssh_key import SSHKey
from src.models.user.user import User
from src.models.user.user_config import UserConfig
@pytest.mark.integration
@@ -39,6 +39,7 @@ def test_expected_tables_are_registered() -> None:
"tool_types",
"user_configs",
"users",
"workspaces",
}
apps/api/tests/integration/test_notifications_api.py
+3 -3
View File
@@ -7,9 +7,9 @@ import pytest
from fastapi.testclient import TestClient
from sqlalchemy.ext.asyncio import AsyncSession
from src.models.user import User
from src.models.user_config import UserConfig
from src.services.notification_service import NotificationService
from src.models.user.user import User
from src.models.user.user_config import UserConfig
from src.services.shared.notification_service import NotificationService
@pytest.fixture
apps/api/tests/integration/test_notifications_lifecycle.py
+28 -20
View File
@@ -9,14 +9,14 @@ import pytest_asyncio
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from src.models.git_repository import GitRepository
from src.models.notification import Notification
from src.models.project import Project
from src.models.tool_instance import ToolInstance
from src.models.tool_type import ToolType
from src.models.user import User
from src.services.event_bus import InstanceEventBus, InstanceEventPayload
from src.services.health_monitor import HealthSnapshot
from src.models.project.git_repository import GitRepository
from src.models.system.notification import Notification
from src.models.project.project import Project
from src.models.tool.tool_instance import ToolInstance
from src.models.tool.tool_type import ToolType
from src.models.user.user import User
from src.services.instance.event_bus import InstanceEventBus, InstanceEventPayload
from src.services.instance.health_monitor import HealthSnapshot
@pytest.fixture
@@ -78,6 +78,7 @@ async def test_instance(db_session: AsyncSession) -> ToolInstance:
project_id=project.id,
owner_id=user.id,
status="running",
container_id="container123",
compose_path="/tmp/test-compose.yml",
port=8080,
)
@@ -101,7 +102,7 @@ async def test_lifecycle_started_intermediate_skips_notification(
event_bus.subscribe("instance.started", subscriber)
from src.services.lifecycle_hooks import publish_lifecycle_event
from src.services.instance.lifecycle_hooks import publish_lifecycle_event
await publish_lifecycle_event(
event_bus=event_bus,
@@ -131,7 +132,7 @@ async def test_lifecycle_running_creates_notification(
test_instance: ToolInstance,
) -> None:
"""Successful terminal state (running) creates a notification."""
from src.services.lifecycle_hooks import publish_lifecycle_event
from src.services.instance.lifecycle_hooks import publish_lifecycle_event
await publish_lifecycle_event(
event_bus=event_bus,
@@ -162,10 +163,17 @@ async def test_health_monitor_error_creates_notification(
event_bus: InstanceEventBus,
test_instance: ToolInstance,
) -> None:
"""Simulating a health monitor crash creates an error notification."""
from src.services.health_monitor import HealthMonitor
"""Simulating a container exit creates an error notification."""
from src.services.instance.health_monitor import HealthMonitor
monitor = HealthMonitor(event_bus)
# Seed a different healthy prior state so the exit is treated as a change.
monitor._last_known_state[test_instance.id] = HealthSnapshot(
container_status="running",
container_healthy=None,
tunnel_healthy=True,
exit_code=None,
)
received: list[InstanceEventPayload] = []
@@ -175,7 +183,7 @@ async def test_health_monitor_error_creates_notification(
event_bus.subscribe("instance.error", subscriber)
with patch(
"src.services.health_monitor.get_container_status",
"src.services.instance.health_monitor.get_container_status",
return_value={"status": "exited", "exit_code": 137, "health": None},
):
await monitor._check_instance(db_session, test_instance)
@@ -211,10 +219,10 @@ async def test_notification_failure_does_not_block_event_pipeline(
event_bus.subscribe("instance.started", subscriber)
from src.services.lifecycle_hooks import publish_lifecycle_event
from src.services.instance.lifecycle_hooks import publish_lifecycle_event
with patch(
"src.services.lifecycle_hooks.notification_service.create_notification",
"src.services.instance.lifecycle_hooks.notification_service.create_notification",
side_effect=RuntimeError("DB is down"),
):
# Should not raise
@@ -309,7 +317,7 @@ async def test_notification_ownership_matches_instance_owner(
db_session.add(instance)
await db_session.commit()
from src.services.lifecycle_hooks import publish_lifecycle_event
from src.services.instance.lifecycle_hooks import publish_lifecycle_event
await publish_lifecycle_event(
event_bus=event_bus,
@@ -336,7 +344,7 @@ async def test_lifecycle_error_creates_error_notification(
test_instance: ToolInstance,
) -> None:
"""An instance.error lifecycle event creates a severity=error notification."""
from src.services.lifecycle_hooks import publish_lifecycle_event
from src.services.instance.lifecycle_hooks import publish_lifecycle_event
await publish_lifecycle_event(
event_bus=event_bus,
@@ -363,7 +371,7 @@ async def test_health_monitor_unhealthy_creates_warning_notification(
test_instance: ToolInstance,
) -> None:
"""Health monitor marking instance unhealthy creates severity=warning notification."""
from src.services.health_monitor import HealthMonitor
from src.services.instance.health_monitor import HealthMonitor
monitor = HealthMonitor(event_bus)
monitor._last_known_state[test_instance.id] = HealthSnapshot(
@@ -376,11 +384,11 @@ async def test_health_monitor_unhealthy_creates_warning_notification(
with (
patch(
"src.services.health_monitor.get_container_status",
"src.services.instance.health_monitor.get_container_status",
return_value={"status": "running", "exit_code": None, "health": "healthy"},
),
patch(
"src.services.health_monitor.check_tunnel_health",
"src.services.instance.health_monitor.check_tunnel_health",
return_value={"healthy": False, "tunnel_status": "error_response"},
),
):
apps/api/tests/integration/test_projects_api.py
+28 -8
View File
@@ -10,8 +10,31 @@ from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from src.auth.session import create_session_cookie
from src.config import Settings, build_database_url
from src.models import Base
from src.models.project import Project
from src.models.user import User
from src.models.project.project import Project
from src.models.user.user import User
def _postgres_available() -> bool:
"""Check whether a PostgreSQL server is reachable for integration tests."""
import asyncpg
async def _check() -> bool:
try:
conn = await asyncpg.connect(
host="localhost", port=5432, user="headquarter", password="headquarter", database="headquarter"
)
await conn.close()
return True
except Exception:
return False
return asyncio.run(_check())
pytestmark = pytest.mark.skipif(
not _postgres_available(),
reason="PostgreSQL not available on localhost:5432",
)
def _prepare_test_db() -> None:
@@ -36,8 +59,8 @@ def _prepare_test_db() -> None:
def _load_app():
import importlib
import src.database as database_module
import src.api.auth as auth_module
import src.api.projects as projects_module
import src.api.user.auth as auth_module
import src.api.project.projects as projects_module
import src.main as main_module
# Dispose old engine connections before reload to prevent pool exhaustion
@@ -56,10 +79,7 @@ def _mint_token(user_id: str) -> str:
settings = Settings()
return create_session_cookie(
settings=settings,
subject=user_id,
email="test@headquarter.local",
name="Test User",
expires_at=datetime.now(UTC) + timedelta(minutes=15),
user_id=user_id,
)
apps/api/tests/integration/test_ssh_keys_api.py
+5 -15
View File
@@ -1,24 +1,14 @@
import pytest
from httpx import AsyncClient
from src.main import app
from fastapi.testclient import TestClient
@pytest.fixture
async def async_client():
async with AsyncClient(app=app, base_url="http://test") as client:
yield client
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_ssh_key_requires_authentication(async_client: AsyncClient) -> None:
response = await async_client.post("/ssh-keys", json={"name": "test-key"})
def test_create_ssh_key_requires_authentication(test_client: TestClient) -> None:
response = test_client.post("/ssh-keys", json={"name": "test-key"})
assert response.status_code == 401
@pytest.mark.asyncio
@pytest.mark.integration
async def test_list_ssh_keys_requires_authentication(async_client: AsyncClient) -> None:
response = await async_client.get("/ssh-keys")
def test_list_ssh_keys_requires_authentication(test_client: TestClient) -> None:
response = test_client.get("/ssh-keys")
assert response.status_code == 401
apps/api/tests/integration/test_tool_types_api.py
+27 -4
View File
@@ -10,8 +10,31 @@ from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from src.auth.session import create_session_cookie
from src.config import Settings, build_database_url
from src.models import Base
from src.models.tool_type import ToolType
from src.models.user import User
from src.models.tool.tool_type import ToolType
from src.models.user.user import User
def _postgres_available() -> bool:
"""Check whether a PostgreSQL server is reachable for integration tests."""
import asyncpg
async def _check() -> bool:
try:
conn = await asyncpg.connect(
host="localhost", port=5432, user="headquarter", password="headquarter", database="headquarter"
)
await conn.close()
return True
except Exception:
return False
return asyncio.run(_check())
pytestmark = pytest.mark.skipif(
not _postgres_available(),
reason="PostgreSQL not available on localhost:5432",
)
def _prepare_test_db() -> None:
@@ -36,8 +59,8 @@ def _prepare_test_db() -> None:
def _load_app():
import importlib
import src.database as database_module
import src.api.auth as auth_module
import src.api.tool_types as tool_types_module
import src.api.user.auth as auth_module
import src.api.tool.tool_types as tool_types_module
import src.main as main_module
# Dispose old engine connections before reload to prevent pool exhaustion
apps/api/tests/integration/test_tool_types_api_extended.py
+12 -11
View File
@@ -181,7 +181,7 @@ class TestToolTypesAPIExtended:
"name": "full-tool",
"display_name": "Full Tool",
"category": "editor",
"interfaces": ["web", "terminal"],
"interface_type": "web",
"default_port": 8443,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: code-server\n command: --bind-addr 0.0.0.0:8443\n ports:\n - '8443:8443'\n volumes:\n - \"{{REPO_PATH}}:/workspace\"",
@@ -201,20 +201,20 @@ class TestToolTypesAPIExtended:
data = response.json()
assert data["definition_type"] == "compose"
assert data["category"] == "editor"
assert data["interfaces"] == ["web", "terminal"]
assert data["interface_type"] == "web"
assert "readiness_probe" in data
def test_create_tool_type_without_port_fails(
self, authenticated_client: TestClient
) -> None:
"""Test that creating a tool type without default_port fails validation."""
"""Test that creating a tool type requiring a port with default_port=0 fails validation."""
response = authenticated_client.post(
"/tool-types",
json={
"name": "no-port-tool",
"display_name": "No Port Tool",
"category": "utility",
"interfaces": ["web"],
"interface_type": "web",
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx\n ports:\n - '8080:8080'",
"required_variables": [],
@@ -256,9 +256,9 @@ class TestToolTypesAPIExtended:
"category": "utility",
"interface_type": "terminal",
"requires_port": False,
"default_port": 0,
"default_port": 8080,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: alpine",
"compose_template": "version: '3.8'\nservices:\n app:\n image: alpine\n command: sleep 3600",
"startup_command": "cd /workspace && ls",
"required_variables": [],
},
@@ -280,9 +280,9 @@ class TestToolTypesAPIExtended:
"display_name": "Update Startup Tool",
"interface_type": "terminal",
"requires_port": False,
"default_port": 0,
"default_port": 8080,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: alpine",
"compose_template": "version: '3.8'\nservices:\n app:\n image: alpine\n command: sleep 3600",
"required_variables": [],
},
)
@@ -293,6 +293,8 @@ class TestToolTypesAPIExtended:
f"/tool-types/{tool_id}",
json={
"startup_command": "source /etc/profile",
"requires_port": False,
"default_port": 8080,
},
)
assert response.status_code == 200
@@ -310,9 +312,9 @@ class TestToolTypesAPIExtended:
"display_name": "Get Startup Tool",
"interface_type": "terminal",
"requires_port": False,
"default_port": 0,
"default_port": 8080,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: alpine",
"compose_template": "version: '3.8'\nservices:\n app:\n image: alpine\n command: sleep 3600",
"startup_command": "echo hello",
"required_variables": [],
},
@@ -323,4 +325,3 @@ class TestToolTypesAPIExtended:
assert response.status_code == 200
data = response.json()
assert data["startup_command"] == "echo hello"
assert "Port 9999 is not exposed" in str(data)
apps/api/tests/integration/test_users_api.py
+26 -6
View File
@@ -11,7 +11,30 @@ from sqlalchemy.ext.asyncio import create_async_engine
from src.auth.session import create_session_cookie
from src.config import Settings, build_database_url
from src.models import Base
from src.models.user import User
from src.models.user.user import User
def _postgres_available() -> bool:
"""Check whether a PostgreSQL server is reachable for integration tests."""
import asyncpg
async def _check() -> bool:
try:
conn = await asyncpg.connect(
host="localhost", port=5432, user="headquarter", password="headquarter", database="headquarter"
)
await conn.close()
return True
except Exception:
return False
return asyncio.run(_check())
pytestmark = pytest.mark.skipif(
not _postgres_available(),
reason="PostgreSQL not available on localhost:5432",
)
def _prepare_users_test_db() -> None:
@@ -36,7 +59,7 @@ def _prepare_users_test_db() -> None:
def _load_app():
import importlib
import src.database as database_module
import src.api.users as users_module
import src.api.user.users as users_module
import src.main as main_module
importlib.reload(database_module)
@@ -80,10 +103,7 @@ def _create_auth_cookie(user_id: str) -> str:
settings = Settings()
return create_session_cookie(
settings=settings,
subject=user_id,
email="test@headquarter.local",
name="Test User",
expires_at=datetime.now(UTC) + timedelta(minutes=15),
user_id=user_id,
)
apps/api/tests/services/test_terminal_manager_multi.py
+2 -2
View File
@@ -6,8 +6,8 @@ from unittest.mock import AsyncMock, patch
import pytest
from src.services.terminal_manager import MaxSessionsExceededError, TerminalManager
from src.services.terminal_session import TerminalSession
from src.services.terminal.terminal_manager import MaxSessionsExceededError, TerminalManager
from src.services.terminal.terminal_session import TerminalSession
@pytest.fixture
apps/api/tests/unit/test_config.py
+4 -4
View File
@@ -5,12 +5,12 @@ from src.database import build_database_url
@pytest.mark.unit
def test_settings_default_database_url_uses_asyncpg() -> None:
def test_settings_default_database_url_uses_asyncpg(monkeypatch) -> None:
"""Test that default database URL uses asyncpg driver and correct defaults."""
monkeypatch.delenv("DATABASE_URL", raising=False)
settings = Settings()
# When DATABASE_URL env var is set (by conftest), it overrides the defaults
# This test verifies the URL format when built from defaults
expected = "postgresql+asyncpg://headquarter:headquarter@localhost:5432/headquarter"
# When DATABASE_URL env var is not set, the URL is built from defaults.
expected = "postgresql+asyncpg://headquarter:headquarter@postgres:5432/headquarter"
assert settings.database_url == expected
apps/api/tests/unit/test_config_profile_resolver.py
+4 -4
View File
@@ -2,8 +2,8 @@ import uuid
import pytest
from sqlalchemy.ext.asyncio import AsyncSession
from src.models.config_profile import ConfigProfile, ConfigProfileInclude
from src.services.config_profile_resolver import (
from src.models.config.config_profile import ConfigProfile, ConfigProfileInclude
from src.services.config.config_profile_resolver import (
ConfigProfileCycleError,
ConfigProfileNotFoundError,
ResolvedMount,
@@ -77,7 +77,7 @@ class TestMergeFunctions:
def test_merge_mounts_file_override(self) -> None:
"""Test mount file map merging with overrides."""
from src.services.config_profile_resolver import ResolvedMount
from src.services.config.config_profile_resolver import ResolvedMount
result = _merge_mounts(
{"/app": ResolvedMount(target="/app", mode="rw", files={"a.txt": "old"})},
@@ -89,7 +89,7 @@ class TestMergeFunctions:
def test_merge_mounts_mode_conflict(self) -> None:
"""Test that mount mode conflicts are resolved (later wins)."""
from src.services.config_profile_resolver import ResolvedMount
from src.services.config.config_profile_resolver import ResolvedMount
overrides = {}
result = _merge_mounts(
apps/api/tests/unit/test_docker_build.py
+1 -1
View File
@@ -7,7 +7,7 @@ from unittest.mock import MagicMock, patch
import pytest
from src.services.docker_build import build_image
from src.services.build.docker_build import build_image
class TestBuildImage:
apps/api/tests/unit/test_docker_service.py
+7 -7
View File
@@ -4,11 +4,11 @@ from unittest.mock import MagicMock, patch
import logging
from src.services.docker import (
from src.services.docker.container import (
get_container_id,
get_container_name,
sort_volumes_by_specificity,
)
from src.services.docker.compose import sort_volumes_by_specificity
class TestGetContainerId:
@@ -16,15 +16,15 @@ class TestGetContainerId:
@patch("subprocess.run")
def test_lowercases_name_for_filter(self, mock_run) -> None:
"""Docker ps name filter is case-sensitive; we must lowercase."""
"""Docker inspect is case-sensitive; we must lowercase the name."""
mock_run.return_value = MagicMock(returncode=0, stdout="abc123\n")
result = get_container_id("MyContainer-ABC")
assert result == "abc123"
call_args = mock_run.call_args[0][0]
# The filter must use lowercase
assert "name=mycontainer-abc" in call_args
# Exact inspect call uses lowercase
assert call_args == ["docker", "inspect", "-f", "{{.Id}}", "mycontainer-abc"]
@patch("subprocess.run")
def test_returns_none_when_not_found(self, mock_run) -> None:
@@ -40,14 +40,14 @@ class TestGetContainerName:
@patch("subprocess.run")
def test_lowercases_name_for_filter(self, mock_run) -> None:
"""Docker ps name filter is case-sensitive; we must lowercase."""
"""Docker inspect is case-sensitive; we must lowercase the name."""
mock_run.return_value = MagicMock(returncode=0, stdout="mycontainer-abc\n")
result = get_container_name("MyContainer-ABC")
assert result == "mycontainer-abc"
call_args = mock_run.call_args[0][0]
assert "name=mycontainer-abc" in call_args
assert call_args == ["docker", "inspect", "-f", "{{.Name}}", "mycontainer-abc"]
@patch("subprocess.run")
def test_returns_none_when_not_found(self, mock_run) -> None:
apps/api/tests/unit/test_event_bus.py
+1 -1
View File
@@ -6,7 +6,7 @@ from typing import Any
import pytest
from src.services.event_bus import InstanceEventBus, InstanceEventPayload
from src.services.instance.event_bus import InstanceEventBus, InstanceEventPayload
@pytest.fixture
apps/api/tests/unit/test_git_mount_resolution.py
-124
View File
@@ -1,124 +0,0 @@
"""Unit tests for git mount resolution in tool instances."""
import os
from pathlib import Path
import pytest
from src.api.tool_instances import (
_checkout_branch,
_expand_glob_source,
_resolve_single_git_mount,
)
class TestExpandGlobSource:
"""Unit tests for glob pattern expansion."""
def test_no_glob_single_file(self, tmp_path: Path) -> None:
"""Test non-glob path returns single file."""
test_file = tmp_path / "test.txt"
test_file.write_text("content")
result = _expand_glob_source(str(test_file), str(tmp_path))
assert len(result) == 1
assert result[0] == str(test_file)
def test_no_glob_missing_file(self, tmp_path: Path) -> None:
"""Test non-glob missing file returns empty list."""
missing_file = tmp_path / "missing.txt"
result = _expand_glob_source(str(missing_file), str(tmp_path))
assert len(result) == 0
def test_glob_pattern(self, tmp_path: Path) -> None:
"""Test glob pattern matches files."""
(tmp_path / "file1.txt").write_text("content1")
(tmp_path / "file2.txt").write_text("content2")
(tmp_path / "other.py").write_text("code")
result = _expand_glob_source(str(tmp_path / "*.txt"), str(tmp_path))
assert len(result) == 2
assert all(f.endswith(".txt") for f in result)
def test_glob_recursive(self, tmp_path: Path) -> None:
"""Test recursive glob pattern."""
subdir = tmp_path / "subdir"
subdir.mkdir()
(subdir / "nested.txt").write_text("content")
result = _expand_glob_source(str(tmp_path / "**" / "*.txt"), str(tmp_path))
assert len(result) == 1
assert "nested.txt" in result[0]
def test_glob_limit_enforced(self, tmp_path: Path) -> None:
"""Test that glob matches are limited to prevent abuse."""
# Create more than 100 files
for i in range(105):
(tmp_path / f"file{i}.txt").write_text("content")
result = _expand_glob_source(str(tmp_path / "*.txt"), str(tmp_path))
assert len(result) == 100 # MAX_GLOB_MATCHES limit
def test_glob_escapes_repo(self, tmp_path: Path) -> None:
"""Test that glob results outside repo are filtered."""
other_dir = tmp_path.parent / "other"
other_dir.mkdir(exist_ok=True)
(other_dir / "outside.txt").write_text("content")
result = _expand_glob_source(str(tmp_path.parent / "*" / "*.txt"), str(tmp_path))
# Should only include files within tmp_path, not other_dir
assert all(r.startswith(str(tmp_path)) for r in result)
class TestCheckoutBranch:
"""Unit tests for branch checkout."""
def test_checkout_existing_branch(self, tmp_path: Path) -> None:
"""Test checking out an existing branch."""
# Initialize git repo
os.system(f"cd {tmp_path} && git init && git config user.email 'test@test.com' && git config user.name 'Test'")
(tmp_path / "file.txt").write_text("content")
os.system(f"cd {tmp_path} && git add . && git commit -m 'initial'")
os.system(f"cd {tmp_path} && git branch feature")
_checkout_branch(str(tmp_path), "feature")
# Verify we're on feature branch
result = os.popen(f"cd {tmp_path} && git branch --show-current").read().strip()
assert result == "feature"
def test_checkout_nonexistent_branch(self, tmp_path: Path) -> None:
"""Test checking out a non-existent branch returns False."""
os.system(f"cd {tmp_path} && git init && git config user.email 'test@test.com' && git config user.name 'Test'")
(tmp_path / "file.txt").write_text("content")
os.system(f"cd {tmp_path} && git add . && git commit -m 'initial'")
result = _checkout_branch(str(tmp_path), "nonexistent")
assert result is False
class TestResolveSingleGitMount:
"""Unit tests for resolving a single git mount."""
@pytest.mark.asyncio
async def test_resolve_missing_remote_url(self, db_session) -> None:
"""Test that missing remote_url returns empty list."""
git_mount = {
"source_path": ".",
"target_path": "/app",
}
result = await _resolve_single_git_mount(db_session, git_mount)
assert result == []
@pytest.mark.asyncio
async def test_resolve_missing_target_path(self, db_session) -> None:
"""Test that missing target path returns empty list."""
git_mount = {
"remote_url": "https://github.com/user/repo.git",
"source_path": ".",
}
result = await _resolve_single_git_mount(db_session, git_mount)
assert result == []
apps/api/tests/unit/test_git_mounts.py
-219
View File
@@ -1,219 +0,0 @@
"""Unit tests for git mount resolution with multi-mapping support."""
import os
import tempfile
from unittest.mock import MagicMock, patch
import pytest
from src.api.tool_instances import (
_clone_git_repo,
_expand_glob_source,
_normalize_git_mount,
_resolve_git_mount_mappings,
_resolve_single_git_mount,
)
class TestNormalizeGitMount:
"""Tests for _normalize_git_mount."""
def test_legacy_to_mappings(self) -> None:
"""Legacy source_path + target_path becomes mappings array."""
entry = {
"remote_url": "https://github.com/user/repo.git",
"source_path": "packages/api",
"target_path": "/app/api",
"branch": "main",
}
result = _normalize_git_mount(entry)
assert "mappings" in result
assert result["mappings"] == [
{"source_path": "packages/api", "target_path": "/app/api"}
]
assert "source_path" not in result
assert "target_path" not in result
assert result["remote_url"] == "https://github.com/user/repo.git"
assert result["branch"] == "main"
def test_already_mappings(self) -> None:
"""Entry already with mappings is left unchanged."""
entry = {
"remote_url": "https://github.com/user/repo.git",
"branch": "main",
"mappings": [
{"source_path": "a", "target_path": "/a"},
{"source_path": "b", "target_path": "/b"},
],
}
result = _normalize_git_mount(entry)
assert result["mappings"] == [
{"source_path": "a", "target_path": "/a"},
{"source_path": "b", "target_path": "/b"},
]
assert "source_path" not in result
assert "target_path" not in result
def test_missing_target_path_no_mappings(self) -> None:
"""Entry with source_path but no target_path creates empty mappings."""
entry = {
"remote_url": "https://github.com/user/repo.git",
"source_path": "src",
}
result = _normalize_git_mount(entry)
assert "mappings" not in result
class TestResolveGitMountMappings:
"""Tests for _resolve_git_mount_mappings."""
def test_single_mapping(self) -> None:
"""A single mapping produces one volume mount."""
with tempfile.TemporaryDirectory() as repo_path:
os.makedirs(os.path.join(repo_path, "packages", "api"))
mappings = [
{"source_path": "packages/api", "target_path": "/app/api"},
]
result = _resolve_git_mount_mappings(repo_path, mappings, None)
assert len(result) == 1
assert result[0]["source"] == os.path.join(repo_path, "packages", "api")
assert result[0]["target"] == "/app/api"
assert result[0]["type"] == "bind"
def test_multiple_mappings(self) -> None:
"""Multiple mappings from same repo produce multiple mounts."""
with tempfile.TemporaryDirectory() as repo_path:
os.makedirs(os.path.join(repo_path, "packages", "api"))
os.makedirs(os.path.join(repo_path, "packages", "web"))
mappings = [
{"source_path": "packages/api", "target_path": "/app/api"},
{"source_path": "packages/web", "target_path": "/app/web"},
]
result = _resolve_git_mount_mappings(repo_path, mappings, None)
assert len(result) == 2
targets = {r["target"] for r in result}
assert targets == {"/app/api", "/app/web"}
def test_relative_target_path(self) -> None:
"""Relative target_path is resolved against working_directory."""
with tempfile.TemporaryDirectory() as repo_path:
os.makedirs(os.path.join(repo_path, "src"))
mappings = [
{"source_path": "src", "target_path": "code"},
]
result = _resolve_git_mount_mappings(repo_path, mappings, "/workspace")
assert len(result) == 1
assert result[0]["target"] == "/workspace/code"
def test_glob_expansion(self) -> None:
"""Glob patterns in source_path are expanded."""
with tempfile.TemporaryDirectory() as repo_path:
os.makedirs(os.path.join(repo_path, "packages", "api"))
os.makedirs(os.path.join(repo_path, "packages", "web"))
mappings = [
{"source_path": "packages/*", "target_path": "/app/packages"},
]
result = _resolve_git_mount_mappings(repo_path, mappings, None)
assert len(result) == 2
targets = {r["target"] for r in result}
assert targets == {
os.path.join("/app/packages", "packages", "api"),
os.path.join("/app/packages", "packages", "web"),
}
def test_missing_target_path_skipped(self) -> None:
"""Mapping without target_path is skipped."""
with tempfile.TemporaryDirectory() as repo_path:
mappings = [
{"source_path": "src"},
]
result = _resolve_git_mount_mappings(repo_path, mappings, None)
assert len(result) == 0
def test_no_working_directory_for_relative_target(self) -> None:
"""Relative target without working_directory is skipped."""
with tempfile.TemporaryDirectory() as repo_path:
os.makedirs(os.path.join(repo_path, "src"))
mappings = [
{"source_path": "src", "target_path": "code"},
]
result = _resolve_git_mount_mappings(repo_path, mappings, None)
assert len(result) == 0
class TestResolveSingleGitMount:
"""Tests for _resolve_single_git_mount."""
@pytest.mark.asyncio
async def test_missing_remote_url(self) -> None:
"""Git mount without remote_url returns empty list."""
result = await _resolve_single_git_mount(
MagicMock(),
{"mappings": [{"source_path": ".", "target_path": "/app"}]},
"/tmp",
None,
)
assert result == []
@pytest.mark.asyncio
async def test_missing_instance_dir(self) -> None:
"""Git mount without instance_dir returns empty list."""
result = await _resolve_single_git_mount(
MagicMock(),
{
"remote_url": "https://github.com/user/repo.git",
"mappings": [{"source_path": ".", "target_path": "/app"}],
},
None,
None,
)
assert result == []
@pytest.mark.asyncio
async def test_legacy_format_normalized(self) -> None:
"""Legacy format is normalized and resolved."""
with tempfile.TemporaryDirectory() as instance_dir:
with patch(
"src.api.tool_instances._clone_git_repo",
return_value=os.path.join(instance_dir, "repo-clone"),
):
os.makedirs(os.path.join(instance_dir, "repo-clone", "src"))
result = await _resolve_single_git_mount(
MagicMock(),
{
"remote_url": "https://github.com/user/repo.git",
"source_path": "src",
"target_path": "/app/src",
},
instance_dir,
None,
)
assert len(result) == 1
assert result[0]["target"] == "/app/src"
class TestExpandGlobSource:
"""Tests for _expand_glob_source."""
def test_no_glob(self) -> None:
"""Non-glob path returns single item if exists."""
with tempfile.TemporaryDirectory() as tmp:
path = os.path.join(tmp, "file.txt")
open(path, "w").close()
result = _expand_glob_source(path, tmp)
assert result == [path]
def test_no_glob_missing(self) -> None:
"""Non-glob path that doesn't exist returns empty list."""
with tempfile.TemporaryDirectory() as tmp:
path = os.path.join(tmp, "missing.txt")
result = _expand_glob_source(path, tmp)
assert result == []
def test_glob_pattern(self) -> None:
"""Glob pattern expands to matched paths."""
with tempfile.TemporaryDirectory() as tmp:
open(os.path.join(tmp, "a.txt"), "w").close()
open(os.path.join(tmp, "b.txt"), "w").close()
result = _expand_glob_source(os.path.join(tmp, "*.txt"), tmp)
assert len(result) == 2
apps/api/tests/unit/test_git_repository_clone_preflight.py
-28
View File
@@ -1,28 +0,0 @@
from unittest.mock import Mock, patch
import pytest
from fastapi import HTTPException
from src.api.git_repositories import _build_provider_clone_url, _preflight_remote_repository
def test_build_provider_clone_url_uses_fixed_host() -> None:
assert _build_provider_clone_url("alice", "demo") == "git@git.commumedia.org:alice/demo.git"
def test_preflight_remote_repository_allows_accessible_repo() -> None:
completed = Mock(returncode=0)
with patch("src.api.git_repositories.subprocess.run", return_value=completed) as run_mock:
_preflight_remote_repository("git@git.commumedia.org:alice/demo.git")
run_mock.assert_called_once()
def test_preflight_remote_repository_rejects_missing_repo() -> None:
completed = Mock(returncode=128)
with patch("src.api.git_repositories.subprocess.run", return_value=completed):
with pytest.raises(HTTPException) as exc_info:
_preflight_remote_repository("git@git.commumedia.org:alice/missing.git")
assert exc_info.value.status_code == 400
assert exc_info.value.detail == "repository not found or inaccessible"
apps/api/tests/unit/test_git_repository_working_clones.py
-64
View File
@@ -1,64 +0,0 @@
from unittest.mock import Mock, patch
import pytest
from fastapi import HTTPException
from src.api.git_repositories import _clone_working_repository, _init_working_repository
from src.utils.git_control import create_branch
def test_clone_working_repository_uses_normal_clone() -> None:
completed = Mock(returncode=0, stderr="")
with patch("src.api.git_repositories.subprocess.run", return_value=completed) as run_mock:
_clone_working_repository("git@git.commumedia.org:alice/demo.git", "/tmp/demo.git")
run_mock.assert_called_once()
assert run_mock.call_args.args[0] == ["git", "clone", "git@git.commumedia.org:alice/demo.git", "/tmp/demo.git"]
def test_clone_working_repository_raises_on_failure() -> None:
completed = Mock(returncode=128, stderr="fatal: repository not found")
with patch("src.api.git_repositories.subprocess.run", return_value=completed):
with pytest.raises(HTTPException) as exc_info:
_clone_working_repository("git@git.commumedia.org:alice/missing.git", "/tmp/missing.git")
assert exc_info.value.status_code == 400
assert "failed to clone repository" in exc_info.value.detail
def test_init_working_repository_prefers_init_b() -> None:
init_b = Mock(returncode=0, stderr="")
with patch("src.api.git_repositories.subprocess.run", return_value=init_b) as run_mock:
_init_working_repository("/tmp/new-repo")
assert run_mock.call_args.args[0] == ["git", "init", "-b", "main", "/tmp/new-repo"]
def test_init_working_repository_falls_back_to_symbolic_ref() -> None:
init_b = Mock(returncode=1, stderr="unknown switch `b'")
init_ok = Mock(returncode=0, stderr="")
symbolic_ref = Mock(returncode=0, stderr="")
with patch("src.api.git_repositories.subprocess.run", side_effect=[init_b, init_ok, symbolic_ref]) as run_mock:
_init_working_repository("/tmp/new-repo")
assert run_mock.call_args_list[0].args[0] == ["git", "init", "-b", "main", "/tmp/new-repo"]
assert run_mock.call_args_list[1].args[0] == ["git", "init", "/tmp/new-repo"]
assert run_mock.call_args_list[2].args[0] == ["git", "-C", "/tmp/new-repo", "symbolic-ref", "HEAD", "refs/heads/main"]
def test_create_branch_uses_orphan_checkout_when_head_is_unborn() -> None:
call_count = 0
def mock_run(repo_path: str, *args: str) -> str:
nonlocal call_count
call_count += 1
if call_count == 1:
raise RuntimeError("fatal: Needed a single revision")
return ""
with patch("src.utils.git_control._run_git_command", side_effect=mock_run) as run_mock:
create_branch("/tmp/new-repo", "feature/test")
assert run_mock.call_args_list[0].args[1:] == ("rev-parse", "--verify", "HEAD^{commit}")
assert run_mock.call_args_list[1].args[1:] == ("checkout", "--orphan", "feature/test")
apps/api/tests/unit/test_git_service.py
+5 -1
View File
@@ -5,7 +5,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from src.services.git_service import GitService
from src.services.git.git_service import GitService
class TestGitServiceClone:
@@ -35,6 +35,7 @@ class TestGitServiceClone:
"/tmp/ws",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
env=None,
)
@pytest.mark.asyncio
@@ -72,6 +73,7 @@ class TestGitServiceFetch:
"origin",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
env=None,
)
@pytest.mark.asyncio
@@ -110,6 +112,7 @@ class TestGitServicePull:
"feature-branch",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
env=None,
)
@@ -130,6 +133,7 @@ class TestGitServiceBranchExistsRemotely:
["git", "-C", "/tmp/repo", "ls-remote", "--heads", "origin", "main"],
capture_output=True,
text=True,
env=None,
)
def test_branch_not_exists(self):
apps/api/tests/unit/test_health_monitor.py
+24 -14
View File
@@ -8,11 +8,11 @@ from unittest.mock import patch
import pytest
from sqlalchemy import select
from src.models.health_check import HealthCheck
from src.models.tool_instance import ToolInstance
from src.models.user import User
from src.services.event_bus import InstanceEventBus, InstanceEventPayload
from src.services.health_monitor import HealthMonitor, HealthSnapshot
from src.models.system.health_check import HealthCheck
from src.models.tool.tool_instance import ToolInstance
from src.models.user.user import User
from src.services.instance.event_bus import InstanceEventBus, InstanceEventPayload
from src.services.instance.health_monitor import HealthMonitor, HealthSnapshot
@pytest.fixture
@@ -77,11 +77,11 @@ async def test_detects_container_crash(
with (
patch(
"src.services.health_monitor.get_container_status",
"src.services.instance.health_monitor.get_container_status",
return_value={"status": "exited", "exit_code": 137, "health": None},
),
patch(
"src.services.health_monitor.check_tunnel_health",
"src.services.instance.health_monitor.check_tunnel_health",
return_value={"healthy": False, "tunnel_status": "not_applicable"},
),
):
@@ -124,11 +124,11 @@ async def test_detects_tunnel_failure(
with (
patch(
"src.services.health_monitor.get_container_status",
"src.services.instance.health_monitor.get_container_status",
return_value={"status": "running", "exit_code": None, "health": "healthy"},
),
patch(
"src.services.health_monitor.check_tunnel_health",
"src.services.instance.health_monitor.check_tunnel_health",
return_value={
"healthy": False,
"tunnel_status": "error_response",
@@ -181,11 +181,11 @@ async def test_detects_recovery(
with (
patch(
"src.services.health_monitor.get_container_status",
"src.services.instance.health_monitor.get_container_status",
return_value={"status": "running", "exit_code": None, "health": None},
),
patch(
"src.services.health_monitor.check_tunnel_health",
"src.services.instance.health_monitor.check_tunnel_health",
return_value={
"healthy": True,
"tunnel_status": "healthy",
@@ -218,13 +218,22 @@ async def test_skips_writes_when_no_state_change(
"""Two identical polls should result in only one health_checks row."""
instance = await _create_running_instance(db_session)
# Seed a different prior snapshot so the first poll writes a row, then the
# second identical poll skips because the snapshot is unchanged.
health_monitor._last_known_state[instance.id] = HealthSnapshot(
container_status="running",
container_healthy=None,
tunnel_healthy=False,
exit_code=None,
)
with (
patch(
"src.services.health_monitor.get_container_status",
"src.services.instance.health_monitor.get_container_status",
return_value={"status": "running", "exit_code": None, "health": None},
),
patch(
"src.services.health_monitor.check_tunnel_health",
"src.services.instance.health_monitor.check_tunnel_health",
return_value={
"healthy": True,
"tunnel_status": "healthy",
@@ -233,6 +242,7 @@ async def test_skips_writes_when_no_state_change(
),
):
await health_monitor._check_instance(db_session, instance)
# Second identical poll should skip writes because the snapshot is unchanged.
await health_monitor._check_instance(db_session, instance)
result = await db_session.execute(
@@ -259,7 +269,7 @@ async def test_docker_exception_resilience(
event_bus.subscribe("instance.health_changed", capture_event)
with patch(
"src.services.health_monitor.get_container_status",
"src.services.instance.health_monitor.get_container_status",
side_effect=RuntimeError("docker exploded"),
):
# Should not raise
apps/api/tests/unit/test_home_path_expansion.py
+2 -36
View File
@@ -2,9 +2,8 @@
import pytest
from src.api.tool_instances import _resolve_git_mount_mappings
from src.services.config_profile_resolver import expand_container_path
from src.services.manifest_compiler import get_manifest_home_dir
from src.services.config.config_profile_resolver import expand_container_path
from src.services.build.manifest_compiler import get_manifest_home_dir
class TestExpandContainerPath:
@@ -75,36 +74,3 @@ class TestGetManifestHomeDir:
manifest = {"user": {"name": None, "uid": 1000, "gid": 1000}}
assert get_manifest_home_dir(manifest) == "/root"
class TestResolveGitMountMappingsExpansion:
"""Tests that git mount mapping targets expand ~ and $HOME."""
def test_tilde_target_expansion(self, tmp_path) -> None:
"""Mapping with ~/repo target expands to home dir."""
(tmp_path / "src").mkdir()
mappings = [{"source_path": "src", "target_path": "~/repo"}]
result = _resolve_git_mount_mappings(
str(tmp_path), mappings, None, "/home/user"
)
assert len(result) == 1
assert result[0]["target"] == "/home/user/repo"
def test_dollar_home_target_expansion(self, tmp_path) -> None:
"""Mapping with $HOME/repo target expands to home dir."""
(tmp_path / "src").mkdir()
mappings = [{"source_path": "src", "target_path": "$HOME/repo"}]
result = _resolve_git_mount_mappings(
str(tmp_path), mappings, None, "/home/user"
)
assert len(result) == 1
assert result[0]["target"] == "/home/user/repo"
def test_absolute_target_unchanged(self, tmp_path) -> None:
"""Absolute target paths are not modified."""
(tmp_path / "src").mkdir()
mappings = [{"source_path": "src", "target_path": "/app/src"}]
result = _resolve_git_mount_mappings(
str(tmp_path), mappings, None, "/home/user"
)
assert len(result) == 1
assert result[0]["target"] == "/app/src"
apps/api/tests/unit/test_lifecycle_hooks.py
+1 -1
View File
@@ -2,7 +2,7 @@
import pytest
from src.services.lifecycle_hooks import _derive_title, _should_notify
from src.services.instance.lifecycle_hooks import _derive_title, _should_notify
class TestDeriveTitle:
apps/api/tests/unit/test_monitoring_models.py
+4 -4
View File
@@ -6,10 +6,10 @@ from datetime import datetime
import pytest
from sqlalchemy import select
from src.models.health_check import HealthCheck
from src.models.instance_event import InstanceEvent
from src.models.tool_instance import ToolInstance
from src.models.user import User
from src.models.system.health_check import HealthCheck
from src.models.system.instance_event import InstanceEvent
from src.models.tool.tool_instance import ToolInstance
from src.models.user.user import User
@pytest.mark.unit
apps/api/tests/unit/test_notification_service.py
+3 -3
View File
@@ -7,9 +7,9 @@ import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from src.models.notification import Notification
from src.models.user import User
from src.services.notification_service import NotificationService
from src.models.system.notification import Notification
from src.models.user.user import User
from src.services.shared.notification_service import NotificationService
@pytest.fixture
apps/api/tests/unit/test_notifications_api_routes.py
+1 -1
View File
@@ -3,7 +3,7 @@
from fastapi import FastAPI
from fastapi.testclient import TestClient
from src.api.notifications import router as notifications_router
from src.api.system.notifications import router as notifications_router
def test_delete_notifications_route_order() -> None:
apps/api/tests/unit/test_permission_fixer.py
+10 -10
View File
@@ -4,7 +4,7 @@ from unittest.mock import MagicMock, patch
import pytest
from src.services.permission_fixer import (
from src.services.shared.permission_fixer import (
PermissionFixError,
apply_mount_permissions,
apply_ssh_permissions,
@@ -16,7 +16,7 @@ from src.services.permission_fixer import (
class TestApplyMountPermissions:
"""Tests for apply_mount_permissions."""
@patch("src.services.permission_fixer._run_in_container")
@patch("src.services.shared.permission_fixer._run_in_container")
def test_applies_chown_when_owner_declared(self, mock_run) -> None:
mounts = [
{"name": "workspace", "target": "/workspace", "owner": "user"},
@@ -31,7 +31,7 @@ class TestApplyMountPermissions:
assert args[0] == "abc123"
assert args[1] == ["chown", "-R", "user:user", "/workspace"]
@patch("src.services.permission_fixer._run_in_container")
@patch("src.services.shared.permission_fixer._run_in_container")
def test_applies_chmod_when_mode_declared(self, mock_run) -> None:
mounts = [
{"name": "ssh", "target": "/home/user/.ssh", "mode": "0700"},
@@ -44,7 +44,7 @@ class TestApplyMountPermissions:
chmod_call = mock_run.call_args_list[0]
assert chmod_call[0][1] == ["chmod", "0700", "/home/user/.ssh"]
@patch("src.services.permission_fixer._run_in_container")
@patch("src.services.shared.permission_fixer._run_in_container")
def test_applies_file_mode_when_declared(self, mock_run) -> None:
mounts = [
{
@@ -64,7 +64,7 @@ class TestApplyMountPermissions:
"find /home/user/.ssh -type f -exec chmod 0600" in file_mode_call[0][1][2]
)
@patch("src.services.permission_fixer._run_in_container")
@patch("src.services.shared.permission_fixer._run_in_container")
def test_skips_readonly_mount(self, mock_run) -> None:
mounts = [
{
@@ -82,7 +82,7 @@ class TestApplyMountPermissions:
assert results[0]["success"] is True
mock_run.assert_not_called()
@patch("src.services.permission_fixer._run_in_container")
@patch("src.services.shared.permission_fixer._run_in_container")
def test_skips_mount_with_no_policy(self, mock_run) -> None:
mounts = [
{"name": "workspace", "target": "/workspace", "writable": True},
@@ -93,7 +93,7 @@ class TestApplyMountPermissions:
assert results[0]["success"] is True
mock_run.assert_not_called()
@patch("src.services.permission_fixer._run_in_container")
@patch("src.services.shared.permission_fixer._run_in_container")
def test_reports_failure_on_command_error(self, mock_run) -> None:
mock_run.side_effect = PermissionFixError("chown failed")
@@ -105,7 +105,7 @@ class TestApplyMountPermissions:
assert results[0]["success"] is False
assert "chown failed" in results[0]["error"]
@patch("src.services.permission_fixer._run_in_container")
@patch("src.services.shared.permission_fixer._run_in_container")
def test_stops_on_first_failure(self, mock_run) -> None:
"""If chown fails, chmod and file_mode should not run."""
mock_run.side_effect = PermissionFixError("chown failed")
@@ -227,11 +227,11 @@ class TestApplySshPermissions:
class TestCheckRootUserAvailable:
"""Tests for check_root_user_available."""
@patch("src.services.permission_fixer._run_in_container")
@patch("src.services.shared.permission_fixer._run_in_container")
def test_returns_true_when_root_exists(self, mock_run) -> None:
assert check_root_user_available("abc123") is True
@patch("src.services.permission_fixer._run_in_container")
@patch("src.services.shared.permission_fixer._run_in_container")
def test_returns_false_when_root_missing(self, mock_run) -> None:
mock_run.side_effect = PermissionFixError("no such user")
assert check_root_user_available("abc123") is False
apps/api/tests/unit/test_readiness_probe.py
+1 -1
View File
@@ -3,7 +3,7 @@
from unittest.mock import MagicMock, patch
from src.services.readiness_probe import execute_probe
from src.services.shared.readiness_probe import execute_probe
class TestExecuteProbe:
apps/api/tests/unit/test_ssh_keys.py
+4 -4
View File
@@ -6,13 +6,13 @@ from unittest.mock import MagicMock, patch
import pytest
from src.services.ssh_keys import prepare_ssh_key_files
from src.services.shared.ssh_keys import prepare_ssh_key_files
class TestPrepareSshKeyFiles:
"""Tests for prepare_ssh_key_files."""
@patch("src.services.ssh_keys._get_fernet")
@patch("src.services.shared.ssh_keys._get_fernet")
def test_creates_files_with_default_permissions(
self, mock_fernet, tmp_path
) -> None:
@@ -29,7 +29,7 @@ class TestPrepareSshKeyFiles:
assert (Path(ssh_dir) / "config").exists()
assert oct(os.stat(Path(ssh_dir) / "id_ed25519").st_mode)[-3:] == "600"
@patch("src.services.ssh_keys._get_fernet")
@patch("src.services.shared.ssh_keys._get_fernet")
def test_sets_ownership_when_uid_gid_provided(self, mock_fernet, tmp_path) -> None:
mock_fernet.return_value.decrypt.return_value = b"private-key-content"
ssh_key = MagicMock()
@@ -45,7 +45,7 @@ class TestPrepareSshKeyFiles:
assert mock_chown.call_args_list[0][0][1] == 1001
assert mock_chown.call_args_list[0][0][2] == 1001
@patch("src.services.ssh_keys._get_fernet")
@patch("src.services.shared.ssh_keys._get_fernet")
def test_gracefully_handles_permission_error_on_chown(
self, mock_fernet, tmp_path
) -> None: