smart-org/apps/api/tests/test_git_workspace.py

"""§v2-A3.2 — the platform's git adapter, exercised against real git.

These tests run a real ``git`` binary against a local bare repository standing in
for the remote. Nothing is faked except the *remote's location*: the flags, the
locking, the worktree dance and the credential plumbing are the real thing,
because those are precisely the parts that cannot be verified by a mock.

``protocol.file.allow=user`` is passed explicitly. The production default is
``never`` — a repository must not be able to make git read a local path through a
submodule or ``--upload-pack`` — but a local bare repo *is* a local path, so the
hardening has to be relaxed for the test to be able to run at all. That is why
the setting exists rather than being hard-coded.
"""

from __future__ import annotations

import asyncio
import os

import pytest

from app.core.errors import NotFoundError
from app.domain.platform.rules import RuleViolation, assert_remote_host_allowed, is_blocked_address
from app.infrastructure.adapters.git_workspace import (
    LocalGitWorkspaceAdapter,
    ProjectGitSource,
)
from app.infrastructure.adapters.process import ProcessResult, run_process

# ---------------------------------------------------------------------------
# helpers
# ---------------------------------------------------------------------------


async def _git(args: list[str], *, cwd: str, ok: bool = True) -> ProcessResult:
    result = await run_process(
        ["git", *args],
        cwd=cwd,
        timeout_seconds=60,
        env={
            "GIT_CONFIG_NOSYSTEM": "1",
            "GIT_TERMINAL_PROMPT": "0",
            "GIT_AUTHOR_NAME": "t",
            "GIT_AUTHOR_EMAIL": "t@example.com",
            "GIT_COMMITTER_NAME": "t",
            "GIT_COMMITTER_EMAIL": "t@example.com",
        },
    )
    if ok:
        assert result.ok, f"git {' '.join(args)} failed: {result.stderr}"
    return result


async def _make_remote(tmp_path) -> str:
    """A local bare repository with one commit on ``main``; returns its path."""
    remote = tmp_path / "remote.git"
    seed = tmp_path / "seed"
    await _git(["init", "--quiet", "--bare", str(remote)], cwd=str(tmp_path))
    await _git(["init", "--quiet", str(seed)], cwd=str(tmp_path))
    (seed / "README.md").write_text("hello\n")
    await _git(["add", "-A"], cwd=str(seed))
    await _git(["commit", "--quiet", "-m", "init"], cwd=str(seed))
    await _git(["branch", "-M", "main"], cwd=str(seed))
    await _git(["remote", "add", "origin", str(remote)], cwd=str(seed))
    await _git(["push", "--quiet", "origin", "main"], cwd=str(seed))
    return str(remote)


def _resolver(clone_url: str, *, auth_type: str = "none", secret: str | None = None,
              username: str | None = None):
    async def resolve(project_id: str) -> ProjectGitSource | None:
        if project_id == "missing":
            return None
        return ProjectGitSource(
            project_id=project_id,
            organization_id="org1",
            clone_url=clone_url,
            default_branch="main",
            auth_type=auth_type,
            username=username,
            secret=secret,
        )

    return resolve


def _adapter(tmp_path, clone_url: str, **kwargs) -> LocalGitWorkspaceAdapter:
    kwargs.setdefault("allow_file_protocol", True)
    kwargs.setdefault("runner", run_process)
    return LocalGitWorkspaceAdapter(
        resolver=_resolver(clone_url, **kwargs.pop("credentials", {})),
        root=str(tmp_path / "root"),
        **kwargs,
    )


# ---------------------------------------------------------------------------
# ensure_workspace
# ---------------------------------------------------------------------------


async def test_ensure_workspace_clones_and_creates_dev(tmp_path):
    remote = await _make_remote(tmp_path)
    adapter = _adapter(tmp_path, remote)

    ref = await adapter.ensure_workspace(project_id="proj1")

    assert ref.default_branch == "dev"
    assert os.path.isdir(ref.host_path), "the work tree must exist for the sandbox mount"
    assert (tmp_path / "root" / "org1" / "proj1" / "work" / "README.md").is_file()
    assert await adapter.head(project_id="proj1", branch="dev") is not None
    # `dev` was pushed to the platform remote because it did not exist there.
    assert await adapter.head(project_id="proj1", branch="dev") is not None


