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

merge: align dev branch with main

Browse Source
This commit is contained in:
Developer
2026-06-03 08:51:02 +00:00
parent 51a399c775
commit b39d6ce5f4
319 changed files with 30221 additions and 9221 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/tests/conftest.py
+6 -86
View File
@@ -8,14 +8,16 @@ from unittest.mock import patch
import pytest
import pytest_asyncio
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, text
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import sessionmaker
# Set test environment BEFORE importing app modules
os.environ["APP_ENV"] = "testing"
os.environ["SECRET_KEY"] = "test-secret-key-for-testing-only-do-not-use-in-production"
os.environ["DATABASE_URL"] = "sqlite+aiosqlite:///:memory:"
from src.config import Settings
from src.config import Settings, build_database_url
from src.models.base import Base
from src.main import app
from src.auth.dependencies import get_db_session
@@ -45,8 +47,10 @@ def test_client() -> Generator[TestClient, None, None]:
app.dependency_overrides[get_db_session] = override_get_db_session
# Patch startup events to prevent PostgreSQL connection attempts
with patch("src.main.init_database") as mock_init:
with patch("src.main.init_database") as mock_init, \
patch("src.main.seed_builtin_tool_types") as mock_seed:
mock_init.return_value = True
mock_seed.return_value = None
try:
with TestClient(app) as client:
@@ -57,31 +61,6 @@ def test_client() -> Generator[TestClient, None, None]:
asyncio.run(engine.dispose())
@pytest_asyncio.fixture
async def db_session(test_client) -> AsyncGenerator[AsyncSession, None]:
"""Provide an async database session for unit tests."""
# Get the override function from the test_client fixture
override_fn = app.dependency_overrides.get(get_db_session)
if override_fn:
gen = override_fn()
session = await gen.asend(None)
try:
yield session
finally:
await gen.aclose()
else:
# Fallback: create a new engine and session
engine = create_async_engine(
"sqlite+aiosqlite:///:memory:",
connect_args={"check_same_thread": False},
)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async with async_sessionmaker(engine, expire_on_commit=False)() as session:
yield session
await engine.dispose()
@pytest.fixture
def authenticated_client(test_client) -> Generator[TestClient, None, None]:
"""Provide an authenticated test client with a test user."""
@@ -131,65 +110,6 @@ def authenticated_client(test_client) -> Generator[TestClient, None, None]:
yield test_client
@pytest.fixture
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
project_id = uuid.uuid4()
repo_id = uuid.uuid4()
user_id = None
# Get user ID from session
async def get_user_id():
nonlocal user_id
from src.auth.session import decode_session_cookie
settings = Settings()
session_cookie = authenticated_client.cookies.get("session")
if session_cookie:
session = decode_session_cookie(settings=settings, cookie_value=session_cookie)
if session:
user_id = uuid.UUID(session["user_id"])
asyncio.run(get_user_id())
if not user_id:
raise RuntimeError("Could not get user ID from authenticated client")
async def create_project_and_repo():
override_fn = app.dependency_overrides.get(get_db_session)
if override_fn:
gen = override_fn()
session = await gen.asend(None)
try:
project = Project(
id=project_id,
name="test-project",
description="Test project",
owner_id=user_id,
)
session.add(project)
repo = GitRepository(
id=repo_id,
name="test-repo",
path="/tmp/test-repo",
project_id=project_id,
owner_id=user_id,
remote_url="https://github.com/test/repo.git",
)
session.add(repo)
await session.commit()
finally:
await gen.aclose()
asyncio.run(create_project_and_repo())
return str(project_id), str(repo_id)
@pytest.fixture
def admin_client(test_client) -> Generator[TestClient, None, None]:
"""Provide an authenticated test client with an admin user."""
apps/api/tests/integration/test_config_folders_api.py
+255
View File
@@ -0,0 +1,255 @@
import uuid
import pytest
from fastapi.testclient import TestClient
@pytest.mark.integration
class TestConfigFoldersAPI:
"""Integration tests for config folders API."""
def test_list_config_folders_requires_authentication(self, test_client: TestClient) -> None:
"""Test that listing config folders requires authentication."""
response = test_client.get("/config-folders")
assert response.status_code == 401
def test_list_config_folders_returns_user_folders(self, authenticated_client: TestClient) -> None:
"""Test that authenticated users can list their folders."""
response = authenticated_client.get("/config-folders")
assert response.status_code == 200
data = response.json()
assert isinstance(data, dict)
assert "folders" in data
assert isinstance(data["folders"], list)
def test_create_config_folder_successfully(self, authenticated_client: TestClient) -> None:
"""Test creating a config folder."""
response = authenticated_client.post(
"/config-folders",
json={
"name": "test-folder",
"description": "Test folder",
"mount_path": "/home/user",
"files": {"test.txt": "hello world"},
},
)
assert response.status_code == 201
data = response.json()
assert data["name"] == "test-folder"
assert data["mount_path"] == "/home/user"
assert data["files"] == {"test.txt": "hello world"}
def test_create_config_folder_duplicate_name(self, authenticated_client: TestClient) -> None:
"""Test that duplicate folder names are rejected."""
# Create first folder
response = authenticated_client.post(
"/config-folders",
json={
"name": "duplicate-folder",
"mount_path": "/home/user",
"files": {},
},
)
assert response.status_code == 201
# Try to create second with same name
response = authenticated_client.post(
"/config-folders",
json={
"name": "duplicate-folder",
"mount_path": "/home/user",
"files": {},
},
)
assert response.status_code == 409
def test_create_config_folder_exceeds_size_limit(self, authenticated_client: TestClient) -> None:
"""Test that folders exceeding 10MB are rejected."""
large_content = "x" * (11 * 1024 * 1024) # 11MB
response = authenticated_client.post(
"/config-folders",
json={
"name": "large-folder",
"mount_path": "/home/user",
"files": {"large.txt": large_content},
},
)
assert response.status_code == 422
def test_create_config_folder_path_traversal_attack(self, authenticated_client: TestClient) -> None:
"""Test that path traversal in file paths is prevented."""
response = authenticated_client.post(
"/config-folders",
json={
"name": "bad-folder",
"mount_path": "/home/user",
"files": {"../../../etc/passwd": "malicious"},
},
)
assert response.status_code == 422
def test_get_config_folder_by_id(self, authenticated_client: TestClient) -> None:
"""Test getting a config folder by ID."""
# Create folder first
create_response = authenticated_client.post(
"/config-folders",
json={
"name": "get-test",
"mount_path": "/home/user",
"files": {},
},
)
folder_id = create_response.json()["id"]
# Get it back
response = authenticated_client.get(f"/config-folders/{folder_id}")
assert response.status_code == 200
data = response.json()
assert data["name"] == "get-test"
def test_get_config_folder_not_found(self, authenticated_client: TestClient) -> None:
"""Test getting a non-existent folder."""
response = authenticated_client.get(f"/config-folders/{uuid.uuid4()}")
assert response.status_code == 404
def test_update_config_folder_successfully(self, authenticated_client: TestClient) -> None:
"""Test updating a config folder."""
# Create folder first
create_response = authenticated_client.post(
"/config-folders",
json={
"name": "update-test",
"mount_path": "/home/user",
"files": {},
},
)
folder_id = create_response.json()["id"]
# Update it
response = authenticated_client.put(
f"/config-folders/{folder_id}",
json={
"name": "updated-name",
"mount_path": "/workspace",
"files": {"new.txt": "content"},
},
)
assert response.status_code == 200
data = response.json()
assert data["name"] == "updated-name"
assert data["mount_path"] == "/workspace"
def test_delete_config_folder_successfully(self, authenticated_client: TestClient) -> None:
"""Test deleting a config folder."""
# Create folder first
create_response = authenticated_client.post(
"/config-folders",
json={
"name": "delete-test",
"mount_path": "/home/user",
"files": {},
},
)
folder_id = create_response.json()["id"]
# Delete it
response = authenticated_client.delete(f"/config-folders/{folder_id}")
assert response.status_code == 204
# Verify it's gone
get_response = authenticated_client.get(f"/config-folders/{folder_id}")
assert get_response.status_code == 404
def test_add_project_override_successfully(self, authenticated_client: TestClient) -> None:
"""Test adding a project override."""
# Create folder first
create_response = authenticated_client.post(
"/config-folders",
json={
"name": "override-test",
"mount_path": "/home/user",
"files": {"global.txt": "global"},
},
)
folder_id = create_response.json()["id"]
project_id = str(uuid.uuid4())
# Add override
response = authenticated_client.post(
f"/config-folders/{folder_id}/overrides",
json={
"project_id": project_id,
"mount_path": "/workspace",
"files": {"project.txt": "project"},
},
)
assert response.status_code == 200
data = response.json()
assert project_id in data["project_overrides"]
def test_update_project_override_successfully(self, authenticated_client: TestClient) -> None:
"""Test updating a project override."""
# Create folder with override
create_response = authenticated_client.post(
"/config-folders",
json={
"name": "update-override-test",
"mount_path": "/home/user",
"files": {},
},
)
folder_id = create_response.json()["id"]
project_id = str(uuid.uuid4())
# Add override
authenticated_client.post(
f"/config-folders/{folder_id}/overrides",
json={
"project_id": project_id,
"mount_path": "/workspace",
"files": {"old.txt": "old"},
},
)
# Update override
response = authenticated_client.put(
f"/config-folders/{folder_id}/overrides/{project_id}",
json={
"mount_path": "/app",
"files": {"new.txt": "new"},
},
)
assert response.status_code == 200
data = response.json()
assert data["project_overrides"][project_id]["mount_path"] == "/app"
def test_delete_project_override_successfully(self, authenticated_client: TestClient) -> None:
"""Test deleting a project override."""
# Create folder with override
create_response = authenticated_client.post(
"/config-folders",
json={
"name": "delete-override-test",
"mount_path": "/home/user",
"files": {},
},
)
folder_id = create_response.json()["id"]
project_id = str(uuid.uuid4())
# Add override
authenticated_client.post(
f"/config-folders/{folder_id}/overrides",
json={
"project_id": project_id,
"mount_path": "/workspace",
"files": {},
},
)
# Delete override
response = authenticated_client.delete(
f"/config-folders/{folder_id}/overrides/{project_id}"
)
assert response.status_code == 200
data = response.json()
assert project_id not in data["project_overrides"]
apps/api/tests/integration/test_config_profiles_api.py
+335 -327
View File
@@ -1,4 +1,7 @@
"""Integration tests for config profiles API."""
import uuid
import pytest
from fastapi.testclient import TestClient
@@ -17,7 +20,9 @@ class TestConfigProfilesAPI:
response = authenticated_client.get("/config-profiles")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert isinstance(data, dict)
assert "profiles" in data
assert isinstance(data["profiles"], list)
def test_create_config_profile_successfully(self, authenticated_client: TestClient) -> None:
"""Test creating a config profile."""
@@ -26,99 +31,48 @@ class TestConfigProfilesAPI:
json={
"name": "test-profile",
"description": "Test profile",
"env_vars": {"VAR": "value"},
"runtime_hints": {"start_command": "npm start"},
"mounts": [{"target": "/app", "mode": "rw", "files": {}}],
"files": {"test.txt": "hello"},
},
)
assert response.status_code == 201
data = response.json()
assert data["name"] == "test-profile"
assert data["env_vars"] == {"VAR": "value"}
assert data["files"] == {"test.txt": "hello"}
assert data["mounts"][0]["target"] == "/app"
assert data["description"] == "Test profile"
def test_create_config_profile_duplicate_name(self, authenticated_client: TestClient) -> None:
"""Test that duplicate profile names are rejected."""
# Create first profile
response = authenticated_client.post(
authenticated_client.post(
"/config-profiles",
json={
"name": "duplicate-profile",
"env_vars": {},
"files": {},
},
json={"name": "duplicate-profile"},
)
assert response.status_code == 201
# Try to create second with same name
response = authenticated_client.post(
"/config-profiles",
json={
"name": "duplicate-profile",
"env_vars": {},
"files": {},
},
json={"name": "duplicate-profile"},
)
assert response.status_code == 409
def test_create_config_profile_exceeds_size_limit(self, authenticated_client: TestClient) -> None:
"""Test that profiles exceeding 10MB are rejected."""
large_content = "x" * (11 * 1024 * 1024) # 11MB
def test_create_config_profile_empty_name(self, authenticated_client: TestClient) -> None:
"""Test that empty profile names are rejected."""
response = authenticated_client.post(
"/config-profiles",
json={
"name": "large-profile",
"env_vars": {},
"files": {"large.txt": large_content},
},
)
assert response.status_code == 413
def test_create_config_profile_invalid_file_path(self, authenticated_client: TestClient) -> None:
"""Test that invalid file paths are rejected."""
response = authenticated_client.post(
"/config-profiles",
json={
"name": "bad-profile",
"env_vars": {},
"files": {"../../../etc/passwd": "malicious"},
},
)
assert response.status_code == 422
def test_create_config_profile_invalid_mount_target(self, authenticated_client: TestClient) -> None:
"""Test that invalid mount targets are rejected."""
response = authenticated_client.post(
"/config-profiles",
json={
"name": "bad-mount-profile",
"env_vars": {},
"files": {},
"mounts": [{"target": "relative/path", "mode": "rw", "files": {}}],
},
json={"name": " "},
)
assert response.status_code == 422
def test_get_config_profile_by_id(self, authenticated_client: TestClient) -> None:
"""Test getting a config profile by ID."""
# Create profile first
create_response = authenticated_client.post(
"/config-profiles",
json={
"name": "get-test",
"env_vars": {},
"files": {},
},
json={"name": "get-test"},
)
profile_id = create_response.json()["id"]
# Get it back
response = authenticated_client.get(f"/config-profiles/{profile_id}")
assert response.status_code == 200
data = response.json()
assert data["name"] == "get-test"
assert "includes" in data
assert "mounts" in data
def test_get_config_profile_not_found(self, authenticated_client: TestClient) -> None:
"""Test getting a non-existent profile."""
@@ -127,327 +81,381 @@ class TestConfigProfilesAPI:
def test_update_config_profile_successfully(self, authenticated_client: TestClient) -> None:
"""Test updating a config profile."""
# Create profile first
create_response = authenticated_client.post(
"/config-profiles",
json={
"name": "update-test",
"env_vars": {},
"files": {},
},
json={"name": "update-test"},
)
profile_id = create_response.json()["id"]
# Update it
response = authenticated_client.put(
f"/config-profiles/{profile_id}",
json={
"name": "updated-name",
"env_vars": {"NEW_VAR": "new_value"},
},
json={"name": "updated-name", "description": "updated desc"},
)
assert response.status_code == 200
data = response.json()
assert data["name"] == "updated-name"
assert data["env_vars"] == {"NEW_VAR": "new_value"}
assert data["description"] == "updated desc"
def test_delete_config_profile_successfully(self, authenticated_client: TestClient) -> None:
"""Test deleting a config profile."""
# Create profile first
create_response = authenticated_client.post(
"/config-profiles",
json={
"name": "delete-test",
"env_vars": {},
"files": {},
},
json={"name": "delete-test"},
)
profile_id = create_response.json()["id"]
# Delete it
response = authenticated_client.delete(f"/config-profiles/{profile_id}")
assert response.status_code == 204
# Verify it's gone
get_response = authenticated_client.get(f"/config-profiles/{profile_id}")
assert get_response.status_code == 404
def test_update_profile_includes_successfully(self, authenticated_client: TestClient) -> None:
"""Test updating profile includes."""
# Create base profile
base_response = authenticated_client.post(
def test_profile_access_check(self, authenticated_client: TestClient) -> None:
"""Test that users can only access their own profiles."""
# Create a profile
create_response = authenticated_client.post(
"/config-profiles",
json={
"name": "base-profile",
"env_vars": {"BASE_VAR": "base_value"},
"files": {},
},
json={"name": "access-test"},
)
base_id = base_response.json()["id"]
profile_id = create_response.json()["id"]
# Create child profile
child_response = authenticated_client.post(
"/config-profiles",
json={
"name": "child-profile",
"env_vars": {},
"files": {},
},
)
child_id = child_response.json()["id"]
# Update includes
response = authenticated_client.put(
f"/config-profiles/{child_id}/includes",
json={"includes": [base_id]},
)
# The profile should be accessible
response = authenticated_client.get(f"/config-profiles/{profile_id}")
assert response.status_code == 200
data = response.json()
print(f"Response data: {data}")
print(f"Includes: {data.get('includes', 'NO INCLUDES KEY')}")
assert len(data["includes"]) == 1, f"Expected 1 include, got {len(data.get('includes', []))}: {data.get('includes', [])}"
assert data["includes"][0]["included_profile_id"] == base_id
def test_update_profile_includes_cycle_detection(self, authenticated_client: TestClient) -> None:
"""Test that include cycles are detected."""
# Create profile A
a_response = authenticated_client.post(
@pytest.mark.integration
class TestConfigProfileIncludes:
"""Integration tests for config profile includes."""
def test_add_include_successfully(self, authenticated_client: TestClient) -> None:
"""Test adding an include to a profile."""
# Create two profiles
profile1 = authenticated_client.post(
"/config-profiles",
json={
"name": "profile-a",
"env_vars": {},
"files": {},
},
)
a_id = a_response.json()["id"]
# Create profile B
b_response = authenticated_client.post(
json={"name": "profile-1"},
).json()
profile2 = authenticated_client.post(
"/config-profiles",
json={
"name": "profile-b",
"env_vars": {},
"files": {},
},
)
b_id = b_response.json()["id"]
# Make B include A
authenticated_client.put(
f"/config-profiles/{b_id}/includes",
json={"includes": [a_id]},
)
# Try to make A include B (would create cycle)
response = authenticated_client.put(
f"/config-profiles/{a_id}/includes",
json={"includes": [b_id]},
)
assert response.status_code == 400
def test_preview_config_profile_successfully(self, authenticated_client: TestClient) -> None:
"""Test previewing a resolved config profile."""
# Create base profile
base_response = authenticated_client.post(
"/config-profiles",
json={
"name": "preview-base",
"env_vars": {"BASE_VAR": "base"},
"files": {},
},
)
base_id = base_response.json()["id"]
# Create child profile
child_response = authenticated_client.post(
"/config-profiles",
json={
"name": "preview-child",
"env_vars": {"CHILD_VAR": "child"},
"files": {},
},
)
child_id = child_response.json()["id"]
# Make child include base
authenticated_client.put(
f"/config-profiles/{child_id}/includes",
json={"includes": [base_id]},
)
# Preview child
response = authenticated_client.get(f"/config-profiles/{child_id}/preview")
assert response.status_code == 200
data = response.json()
assert data["profile_name"] == "preview-child"
assert data["env_vars"]["BASE_VAR"] == "base"
assert data["env_vars"]["CHILD_VAR"] == "child"
assert len(data["included_profiles"]) == 1
def test_resolve_default_profile(self, authenticated_client: TestClient) -> None:
"""Test resolving default profile for project/tool."""
# Create a global default profile (no project/tool scoping)
authenticated_client.post(
"/config-profiles",
json={
"name": "default-profile",
"env_vars": {},
"files": {},
"is_default": True,
},
)
# Resolve default with random project/tool (should fall back to global)
project_id = str(uuid.uuid4())
tool_type_id = str(uuid.uuid4())
response = authenticated_client.get(
"/config-profiles/defaults/resolve",
params={"project_id": project_id, "tool_type_id": tool_type_id},
)
assert response.status_code == 200
data = response.json()
assert data["profile_name"] == "default-profile"
def test_resolve_default_profile_no_match(self, authenticated_client: TestClient) -> None:
"""Test resolving default profile when no profiles exist."""
project_id = str(uuid.uuid4())
tool_type_id = str(uuid.uuid4())
response = authenticated_client.get(
"/config-profiles/defaults/resolve",
params={"project_id": project_id, "tool_type_id": tool_type_id},
)
assert response.status_code == 200
data = response.json()
assert data["profile_id"] is None
def test_create_config_profile_with_git_mounts(self, authenticated_client: TestClient, test_project_and_repo) -> None:
"""Test creating a config profile with git mounts."""
_project_id, repo_id = test_project_and_repo
json={"name": "profile-2"},
).json()
# Add include
response = authenticated_client.post(
"/config-profiles",
json={
"name": "git-mount-profile",
"env_vars": {},
"files": {},
"git_mounts": [
{
"remote_url": "https://github.com/user/repo.git",
"source_path": ".",
"target_path": "/app",
"branch": "main",
}
],
},
f"/config-profiles/{profile1['id']}/includes",
json={"included_profile_id": profile2["id"], "order_index": 0},
)
assert response.status_code == 201
data = response.json()
assert data["name"] == "git-mount-profile"
assert len(data["git_mounts"]) == 1
assert data["git_mounts"][0]["target_path"] == "/app"
assert data["git_mounts"][0]["branch"] == "main"
assert data["included_profile_id"] == profile2["id"]
assert data["included_profile_name"] == "profile-2"
def test_update_config_profile_git_mounts(self, authenticated_client: TestClient, test_project_and_repo) -> None:
"""Test updating git mounts on a config profile."""
_project_id, repo_id = test_project_and_repo
# Create profile first
create_response = authenticated_client.post(
def test_add_self_include_rejected(self, authenticated_client: TestClient) -> None:
"""Test that self-includes are rejected."""
profile = authenticated_client.post(
"/config-profiles",
json={
"name": "update-git-mounts",
"env_vars": {},
"files": {},
},
)
profile_id = create_response.json()["id"]
json={"name": "self-include-test"},
).json()
response = authenticated_client.post(
f"/config-profiles/{profile['id']}/includes",
json={"included_profile_id": profile["id"], "order_index": 0},
)
assert response.status_code == 400
def test_add_include_cycle_rejected(self, authenticated_client: TestClient) -> None:
"""Test that circular includes are rejected."""
profile1 = authenticated_client.post(
"/config-profiles",
json={"name": "cycle-1"},
).json()
profile2 = authenticated_client.post(
"/config-profiles",
json={"name": "cycle-2"},
).json()
# Add profile1 includes profile2
authenticated_client.post(
f"/config-profiles/{profile1['id']}/includes",
json={"included_profile_id": profile2["id"], "order_index": 0},
)
# Try to add profile2 includes profile1 (creates cycle)
response = authenticated_client.post(
f"/config-profiles/{profile2['id']}/includes",
json={"included_profile_id": profile1["id"], "order_index": 0},
)
assert response.status_code == 400
def test_add_deep_cycle_rejected(self, authenticated_client: TestClient) -> None:
"""Test that deep circular includes are rejected."""
p1 = authenticated_client.post(
"/config-profiles", json={"name": "deep-1"}
).json()
p2 = authenticated_client.post(
"/config-profiles", json={"name": "deep-2"}
).json()
p3 = authenticated_client.post(
"/config-profiles", json={"name": "deep-3"}
).json()
# p1 -> p2 -> p3
authenticated_client.post(
f"/config-profiles/{p1['id']}/includes",
json={"included_profile_id": p2["id"], "order_index": 0},
)
authenticated_client.post(
f"/config-profiles/{p2['id']}/includes",
json={"included_profile_id": p3["id"], "order_index": 0},
)
# Try p3 -> p1 (creates cycle)
response = authenticated_client.post(
f"/config-profiles/{p3['id']}/includes",
json={"included_profile_id": p1["id"], "order_index": 0},
)
assert response.status_code == 400
def test_add_duplicate_include_rejected(self, authenticated_client: TestClient) -> None:
"""Test that duplicate includes are rejected."""
p1 = authenticated_client.post(
"/config-profiles", json={"name": "dup-1"}
).json()
p2 = authenticated_client.post(
"/config-profiles", json={"name": "dup-2"}
).json()
authenticated_client.post(
f"/config-profiles/{p1['id']}/includes",
json={"included_profile_id": p2["id"], "order_index": 0},
)
response = authenticated_client.post(
f"/config-profiles/{p1['id']}/includes",
json={"included_profile_id": p2["id"], "order_index": 1},
)
assert response.status_code == 409
def test_list_includes(self, authenticated_client: TestClient) -> None:
"""Test listing includes for a profile."""
p1 = authenticated_client.post(
"/config-profiles", json={"name": "list-inc-1"}
).json()
p2 = authenticated_client.post(
"/config-profiles", json={"name": "list-inc-2"}
).json()
authenticated_client.post(
f"/config-profiles/{p1['id']}/includes",
json={"included_profile_id": p2["id"], "order_index": 0},
)
response = authenticated_client.get(f"/config-profiles/{p1['id']}/includes")
assert response.status_code == 200
data = response.json()
assert len(data["includes"]) == 1
def test_update_include_order(self, authenticated_client: TestClient) -> None:
"""Test updating include order index."""
p1 = authenticated_client.post(
"/config-profiles", json={"name": "order-1"}
).json()
p2 = authenticated_client.post(
"/config-profiles", json={"name": "order-2"}
).json()
inc = authenticated_client.post(
f"/config-profiles/{p1['id']}/includes",
json={"included_profile_id": p2["id"], "order_index": 0},
).json()
# Update with git mounts
response = authenticated_client.put(
f"/config-profiles/{profile_id}",
json={
"git_mounts": [
{
"remote_url": "https://github.com/user/repo.git",
"source_path": "config",
"target_path": "/config",
}
],
},
f"/config-profiles/{p1['id']}/includes/{inc['id']}",
json={"order_index": 5},
)
assert response.status_code == 200
data = response.json()
assert len(data["git_mounts"]) == 1
assert data["git_mounts"][0]["source_path"] == "config"
assert response.json()["order_index"] == 5
def test_create_config_profile_invalid_git_mount_source_path(self, authenticated_client: TestClient, test_project_and_repo) -> None:
"""Test that invalid git mount source paths are rejected."""
_project_id, repo_id = test_project_and_repo
def test_remove_include(self, authenticated_client: TestClient) -> None:
"""Test removing an include."""
p1 = authenticated_client.post(
"/config-profiles", json={"name": "rem-1"}
).json()
p2 = authenticated_client.post(
"/config-profiles", json={"name": "rem-2"}
).json()
inc = authenticated_client.post(
f"/config-profiles/{p1['id']}/includes",
json={"included_profile_id": p2["id"], "order_index": 0},
).json()
response = authenticated_client.delete(
f"/config-profiles/{p1['id']}/includes/{inc['id']}"
)
assert response.status_code == 204
@pytest.mark.integration
class TestConfigProfileMounts:
"""Integration tests for config profile mounts."""
def test_add_mount_successfully(self, authenticated_client: TestClient) -> None:
"""Test adding a mount to a profile."""
profile = authenticated_client.post(
"/config-profiles",
json={"name": "mount-test"},
).json()
response = authenticated_client.post(
f"/config-profiles/{profile['id']}/mounts",
json={"target_path": "/etc/config", "files": {"test.txt": "hello"}, "order_index": 0},
)
assert response.status_code == 201
data = response.json()
assert data["target_path"] == "/etc/config"
assert data["files"] == {"test.txt": "hello"}
def test_add_mount_relative_path_rejected(self, authenticated_client: TestClient) -> None:
"""Test that relative mount paths are rejected."""
profile = authenticated_client.post(
"/config-profiles",
json={
"name": "bad-git-mount",
"env_vars": {},
"files": {},
"git_mounts": [
{
"remote_url": "https://github.com/user/repo.git",
"source_path": "/absolute/path",
"target_path": "/app",
}
],
},
json={"name": "rel-path-test"},
).json()
response = authenticated_client.post(
f"/config-profiles/{profile['id']}/mounts",
json={"target_path": "etc/config", "files": {"test.txt": "hello"}},
)
assert response.status_code == 422
def test_create_config_profile_invalid_git_mount_target_path_traversal(self, authenticated_client: TestClient, test_project_and_repo) -> None:
"""Test that git mount target paths with traversal are rejected."""
_project_id, repo_id = test_project_and_repo
def test_add_target_path_traversal_rejected(self, authenticated_client: TestClient) -> None:
"""Test that path traversal in mount paths is rejected."""
profile = authenticated_client.post(
"/config-profiles",
json={"name": "traversal-test"},
).json()
response = authenticated_client.post(
"/config-profiles",
json={
"name": "bad-git-mount-target",
"env_vars": {},
"files": {},
"git_mounts": [
{
"remote_url": "https://github.com/user/repo.git",
"source_path": ".",
"target_path": "../../../etc/passwd",
}
],
},
f"/config-profiles/{profile['id']}/mounts",
json={"target_path": "/etc/../passwd", "files": {"test.txt": "hello"}},
)
assert response.status_code == 422
def test_preview_config_profile_with_git_mounts(self, authenticated_client: TestClient, test_project_and_repo) -> None:
"""Test previewing a profile with git mounts."""
_project_id, repo_id = test_project_and_repo
# Create profile with git mounts
create_response = authenticated_client.post(
def test_add_duplicate_mount_rejected(self, authenticated_client: TestClient) -> None:
"""Test that duplicate mount paths are rejected."""
profile = authenticated_client.post(
"/config-profiles",
json={
"name": "preview-git-mounts",
"env_vars": {},
"files": {},
"git_mounts": [
{
"remote_url": "https://github.com/user/repo.git",
"source_path": ".",
"target_path": "/app",
}
],
},
)
profile_id = create_response.json()["id"]
json={"name": "dup-mount-test"},
).json()
# Preview
response = authenticated_client.get(f"/config-profiles/{profile_id}/preview")
authenticated_client.post(
f"/config-profiles/{profile['id']}/mounts",
json={"target_path": "/etc/config", "files": {"test.txt": "hello"}},
)
response = authenticated_client.post(
f"/config-profiles/{profile['id']}/mounts",
json={"target_path": "/etc/config", "files": {"test.txt": "world"}},
)
assert response.status_code == 409
def test_update_mount(self, authenticated_client: TestClient) -> None:
"""Test updating a mount."""
profile = authenticated_client.post(
"/config-profiles",
json={"name": "update-mount-test"},
).json()
mount = authenticated_client.post(
f"/config-profiles/{profile['id']}/mounts",
json={"target_path": "/old/path", "files": {"test.txt": "old"}},
).json()
response = authenticated_client.put(
f"/config-profiles/{profile['id']}/mounts/{mount['id']}",
json={"target_path": "/new/path", "files": {"test.txt": "new"}, "order_index": 2},
)
assert response.status_code == 200
data = response.json()
assert len(data["git_mounts"]) == 1
assert data["git_mounts"][0]["remote_url"] == "https://github.com/user/repo.git"
assert data["target_path"] == "/new/path"
assert data["files"] == {"test.txt": "new"}
assert data["order_index"] == 2
def test_remove_mount(self, authenticated_client: TestClient) -> None:
"""Test removing a mount."""
profile = authenticated_client.post(
"/config-profiles",
json={"name": "rem-mount-test"},
).json()
mount = authenticated_client.post(
f"/config-profiles/{profile['id']}/mounts",
json={"target_path": "/tmp/test", "files": {"test.txt": "x"}},
).json()
response = authenticated_client.delete(
f"/config-profiles/{profile['id']}/mounts/{mount['id']}"
)
assert response.status_code == 204
@pytest.mark.integration
class TestConfigProfileDefaults:
"""Integration tests for default profile APIs."""
def test_get_default_profiles_empty(self, authenticated_client: TestClient) -> None:
"""Test getting default profiles when none are set."""
response = authenticated_client.get("/config-profiles/defaults")
assert response.status_code == 200
data = response.json()
assert data["default_profiles"] == {}
def test_set_default_profiles(self, authenticated_client: TestClient) -> None:
"""Test setting default profiles."""
profile = authenticated_client.post(
"/config-profiles",
json={"name": "default-test"},
).json()
response = authenticated_client.put(
"/config-profiles/defaults",
json={"default_profiles": {"code-server": profile["id"]}},
)
assert response.status_code == 200
data = response.json()
assert data["default_profiles"]["code-server"] == profile["id"]
def test_set_default_profiles_invalid_profile(self, authenticated_client: TestClient) -> None:
"""Test setting default profiles with invalid profile ID."""
response = authenticated_client.put(
"/config-profiles/defaults",
json={"default_profiles": {"code-server": str(uuid.uuid4())}},
)
assert response.status_code == 404
def test_get_default_profile_for_tool_type(self, authenticated_client: TestClient) -> None:
"""Test getting default profile for a specific tool type."""
profile = authenticated_client.post(
"/config-profiles",
json={"name": "tool-default-test"},
).json()
authenticated_client.put(
"/config-profiles/defaults",
json={"default_profiles": {"jupyter-notebook": profile["id"]}},
)
response = authenticated_client.get("/config-profiles/defaults/jupyter-notebook")
assert response.status_code == 200
data = response.json()
assert data["tool_type_id"] == "jupyter-notebook"
assert data["profile_id"] == profile["id"]
def test_get_default_profile_for_tool_type_not_set(self, authenticated_client: TestClient) -> None:
"""Test getting default profile when not set."""
response = authenticated_client.get("/config-profiles/defaults/opencode")
assert response.status_code == 200
data = response.json()
assert data["tool_type_id"] == "opencode"
assert data["profile_id"] is None
apps/api/tests/integration/test_git_control.py
-14
View File
@@ -70,20 +70,6 @@ def test_get_current_branch_handles_unborn_main() -> None:
assert get_current_branch(tmpdir) == "main"
def test_create_branch_on_bare_repo_with_no_commits() -> None:
with tempfile.TemporaryDirectory() as tmpdir:
os.system(f"git init --bare {tmpdir}/bare.git >/dev/null 2>&1")
create_branch(f"{tmpdir}/bare.git", "main")
assert get_current_branch(f"{tmpdir}/bare.git") == "main"
def test_checkout_branch_on_bare_repo_with_no_commits() -> None:
with tempfile.TemporaryDirectory() as tmpdir:
os.system(f"git init --bare {tmpdir}/bare.git >/dev/null 2>&1")
checkout_branch(f"{tmpdir}/bare.git", "main")
assert get_current_branch(f"{tmpdir}/bare.git") == "main"
class TestBranchOperations:
"""Tests for branch management functions."""
apps/api/tests/integration/test_models.py
+30 -15
View File
@@ -16,6 +16,7 @@ def test_base_metadata_collects_declared_tables() -> None:
@pytest.mark.integration
def test_shared_mixins_define_expected_columns() -> None:
assert "id" in UUIDPrimaryKeyMixin.__dict__
assert "created_at" in TimestampMixin.__dict__
@@ -23,26 +24,20 @@ def test_shared_mixins_define_expected_columns() -> None:
@pytest.mark.integration
def test_expected_tables_are_registered() -> None:
assert set(Base.metadata.tables) == {
"config_profile_includes",
"config_profiles",
"refresh_tokens",
"git_repositories",
"health_checks",
"instance_events",
"notifications",
"projects",
"ssh_keys",
"terminal_sessions",
"tool_definition_manifests",
"tool_instances",
"tool_types",
"user_configs",
"users",
}
@pytest.mark.integration
def test_user_table_has_required_columns() -> None:
columns = User.__table__.columns
@@ -61,6 +56,7 @@ def test_user_table_has_required_columns() -> None:
@pytest.mark.integration
def test_project_relationships_point_to_owner_and_default_ssh_key() -> None:
owner_fk = next(iter(Project.__table__.c.owner_id.foreign_keys))
ssh_fk = next(iter(Project.__table__.c.default_ssh_key_id.foreign_keys))
@@ -72,6 +68,7 @@ def test_project_relationships_point_to_owner_and_default_ssh_key() -> None:
@pytest.mark.integration
def test_repository_and_user_config_relationships_are_registered() -> None:
project_fk = next(iter(GitRepository.__table__.c.project_id.foreign_keys))
owner_fk = next(iter(GitRepository.__table__.c.owner_id.foreign_keys))
@@ -85,15 +82,33 @@ def test_repository_and_user_config_relationships_are_registered() -> None:
assert UserConfig.user.property.mapper.class_ is User
@pytest.mark.integration
def test_refresh_token_table_has_required_columns_and_relationships() -> None:
columns = RefreshToken.__table__.columns
user_fk = next(iter(RefreshToken.__table__.c.user_id.foreign_keys))
assert set(columns.keys()) == {
"id",
"user_id",
"token_hash",
"expires_at",
"revoked_at",
"user_agent",
"ip_address",
"created_at",
}
assert columns["token_hash"].unique is True
assert columns["revoked_at"].nullable is True
assert user_fk.target_fullname == "users.id"
assert RefreshToken.user.property.mapper.class_ is User
@pytest.mark.asyncio
@pytest.mark.integration
async def test_async_session_can_insert_and_load_user(db_session: AsyncSession) -> None:
user = User(
email="dev@headquarter.local",
name="Dev User",
authentik_id="dev-user",
avatar_url=None,
)
user = User(email="dev@headquarter.local", name="Dev User", authentik_id="dev-user", avatar_url=None)
db_session.add(user)
await db_session.commit()
apps/api/tests/integration/test_projects_api.py
+2 -3
View File
@@ -1,9 +1,8 @@
import uuid
from datetime import UTC, datetime, timedelta
from datetime import datetime, timedelta, timezone
import asyncio
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
@@ -59,7 +58,7 @@ def _mint_token(user_id: str) -> str:
subject=user_id,
email="test@headquarter.local",
name="Test User",
expires_at=datetime.now(UTC) + timedelta(minutes=15),
expires_at=datetime.now(timezone.utc) + timedelta(minutes=15),
)
apps/api/tests/integration/test_tool_configs_api_extended.py
+256
View File
@@ -0,0 +1,256 @@
import uuid
import pytest
from fastapi.testclient import TestClient
@pytest.mark.integration
class TestToolConfigsAPIExtended:
"""Integration tests for tool configs API with new fields."""
def test_create_tool_config_with_new_fields(self, authenticated_client: TestClient) -> None:
"""Test creating a tool config with all new fields."""
# Create a tool type first
tool_response = authenticated_client.post(
"/tool-types",
json={
"name": "config-test-tool",
"display_name": "Config Test Tool",
"default_port": 8080,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx",
"required_variables": [],
},
)
tool_id = tool_response.json()["id"]
# Create config with new fields
response = authenticated_client.post(
"/tool-configs",
json={
"tool_type_id": tool_id,
"key": "ADVANCED_CONFIG",
"value": "test-value",
"config_type": "env",
"port_override": 9090,
"start_command": "python app.py",
"working_directory": "/app",
"environment_variables": {"DEBUG": "true", "LOG_LEVEL": "debug"},
"volumes": [
{"source": "data", "target": "/data", "type": "bind"}
],
},
)
assert response.status_code == 201
data = response.json()
assert data["key"] == "ADVANCED_CONFIG"
assert data["port_override"] == 9090
assert data["start_command"] == "python app.py"
assert data["working_directory"] == "/app"
assert data["environment_variables"] == {"DEBUG": "true", "LOG_LEVEL": "debug"}
assert data["volumes"] == [{"source": "data", "target": "/data", "type": "bind"}]
def test_create_tool_config_invalid_port(self, authenticated_client: TestClient) -> None:
"""Test that invalid port numbers are rejected."""
# Create a tool type first
tool_response = authenticated_client.post(
"/tool-types",
json={
"name": "port-test-tool",
"display_name": "Port Test Tool",
"default_port": 8080,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx",
"required_variables": [],
},
)
tool_id = tool_response.json()["id"]
# Try to create config with invalid port
response = authenticated_client.post(
"/tool-configs",
json={
"tool_type_id": tool_id,
"key": "BAD_PORT",
"value": "test",
"config_type": "env",
"port_override": 99999,
},
)
assert response.status_code == 422
def test_create_tool_config_invalid_volume_structure(self, authenticated_client: TestClient) -> None:
"""Test that invalid volume structures are rejected."""
# Create a tool type first
tool_response = authenticated_client.post(
"/tool-types",
json={
"name": "volume-test-tool",
"display_name": "Volume Test Tool",
"default_port": 8080,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx",
"required_variables": [],
},
)
tool_id = tool_response.json()["id"]
# Try to create config with invalid volume
response = authenticated_client.post(
"/tool-configs",
json={
"tool_type_id": tool_id,
"key": "BAD_VOLUME",
"value": "test",
"config_type": "env",
"volumes": [{"invalid": "structure"}],
},
)
assert response.status_code == 422
def test_update_tool_config_with_new_fields(self, authenticated_client: TestClient) -> None:
"""Test updating a tool config with new fields."""
# Create a tool type first
tool_response = authenticated_client.post(
"/tool-types",
json={
"name": "update-config-tool",
"display_name": "Update Config Tool",
"default_port": 8080,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx",
"required_variables": [],
},
)
tool_id = tool_response.json()["id"]
# Create config
create_response = authenticated_client.post(
"/tool-configs",
json={
"tool_type_id": tool_id,
"key": "UPDATE_TEST",
"value": "original",
"config_type": "env",
},
)
config_id = create_response.json()["id"]
# Update with new fields
response = authenticated_client.put(
f"/tool-configs/{config_id}",
json={
"value": "updated",
"port_override": 3000,
"start_command": "npm start",
"working_directory": "/workspace",
"environment_variables": {"NODE_ENV": "production"},
"volumes": [{"source": "src", "target": "/app/src", "type": "bind"}],
},
)
assert response.status_code == 200
data = response.json()
assert data["value"] == "updated"
assert data["port_override"] == 3000
assert data["start_command"] == "npm start"
assert data["working_directory"] == "/workspace"
assert data["environment_variables"] == {"NODE_ENV": "production"}
def test_list_tool_configs_returns_new_fields(self, authenticated_client: TestClient) -> None:
"""Test that listing configs returns new fields."""
# Create a tool type first
tool_response = authenticated_client.post(
"/tool-types",
json={
"name": "list-config-tool",
"display_name": "List Config Tool",
"default_port": 8080,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx",
"required_variables": [],
},
)
tool_id = tool_response.json()["id"]
# Create config with new fields
authenticated_client.post(
"/tool-configs",
json={
"tool_type_id": tool_id,
"key": "LIST_TEST",
"value": "test",
"config_type": "env",
"port_override": 5000,
"environment_variables": {"TEST": "true"},
},
)
# List configs
response = authenticated_client.get("/tool-configs")
assert response.status_code == 200
data = response.json()
assert len(data) > 0
config = data[0]
assert "port_override" in config
assert "start_command" in config
assert "working_directory" in config
assert "environment_variables" in config
assert "volumes" in config
def test_get_tool_config_defaults(self, authenticated_client: TestClient) -> None:
"""Test getting tool config defaults."""
# Create a tool type first
tool_response = authenticated_client.post(
"/tool-types",
json={
"name": "defaults-tool",
"display_name": "Defaults Tool",
"default_port": 8080,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx\n volumes:\n - \"{{REPO_PATH}}:/workspace\"\n",
"required_variables": ["REPO_PATH"],
},
)
tool_id = tool_response.json()["id"]
# Get defaults
response = authenticated_client.get(f"/tool-configs/defaults/{tool_id}")
assert response.status_code == 200
data = response.json()
assert data["tool_type_id"] == tool_id
assert "suggested_configs" in data
def test_tool_config_backward_compatibility(self, authenticated_client: TestClient) -> None:
"""Test that old configs without new fields still work."""
# Create a tool type first
tool_response = authenticated_client.post(
"/tool-types",
json={
"name": "backward-compat-tool",
"display_name": "Backward Compat Tool",
"default_port": 8080,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx",
"required_variables": [],
},
)
tool_id = tool_response.json()["id"]
# Create config without new fields (simulating old client)
response = authenticated_client.post(
"/tool-configs",
json={
"tool_type_id": tool_id,
"key": "OLD_STYLE",
"value": "value",
"config_type": "env",
},
)
assert response.status_code == 201
data = response.json()
assert data["key"] == "OLD_STYLE"
# New fields should have default values
assert data["port_override"] is None
assert data["start_command"] is None
assert data["working_directory"] is None
assert data["environment_variables"] is None
assert data["volumes"] is None
apps/api/tests/integration/test_tool_types_api.py
+75 -2
View File
@@ -1,5 +1,5 @@
import uuid
from datetime import UTC, datetime, timedelta
from datetime import datetime, timedelta, timezone
import asyncio
import pytest
@@ -59,7 +59,7 @@ def _mint_token(user_id: str) -> str:
subject=user_id,
email="test@headquarter.local",
name="Test User",
expires_at=datetime.now(UTC) + timedelta(minutes=15),
expires_at=datetime.now(timezone.utc) + timedelta(minutes=15),
)
@@ -98,6 +98,7 @@ def _insert_tool_type(
name: str,
display_name: str,
compose_template: str,
is_builtin: bool = False,
created_by_id: str | None = None,
) -> None:
async def _run() -> None:
@@ -119,6 +120,7 @@ def _insert_tool_type(
description="A test tool type",
compose_template=compose_template,
required_variables=["REPO_PATH", "TOOL_NAME"],
is_builtin=is_builtin,
created_by_id=uuid.UUID(created_by_id) if created_by_id else None,
)
await session.merge(tool_type)
@@ -232,6 +234,7 @@ def test_create_tool_type_successfully() -> None:
assert data["name"] == "my-custom-tool"
assert data["display_name"] == "My Custom Tool"
assert data["description"] == "A custom development tool"
assert data["is_builtin"] == False
assert data["created_by_id"] == user_id
assert "id" in data
@@ -373,7 +376,28 @@ def test_update_tool_type_not_found() -> None:
assert response.status_code == 404
@pytest.mark.integration
def test_update_builtin_tool_type_fails() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
tool_type_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
_insert_user(user_id)
_insert_tool_type(
tool_type_id,
"builtin-tool",
"Built-in Tool",
"version: '3.8'\nservices:\n app:\n image: builtin",
is_builtin=True,
)
app = _load_app()
client = TestClient(app)
client.cookies.set("access_token", _mint_token(user_id))
payload = {"display_name": "Updated"}
response = client.put(f"/tool-types/{tool_type_id}", json=payload)
assert response.status_code == 403
@pytest.mark.integration
@@ -418,4 +442,53 @@ def test_delete_tool_type_not_found() -> None:
assert response.status_code == 404
@pytest.mark.integration
def test_delete_builtin_tool_type_fails() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
tool_type_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
_insert_user(user_id)
_insert_tool_type(
tool_type_id,
"builtin-tool",
"Built-in Tool",
"version: '3.8'\nservices:\n app:\n image: builtin",
is_builtin=True,
)
app = _load_app()
client = TestClient(app)
client.cookies.set("access_token", _mint_token(user_id))
response = client.delete(f"/tool-types/{tool_type_id}")
assert response.status_code == 403
@pytest.mark.integration
def test_builtin_tool_types_seeded_on_startup() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
_insert_user(user_id)
# Load app triggers startup event which seeds built-in types
app = _load_app()
client = TestClient(app)
client.cookies.set("access_token", _mint_token(user_id))
response = client.get("/tool-types")
assert response.status_code == 200
data = response.json()
# Check that built-in types exist
builtin_names = [t["name"] for t in data if t["is_builtin"]]
assert "code-server" in builtin_names
assert "jupyter-notebook" in builtin_names
# Verify built-in types have correct attributes
code_server = next((t for t in data if t["name"] == "code-server"), None)
assert code_server is not None
assert code_server["display_name"] == "VS Code Server"
assert "services" in code_server["compose_template"]
assert code_server["required_variables"] == ["REPO_PATH", "TOOL_NAME"]
apps/api/tests/integration/test_tool_types_api_extended.py
+13 -151
View File
@@ -1,3 +1,4 @@
import uuid
import pytest
from fastapi.testclient import TestClient
@@ -6,9 +7,7 @@ from fastapi.testclient import TestClient
class TestToolTypesAPIExtended:
"""Integration tests for tool types API with new fields."""
def test_create_tool_type_with_dockerfile(
self, authenticated_client: TestClient
) -> None:
def test_create_tool_type_with_dockerfile(self, authenticated_client: TestClient) -> None:
"""Test creating a tool type with dockerfile definition."""
response = authenticated_client.post(
"/tool-types",
@@ -29,9 +28,7 @@ class TestToolTypesAPIExtended:
assert data["definition_type"] == "dockerfile"
assert data["dockerfile_template"] == "FROM python:3.11\nRUN pip install flask"
def test_create_tool_type_with_readiness_probe(
self, authenticated_client: TestClient
) -> None:
def test_create_tool_type_with_readiness_probe(self, authenticated_client: TestClient) -> None:
"""Test creating a tool type with readiness probe."""
response = authenticated_client.post(
"/tool-types",
@@ -42,7 +39,7 @@ class TestToolTypesAPIExtended:
"interfaces": ["web"],
"default_port": 8080,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx\n ports:\n - '8080:8080'",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx",
"readiness_probe": {
"command": "curl -f http://localhost:8080",
"timeout": 30,
@@ -56,9 +53,7 @@ class TestToolTypesAPIExtended:
assert data["readiness_probe"]["command"] == "curl -f http://localhost:8080"
assert data["readiness_probe"]["timeout"] == 30
def test_create_tool_type_invalid_definition_type(
self, authenticated_client: TestClient
) -> None:
def test_create_tool_type_invalid_definition_type(self, authenticated_client: TestClient) -> None:
"""Test that invalid definition types are rejected."""
response = authenticated_client.post(
"/tool-types",
@@ -73,9 +68,7 @@ class TestToolTypesAPIExtended:
)
assert response.status_code == 422
def test_create_tool_type_dockerfile_without_template(
self, authenticated_client: TestClient
) -> None:
def test_create_tool_type_dockerfile_without_template(self, authenticated_client: TestClient) -> None:
"""Test that dockerfile type requires dockerfile_template."""
response = authenticated_client.post(
"/tool-types",
@@ -89,9 +82,7 @@ class TestToolTypesAPIExtended:
)
assert response.status_code == 422
def test_update_tool_type_with_new_fields(
self, authenticated_client: TestClient
) -> None:
def test_update_tool_type_with_new_fields(self, authenticated_client: TestClient) -> None:
"""Test updating a tool type with new fields."""
# Create tool type first
create_response = authenticated_client.post(
@@ -101,7 +92,7 @@ class TestToolTypesAPIExtended:
"display_name": "Update Test Tool",
"default_port": 8080,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx\n ports:\n - '8080:8080'",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx",
"required_variables": [],
},
)
@@ -122,9 +113,7 @@ class TestToolTypesAPIExtended:
assert response.status_code == 200
data = response.json()
assert data["display_name"] == "Updated Name"
assert (
data["readiness_probe"]["command"] == "curl -f http://localhost:8080/health"
)
assert data["readiness_probe"]["command"] == "curl -f http://localhost:8080/health"
def test_validate_tool_type_compose(self, authenticated_client: TestClient) -> None:
"""Test validating compose template."""
@@ -139,9 +128,7 @@ class TestToolTypesAPIExtended:
data = response.json()
assert data["valid"] is True
def test_validate_tool_type_invalid_compose(
self, authenticated_client: TestClient
) -> None:
def test_validate_tool_type_invalid_compose(self, authenticated_client: TestClient) -> None:
"""Test validating invalid compose template."""
response = authenticated_client.post(
"/tool-types/validate",
@@ -155,9 +142,7 @@ class TestToolTypesAPIExtended:
assert data["valid"] is False
assert "errors" in data
def test_validate_tool_type_dockerfile(
self, authenticated_client: TestClient
) -> None:
def test_validate_tool_type_dockerfile(self, authenticated_client: TestClient) -> None:
"""Test validating dockerfile template."""
response = authenticated_client.post(
"/tool-types/validate",
@@ -170,9 +155,7 @@ class TestToolTypesAPIExtended:
data = response.json()
assert data["valid"] is True
def test_get_tool_type_returns_new_fields(
self, authenticated_client: TestClient
) -> None:
def test_get_tool_type_returns_new_fields(self, authenticated_client: TestClient) -> None:
"""Test that GET returns new fields."""
# Create tool type with all fields
create_response = authenticated_client.post(
@@ -184,7 +167,7 @@ class TestToolTypesAPIExtended:
"interfaces": ["web", "terminal"],
"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\"",
"compose_template": "version: '3.8'\nservices:\n app:\n image: code-server\n volumes:\n - \"{{REPO_PATH}}:/workspace\"",
"readiness_probe": {
"command": "curl -f http://localhost:8443",
"timeout": 30,
@@ -203,124 +186,3 @@ class TestToolTypesAPIExtended:
assert data["category"] == "editor"
assert data["interfaces"] == ["web", "terminal"]
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."""
response = authenticated_client.post(
"/tool-types",
json={
"name": "no-port-tool",
"display_name": "No Port Tool",
"category": "utility",
"interfaces": ["web"],
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx\n ports:\n - '8080:8080'",
"required_variables": [],
},
)
assert response.status_code == 422
data = response.json()
assert "default_port" in str(data)
def test_create_tool_type_with_port_mismatch_fails(
self, authenticated_client: TestClient
) -> None:
"""Test that port mismatch between default_port and compose template fails."""
response = authenticated_client.post(
"/tool-types",
json={
"name": "port-mismatch-tool",
"display_name": "Port Mismatch Tool",
"category": "utility",
"interfaces": ["web"],
"default_port": 9999,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx\n ports:\n - '8080:8080'",
"required_variables": [],
},
)
assert response.status_code == 422
_ = response.json()
def test_create_tool_type_with_startup_command(
self, authenticated_client: TestClient
) -> None:
"""Test creating a tool type with startup_command."""
response = authenticated_client.post(
"/tool-types",
json={
"name": "startup-tool",
"display_name": "Startup Tool",
"category": "utility",
"interface_type": "terminal",
"requires_port": False,
"default_port": 0,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: alpine",
"startup_command": "cd /workspace && ls",
"required_variables": [],
},
)
assert response.status_code == 201
data = response.json()
assert data["startup_command"] == "cd /workspace && ls"
assert data["interface_type"] == "terminal"
def test_update_tool_type_startup_command(
self, authenticated_client: TestClient
) -> None:
"""Test updating a tool type's startup_command."""
# Create tool type first
create_response = authenticated_client.post(
"/tool-types",
json={
"name": "update-startup-tool",
"display_name": "Update Startup Tool",
"interface_type": "terminal",
"requires_port": False,
"default_port": 0,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: alpine",
"required_variables": [],
},
)
tool_id = create_response.json()["id"]
# Update with startup_command
response = authenticated_client.put(
f"/tool-types/{tool_id}",
json={
"startup_command": "source /etc/profile",
},
)
assert response.status_code == 200
data = response.json()
assert data["startup_command"] == "source /etc/profile"
def test_get_tool_type_returns_startup_command(
self, authenticated_client: TestClient
) -> None:
"""Test that GET returns startup_command."""
create_response = authenticated_client.post(
"/tool-types",
json={
"name": "get-startup-tool",
"display_name": "Get Startup Tool",
"interface_type": "terminal",
"requires_port": False,
"default_port": 0,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: alpine",
"startup_command": "echo hello",
"required_variables": [],
},
)
tool_id = create_response.json()["id"]
response = authenticated_client.get(f"/tool-types/{tool_id}")
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
+2 -2
View File
@@ -1,5 +1,5 @@
import uuid
from datetime import UTC, datetime, timedelta
from datetime import datetime, timedelta, timezone
import asyncio
import io
@@ -83,7 +83,7 @@ def _create_auth_cookie(user_id: str) -> str:
subject=user_id,
email="test@headquarter.local",
name="Test User",
expires_at=datetime.now(UTC) + timedelta(minutes=15),
expires_at=datetime.now(timezone.utc) + timedelta(minutes=15),
)
apps/api/tests/unit/test_git_url_parser.py
+1
View File
@@ -1,5 +1,6 @@
"""Tests for git URL parsing utilities."""
import pytest
from src.utils.git_url_parser import extract_base_repo_url, is_valid_clone_url, parse_git_url
apps/api/tests/unit/test_migration_metadata.py
+15
View File
@@ -39,3 +39,18 @@ def test_refresh_tokens_migration_has_expected_revision_chain() -> None:
assert module.revision == "0002_refresh_tokens"
assert module.down_revision == "0001_initial_schema"
@pytest.mark.unit
def test_config_profiles_migration_has_expected_revision_chain() -> None:
migration_path = Path(__file__).resolve().parents[2] / "alembic" / "versions" / "0013_add_config_profiles.py"
spec = spec_from_file_location("add_config_profiles", migration_path)
assert spec is not None
assert spec.loader is not None
module = module_from_spec(spec)
spec.loader.exec_module(module)
assert module.revision == "0013_add_config_profiles"
assert module.down_revision == "0012_default_port_req"
apps/api/tests/unit/test_profile_resolver.py
+463
View File
@@ -0,0 +1,463 @@
"""Unit tests for the profile resolver service."""
import uuid
from unittest.mock import MagicMock
import pytest
from src.services.profile_resolver import (
ProfileCycleError,
ResolvedProfileOutput,
resolve_profile,
)
def _make_profile(
name: str,
env_vars: dict[str, str] | None = None,
start_command: str | None = None,
working_directory: str | None = None,
port: int | None = None,
mounts: list[MagicMock] | None = None,
includes: list[MagicMock] | None = None,
) -> MagicMock:
"""Create a mock ConfigProfile for testing."""
profile = MagicMock()
profile.id = uuid.uuid4()
profile.name = name
profile.environment_variables = env_vars or {}
profile.start_command = start_command
profile.working_directory = working_directory
profile.port = port
profile.mounts = mounts or []
profile.includes = includes or []
return profile
def _make_include(included_profile: MagicMock, order_index: int = 0) -> MagicMock:
"""Create a mock ConfigInclude for testing."""
include = MagicMock()
include.included_profile = included_profile
include.order_index = order_index
return include
def _make_mount(
target_path: str,
mode: str = "rw",
files: dict[str, str] | None = None,
order_index: int = 0,
) -> MagicMock:
"""Create a mock ConfigMount for testing."""
mount = MagicMock()
mount.target_path = target_path
mount.mode = mode
mount.files = files or {}
mount.order_index = order_index
return mount
class TestResolveProfileBasic:
"""Tests for basic profile resolution without includes."""
def test_empty_profile(self) -> None:
"""Resolving an empty profile returns empty output."""
profile = _make_profile("empty")
result = resolve_profile(profile)
assert isinstance(result, ResolvedProfileOutput)
assert result.profile_name == "empty"
assert result.environment_variables == {}
assert result.runtime_hints.start_command is None
assert result.runtime_hints.working_directory is None
assert result.runtime_hints.port is None
assert result.mounts == {}
assert result.resolution_order == ["empty"]
def test_env_vars_only(self) -> None:
"""Profile with env vars resolves correctly."""
profile = _make_profile(
"env-only",
env_vars={"FOO": "bar", "BAZ": "qux"},
)
result = resolve_profile(profile)
assert result.environment_variables == {"FOO": "bar", "BAZ": "qux"}
assert result.env_var_sources == {
"FOO": ["env-only"],
"BAZ": ["env-only"],
}
def test_runtime_hints_only(self) -> None:
"""Profile with runtime hints resolves correctly."""
profile = _make_profile(
"hints-only",
start_command="python app.py",
working_directory="/app",
port=8080,
)
result = resolve_profile(profile)
assert result.runtime_hints.start_command == "python app.py"
assert result.runtime_hints.working_directory == "/app"
assert result.runtime_hints.port == 8080
assert result.runtime_hints.overridden_hints == {
"start_command": "hints-only",
"working_directory": "hints-only",
"port": "hints-only",
}
def test_mounts_only(self) -> None:
"""Profile with mounts resolves correctly."""
profile = _make_profile(
"mounts-only",
mounts=[
_make_mount(
"/config",
mode="ro",
files={"settings.json": '{"key": "value"}'},
),
],
)
result = resolve_profile(profile)
assert "/config" in result.mounts
mount = result.mounts["/config"]
assert mount.target_path == "/config"
assert mount.mode == "ro"
assert mount.files == {"settings.json": '{"key": "value"}'}
class TestResolveProfileIncludes:
"""Tests for profile resolution with includes."""
def test_single_include(self) -> None:
"""Profile with one include resolves in correct order."""
base = _make_profile("base", env_vars={"FOO": "base"})
derived = _make_profile(
"derived",
env_vars={"BAR": "derived"},
includes=[_make_include(base, order_index=0)],
)
result = resolve_profile(derived)
assert result.resolution_order == ["derived", "base"]
assert result.environment_variables == {
"FOO": "base",
"BAR": "derived",
}
def test_multiple_includes_ordered(self) -> None:
"""Multiple includes are resolved in order_index order."""
first = _make_profile("first", env_vars={"KEY": "first"})
second = _make_profile("second", env_vars={"KEY": "second"})
main = _make_profile(
"main",
includes=[
_make_include(first, order_index=0),
_make_include(second, order_index=1),
],
)
result = resolve_profile(main)
assert result.resolution_order == ["main", "first", "second"]
# second overrides first
assert result.environment_variables == {"KEY": "second"}
assert result.env_var_sources["KEY"] == ["first", "second"]
def test_include_order_matters(self) -> None:
"""Changing include order changes resolution."""
a = _make_profile("a", env_vars={"KEY": "a"})
b = _make_profile("b", env_vars={"KEY": "b"})
main1 = _make_profile(
"main",
includes=[
_make_include(a, order_index=0),
_make_include(b, order_index=1),
],
)
main2 = _make_profile(
"main",
includes=[
_make_include(b, order_index=0),
_make_include(a, order_index=1),
],
)
result1 = resolve_profile(main1)
result2 = resolve_profile(main2)
assert result1.environment_variables["KEY"] == "b"
assert result2.environment_variables["KEY"] == "a"
def test_nested_includes(self) -> None:
"""Deeply nested includes resolve recursively."""
deep = _make_profile("deep", env_vars={"DEEP": "value"})
mid = _make_profile(
"mid",
env_vars={"MID": "value"},
includes=[_make_include(deep, order_index=0)],
)
top = _make_profile(
"top",
env_vars={"TOP": "value"},
includes=[_make_include(mid, order_index=0)],
)
result = resolve_profile(top)
assert result.resolution_order == ["top", "mid", "deep"]
assert result.environment_variables == {
"TOP": "value",
"MID": "value",
"DEEP": "value",
}
class TestResolveProfileOverrides:
"""Tests for deterministic override rules."""
def test_env_var_override(self) -> None:
"""Later layers override earlier env vars."""
base = _make_profile("base", env_vars={"KEY": "base"})
override = _make_profile("override", env_vars={"KEY": "override"})
main = _make_profile(
"main",
includes=[
_make_include(base, order_index=0),
_make_include(override, order_index=1),
],
)
result = resolve_profile(main)
assert result.environment_variables["KEY"] == "override"
assert result.env_var_sources["KEY"] == ["base", "override"]
def test_main_profile_wins_over_includes(self) -> None:
"""The main profile itself wins over all includes."""
base = _make_profile("base", env_vars={"KEY": "base"})
main = _make_profile(
"main",
env_vars={"KEY": "main"},
includes=[_make_include(base, order_index=0)],
)
result = resolve_profile(main)
assert result.environment_variables["KEY"] == "main"
assert result.env_var_sources["KEY"] == ["base", "main"]
def test_runtime_hint_override(self) -> None:
"""Later layers override earlier runtime hints."""
base = _make_profile("base", start_command="python old.py")
override = _make_profile("override", start_command="python new.py")
main = _make_profile(
"main",
includes=[
_make_include(base, order_index=0),
_make_include(override, order_index=1),
],
)
result = resolve_profile(main)
assert result.runtime_hints.start_command == "python new.py"
assert result.runtime_hints.overridden_hints["start_command"] == "override"
def test_mount_file_override(self) -> None:
"""Later layers override earlier files in the same mount."""
base = _make_profile(
"base",
mounts=[
_make_mount(
"/config",
files={"app.json": '{"v": 1}'},
),
],
)
override = _make_profile(
"override",
mounts=[
_make_mount(
"/config",
files={"app.json": '{"v": 2}'},
),
],
)
main = _make_profile(
"main",
includes=[
_make_include(base, order_index=0),
_make_include(override, order_index=1),
],
)
result = resolve_profile(main)
mount = result.mounts["/config"]
assert mount.files["app.json"] == '{"v": 2}'
assert mount.overridden_files["app.json"] == ["override"]
def test_mount_mode_override(self) -> None:
"""Later layers override mount mode."""
base = _make_profile(
"base",
mounts=[_make_mount("/data", mode="ro")],
)
override = _make_profile(
"override",
mounts=[_make_mount("/data", mode="rw")],
)
main = _make_profile(
"main",
includes=[
_make_include(base, order_index=0),
_make_include(override, order_index=1),
],
)
result = resolve_profile(main)
assert result.mounts["/data"].mode == "rw"
assert result.mounts["/data"].mode_overridden_by == "override"
def test_mount_file_merge(self) -> None:
"""Different files in the same mount are merged."""
base = _make_profile(
"base",
mounts=[
_make_mount(
"/config",
files={"a.json": "1"},
),
],
)
override = _make_profile(
"override",
mounts=[
_make_mount(
"/config",
files={"b.json": "2"},
),
],
)
main = _make_profile(
"main",
includes=[
_make_include(base, order_index=0),
_make_include(override, order_index=1),
],
)
result = resolve_profile(main)
mount = result.mounts["/config"]
assert mount.files == {"a.json": "1", "b.json": "2"}
class TestResolveProfileCycles:
"""Tests for cycle detection during resolution."""
def test_direct_cycle(self) -> None:
"""A -> B -> A is detected."""
a = _make_profile("a")
b = _make_profile("b", includes=[_make_include(a, order_index=0)])
a.includes = [_make_include(b, order_index=0)]
with pytest.raises(ProfileCycleError) as exc_info:
resolve_profile(a)
assert "a" in exc_info.value.cycle_path
assert "b" in exc_info.value.cycle_path
def test_indirect_cycle(self) -> None:
"""A -> B -> C -> A is detected."""
a = _make_profile("a")
c = _make_profile("c")
b = _make_profile("b", includes=[_make_include(c, order_index=0)])
a.includes = [_make_include(b, order_index=0)]
c.includes = [_make_include(a, order_index=0)]
with pytest.raises(ProfileCycleError) as exc_info:
resolve_profile(a)
assert "a" in exc_info.value.cycle_path
assert "b" in exc_info.value.cycle_path
assert "c" in exc_info.value.cycle_path
def test_self_cycle(self) -> None:
"""A -> A is detected."""
a = _make_profile("a")
a.includes = [_make_include(a, order_index=0)]
with pytest.raises(ProfileCycleError) as exc_info:
resolve_profile(a)
assert exc_info.value.cycle_path == ["a", "a"]
def test_cycle_does_not_partially_resolve(self) -> None:
"""Cycle detection prevents any partial resolution."""
a = _make_profile("a", env_vars={"A": "a"})
b = _make_profile("b", env_vars={"B": "b"})
a.includes = [_make_include(b, order_index=0)]
b.includes = [_make_include(a, order_index=0)]
with pytest.raises(ProfileCycleError):
resolve_profile(a)
class TestResolveProfileDiamond:
"""Tests for diamond-shaped include graphs."""
def test_diamond_resolution(self) -> None:
"""Diamond graph resolves correctly without duplication issues."""
base = _make_profile("base", env_vars={"BASE": "base"})
left = _make_profile(
"left",
env_vars={"LEFT": "left"},
includes=[_make_include(base, order_index=0)],
)
right = _make_profile(
"right",
env_vars={"RIGHT": "right"},
includes=[_make_include(base, order_index=0)],
)
top = _make_profile(
"top",
env_vars={"TOP": "top"},
includes=[
_make_include(left, order_index=0),
_make_include(right, order_index=1),
],
)
result = resolve_profile(top)
# base should appear once (via left, then right skips because visited)
assert result.resolution_order == ["top", "left", "base", "right"]
assert result.environment_variables == {
"TOP": "top",
"LEFT": "left",
"RIGHT": "right",
"BASE": "base",
}
def test_diamond_override(self) -> None:
"""Diamond graph with conflicting overrides resolves correctly."""
base = _make_profile("base", env_vars={"KEY": "base"})
left = _make_profile(
"left",
env_vars={"KEY": "left"},
includes=[_make_include(base, order_index=0)],
)
right = _make_profile(
"right",
env_vars={"KEY": "right"},
includes=[_make_include(base, order_index=0)],
)
top = _make_profile(
"top",
includes=[
_make_include(left, order_index=0),
_make_include(right, order_index=1),
],
)
result = resolve_profile(top)
# right wins because it's later
assert result.environment_variables["KEY"] == "right"
assert result.env_var_sources["KEY"] == ["base", "left", "right"]
# Note: base appears once because visited set skips duplicate resolution in diamond graphs
apps/api/tests/unit/test_readiness_probe.py
+2
View File
@@ -1,7 +1,9 @@
"""Unit tests for readiness probe service."""
import asyncio
from unittest.mock import MagicMock, patch
import pytest
from src.services.readiness_probe import execute_probe
apps/api/tests/unit/test_terminal_manager.py
+112
View File
@@ -0,0 +1,112 @@
"""Unit tests for TerminalManager."""
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from src.services.terminal_manager import TerminalManager
from src.services.terminal_session import TerminalSession
@pytest.fixture
def manager():
return TerminalManager()
@pytest.fixture
def mock_websocket():
ws = AsyncMock()
ws.send_bytes = AsyncMock()
ws.send_json = AsyncMock()
ws.close = AsyncMock()
ws.receive = AsyncMock()
return ws
@pytest.fixture
def mock_session():
session = MagicMock(spec=TerminalSession)
session.session_id = "sess-123"
session.is_alive.return_value = True
session._closed = False
session.read_output = AsyncMock(return_value=b"")
session.write_input = AsyncMock()
session.resize = AsyncMock()
session.close = AsyncMock()
session.get_exit_reason.return_value = None
return session
class TestCreateSession:
@patch("src.services.terminal_manager.asyncio.create_task")
@patch("src.services.terminal_manager.uuid.uuid4", return_value="sess-123")
async def test_create_session_registers_and_starts_loops(
self, mock_uuid, mock_create_task, manager, mock_websocket
):
instance_id = __import__("uuid").uuid4()
mock_sess = MagicMock()
mock_sess.session_id = "sess-123"
mock_sess.is_alive.return_value = True
mock_sess._closed = False
mock_sess.start = AsyncMock()
mock_sess.read_output = AsyncMock(return_value=b"")
mock_sess.write_input = AsyncMock()
mock_sess.resize = AsyncMock()
mock_sess.close = AsyncMock()
mock_sess.get_exit_reason.return_value = None
with (
patch.object(manager, "_read_loop", new=AsyncMock()),
patch.object(manager, "_write_loop", new=AsyncMock()),
patch.object(manager, "_heartbeat_loop", new=AsyncMock()),
patch(
"src.services.terminal_manager.TerminalSession",
return_value=mock_sess,
),
):
session = await manager.create_session(
instance_id, "container-abc", mock_websocket
)
assert session.session_id == "sess-123"
assert "sess-123" in manager._sessions
assert "sess-123" in manager._last_client_message
class TestHandleControlMessage:
async def test_handle_resize(self, manager, mock_session, mock_websocket):
ctrl = {"type": "resize", "cols": 120, "rows": 40}
await manager._handle_control_message(mock_session, mock_websocket, ctrl)
mock_session.resize.assert_awaited_once_with(120, 40)
async def test_handle_ping(self, manager, mock_session, mock_websocket):
ctrl = {"type": "ping", "id": 42}
await manager._handle_control_message(mock_session, mock_websocket, ctrl)
mock_websocket.send_json.assert_awaited_once_with({"type": "pong", "id": 42})
async def test_handle_unknown_type(self, manager, mock_session, mock_websocket):
ctrl = {"type": "unknown", "data": "test"}
await manager._handle_control_message(mock_session, mock_websocket, ctrl)
mock_websocket.send_json.assert_not_awaited()
mock_session.resize.assert_not_awaited()
class TestCleanupSession:
async def test_cleanup_removes_session(self, manager, mock_session):
manager._sessions["sess-123"] = mock_session
manager._last_client_message["sess-123"] = 123.0
await manager._cleanup_session(mock_session)
assert "sess-123" not in manager._sessions
assert "sess-123" not in manager._last_client_message
mock_session.close.assert_awaited_once()
class TestCloseAll:
async def test_close_all_clears_sessions(self, manager, mock_session):
manager._sessions["sess-123"] = mock_session
manager._last_client_message["sess-123"] = 123.0
await manager.close_all()
assert len(manager._sessions) == 0
assert len(manager._last_client_message) == 0
mock_session.close.assert_awaited_once()
apps/api/tests/unit/test_terminal_session.py
+168
View File
@@ -0,0 +1,168 @@
"""Unit tests for TerminalSession."""
import asyncio
from unittest.mock import MagicMock, patch
import pytest
from src.services.terminal_session import TerminalSession
@pytest.fixture
def mock_pty():
"""Mock pty.openpty to return predictable fds."""
master_fd = 10
slave_fd = 11
with (
patch(
"src.services.terminal_session.pty.openpty",
return_value=(master_fd, slave_fd),
),
patch("src.services.terminal_session.os.close") as mock_close,
):
yield master_fd, slave_fd, mock_close
class TestTerminalSessionStart:
def test_init_state(self, mock_pty):
session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
assert session.session_id == "sess-1"
assert session.container_id == "container-abc"
assert session._echo_enabled is True
assert session._exit_reason is None
class TestTerminalSessionEchoDetection:
@patch("src.services.terminal_session.termios.tcgetattr")
def test_detect_echo_state_enabled(self, mock_tcgetattr):
session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
session._master_fd = 10
# termios.ECHO flag set
attrs = [[], [], [], __import__("termios").ECHO, [], [], []]
mock_tcgetattr.return_value = attrs
result = session._detect_echo_state()
assert result is True
@patch("src.services.terminal_session.termios.tcgetattr")
def test_detect_echo_state_disabled(self, mock_tcgetattr):
session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
session._master_fd = 10
# termios.ECHO flag NOT set
attrs = [[], [], [], 0, [], [], []]
mock_tcgetattr.return_value = attrs
result = session._detect_echo_state()
assert result is False
def test_detect_echo_state_no_master_fd(self):
session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
session._master_fd = None
result = session._detect_echo_state()
assert result is True # default
class TestTerminalSessionResize:
@patch("src.services.terminal_session.fcntl.ioctl")
def test_resize_sets_size(self, mock_ioctl):
session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
session._master_fd = 10
# Should not raise
asyncio.run(session.resize(120, 40))
mock_ioctl.assert_called_once()
def test_resize_when_closed(self):
session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
session._closed = True
# Should not raise
asyncio.run(session.resize(120, 40))
class TestTerminalSessionWriteInput:
@patch("src.services.terminal_session.os.write")
def test_write_input(self, mock_write):
session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
session._master_fd = 10
asyncio.run(session.write_input(b"hello"))
mock_write.assert_called_once_with(10, b"hello")
def test_write_input_when_closed(self):
session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
session._closed = True
# Should not raise
asyncio.run(session.write_input(b"hello"))
class TestTerminalSessionReadOutput:
@patch("src.services.terminal_session.select.select")
@patch("src.services.terminal_session.os.read")
def test_read_output_with_data(self, mock_read, mock_select):
session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
session._master_fd = 10
mock_select.return_value = ([10], [], [])
mock_read.return_value = b"output"
result = asyncio.run(session.read_output())
assert result == b"output"
@patch("src.services.terminal_session.select.select")
def test_read_output_no_data(self, mock_select):
session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
session._master_fd = 10
mock_select.return_value = ([], [], [])
result = asyncio.run(session.read_output())
assert result == b""
class TestTerminalSessionClose:
@patch("src.services.terminal_session.os.close")
@patch("src.services.terminal_session.asyncio.wait_for")
async def test_close_sets_exit_reason(self, mock_wait_for, mock_close):
session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
session._master_fd = 10
session.process = MagicMock()
session.process.returncode = 0
await session.close()
assert session._exit_reason == "process_exit"
assert session._closed is True
async def test_close_idempotent(self):
session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
session._closed = True
# Should not raise
await session.close()
class TestTerminalSessionIsAlive:
def test_is_alive_with_running_process(self):
session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
session.process = MagicMock()
session.process.returncode = None
assert session.is_alive() is True
def test_is_alive_with_exited_process(self):
session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
session.process = MagicMock()
session.process.returncode = 0
assert session.is_alive() is False
def test_is_alive_no_process(self):
session = TerminalSession("sess-1", __import__("uuid").uuid4(), "container-abc")
session.process = None
assert session.is_alive() is False