headquarter/apps/api/tests/unit/test_instance_service.py

"""Unit tests for the tool instance service."""

from unittest.mock import MagicMock, AsyncMock

import pytest

from src.services.tool.instance_service import (
    _get_repository_mount_name,
    _stack_profile_mounts_with_git_mounts,
    modify_compose_file,
    prepare_manifest_instance,
)


@pytest.mark.unit
class TestModifyComposeFile:
    """Tests for modify_compose_file home-directory expansion."""

    def test_extra_volumes_expand_home_dir(self, tmp_path):
        compose_path = tmp_path / "docker-compose.yml"
        compose_path.write_text(
            "services:\n  app:\n    image: test:latest\n    volumes: []\n"
        )

        modify_compose_file(
            str(compose_path),
            extra_volumes=[
                {"source": "/host/config", "target": "~/.config", "type": "bind"},
                {"source": "/host/code", "target": "$HOME/code", "type": "bind"},
            ],
            home_dir="/home/user",
        )

        content = compose_path.read_text()
        assert "/host/config:/home/user/.config" in content
        assert "/host/code:/home/user/code" in content

    def test_working_directory_expands_home_dir(self, tmp_path):
        compose_path = tmp_path / "docker-compose.yml"
        compose_path.write_text("services:\n  app:\n    image: test:latest\n")

        modify_compose_file(
            str(compose_path),
            working_directory="~/workspace",
            home_dir="/home/user",
        )

        content = compose_path.read_text()
        assert "working_dir: /home/user/workspace" in content


@pytest.mark.unit
class TestGetRepositoryMountName:
    """Tests for _get_repository_mount_name."""

    def test_prefers_remote_url_name_over_user_provided_name(self):
        repo = MagicMock()
        repo.name = "src"
        repo.remote_url = "git@git.example.com:acme/headquarter.git"
        assert _get_repository_mount_name(repo) == "headquarter"

    def test_parses_browser_url_to_repo_name(self):
        repo = MagicMock()
        repo.name = "src"
        repo.remote_url = "https://github.com/acme/headquarter/tree/main"
        assert _get_repository_mount_name(repo) == "headquarter"

    def test_uses_workspace_path_basename_when_workspace_provided(self):
        repo = MagicMock()
        repo.name = "src"
        repo.remote_url = "git@git.example.com:acme/headquarter.git"
        workspace = MagicMock()
        workspace.path = "/data/working-copies/uuid/headquarter"
        assert _get_repository_mount_name(repo, workspace) == "headquarter"

    def test_falls_back_to_repo_name_when_remote_url_missing(self):
        repo = MagicMock()
        repo.name = "my-cool-repo"
        repo.remote_url = None
        assert _get_repository_mount_name(repo) == "my-cool-repo"

    def test_falls_back_to_repo_name_for_unparseable_url(self):
        repo = MagicMock()
        repo.name = "my-cool-repo"
        repo.remote_url = ""
        assert _get_repository_mount_name(repo) == "my-cool-repo"


@pytest.mark.unit
class TestStackProfileMountsWithGitMounts:
    """Tests for _stack_profile_mounts_with_git_mounts."""

    def test_exact_overlap_merges_profile_files_into_git_source(
        self, tmp_path
    ) -> None:
        """When a profile mount targets the same directory as a git mount,
        the profile files should be copied into the git-mount source so the
        container sees both sets of files through one bind mount."""
        git_source = tmp_path / "git" / "repo-clone"
        git_source.mkdir(parents=True)
        (git_source / "existing.txt").write_text("from git")

        profile_source = tmp_path / "profile" / "home_user_.pi"
        profile_source.mkdir(parents=True)
        (profile_source / "settings.json").write_text("{}")

        profile_mounts = [
            {
                "source": str(profile_source),
                "target": "/home/user/.pi",
                "type": "bind",
                "readonly": False,
            }
        ]
        git_mount_volumes = [
            {"source": str(git_source), "target": "/home/user/.pi", "type": "bind"}
        ]

        result = _stack_profile_mounts_with_git_mounts(
            profile_mounts, git_mount_volumes
        )

        assert result == []
        assert (git_source / "existing.txt").read_text() == "from git"
        assert (git_source / "settings.json").read_text() == "{}"

    def test_descendant_overlap_copies_into_subdirectory(self, tmp_path) -> None:
        """Profile mounts targeting a child directory are copied into the
        corresponding subdirectory of the git-mount source."""
        git_source = tmp_path / "git"
        git_source.mkdir()
        (git_source / "README").write_text("repo")

        profile_source = tmp_path / "profile" / "agent"
        profile_source.mkdir(parents=True)
        (profile_source / "settings.json").write_text("x")

        profile_mounts = [
            {
                "source": str(profile_source),
                "target": "/home/user/.pi/agent",
                "type": "bind",
            }
        ]
        git_mount_volumes = [
            {"source": str(git_source), "target": "/home/user/.pi", "type": "bind"}
        ]

        result = _stack_profile_mounts_with_git_mounts(
            profile_mounts, git_mount_volumes
        )

        assert result == []
        assert (git_source / "agent" / "settings.json").read_text() == "x"
        assert (git_source / "README").read_text() == "repo"

    def test_non_overlapping_mounts_left_untouched(self, tmp_path) -> None:
        """Profile mounts that do not overlap a git mount are returned as-is."""
        git_source = tmp_path / "git"
        git_source.mkdir()

        profile_source = tmp_path / "profile"
        profile_source.mkdir()
        (profile_source / "config").write_text("c")

        profile_mounts = [
            {
                "source": str(profile_source),
                "target": "/home/user/.config",
                "type": "bind",
            }
        ]
        git_mount_volumes = [
            {"source": str(git_source), "target": "/home/user/.pi", "type": "bind"}
        ]

        result = _stack_profile_mounts_with_git_mounts(
            profile_mounts, git_mount_volumes
        )

        assert result == profile_mounts

    def test_git_source_file_does_not_consume_profile_mount(self, tmp_path) -> None:
        """If the overlapping git-mount source is a file, the profile mount
        cannot be merged and must be kept."""
        git_source = tmp_path / "file.txt"
        git_source.write_text("file")

        profile_source = tmp_path / "profile"
        profile_source.mkdir()
        (profile_source / "settings.json").write_text("{}")

        profile_mounts = [
            {
                "source": str(profile_source),
                "target": "/home/user/.pi",
                "type": "bind",
            }
        ]
        git_mount_volumes = [
            {
                "source": str(git_source),
                "target": "/home/user/.pi/file.txt",
                "type": "bind",
            }
        ]

        result = _stack_profile_mounts_with_git_mounts(
            profile_mounts, git_mount_volumes
        )

        assert result == profile_mounts

    def test_profile_source_file_copied_into_git_source(self, tmp_path) -> None:
        """A profile mount that supplies a single file is copied into the
        git-mount source directory."""
        git_source = tmp_path / "git"
        git_source.mkdir()

        profile_source = tmp_path / "settings.json"
        profile_source.write_text("{}")

        profile_mounts = [
            {
                "source": str(profile_source),
                "target": "/home/user/.pi/settings.json",
                "type": "bind",
            }
        ]
        git_mount_volumes = [
            {"source": str(git_source), "target": "/home/user/.pi", "type": "bind"}
        ]

        result = _stack_profile_mounts_with_git_mounts(
            profile_mounts, git_mount_volumes
        )

        assert result == []
        assert (git_source / "settings.json").read_text() == "{}"


@pytest.mark.unit
async def test_prepare_manifest_instance_uses_workspace_path_basename():
    """WORKSPACE_NAME must be the repo-named workspace directory, not workspace.name."""
    repo = MagicMock()
    repo.name = "src"
    repo.remote_url = "git@git.example.com:acme/headquarter.git"
    repo.path = "/data/repos/main"

    workspace = MagicMock()
    workspace.id = "workspace-uuid"
    workspace.repo_id = "repo-uuid"
    workspace.name = "main"
    workspace.path = "/data/working-copies/workspace-uuid/headquarter"

    tool_type = MagicMock()
    tool_type.name = "pi-agent"
    tool_type.manifest_id = "manifest-uuid"

    manifest_def = MagicMock()
    manifest_def.id = "manifest-uuid"
    manifest_def.manifest = {
        "base_image": "ubuntu:24.04",
        "interface_type": "terminal",
        "user": {"name": "user", "uid": 1001, "gid": 1001},
    }
    manifest_def.base_definition_id = None

    instance = MagicMock()
    instance.id = "instance-uuid"
    instance.name = "pi-agent-headquarter-abc123"
    instance.repository_id = "repo-uuid"
    instance.tool_type_id = "tooltype-uuid"
    instance.port = 0
    instance.selected_config_profile_id = None
    instance.workspace_id = "workspace-uuid"

    session = AsyncMock()

    async def session_get(model, obj_id):
        if model.__name__ == "ToolType":
            return tool_type
        if model.__name__ == "ToolDefinitionManifest":
            return manifest_def
        if model.__name__ == "GitRepository":
            return repo
        if model.__name__ == "Workspace":
            return workspace
        return None

    session.get.side_effect = session_get

    # Patch docker images check to report the image already exists so we skip
    # the actual Docker build.
    import subprocess
    from src.services.tool import instance_service

    original_run = subprocess.run

    def fake_run(cmd, **kwargs):
        class Result:
            returncode = 0
            stdout = "image-id"
            stderr = ""

        return Result()

    instance_service.subprocess.run = fake_run
    try:
        (
            image_tag,
            compose_content,
            manifest,
            home_dir,
        ) = await prepare_manifest_instance(
            session=session,
            instance=instance,
            instance_dir="/tmp/instance",
            repo_path="/data/working-copies/workspace-uuid/headquarter",
            env_vars={},
            extra_volumes=[],
            working_directory=None,
        )
    finally:
        instance_service.subprocess.run = original_run

    assert "WORKSPACE_NAME: headquarter" in compose_content
    assert "/data/working-copies/workspace-uuid/headquarter:/home/user/headquarter" in compose_content