async def test_ensure_workspace_is_idempotent(tmp_path):
    remote = await _make_remote(tmp_path)
    adapter = _adapter(tmp_path, remote)

    first = await adapter.ensure_workspace(project_id="proj1")
    (tmp_path / "root" / "org1" / "proj1" / "work" / "new.txt").write_text("x")
    second = await adapter.ensure_workspace(project_id="proj1")

    assert first.host_path == second.host_path
    assert (tmp_path / "root" / "org1" / "proj1" / "work" / "new.txt").is_file()


async def test_ensure_workspace_rejects_an_unknown_project(tmp_path):
    adapter = _adapter(tmp_path, str(tmp_path / "nothing.git"))
    with pytest.raises(NotFoundError):
        await adapter.ensure_workspace(project_id="missing")


async def test_gitdir_is_not_inside_the_work_tree(tmp_path):
    """The git dir must never be reachable from the sandbox mount (D6)."""
    remote = await _make_remote(tmp_path)
    adapter = _adapter(tmp_path, remote)
    ref = await adapter.ensure_workspace(project_id="proj1")

    gitdir = tmp_path / "root" / "org1" / "proj1" / "git"
    assert gitdir.is_dir()
    assert not os.path.exists(os.path.join(ref.host_path, ".git"))
    assert not str(gitdir).startswith(ref.host_path + os.sep)


# ---------------------------------------------------------------------------
# worktrees
# ---------------------------------------------------------------------------


async def test_task_worktree_is_created_and_removed(tmp_path):
    remote = await _make_remote(tmp_path)
    adapter = _adapter(tmp_path, remote)
    await adapter.ensure_workspace(project_id="proj1")

    worktree = await adapter.create_task_worktree(project_id="proj1", task_id="t1")

    assert worktree.branch == "task/t1"
    assert worktree.sandbox_path == "/workspace/.worktrees/t1"
    assert os.path.isdir(worktree.host_path)

    await adapter.remove_worktree(worktree=worktree)
    assert not os.path.isdir(worktree.host_path)


async def test_task_worktree_is_invisible_in_the_main_work_tree(tmp_path):
    """`info/exclude` keeps .worktrees out of `status` without touching .gitignore."""
    remote = await _make_remote(tmp_path)
    adapter = _adapter(tmp_path, remote)
    ref = await adapter.ensure_workspace(project_id="proj1")
    await adapter.create_task_worktree(project_id="proj1", task_id="t1")

    result = await _git(
        ["--git-dir", str(tmp_path / "root" / "org1" / "proj1" / "git"),
         "--work-tree", ref.host_path,
         "status", "--porcelain=v1", "-uall"],
        cwd=ref.host_path,
    )
    assert result.stdout.strip() == "", "the worktree dir must not show up as untracked"
    assert not os.path.exists(os.path.join(ref.host_path, ".gitignore"))


async def test_create_task_worktree_is_idempotent(tmp_path):
    remote = await _make_remote(tmp_path)
    adapter = _adapter(tmp_path, remote)
    await adapter.ensure_workspace(project_id="proj1")

    first = await adapter.create_task_worktree(project_id="proj1", task_id="t1")
    second = await adapter.create_task_worktree(project_id="proj1", task_id="t1")
    assert first.host_path == second.host_path


# ---------------------------------------------------------------------------
# diff / commit
# ---------------------------------------------------------------------------


async def test_diff_reports_a_new_directory_thanks_to_uall(tmp_path):
    """`git status` without ``-uall`` hides new directories — a real agybot bug."""
    remote = await _make_remote(tmp_path)
    adapter = _adapter(tmp_path, remote)
    await adapter.ensure_workspace(project_id="proj1")
    worktree = await adapter.create_task_worktree(project_id="proj1", task_id="t1")

    (tmp_path / "root" / "org1" / "proj1" / "work" / ".worktrees" / "t1" / "src").mkdir()
    (
        tmp_path / "root" / "org1" / "proj1" / "work" / ".worktrees" / "t1" / "src" / "a.py"
    ).write_text("print('hi')\n")

    diff = await adapter.diff(worktree=worktree)

    assert diff.changed_files == ["src/a.py"]
    assert diff.is_empty is False


async def test_diff_detects_an_escaping_symlink(tmp_path):
    remote = await _make_remote(tmp_path)
    adapter = _adapter(tmp_path, remote)
    await adapter.ensure_workspace(project_id="proj1")
    worktree = await adapter.create_task_worktree(project_id="proj1", task_id="t1")

    outside = tmp_path / "outside.txt"
    outside.write_text("secret")
    os.symlink(outside, os.path.join(worktree.host_path, "escape.txt"))

    diff = await adapter.diff(worktree=worktree)

    assert diff.escaping_symlinks == ["escape.txt"]


async def test_commit_returns_none_when_there_is_nothing_to_commit(tmp_path):
    remote = await _make_remote(tmp_path)
    adapter = _adapter(tmp_path, remote)
    await adapter.ensure_workspace(project_id="proj1")
    worktree = await adapter.create_task_worktree(project_id="proj1", task_id="t1")

    assert await adapter.commit(worktree=worktree, message="empty", author="AI") is None


async def test_commit_records_the_author_and_returns_a_sha(tmp_path):
    remote = await _make_remote(tmp_path)
    adapter = _adapter(tmp_path, remote)
    await adapter.ensure_workspace(project_id="proj1")
    worktree = await adapter.create_task_worktree(project_id="proj1", task_id="t1")

    target = os.path.join(worktree.host_path, "feature.py")
    with open(target, "w", encoding="utf-8") as handle:
        handle.write("x = 1\n")

    sha = await adapter.commit(worktree=worktree, message="task/t1: add feature", author="Alice")

    assert sha and len(sha) == 40
    shared_gitdir = tmp_path / "root" / "org1" / "proj1" / "git"
    log = await _git(
        ["--git-dir", str(shared_gitdir), "log", "-1", "--format=%an <%ae>|%s", "task/t1"],
        cwd=str(tmp_path),
    )
    assert log.stdout.strip() == "Alice <alice@smart-org.local>|task/t1: add feature"

    # ...and specifically to `task/t1`, not to `dev`. A shared gitdir makes this
    # the easiest thing in the world to get wrong.
    task_head = await _git(
        ["--git-dir", str(shared_gitdir), "rev-parse", "task/t1"], cwd=str(tmp_path)
    )
    assert task_head.stdout.strip() == sha
    dev_head = await _git(
        ["--git-dir", str(shared_gitdir), "rev-parse", "dev"], cwd=str(tmp_path)
    )
    assert dev_head.stdout.strip() != sha, "the commit must not land on dev directly"


# ---------------------------------------------------------------------------
# merge / push / head
# ---------------------------------------------------------------------------


async def test_merge_into_dev_and_push(tmp_path):
    remote = await _make_remote(tmp_path)
    adapter = _adapter(tmp_path, remote)
    await adapter.ensure_workspace(project_id="proj1")
    worktree = await adapter.create_task_worktree(project_id="proj1", task_id="t1")

    with open(os.path.join(worktree.host_path, "a.txt"), "w", encoding="utf-8") as handle:
        handle.write("a\n")
    sha = await adapter.commit(worktree=worktree, message="task/t1: a", author="AI")

    merged = await adapter.merge_into(project_id="proj1", branch="task/t1", target="dev")
    assert merged.ok is True, merged.message

    # `--no-ff` means dev gets a merge *commit*, not the task commit itself, so
    # the task sha is a parent of dev's new head rather than dev's head.
    dev_head = await adapter.head(project_id="proj1", branch="dev")
    assert dev_head == merged.sha
    assert dev_head != sha
    ancestry = await _git(
        ["--git-dir", str(tmp_path / "root" / "org1" / "proj1" / "git"),
         "rev-list", "--parents", "-1", "dev"],
        cwd=str(tmp_path),
    )
    assert sha in ancestry.stdout.split()

    pushed = await adapter.push(project_id="proj1", branch="dev")
    assert pushed.ok is True, pushed.message

    # The remote really received it.
    ls = await _git(["--git-dir", remote, "rev-parse", "dev"], cwd=str(tmp_path))
    assert ls.stdout.strip() == dev_head


