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

fix: resolve stale backend test imports and schema drift

Browse Source 
- Delete 4 obsolete unit tests tied to removed git mount/clone models
- Update imports and assertions across unit/integration/service tests
- Fix Settings defaults (postgres host, JWT props, cookie_samesite)
- Add skip guards for PostgreSQL-dependent integration tests
- Fix GitService env assertions and HealthMonitor state-change tests
- Repair docker/container inspect assertions in test_docker_service
- Fix ToolTypeCreate default_port validator ordering bug
- Fix check_port_exposed substring false-positive for port 0
- Update test_tool_types_api_extended to use interface_type field

Quality gates: pytest 311 passed, 34 skipped; npm typecheck/lint/test 87 passed
This commit is contained in:
Developer
2026-06-12 20:23:17 +00:00
parent 79be4eb525
commit 81b9a66ef5
35 changed files with 268 additions and 625 deletions
Download Patch File Download Diff File Expand all files Collapse all files
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: