AegisGitea-MCP/tests/integration/test_harness.py

"""Smoke tests of the integration harness itself (needs Docker, opt-in)."""

from __future__ import annotations

from pathlib import Path

import pytest

from tests.integration import harness
from tests.integration.conftest import RegularUser
from tests.integration.harness import GiteaInstance, HarnessError, RunResources


def _names(kind: str, run_id: str) -> str:
    listing = {
        "container": ["ps", "-a", "--filter", f"name={harness.NAME_PREFIX}{run_id}"],
        "network": ["network", "ls", "--filter", f"name={harness.NAME_PREFIX}{run_id}"],
    }[kind]
    fmt = "{{.Names}}" if kind == "container" else "{{.Name}}"
    return harness.run_docker(*listing, "--format", fmt).stdout.strip()


def test_image_is_digest_pinned() -> None:
    assert "@sha256:" in harness.GITEA_IMAGE
    assert harness.GITEA_IMAGE.startswith("gitea/gitea:28.1.0@")


def test_version_endpoint_reachable(gitea: GiteaInstance) -> None:
    response = gitea.api("GET", "/version", anonymous=True)
    assert response.status_code == 200
    assert response.json()["version"].startswith("28.1.0")


def test_admin_token_is_admin(gitea: GiteaInstance) -> None:
    response = gitea.api("GET", "/user")
    assert response.status_code == 200
    body = response.json()
    assert body["login"] == gitea.admin_login
    assert body["is_admin"] is True


def test_admin_token_has_all_scope(gitea: GiteaInstance) -> None:
    response = gitea.api("GET", "/token")
    assert response.status_code == 200
    assert "all" in response.json()["scopes"]


def test_regular_user_is_not_admin(gitea: GiteaInstance, regular_user: RegularUser) -> None:
    me = gitea.api("GET", "/user", token=regular_user.token)
    assert me.status_code == 200
    assert me.json()["login"] == regular_user.login
    assert me.json()["is_admin"] is False
    forbidden = gitea.api("GET", "/admin/users", token=regular_user.token)
    assert forbidden.status_code == 403


def test_anonymous_call_is_rejected(gitea: GiteaInstance) -> None:
    assert gitea.api("GET", "/user", anonymous=True).status_code == 401


def test_create_user_can_be_admin(gitea: GiteaInstance, unique_name: str) -> None:
    login, password = gitea.create_user(unique_name, admin=True)
    token = gitea.mint_token(login, password, ["read:user"])
    assert gitea.api("GET", "/user", token=token).json()["is_admin"] is True


def test_create_org_and_repos(gitea: GiteaInstance, unique_name: str) -> None:
    org = gitea.create_org(unique_name)
    assert org["name"] == unique_name

    org_repo = gitea.create_repo(unique_name, "org-repo", org=True)
    assert org_repo["full_name"] == f"{unique_name}/org-repo"
    assert org_repo["empty"] is False

    user_repo = gitea.create_repo(gitea.admin_login, f"{unique_name}-u", auto_init=False)
    assert user_repo["owner"]["login"] == gitea.admin_login
    assert user_repo["empty"] is True

    fetched = gitea.api("GET", f"/repos/{unique_name}/org-repo")
    assert fetched.status_code == 200


def test_unique_names_differ(gitea: GiteaInstance) -> None:
    assert gitea.unique("x") != gitea.unique("x")
    assert gitea.unique("x").startswith("x-")


def test_secrets_never_in_repr_or_errors(
    gitea: GiteaInstance,
    regular_user: RegularUser,
    capfd: pytest.CaptureFixture[str],
) -> None:
    secrets_ = [
        gitea.admin_token,
        gitea.admin_password,
        regular_user.token,
        regular_user.password,
    ]
    rendered = repr(gitea) + repr(regular_user) + str(gitea) + str(regular_user)
    with pytest.raises(HarnessError) as excinfo:
        gitea.mint_token(regular_user.login, "definitely-wrong-password", ["all"])
    gitea.create_org(gitea.unique("quiet"))
    out, err = capfd.readouterr()
    for secret in secrets_:
        assert secret not in rendered
        assert secret not in str(excinfo.value)
        assert secret not in out
        assert secret not in err


def test_cleanup_removes_container_and_network() -> None:
    run_id = harness.secrets.token_hex(4)
    name = f"{harness.NAME_PREFIX}{run_id}"
    res = RunResources(run_id=run_id, container=name, network=name)
    harness.register_resources(res)
    harness.run_docker("network", "create", name)
    harness.run_docker("create", "--name", name, "--network", name, harness.GITEA_IMAGE)
    assert _names("container", run_id) == name
    assert _names("network", run_id) == name

    harness.cleanup_resources(res)

    assert _names("container", run_id) == ""
    assert _names("network", run_id) == ""
    # Idempotent, and no longer tracked for atexit.
    harness.cleanup_resources(res)
    assert run_id not in harness._ACTIVE


def test_cleanup_all_handles_registered_runs() -> None:
    run_id = harness.secrets.token_hex(4)
    name = f"{harness.NAME_PREFIX}{run_id}"
    harness.register_resources(RunResources(run_id=run_id, container=name, network=name))
    harness.run_docker("network", "create", name)
    harness.cleanup_all()
    assert _names("network", run_id) == ""


def test_missing_docker_cli_gives_clear_error(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("PATH", "")
    with pytest.raises(HarnessError, match="Docker is unavailable"):
        harness.require_docker()


def test_container_id_candidates_fall_back_to_proc(tmp_path: Path) -> None:
    cid = "ab" * 32
    (tmp_path / "etc").mkdir()
    (tmp_path / "proc/self").mkdir(parents=True)
    (tmp_path / "etc/hostname").write_text("custom-host\n")
    (tmp_path / "proc/self/mountinfo").write_text(
        f"1 2 0:1 /var/lib/docker/containers/{cid}/hostname /etc/hostname rw - ext4 /dev/x rw\n"
    )
    (tmp_path / "proc/self/cgroup").write_text(f"0::/docker/{cid}\n")
    candidates = harness.container_id_candidates({"HOSTNAME": "job-host"}, proc_root=tmp_path)
    assert candidates == ["job-host", "custom-host", cid]


def test_container_id_candidates_tolerate_missing_files(tmp_path: Path) -> None:
    assert harness.container_id_candidates({}, proc_root=tmp_path) == []