async def test_merge_conflict_leaves_a_clean_tree(tmp_path):
    """A half-merge would break every later operation (agybot bug H3).

    The task branch has to fork *before* ``dev`` moves, otherwise the merge is a
    trivial fast-forward and there is nothing to conflict about.
    """
    remote = await _make_remote(tmp_path)
    adapter = _adapter(tmp_path, remote)
    await adapter.ensure_workspace(project_id="proj1")

    # 1. The task branches off dev as it stands.
    worktree = await adapter.create_task_worktree(project_id="proj1", task_id="t1")
    with open(os.path.join(worktree.host_path, "README.md"), "w", encoding="utf-8") as handle:
        handle.write("task version\n")
    await adapter.commit(worktree=worktree, message="task/t1: edit", author="AI")

    # 2. dev moves on, touching the same file.
    main_work = tmp_path / "root" / "org1" / "proj1" / "work"
    (main_work / "README.md").write_text("dev version\n")
    await _git(
        ["--git-dir", str(tmp_path / "root" / "org1" / "proj1" / "git"),
         "--work-tree", str(main_work), "add", "-A"],
        cwd=str(main_work),
    )
    await _git(
        ["--git-dir", str(tmp_path / "root" / "org1" / "proj1" / "git"),
         "--work-tree", str(main_work), "commit", "--quiet", "-m", "dev edit"],
        cwd=str(main_work),
    )

    outcome = await adapter.merge_into(project_id="proj1", branch="task/t1", target="dev")

    assert outcome.ok is False
    assert outcome.conflict_files == ["README.md"]

    # No merge in progress: the next status is clean.
    status = await _git(
        ["--git-dir", str(tmp_path / "root" / "org1" / "proj1" / "git"),
         "--work-tree", str(main_work), "status", "--porcelain=v1", "-uall"],
        cwd=str(main_work),
    )
    assert status.stdout.strip() == ""
    assert not os.path.exists(
        tmp_path / "root" / "org1" / "proj1" / "git" / "MERGE_HEAD"
    )


async def test_sync_with_base_reports_that_the_base_moved(tmp_path):
    remote = await _make_remote(tmp_path)
    adapter = _adapter(tmp_path, remote)
    await adapter.ensure_workspace(project_id="proj1")
    worktree = await adapter.create_task_worktree(project_id="proj1", task_id="t1")

    # Nothing moved yet.
    quiet = await adapter.sync_with_base(worktree=worktree)
    assert quiet.ok is True and quiet.base_changed is False

    # Move dev forward.
    main_work = tmp_path / "root" / "org1" / "proj1" / "work"
    (main_work / "other.txt").write_text("x\n")
    await _git(
        ["--git-dir", str(tmp_path / "root" / "org1" / "proj1" / "git"),
         "--work-tree", str(main_work), "add", "-A"],
        cwd=str(main_work),
    )
    await _git(
        ["--git-dir", str(tmp_path / "root" / "org1" / "proj1" / "git"),
         "--work-tree", str(main_work), "commit", "--quiet", "-m", "dev moved"],
        cwd=str(main_work),
    )

    moved = await adapter.sync_with_base(worktree=worktree)
    assert moved.ok is True
    assert moved.base_changed is True
    assert os.path.isfile(os.path.join(worktree.host_path, "other.txt"))


