merge: integrate main restructuring into dev
- Resolve 57 merge conflicts from codebase restructure - Port dev feature code to new directory structure: * Update import paths to use @/ aliases * Add backward-compatible API signatures (createInstance, startInstance, deleteInstance) * Add missing type exports (ProjectWithRepos, InstanceHealth, Branch, BranchesResponse) * Extend Session and GitRepository types for dev features * Extend TerminalComponent props for mobile terminal wrapper * Add missing icon names (bell, drag, undo) Quality gates: tsc pass (0 errors), build pass, 127/131 tests pass (4 pre-existing failures unrelated to merge)
This commit is contained in:
Developer
@@ -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
|
||||
|
||||
@@ -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),
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -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),
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -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),
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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
|
||||
@@ -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()
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user