187 lines
6.8 KiB
Python
187 lines
6.8 KiB
Python
"""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(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
# cleanup_all removes every tracked run, including the session's own Gitea, so this
|
|
# test works on a private registry instead of the real one.
|
|
monkeypatch.setattr(harness, "_ACTIVE", {})
|
|
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) == []
|
|
|
|
|
|
def test_run_docker_passes_secrets_through_the_environment_not_argv(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
seen: dict[str, object] = {}
|
|
|
|
def fake_run(argv: list[str], **kwargs: object) -> object:
|
|
seen["argv"] = argv
|
|
seen["env"] = kwargs.get("env")
|
|
|
|
class Done:
|
|
returncode = 0
|
|
stdout = ""
|
|
stderr = ""
|
|
|
|
return Done()
|
|
|
|
monkeypatch.setattr(harness.subprocess, "run", fake_run)
|
|
harness.run_docker("exec", "-e", "SECRET_VAR", "box", env={"SECRET_VAR": "s3cret-value"})
|
|
|
|
assert "s3cret-value" not in " ".join(seen["argv"]) # type: ignore[call-overload]
|
|
assert seen["env"]["SECRET_VAR"] == "s3cret-value" # type: ignore[index]
|
|
assert "PATH" in seen["env"] # type: ignore[operator]
|