async def test_concurrent_merges_are_serialised(tmp_path):
    """Without the per-project lock two merges corrupt the index."""
    remote = await _make_remote(tmp_path)
    adapter = _adapter(tmp_path, remote)
    await adapter.ensure_workspace(project_id="proj1")

    for task in ("t1", "t2"):
        worktree = await adapter.create_task_worktree(project_id="proj1", task_id=task)
        with open(
            os.path.join(worktree.host_path, f"{task}.txt"), "w", encoding="utf-8"
        ) as handle:
            handle.write(f"{task}\n")
        await adapter.commit(worktree=worktree, message=f"task/{task}", author="AI")

    first, second = await asyncio.gather(
        adapter.merge_into(project_id="proj1", branch="task/t1", target="dev"),
        adapter.merge_into(project_id="proj1", branch="task/t2", target="dev"),
    )

    assert first.ok is True and second.ok is True
    status = await _git(
        ["--git-dir", str(tmp_path / "root" / "org1" / "proj1" / "git"),
         "--work-tree", str(tmp_path / "root" / "org1" / "proj1" / "work"),
         "status", "--porcelain=v1", "-uall"],
        cwd=str(tmp_path),
    )
    assert status.stdout.strip() == ""


async def test_push_reports_a_rejection_without_forcing(tmp_path):
    remote = await _make_remote(tmp_path)
    adapter = _adapter(tmp_path, remote)
    await adapter.ensure_workspace(project_id="proj1")

    # Someone else advances the remote's dev, so our push is no longer a
    # fast-forward. `origin/dev` is what the clone checked out, not a fresh
    # branch off the default one.
    other = tmp_path / "other"
    await _git(["clone", "--quiet", remote, str(other)], cwd=str(tmp_path))
    await _git(["checkout", "--quiet", "-B", "dev", "origin/dev"], cwd=str(other))
    (other / "divergent.txt").write_text("other\n")
    await _git(["add", "-A"], cwd=str(other))
    await _git(["commit", "--quiet", "-m", "divergent"], cwd=str(other))
    await _git(["push", "--quiet", "origin", "dev"], cwd=str(other))

    # Local dev moves a different way.
    main_work = tmp_path / "root" / "org1" / "proj1" / "work"
    (main_work / "mine.txt").write_text("mine\n")
    await _git(
        ["--git-dir", str(tmp_path / "root" / "org1" / "proj1" / "git"),
         "--work-tree", str(main_work), "add", "-A"],
        cwd=str(main_work),
    )
    await _git(
        ["--git-dir", str(tmp_path / "root" / "org1" / "proj1" / "git"),
         "--work-tree", str(main_work), "commit", "--quiet", "-m", "mine"],
        cwd=str(main_work),
    )

    outcome = await adapter.push(project_id="proj1", branch="dev")

    assert outcome.ok is False
    assert outcome.rejected is True, outcome.message


# ---------------------------------------------------------------------------
# hardening
# ---------------------------------------------------------------------------


async def test_hooks_are_never_executed(tmp_path):
    """A repository may ship hooks; a hook is arbitrary host code execution."""
    remote = await _make_remote(tmp_path)
    adapter = _adapter(tmp_path, remote)
    await adapter.ensure_workspace(project_id="proj1")

    gitdir = tmp_path / "root" / "org1" / "proj1" / "git"
    hooks = gitdir / "hooks"
    hooks.mkdir(parents=True, exist_ok=True)
    marker = tmp_path / "HOOK_RAN"
    hook = hooks / "post-commit"
    hook.write_text(f"#!/bin/sh\ntouch {marker}\n")
    os.chmod(hook, 0o755)

    worktree = await adapter.create_task_worktree(project_id="proj1", task_id="t1")
    with open(os.path.join(worktree.host_path, "a.txt"), "w", encoding="utf-8") as handle:
        handle.write("a\n")
    await adapter.commit(worktree=worktree, message="task/t1", author="AI")

    assert not marker.exists(), "a post-commit hook must not run"


