"""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.clone_mode = "mount" data.branch = None data.new_branch = None data.config_profile_id = None 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.clone_mode = "mount" data.branch = None data.new_branch = None data.config_profile_id = None 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.clone_mode = "mount" data.branch = None data.new_branch = None data.config_profile_id = None 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.clone_mode = "mount" data.branch = None data.new_branch = None data.config_profile_id = None 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.get_container_name") @patch("src.api.tool_instances.connect_container_to_network") @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_connect_network, mock_get_container_name, 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_get_container_name.return_value = "test-container" 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.get_container_name") @patch("src.api.tool_instances.connect_container_to_network") @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_connect_network, mock_get_container_name, 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_get_container_name.return_value = "test-container" 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.get_container_name") @patch("src.api.tool_instances.connect_container_to_network") @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_connect_network, mock_get_container_name, 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_get_container_name.return_value = "test-container" 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 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.get_container_name") @patch("src.api.tool_instances.connect_container_to_network") @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_connect_network, mock_get_container_name, 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_get_container_name.return_value = "test-container" 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()