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

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

import hashlib
import uuid
from pathlib import Path
from unittest.mock import MagicMock, AsyncMock

import pytest

from src.services.tool.instance_service import (
    _chown_staged_mounts,
    _get_repository_mount_name,
    _stack_profile_mounts_with_git_mounts,
    _stage_profile_mounts,
    clone_git_repo,
    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_uses_project_name(self):
        project = MagicMock()
        project.name = "My Project"
        repo = MagicMock()
        assert _get_repository_mount_name(project, repo) == "my-project"

    def test_slugifies_project_name(self):
        project = MagicMock()
        project.name = "Project v2.0!"
        repo = MagicMock()
        assert _get_repository_mount_name(project, repo) == "project-v2-0"


@pytest.mark.unit
class TestCloneGitRepo:
    """Regression tests for reusable config-profile git mount clones."""

    def test_reuses_existing_clone(self, monkeypatch, tmp_path) -> None:
        from src.services.tool import instance_service

        remote_url = "https://gitlab.com/example/dotfiles"
        branch = None
        clone_parent = str(tmp_path)
        url_hash = hashlib.md5(f"{remote_url}:default".encode()).hexdigest()[:12]
        repo_path = tmp_path / "git-mounts" / f"dotfiles-{url_hash}" / "repo-clone"
        (repo_path / ".git").mkdir(parents=True)

        clone = MagicMock(side_effect=AssertionError("existing clone must be reused"))
        pull = MagicMock()
        monkeypatch.setattr(instance_service, "clone_repository", clone)
        monkeypatch.setattr(instance_service, "pull_repository_updates", pull)

        result = clone_git_repo(remote_url, branch, clone_parent)

        assert result == str(repo_path)
        clone.assert_not_called()
        pull.assert_called_once_with(str(repo_path), remote_url)

    def test_replaces_incomplete_clone_before_retry(
        self, monkeypatch, tmp_path
    ) -> None:
        from src.services.tool import instance_service

        remote_url = "https://gitlab.com/example/dotfiles"
        url_hash = hashlib.md5(f"{remote_url}:main".encode()).hexdigest()[:12]
        clone_dir = tmp_path / "git-mounts" / f"dotfiles-{url_hash}"
        repo_path = clone_dir / "repo-clone"
        repo_path.mkdir(parents=True)
        (repo_path / "partial-file").write_text("incomplete")

        def clone(_url, _key, destination, _branch, project_name=None):
            assert destination == str(clone_dir)
            assert project_name is None
            assert not repo_path.exists()
            (repo_path / ".git").mkdir(parents=True)
            return str(repo_path)

        monkeypatch.setattr(instance_service, "clone_repository", clone)

        result = clone_git_repo(remote_url, "main", str(tmp_path))

        assert result == str(repo_path)
        assert (repo_path / ".git").is_dir()


@pytest.mark.unit
class TestStackProfileMountsWithGitMounts:
    """Tests for composing profile and Git mounts without Docker masking."""

    def test_stages_profile_source_under_instance_directory(self, tmp_path) -> None:
        """Profile sources must be instance-local before ownership is fixed."""
        instance_dir = tmp_path / "instance"
        profile_source = tmp_path / "profile" / "settings.json"
        profile_source.parent.mkdir(parents=True)
        profile_source.write_text("{}")

        staged = _stage_profile_mounts(
            [{"source": str(profile_source), "target": "/home/user/.pi/settings.json"}],
            str(instance_dir),
        )

        assert staged[0]["source"].startswith(str(instance_dir))
        assert staged[0]["source"] != str(profile_source)
        assert Path(staged[0]["source"]).read_text() == "{}"

    def test_staged_profile_source_is_chowned_for_container_user(
        self, monkeypatch, tmp_path
    ) -> None:
        """The ownership pass must include the instance-local profile copy."""
        from src.services.tool import instance_service

        instance_dir = tmp_path / "instance"
        profile_source = tmp_path / "profile"
        profile_source.mkdir()
        (profile_source / "settings.json").write_text("{}")
        staged = _stage_profile_mounts(
            [{"source": str(profile_source), "target": "/home/user/.pi"}],
            str(instance_dir),
        )
        chown = MagicMock()
        monkeypatch.setattr(instance_service, "_chown_path", chown)

        _chown_staged_mounts(staged, str(instance_dir), 1000, 1000)

        chown.assert_called_once_with(staged[0]["source"], 1000, 1000)

    def test_exact_overlap_creates_instance_local_composite(self, tmp_path) -> None:
        """Profile files extend a Git root without mutating its shared clone."""
        instance_dir = tmp_path / "instance"
        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"
        profile_source.mkdir()
        (profile_source / "settings.json").write_text("{}")

        profile_mounts = _stage_profile_mounts(
            [
                {
                    "source": str(profile_source),
                    "target": "/home/user/.pi",
                    "type": "bind",
                }
            ],
            str(instance_dir),
        )
        result = _stack_profile_mounts_with_git_mounts(
            profile_mounts,
            [{"source": str(git_source), "target": "/home/user/.pi", "type": "bind"}],
            str(instance_dir),
        )

        assert len(result) == 1
        composite = result[0]
        assert composite["target"] == "/home/user/.pi"
        assert composite["source"].startswith(str(instance_dir))
        assert not (tmp_path / "git" / "repo-clone" / "settings.json").exists()
        assert (
            tmp_path / "git" / "repo-clone" / "existing.txt"
        ).read_text() == "from git"
        assert (tmp_path / "instance" / "mounts" / "composites").exists()
        assert (Path(composite["source"]) / "existing.txt").read_text() == "from git"
        assert (Path(composite["source"]) / "settings.json").read_text() == "{}"

    def test_descendant_profile_mount_extends_git_root(self, tmp_path) -> None:
        """Nested targets are composed at the Git root, preserving siblings."""
        instance_dir = tmp_path / "instance"
        git_source = tmp_path / "git"
        git_source.mkdir()
        (git_source / "README").write_text("repo")
        profile_source = tmp_path / "profile"
        profile_source.mkdir()
        (profile_source / "settings.json").write_text("x")

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

        assert len(result) == 1
        composite = Path(result[0]["source"])
        assert result[0]["target"] == "/home/user/.pi"
        assert (composite / "README").read_text() == "repo"
        assert (composite / "agent" / "settings.json").read_text() == "x"
        assert not (git_source / "agent").exists()

    def test_file_profile_mount_extends_git_root(self, tmp_path) -> None:
        """A file bind mount is composed into the Git directory, not masked."""
        instance_dir = tmp_path / "instance"
        git_source = tmp_path / "git"
        git_source.mkdir()
        (git_source / "README").write_text("repo")
        profile_source = tmp_path / "settings.json"
        profile_source.write_text("{}")

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

        assert len(result) == 1
        composite = Path(result[0]["source"])
        assert result[0]["target"] == "/home/user/.pi"
        assert (composite / "README").read_text() == "repo"
        assert (composite / "settings.json").read_text() == "{}"

    def test_parent_profile_mount_extends_nested_git_mount(self, tmp_path) -> None:
        """A profile parent mount keeps Git content and its own sibling files."""
        instance_dir = tmp_path / "instance"
        git_source = tmp_path / "git"
        git_source.mkdir()
        (git_source / "plugin.toml").write_text("git")
        profile_source = tmp_path / "profile"
        profile_source.mkdir()
        (profile_source / "config.toml").write_text("profile")

        profile_mounts = _stage_profile_mounts(
            [{"source": str(profile_source), "target": "/home/user"}], str(instance_dir)
        )
        result = _stack_profile_mounts_with_git_mounts(
            profile_mounts,
            [{"source": str(git_source), "target": "/home/user/.pi", "type": "bind"}],
            str(instance_dir),
        )

        assert len(result) == 1
        composite = Path(result[0]["source"])
        assert result[0]["target"] == "/home/user"
        assert (composite / "config.toml").read_text() == "profile"
        assert (composite / ".pi" / "plugin.toml").read_text() == "git"

    def test_non_overlapping_mounts_remain_separate(self, tmp_path) -> None:
        """Unrelated profile and Git mounts retain their independent sources."""
        instance_dir = tmp_path / "instance"
        git_source = tmp_path / "git"
        git_source.mkdir()
        profile_source = tmp_path / "profile"
        profile_source.mkdir()

        profile_mounts = _stage_profile_mounts(
            [{"source": str(profile_source), "target": "/home/user/.config"}],
            str(instance_dir),
        )
        git_mounts = [{"source": str(git_source), "target": "/home/user/.pi"}]

        assert (
            _stack_profile_mounts_with_git_mounts(
                profile_mounts, git_mounts, str(instance_dir)
            )
            == profile_mounts + git_mounts
        )


@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()

    project = MagicMock()
    project.name = "acme"

    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
        if model.__name__ == "Project":
            return project
        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: acme" in compose_content
    assert (
        "/data/working-copies/workspace-uuid/headquarter:/home/user/acme"
        in compose_content
    )


@pytest.mark.unit
async def test_create_tool_instance_resolves_project_when_display_name_given(
    monkeypatch, tmp_path
):
    """Regression: create_tool_instance must resolve `project` even when a
    display_name is supplied.

    Previously `project` was only fetched inside the auto-display-name branch,
    so providing display_name left it unbound and the manifest/dockerfile/compose
    branches crashed with UnboundLocalError at `_get_repository_mount_name`.
    The mount name must derive from the project name regardless.
    """
    from src.schemas.tool import CreateInstanceRequest
    from src.services.tool import instance_service

    project_id = uuid.uuid4()
    repo_id = uuid.uuid4()
    user_id = uuid.uuid4()
    tool_type_id = uuid.uuid4()
    manifest_id = uuid.uuid4()

    repo = MagicMock()
    repo.project_id = project_id
    repo.name = "src"
    repo.path = "/data/repos/src"

    project = MagicMock()
    project.name = "My Project"  # slugifies to "my-project"

    tool_type = MagicMock()
    tool_type.id = tool_type_id
    tool_type.name = "pi-agent"
    tool_type.definition_type = "manifest"
    tool_type.manifest_id = manifest_id
    tool_type.home_directory = None

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

    async def session_get(model, _obj_id):
        name = getattr(model, "__name__", "")
        if name == "GitRepository":
            return repo
        if name == "Project":
            return project
        if name == "ToolType":
            return tool_type
        if name == "ToolDefinitionManifest":
            return manifest_def
        return None

    session = AsyncMock()
    session.get.side_effect = session_get
    session.add = MagicMock()

    monkeypatch.setattr(
        instance_service, "validate_config_profile", AsyncMock(return_value=None)
    )
    monkeypatch.setattr(
        instance_service, "ensure_instance_directory", lambda _name: str(tmp_path)
    )
    monkeypatch.setattr(instance_service, "find_free_port", lambda: 12345)
    monkeypatch.setattr(
        instance_service, "compute_image_tag", lambda *a, **k: "img:tag"
    )
    compile_compose = MagicMock(return_value="services:\n  app: {}")
    monkeypatch.setattr(instance_service, "compile_compose", compile_compose)
    monkeypatch.setattr(instance_service, "write_compose_file", MagicMock())
    monkeypatch.setattr(instance_service, "publish_lifecycle_event", AsyncMock())

    data = CreateInstanceRequest(
        tool_type_id=str(tool_type_id), display_name="Custom Name"
    )

    instance = await instance_service.create_tool_instance(
        session, user_id, project_id, repo_id, data
    )

    # Must not raise UnboundLocalError; instance returned.
    assert instance is not None

    # The mount name must derive from project.name, proving `project` was
    # resolved on the display_name path (the previous crash site).
    variables = compile_compose.call_args.args[1]
    assert variables["REPO_NAME"] == "my-project"
    assert variables["WORKSPACE_NAME"] == "my-project"