async def test_a_tampered_dot_git_file_is_ignored(tmp_path):
    """The `.git` pointer inside a worktree is attacker-controlled text."""
    remote = await _make_remote(tmp_path)
    adapter = _adapter(tmp_path, remote)
    await adapter.ensure_workspace(project_id="proj1")
    worktree = await adapter.create_task_worktree(project_id="proj1", task_id="t1")

    decoy = tmp_path / "decoy"
    await _git(["init", "--quiet", str(decoy)], cwd=str(tmp_path))
    with open(os.path.join(worktree.host_path, ".git"), "w", encoding="utf-8") as handle:
        handle.write(f"gitdir: {decoy}\n")

    with open(os.path.join(worktree.host_path, "a.txt"), "w", encoding="utf-8") as handle:
        handle.write("a\n")
    sha = await adapter.commit(worktree=worktree, message="task/t1", author="AI")

    assert sha is not None
    # The commit landed on OUR task branch, not in the decoy.
    shared_gitdir = tmp_path / "root" / "org1" / "proj1" / "git"
    real = await _git(
        ["--git-dir", str(shared_gitdir), "rev-parse", "task/t1"], cwd=str(tmp_path)
    )
    assert real.stdout.strip() == sha
    decoy_log = await _git(
        ["--git-dir", str(decoy), "log", "--oneline"], cwd=str(tmp_path), ok=False
    )
    assert decoy_log.exit_code != 0 or "task/t1" not in decoy_log.stdout


async def test_credentials_never_appear_in_argv(tmp_path):
    """argv is world-readable via `ps`; a secret must travel another way."""
    secret = "ghp_thisisasecrettokenvalue0123456789"
    seen: list[tuple[str, ...]] = []

    async def recording_runner(argv, **kwargs):  # noqa: ANN001, ANN003
        seen.append(tuple(argv))
        return await run_process(argv, **kwargs)

    remote = await _make_remote(tmp_path)
    adapter = LocalGitWorkspaceAdapter(
        resolver=_resolver(remote, auth_type="token", secret=secret, username="git"),
        root=str(tmp_path / "root"),
        allow_file_protocol=True,
        runner=recording_runner,
    )

    # A token against a local path fails, but the *attempt* is what we assert on.
    await adapter.ensure_workspace(project_id="proj1")

    assert seen, "the adapter must have invoked git"
    for argv in seen:
        joined = " ".join(argv)
        assert secret not in joined, f"secret leaked into argv: {argv}"


async def test_the_credential_file_is_removed_after_the_call(tmp_path):
    remote = await _make_remote(tmp_path)
    adapter = LocalGitWorkspaceAdapter(
        resolver=_resolver(remote, auth_type="token", secret="s3cret-token-value", username="git"),
        root=str(tmp_path / "root"),
        allow_file_protocol=True,
    )
    await adapter.ensure_workspace(project_id="proj1")

    private = tmp_path / "root" / "org1" / "proj1" / "private"
    assert private.is_dir()
    assert not (private / "credentials").exists(), "the 0600 secret file must be deleted"
    assert not (private / "id_key").exists()


async def test_the_askpass_helper_is_private_and_holds_no_secret(tmp_path):
    remote = await _make_remote(tmp_path)
    adapter = LocalGitWorkspaceAdapter(
        resolver=_resolver(remote, auth_type="token", secret="s3cret-token-value", username="git"),
        root=str(tmp_path / "root"),
        allow_file_protocol=True,
    )
    await adapter.ensure_workspace(project_id="proj1")

    helper = tmp_path / "root" / "org1" / "proj1" / "private" / "askpass.sh"
    assert helper.is_file()
    assert oct(helper.stat().st_mode)[-3:] == "700"
    assert "s3cret-token-value" not in helper.read_text()


# ---------------------------------------------------------------------------
# SSRF: the shared host check
# ---------------------------------------------------------------------------


@pytest.mark.parametrize(
    "address",
    [
        "127.0.0.1",
        "10.0.0.1",
        "172.16.0.1",
        "192.168.1.1",
        "169.254.169.254",
        "::1",
        "fd00::1",
        "fe80::1",
        "0.0.0.0",
    ],
)
def test_is_blocked_address_covers_the_dangerous_ranges(address: str):
    assert is_blocked_address(address) is True


