alex/headquarter
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
6aea83bf17c4a9304765ddd4421b30a5db6fe0c0
headquarter/apps/api/tests/unit/test_tool_instances_legacy.py
T
Alex Blank c6b804bf0a feat: fix SSH key mounting with multi-key support and unique filenames 
SSH key mounting was broken because:
1. Each selected key was mounted to a separate source dir but all targeted
   the same ~/.ssh path in the container, causing Docker Compose's
   last-mount-wins behavior
2. All keys were named id_ed25519, so they'd overwrite each other

Changes:
- ssh_keys.py: add key_filename param to prepare_ssh_key_files for unique
  key names; add write_ssh_config for combined multi-key config
- tool_instances.py: collect all selected keys into a single ~/.ssh mount
  with sanitized unique filenames (id_ed25519_<name>); generate combined
  SSH config with all IdentityFile entries
- tests: add os.makedirs mock for SSH permission tests

Quality gates: pytest (19 passed, 1 skipped)
2026-06-02 15:03:26 +02:00

1102 lines
38 KiB
Python
Raw Blame History

"""Unit tests for legacy tool instance fallback paths.
Verifies that tool types with definition_type "dockerfile", "compose",
and "legacy" continue to use the original startup flow after the
manifest-based flow was introduced.
"""
import os
import uuid
from datetime import datetime
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from src.api.tool_instances import create_instance, start_instance
from src.models.git_repository import GitRepository
from src.models.tool_instance import ToolInstance
from src.models.tool_type import ToolType
from src.models.ssh_key import SSHKey
from src.models.user import User
@pytest.fixture
def fake_user_id() -> uuid.UUID:
return uuid.uuid4()
@pytest.fixture
def fake_project_id() -> uuid.UUID:
return uuid.uuid4()
@pytest.fixture
def fake_repo_id() -> uuid.UUID:
return uuid.uuid4()
@pytest.fixture
def fake_tool_type_id() -> uuid.UUID:
return uuid.uuid4()
@pytest.fixture
def fake_instance_id() -> uuid.UUID:
return uuid.uuid4()
@pytest.fixture
def mock_session(fake_user_id, fake_project_id, fake_repo_id, fake_tool_type_id):
"""Return an async SQLAlchemy session with basic mocks."""
session = AsyncMock()
user = User(id=fake_user_id, email="test@example.com")
project = MagicMock()
project.id = fake_project_id
repo = GitRepository(
id=fake_repo_id,
project_id=fake_project_id,
name="test-repo",
path="/data/repos/test-repo",
remote_url=None,
ssh_key_id=None,
)
async def _get(model, pk):
if model is User and pk == fake_user_id:
return user
if model is GitRepository and pk == fake_repo_id:
return repo
return None
def _add(instance):
if getattr(instance, "created_at", None) is None:
instance.created_at = datetime.now()
if getattr(instance, "updated_at", None) is None:
instance.updated_at = datetime.now()
session.get.side_effect = _get
session.add = MagicMock(side_effect=_add)
session.execute.return_value = MagicMock(
scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[])))
)
return session
class TestCreateInstanceDockerfileLegacy:
"""Legacy dockerfile definition type in create_instance."""
@patch("src.api.tool_instances.ensure_instance_directory")
@patch("src.api.tool_instances.find_free_port")
@patch("src.api.tool_instances.build_image")
@patch("src.api.tool_instances.write_compose_file")
@patch("src.api.tool_instances._get_user")
@patch("src.api.tool_instances._get_owned_project")
async def test_builds_from_dockerfile_template(
self,
mock_get_project,
mock_get_user,
mock_write_compose,
mock_build_image,
mock_find_port,
mock_ensure_dir,
mock_session,
fake_user_id,
fake_project_id,
fake_repo_id,
fake_tool_type_id,
) -> None:
"""When definition_type is 'dockerfile', build_image is called with the template."""
mock_get_user.return_value = AsyncMock()
mock_get_project.return_value = AsyncMock()
mock_find_port.return_value = 12345
mock_ensure_dir.return_value = "/data/instances/test-instance"
mock_build_image.return_value = (0, "built", "")
tool_type = ToolType(
id=fake_tool_type_id,
name="legacy-df-tool",
display_name="Legacy DF Tool",
default_port=8080,
definition_type="dockerfile",
dockerfile_template="FROM python:3.11\nRUN echo hi",
compose_template=None,
)
repo = GitRepository(
id=fake_repo_id,
project_id=fake_project_id,
name="test-repo",
path="/data/repos/test-repo",
remote_url=None,
ssh_key_id=None,
)
async def _get(model, pk):
if model is ToolType and pk == fake_tool_type_id:
return tool_type
if model is GitRepository and pk == fake_repo_id:
return repo
if model is User and pk == fake_user_id:
return User(id=fake_user_id, email="test@example.com")
return None
mock_session.get.side_effect = _get
data = MagicMock()
data.tool_type_id = str(fake_tool_type_id)
data.display_name = None
data.workspace_id = None
data.clone_mode = "mount"
data.branch = None
data.new_branch = None
data.config_profile_id = None
data.ssh_key_ids = []
result = await create_instance(
project_id=fake_project_id,
repo_id=fake_repo_id,
data=data,
user_id=fake_user_id,
session=mock_session,
)
assert result["status"] == "pending"
mock_build_image.assert_called_once()
call_kwargs = mock_build_image.call_args.kwargs
assert call_kwargs["dockerfile"] == "FROM python:3.11\nRUN echo hi"
mock_write_compose.assert_called_once()
@patch("src.api.tool_instances.ensure_instance_directory")
@patch("src.api.tool_instances.find_free_port")
@patch("src.api.tool_instances.build_image")
@patch("src.api.tool_instances.write_compose_file")
@patch("src.api.tool_instances._get_user")
@patch("src.api.tool_instances._get_owned_project")
async def test_dockerfile_build_failure_raises_500(
self,
mock_get_project,
mock_get_user,
mock_write_compose,
mock_build_image,
mock_find_port,
mock_ensure_dir,
mock_session,
fake_user_id,
fake_project_id,
fake_repo_id,
fake_tool_type_id,
) -> None:
"""Failed dockerfile build should raise HTTP 500."""
from fastapi import HTTPException
mock_get_user.return_value = AsyncMock()
mock_get_project.return_value = AsyncMock()
mock_find_port.return_value = 12345
mock_ensure_dir.return_value = "/data/instances/test-instance"
mock_build_image.return_value = (1, "", "build failed")
tool_type = ToolType(
id=fake_tool_type_id,
name="legacy-df-tool",
display_name="Legacy DF Tool",
default_port=8080,
definition_type="dockerfile",
dockerfile_template="FROM invalid",
compose_template=None,
)
repo = GitRepository(
id=fake_repo_id,
project_id=fake_project_id,
name="test-repo",
path="/data/repos/test-repo",
remote_url=None,
ssh_key_id=None,
)
async def _get(model, pk):
if model is ToolType and pk == fake_tool_type_id:
return tool_type
if model is GitRepository and pk == fake_repo_id:
return repo
if model is User and pk == fake_user_id:
return User(id=fake_user_id, email="test@example.com")
return None
mock_session.get.side_effect = _get
data = MagicMock()
data.tool_type_id = str(fake_tool_type_id)
data.display_name = None
data.workspace_id = None
data.clone_mode = "mount"
data.branch = None
data.new_branch = None
data.config_profile_id = None
data.ssh_key_ids = []
with pytest.raises(HTTPException) as exc_info:
await create_instance(
project_id=fake_project_id,
repo_id=fake_repo_id,
data=data,
user_id=fake_user_id,
session=mock_session,
)
assert exc_info.value.status_code == 500
assert "Failed to build Docker image" in exc_info.value.detail
class TestCreateInstanceComposeLegacy:
"""Legacy compose definition type in create_instance."""
@patch("src.api.tool_instances.ensure_instance_directory")
@patch("src.api.tool_instances.find_free_port")
@patch("src.api.tool_instances.render_compose_template")
@patch("src.api.tool_instances.write_compose_file")
@patch("src.api.tool_instances._get_user")
@patch("src.api.tool_instances._get_owned_project")
async def test_renders_compose_template(
self,
mock_get_project,
mock_get_user,
mock_write_compose,
mock_render_compose,
mock_find_port,
mock_ensure_dir,
mock_session,
fake_user_id,
fake_project_id,
fake_repo_id,
fake_tool_type_id,
) -> None:
"""When definition_type is 'compose', render_compose_template is called."""
mock_get_user.return_value = AsyncMock()
mock_get_project.return_value = AsyncMock()
mock_find_port.return_value = 12345
mock_ensure_dir.return_value = "/data/instances/test-instance"
mock_render_compose.return_value = "services:\n app:\n image: nginx"
tool_type = ToolType(
id=fake_tool_type_id,
name="legacy-compose-tool",
display_name="Legacy Compose Tool",
default_port=80,
definition_type="compose",
dockerfile_template=None,
compose_template="services:\n app:\n image: nginx\n ports:\n - ${TOOL_PORT}:80",
)
repo = GitRepository(
id=fake_repo_id,
project_id=fake_project_id,
name="test-repo",
path="/data/repos/test-repo",
remote_url=None,
ssh_key_id=None,
)
async def _get(model, pk):
if model is ToolType and pk == fake_tool_type_id:
return tool_type
if model is GitRepository and pk == fake_repo_id:
return repo
if model is User and pk == fake_user_id:
return User(id=fake_user_id, email="test@example.com")
return None
mock_session.get.side_effect = _get
data = MagicMock()
data.tool_type_id = str(fake_tool_type_id)
data.display_name = None
data.workspace_id = None
data.clone_mode = "mount"
data.branch = None
data.new_branch = None
data.config_profile_id = None
data.ssh_key_ids = []
result = await create_instance(
project_id=fake_project_id,
repo_id=fake_repo_id,
data=data,
user_id=fake_user_id,
session=mock_session,
)
assert result["status"] == "pending"
mock_render_compose.assert_called_once()
args = mock_render_compose.call_args[0]
assert "services:" in args[0]
mock_write_compose.assert_called_once()
class TestCreateInstanceManifestNotCalledForLegacy:
"""Ensure manifest compiler is never invoked for legacy types."""
@patch("src.api.tool_instances.ensure_instance_directory")
@patch("src.api.tool_instances.find_free_port")
@patch("src.api.tool_instances.build_image")
@patch("src.api.tool_instances.write_compose_file")
@patch("src.api.tool_instances._prepare_manifest_instance")
@patch("src.api.tool_instances._get_user")
@patch("src.api.tool_instances._get_owned_project")
async def test_dockerfile_does_not_call_manifest_compiler(
self,
mock_get_project,
mock_get_user,
mock_prepare_manifest,
mock_write_compose,
mock_build_image,
mock_find_port,
mock_ensure_dir,
mock_session,
fake_user_id,
fake_project_id,
fake_repo_id,
fake_tool_type_id,
) -> None:
"""Legacy dockerfile type must not trigger manifest compilation."""
mock_get_user.return_value = AsyncMock()
mock_get_project.return_value = AsyncMock()
mock_find_port.return_value = 12345
mock_ensure_dir.return_value = "/data/instances/test-instance"
mock_build_image.return_value = (0, "built", "")
tool_type = ToolType(
id=fake_tool_type_id,
name="legacy-df",
display_name="Legacy",
default_port=8080,
definition_type="dockerfile",
dockerfile_template="FROM alpine",
compose_template=None,
)
repo = GitRepository(
id=fake_repo_id,
project_id=fake_project_id,
name="test-repo",
path="/data/repos/test-repo",
remote_url=None,
ssh_key_id=None,
)
async def _get(model, pk):
if model is ToolType and pk == fake_tool_type_id:
return tool_type
if model is GitRepository and pk == fake_repo_id:
return repo
if model is User and pk == fake_user_id:
return User(id=fake_user_id, email="test@example.com")
return None
mock_session.get.side_effect = _get
data = MagicMock()
data.tool_type_id = str(fake_tool_type_id)
data.display_name = None
data.workspace_id = None
data.clone_mode = "mount"
data.branch = None
data.new_branch = None
data.config_profile_id = None
data.ssh_key_ids = []
await create_instance(
project_id=fake_project_id,
repo_id=fake_repo_id,
data=data,
user_id=fake_user_id,
session=mock_session,
)
mock_prepare_manifest.assert_not_called()
class TestStartInstanceLegacyFallback:
"""Legacy paths in start_instance must NOT call manifest compiler."""
@patch("src.api.tool_instances.wait_for_container_running")
@patch("src.api.tool_instances.execute_compose_command")
@patch("src.api.tool_instances.get_container_id")
@patch("src.api.tool_instances.connect_container_to_network")
@patch("src.api.tool_instances._ensure_backend_network_in_compose")
@patch("src.api.tool_instances._ensure_container_name_in_compose")
@patch("src.api.tool_instances._ensure_web_bind_address")
@patch("src.api.tool_instances._sanitize_compose_file")
@patch("src.api.tool_instances._prepare_manifest_instance")
@patch("src.api.tool_instances._get_user")
@patch("src.api.tool_instances._get_owned_project")
async def test_legacy_type_skips_manifest_flow(
self,
mock_get_project,
mock_get_user,
mock_prepare_manifest,
mock_sanitize,
mock_ensure_web_bind,
mock_ensure_container_name,
mock_backend_network,
mock_connect_network,
mock_get_container_id,
mock_execute_compose,
mock_wait_container,
mock_session,
fake_user_id,
fake_project_id,
fake_repo_id,
fake_instance_id,
fake_tool_type_id,
) -> None:
"""When definition_type is 'legacy', start_instance uses old flow."""
mock_get_user.return_value = AsyncMock()
mock_get_project.return_value = AsyncMock()
mock_execute_compose.return_value = (0, "started", "")
mock_get_container_id.return_value = "abc123"
mock_connect_network.return_value = True
mock_wait_container.return_value = {
"success": True,
"status": "running",
"waited_seconds": 0.5,
}
instance = ToolInstance(
id=fake_instance_id,
name="legacy-instance",
repository_id=fake_repo_id,
tool_type_id=fake_tool_type_id,
compose_path="/data/instances/legacy-instance/docker-compose.yml",
status="stopped",
clone_mode="mount",
created_at=datetime.now(),
updated_at=datetime.now(),
)
tool_type = ToolType(
id=fake_tool_type_id,
name="legacy-tool",
display_name="Legacy Tool",
default_port=8080,
definition_type="legacy",
manifest_id=None,
dockerfile_template=None,
compose_template="services:\n app:\n image: nginx",
)
repo = GitRepository(
id=fake_repo_id,
project_id=fake_project_id,
name="test-repo",
path="/data/repos/test-repo",
remote_url=None,
ssh_key_id=None,
)
async def _get(model, pk):
if model is ToolInstance and pk == fake_instance_id:
return instance
if model is ToolType and pk == fake_tool_type_id:
return tool_type
if model is GitRepository and pk == fake_repo_id:
return repo
if model is User and pk == fake_user_id:
return User(id=fake_user_id, email="test@example.com")
return None
mock_session.get.side_effect = _get
# Ensure compose file exists so the check passes
with patch("os.path.exists", return_value=True):
result = await start_instance(
project_id=fake_project_id,
repo_id=fake_repo_id,
instance_id=fake_instance_id,
data=None,
user_id=fake_user_id,
session=mock_session,
)
assert result["status"] == "running"
mock_prepare_manifest.assert_not_called()
mock_execute_compose.assert_called_once()
@patch("src.api.tool_instances.wait_for_container_running")
@patch("src.api.tool_instances.execute_compose_command")
@patch("src.api.tool_instances.get_container_id")
@patch("src.api.tool_instances.connect_container_to_network")
@patch("src.api.tool_instances._ensure_backend_network_in_compose")
@patch("src.api.tool_instances._ensure_container_name_in_compose")
@patch("src.api.tool_instances._ensure_web_bind_address")
@patch("src.api.tool_instances._sanitize_compose_file")
@patch("src.api.tool_instances._prepare_manifest_instance")
@patch("src.api.tool_instances._get_user")
@patch("src.api.tool_instances._get_owned_project")
async def test_compose_type_skips_manifest_flow(
self,
mock_get_project,
mock_get_user,
mock_prepare_manifest,
mock_sanitize,
mock_ensure_web_bind,
mock_ensure_container_name,
mock_backend_network,
mock_connect_network,
mock_get_container_id,
mock_execute_compose,
mock_wait_container,
mock_session,
fake_user_id,
fake_project_id,
fake_repo_id,
fake_instance_id,
fake_tool_type_id,
) -> None:
"""When definition_type is 'compose', start_instance uses old flow."""
mock_get_user.return_value = AsyncMock()
mock_get_project.return_value = AsyncMock()
mock_execute_compose.return_value = (0, "started", "")
mock_get_container_id.return_value = "abc123"
mock_connect_network.return_value = True
mock_wait_container.return_value = {
"success": True,
"status": "running",
"waited_seconds": 0.5,
}
instance = ToolInstance(
id=fake_instance_id,
name="compose-instance",
repository_id=fake_repo_id,
tool_type_id=fake_tool_type_id,
compose_path="/data/instances/compose-instance/docker-compose.yml",
status="stopped",
clone_mode="mount",
created_at=datetime.now(),
updated_at=datetime.now(),
)
tool_type = ToolType(
id=fake_tool_type_id,
name="compose-tool",
display_name="Compose Tool",
default_port=8080,
definition_type="compose",
manifest_id=None,
dockerfile_template=None,
compose_template="services:\n app:\n image: nginx",
)
repo = GitRepository(
id=fake_repo_id,
project_id=fake_project_id,
name="test-repo",
path="/data/repos/test-repo",
remote_url=None,
ssh_key_id=None,
)
async def _get(model, pk):
if model is ToolInstance and pk == fake_instance_id:
return instance
if model is ToolType and pk == fake_tool_type_id:
return tool_type
if model is GitRepository and pk == fake_repo_id:
return repo
if model is User and pk == fake_user_id:
return User(id=fake_user_id, email="test@example.com")
return None
mock_session.get.side_effect = _get
with patch("os.path.exists", return_value=True):
result = await start_instance(
project_id=fake_project_id,
repo_id=fake_repo_id,
instance_id=fake_instance_id,
data=None,
user_id=fake_user_id,
session=mock_session,
)
assert result["status"] == "running"
mock_prepare_manifest.assert_not_called()
mock_execute_compose.assert_called_once()
@patch("src.api.tool_instances.wait_for_container_running")
@patch("src.api.tool_instances.execute_compose_command")
@patch("src.api.tool_instances.get_container_id")
@patch("src.api.tool_instances.connect_container_to_network")
@patch("src.api.tool_instances._ensure_backend_network_in_compose")
@patch("src.api.tool_instances._ensure_container_name_in_compose")
@patch("src.api.tool_instances._ensure_web_bind_address")
@patch("src.api.tool_instances._sanitize_compose_file")
@patch("src.api.tool_instances._prepare_manifest_instance")
@patch("src.api.tool_instances._get_user")
@patch("src.api.tool_instances._get_owned_project")
async def test_dockerfile_type_skips_manifest_flow(
self,
mock_get_project,
mock_get_user,
mock_prepare_manifest,
mock_sanitize,
mock_ensure_web_bind,
mock_ensure_container_name,
mock_backend_network,
mock_connect_network,
mock_get_container_id,
mock_execute_compose,
mock_wait_container,
mock_session,
fake_user_id,
fake_project_id,
fake_repo_id,
fake_instance_id,
fake_tool_type_id,
) -> None:
"""When definition_type is 'dockerfile', start_instance uses old flow."""
mock_get_user.return_value = AsyncMock()
mock_get_project.return_value = AsyncMock()
mock_execute_compose.return_value = (0, "started", "")
mock_get_container_id.return_value = "abc123"
mock_connect_network.return_value = True
mock_wait_container.return_value = {
"success": True,
"status": "running",
"waited_seconds": 0.5,
}
instance = ToolInstance(
id=fake_instance_id,
name="df-instance",
repository_id=fake_repo_id,
tool_type_id=fake_tool_type_id,
compose_path="/data/instances/df-instance/docker-compose.yml",
status="stopped",
clone_mode="mount",
created_at=datetime.now(),
updated_at=datetime.now(),
)
tool_type = ToolType(
id=fake_tool_type_id,
name="df-tool",
display_name="Dockerfile Tool",
default_port=8080,
definition_type="dockerfile",
manifest_id=None,
dockerfile_template="FROM alpine",
compose_template=None,
)
repo = GitRepository(
id=fake_repo_id,
project_id=fake_project_id,
name="test-repo",
path="/data/repos/test-repo",
remote_url=None,
ssh_key_id=None,
)
async def _get(model, pk):
if model is ToolInstance and pk == fake_instance_id:
return instance
if model is ToolType and pk == fake_tool_type_id:
return tool_type
if model is GitRepository and pk == fake_repo_id:
return repo
if model is User and pk == fake_user_id:
return User(id=fake_user_id, email="test@example.com")
return None
mock_session.get.side_effect = _get
with patch("os.path.exists", return_value=True):
result = await start_instance(
project_id=fake_project_id,
repo_id=fake_repo_id,
instance_id=fake_instance_id,
data=None,
user_id=fake_user_id,
session=mock_session,
)
assert result["status"] == "running"
mock_prepare_manifest.assert_not_called()
mock_execute_compose.assert_called_once()
class TestStartInstanceSshPermissions:
"""SSH key mounts trigger permission fixes after container starts."""
@patch("src.api.tool_instances.write_compose_file")
@patch("src.api.tool_instances.prepare_ssh_key_files")
@patch("src.api.tool_instances.apply_ssh_permissions")
@patch("src.api.tool_instances.wait_for_container_running")
@patch("src.api.tool_instances.execute_compose_command")
@patch("src.api.tool_instances.get_container_id")
@patch("src.api.tool_instances.connect_container_to_network")
@patch("src.api.tool_instances._ensure_backend_network_in_compose")
@patch("src.api.tool_instances._ensure_container_name_in_compose")
@patch("src.api.tool_instances._ensure_web_bind_address")
@patch("src.api.tool_instances._sanitize_compose_file")
@patch("src.api.tool_instances._get_user")
@patch("src.api.tool_instances._get_owned_project")
async def test_manifest_instance_applies_ssh_permissions(
self,
mock_get_project,
mock_get_user,
mock_sanitize,
mock_ensure_web_bind,
mock_ensure_container_name,
mock_backend_network,
mock_connect_network,
mock_get_container_id,
mock_execute_compose,
mock_wait_container,
mock_apply_ssh,
mock_prepare_ssh,
mock_write_compose,
mock_session,
fake_user_id,
fake_project_id,
fake_repo_id,
fake_instance_id,
fake_tool_type_id,
) -> None:
"""Manifest instance with SSH keys calls apply_ssh_permissions."""
from src.models.tool_definition_manifest import ToolDefinitionManifest
manifest_id = uuid.uuid4()
ssh_key_id = str(uuid.uuid4())
mock_get_user.return_value = AsyncMock()
mock_get_project.return_value = AsyncMock()
mock_execute_compose.return_value = (0, "started", "")
mock_get_container_id.return_value = "abc123"
mock_connect_network.return_value = True
mock_wait_container.return_value = {
"success": True,
"status": "running",
"waited_seconds": 0.5,
}
mock_apply_ssh.return_value = {"success": True, "error": None}
instance = ToolInstance(
id=fake_instance_id,
name="manifest-instance",
repository_id=fake_repo_id,
tool_type_id=fake_tool_type_id,
compose_path="/data/instances/manifest-instance/docker-compose.yml",
status="stopped",
clone_mode="mount",
ssh_key_ids=[ssh_key_id],
created_at=datetime.now(),
updated_at=datetime.now(),
)
tool_type = ToolType(
id=fake_tool_type_id,
name="manifest-tool",
display_name="Manifest Tool",
default_port=8080,
definition_type="manifest",
manifest_id=manifest_id,
dockerfile_template=None,
compose_template=None,
)
repo = GitRepository(
id=fake_repo_id,
project_id=fake_project_id,
name="test-repo",
path="/data/repos/test-repo",
remote_url=None,
ssh_key_id=None,
)
manifest_def = ToolDefinitionManifest(
id=manifest_id,
name="test-manifest",
display_name="Test Manifest",
interface_type="web",
manifest={"user": {"name": "user", "uid": 1001, "gid": 1001}},
)
ssh_key = SSHKey(
id=uuid.UUID(ssh_key_id),
user_id=fake_user_id,
name="test-key",
public_key="ssh-ed25519 AAA test@test",
private_key_encrypted="enc",
)
async def _get(model, pk):
if model is ToolInstance and pk == fake_instance_id:
return instance
if model is ToolType and pk == fake_tool_type_id:
return tool_type
if model is GitRepository and pk == fake_repo_id:
return repo
if model is User and pk == fake_user_id:
return User(id=fake_user_id, email="test@example.com")
if model is ToolDefinitionManifest and pk == manifest_id:
return manifest_def
if model is SSHKey and pk == uuid.UUID(ssh_key_id):
return ssh_key
return None
mock_session.get.side_effect = _get
with patch("os.path.exists", return_value=True):
with patch("os.makedirs"):
with patch(
"src.api.tool_instances._prepare_manifest_instance"
) as mock_prepare:
mock_prepare.return_value = (
"headquarter/test:latest",
"services:\n app:\n image: test",
{"name": "test-manifest", "user": {"name": "user"}},
"/home/user",
)
result = await start_instance(
project_id=fake_project_id,
repo_id=fake_repo_id,
instance_id=fake_instance_id,
data=None,
user_id=fake_user_id,
session=mock_session,
)
assert result["status"] == "running"
mock_apply_ssh.assert_called_once_with("abc123", "/home/user/.ssh", "user")
@patch("src.api.tool_instances.prepare_ssh_key_files")
@patch("src.api.tool_instances.apply_ssh_permissions")
@patch("src.api.tool_instances.wait_for_container_running")
@patch("src.api.tool_instances.execute_compose_command")
@patch("src.api.tool_instances.get_container_id")
@patch("src.api.tool_instances.connect_container_to_network")
@patch("src.api.tool_instances._ensure_backend_network_in_compose")
@patch("src.api.tool_instances._ensure_container_name_in_compose")
@patch("src.api.tool_instances._ensure_web_bind_address")
@patch("src.api.tool_instances._sanitize_compose_file")
@patch("src.api.tool_instances._get_user")
@patch("src.api.tool_instances._get_owned_project")
async def test_legacy_instance_applies_ssh_permissions(
self,
mock_get_project,
mock_get_user,
mock_sanitize,
mock_ensure_web_bind,
mock_ensure_container_name,
mock_backend_network,
mock_connect_network,
mock_get_container_id,
mock_execute_compose,
mock_wait_container,
mock_apply_ssh,
mock_prepare_ssh,
mock_session,
fake_user_id,
fake_project_id,
fake_repo_id,
fake_instance_id,
fake_tool_type_id,
) -> None:
"""Legacy instance with SSH keys calls apply_ssh_permissions."""
ssh_key_id = str(uuid.uuid4())
mock_get_user.return_value = AsyncMock()
mock_get_project.return_value = AsyncMock()
mock_execute_compose.return_value = (0, "started", "")
mock_get_container_id.return_value = "abc123"
mock_connect_network.return_value = True
mock_wait_container.return_value = {
"success": True,
"status": "running",
"waited_seconds": 0.5,
}
mock_apply_ssh.return_value = {"success": True, "error": None}
instance = ToolInstance(
id=fake_instance_id,
name="legacy-instance",
repository_id=fake_repo_id,
tool_type_id=fake_tool_type_id,
compose_path="/data/instances/legacy-instance/docker-compose.yml",
status="stopped",
clone_mode="mount",
ssh_key_ids=[ssh_key_id],
created_at=datetime.now(),
updated_at=datetime.now(),
)
tool_type = ToolType(
id=fake_tool_type_id,
name="legacy-tool",
display_name="Legacy Tool",
default_port=8080,
definition_type="legacy",
manifest_id=None,
dockerfile_template=None,
compose_template="services:\n app:\n image: nginx",
)
repo = GitRepository(
id=fake_repo_id,
project_id=fake_project_id,
name="test-repo",
path="/data/repos/test-repo",
remote_url=None,
ssh_key_id=None,
)
ssh_key = SSHKey(
id=uuid.UUID(ssh_key_id),
user_id=fake_user_id,
name="test-key",
public_key="ssh-ed25519 AAA test@test",
private_key_encrypted="enc",
)
async def _get(model, pk):
if model is ToolInstance and pk == fake_instance_id:
return instance
if model is ToolType and pk == fake_tool_type_id:
return tool_type
if model is GitRepository and pk == fake_repo_id:
return repo
if model is User and pk == fake_user_id:
return User(id=fake_user_id, email="test@example.com")
if model is SSHKey and pk == uuid.UUID(ssh_key_id):
return ssh_key
return None
mock_session.get.side_effect = _get
with patch("os.path.exists", return_value=True):
with patch("os.makedirs"):
with patch("src.api.tool_instances._modify_compose_file"):
result = await start_instance(
project_id=fake_project_id,
repo_id=fake_repo_id,
instance_id=fake_instance_id,
data=None,
user_id=fake_user_id,
session=mock_session,
)
assert result["status"] == "running"
mock_apply_ssh.assert_called_once_with("abc123", "/root/.ssh", "root")
class TestStartInstanceManifestBranch:
"""Manifest branch is taken ONLY when definition_type == 'manifest'."""
@patch("src.api.tool_instances.wait_for_container_running")
@patch("src.api.tool_instances.execute_compose_command")
@patch("src.api.tool_instances.get_container_id")
@patch("src.api.tool_instances.connect_container_to_network")
@patch("src.api.tool_instances._ensure_backend_network_in_compose")
@patch("src.api.tool_instances._ensure_container_name_in_compose")
@patch("src.api.tool_instances._ensure_web_bind_address")
@patch("src.api.tool_instances._sanitize_compose_file")
@patch("src.api.tool_instances._prepare_manifest_instance")
@patch("src.api.tool_instances.write_compose_file")
@patch("src.api.tool_instances._get_user")
@patch("src.api.tool_instances._get_owned_project")
async def test_manifest_type_calls_compiler(
self,
mock_get_project,
mock_get_user,
mock_write_compose,
mock_prepare_manifest,
mock_sanitize,
mock_ensure_web_bind,
mock_ensure_container_name,
mock_backend_network,
mock_connect_network,
mock_get_container_id,
mock_execute_compose,
mock_wait_container,
mock_session,
fake_user_id,
fake_project_id,
fake_repo_id,
fake_instance_id,
fake_tool_type_id,
) -> None:
"""When definition_type is 'manifest' and manifest_id is set, compiler runs."""
from src.models.tool_definition_manifest import ToolDefinitionManifest
manifest_id = uuid.uuid4()
mock_get_user.return_value = AsyncMock()
mock_get_project.return_value = AsyncMock()
mock_execute_compose.return_value = (0, "started", "")
mock_get_container_id.return_value = "abc123"
mock_connect_network.return_value = True
mock_wait_container.return_value = {
"success": True,
"status": "running",
"waited_seconds": 0.5,
}
mock_prepare_manifest.return_value = (
"headquarter/test:latest",
"services:\n app:\n image: test",
{"name": "test-manifest"},
"/root",
)
instance = ToolInstance(
id=fake_instance_id,
name="manifest-instance",
repository_id=fake_repo_id,
tool_type_id=fake_tool_type_id,
compose_path="/data/instances/manifest-instance/docker-compose.yml",
status="stopped",
clone_mode="mount",
created_at=datetime.now(),
updated_at=datetime.now(),
)
tool_type = ToolType(
id=fake_tool_type_id,
name="manifest-tool",
display_name="Manifest Tool",
default_port=8080,
definition_type="manifest",
manifest_id=manifest_id,
dockerfile_template=None,
compose_template=None,
)
repo = GitRepository(
id=fake_repo_id,
project_id=fake_project_id,
name="test-repo",
path="/data/repos/test-repo",
remote_url=None,
ssh_key_id=None,
)
manifest_def = ToolDefinitionManifest(
id=manifest_id,
name="test-manifest",
display_name="Test Manifest",
interface_type="web",
manifest={"base_image": "alpine"},
)
async def _get(model, pk):
if model is ToolInstance and pk == fake_instance_id:
return instance
if model is ToolType and pk == fake_tool_type_id:
return tool_type
if model is GitRepository and pk == fake_repo_id:
return repo
if model is User and pk == fake_user_id:
return User(id=fake_user_id, email="test@example.com")
if model is ToolDefinitionManifest and pk == manifest_id:
return manifest_def
return None
mock_session.get.side_effect = _get
with patch("os.path.exists", return_value=True):
result = await start_instance(
project_id=fake_project_id,
repo_id=fake_repo_id,
instance_id=fake_instance_id,
data=None,
user_id=fake_user_id,
session=mock_session,
)
assert result["status"] == "running"
mock_prepare_manifest.assert_called_once()
mock_execute_compose.assert_called_once()
Reference in New Issue View Git Blame Copy Permalink