@pytest.mark.parametrize("address", ["8.8.8.8", "1.1.1.1", "2606:4700::1111"])
def test_is_blocked_address_allows_public_addresses(address: str):
    assert is_blocked_address(address) is False


@pytest.mark.parametrize(
    "url",
    [
        "http://169.254.169.254/latest/meta-data/",
        "http://127.0.0.1:6379/",
        "http://localhost/repo.git",
        "https://10.0.0.5/repo.git",
        "http://[::1]/repo.git",
        "https://user:pass@example.com/repo.git",
        "file:///etc/passwd",
        "https://example.com:6379/repo.git",
    ],
)
def test_assert_remote_host_allowed_refuses_dangerous_targets(url: str):
    with pytest.raises(RuleViolation):
        assert_remote_host_allowed(url, resolver=lambda *a, **k: [])


def test_assert_remote_host_allowed_accepts_a_public_host():
    def resolver(host, port, **kwargs):  # noqa: ANN001, ANN003
        return [(2, 1, 6, "", ("93.184.216.34", port))]

    host, port = assert_remote_host_allowed(
        "https://git.example.com/team/app.git", resolver=resolver
    )
    assert host == "git.example.com"
    assert port is None


def test_assert_remote_host_allowed_refuses_a_name_that_resolves_private():
    """A public-looking name can resolve into the private network."""

    def resolver(host, port, **kwargs):  # noqa: ANN001, ANN003
        return [(2, 1, 6, "", ("10.1.2.3", port))]

    with pytest.raises(RuleViolation):
        assert_remote_host_allowed("https://internal.example.com/repo.git", resolver=resolver)


def test_assert_remote_host_allowed_can_be_told_to_allow_private():
    host, _ = assert_remote_host_allowed(
        "http://127.0.0.1:8080/repo.git", allow_private=True
    )
    assert host == "127.0.0.1"


def test_assert_remote_host_allowed_accepts_scp_style_ssh():
    def resolver(host, port, **kwargs):  # noqa: ANN001, ANN003
        return [(2, 1, 6, "", ("93.184.216.34", port))]

    host, _ = assert_remote_host_allowed("git@git.example.com:team/app.git", resolver=resolver)
    assert host == "git.example.com"


async def test_test_remote_refuses_a_private_host(tmp_path):
    adapter = _adapter(tmp_path, "https://example.com/x.git")
    from app.application.ports.repo_workspace import RemoteCredentials

    check = await adapter.test_remote(
        url="http://169.254.169.254/repo.git", credentials=RemoteCredentials()
    )
    assert check.ok is False
    assert "داخلی" in check.message


async def test_test_remote_lists_branches_of_a_local_repo(tmp_path, monkeypatch):
    """The ls-remote path, with the SSRF check stubbed out.

    The check itself is covered by the tests above; here the subject is the
    parsing of `git ls-remote --heads` and the branch list it produces. A local
    bare repo is used as the remote because the suite must not need a network.
    """
    import app.infrastructure.adapters.git_workspace as module
    from app.application.ports.repo_workspace import RemoteCredentials

    monkeypatch.setattr(module, "assert_remote_host_allowed", lambda *a, **k: ("local", None))

    remote = await _make_remote(tmp_path)
    adapter = _adapter(tmp_path, remote)

    check = await adapter.test_remote(
        url=f"file://{remote}", credentials=RemoteCredentials()
    )

    assert check.ok is True, check.message
    assert check.branches == ["main"]
    assert check.default_branch == "main"


async def test_test_remote_reports_a_missing_repository(tmp_path, monkeypatch):
    import app.infrastructure.adapters.git_workspace as module
    from app.application.ports.repo_workspace import RemoteCredentials

    monkeypatch.setattr(module, "assert_remote_host_allowed", lambda *a, **k: ("local", None))

    adapter = _adapter(tmp_path, "https://example.com/x.git")
    check = await adapter.test_remote(
        url=f"file://{tmp_path}/does-not-exist.git", credentials=RemoteCredentials()
    )

    assert check.ok is False
    assert check.message