Files
panxiao81 68bb02b20a
test / python (pull_request) Successful in 8s
test / shell (pull_request) Successful in 15s
修复 Runner 身份路径并补充关联日志
2026-09-16 18:00:50 +00:00

58 lines
2.3 KiB
Python

import pytest
from gitea_dynamic_runner.models import RunnerRequest
from gitea_dynamic_runner import pod_worker
def request() -> RunnerRequest:
return RunnerRequest(
job_id=47,
run_id=12,
backend="pod",
repository="panxiao81/example",
job_name="publish image",
labels=("self-hosted", "pod"),
)
def test_identity_path_is_meaningful_and_uri_safe():
path = pod_worker.identity_path("panxiao81/example", "publish image")
assert path.startswith("panxiao81/example/publish-image-")
assert "%" not in path
assert all(pod_worker.SPIFFE_PATH_SEGMENT.fullmatch(part) for part in path.split("/"))
assert pod_worker.identity_path("panxiao81/example", "lint") == "panxiao81/example/lint"
assert path == pod_worker.identity_path("panxiao81/example", "publish image")
assert path != pod_worker.identity_path("panxiao81/example", "publish-image")
real_path = pod_worker.identity_path(
"panxiao81/postgresql-tenant-operator", "Run on Ubuntu"
)
assert real_path.startswith(
"panxiao81/postgresql-tenant-operator/Run-on-Ubuntu-"
)
assert all(
pod_worker.SPIFFE_PATH_SEGMENT.fullmatch(part)
for part in real_path.split("/")
)
with pytest.raises(ValueError):
pod_worker.identity_path("invalid", "test")
def test_pod_manifest_is_disposable_host_runner(monkeypatch):
monkeypatch.setattr(pod_worker, "NAMESPACE", "ci")
manifest = pod_worker.pod_manifest(request(), "gitea-pod-abcd")
assert manifest["metadata"]["name"] == "gitea-pod-abcd"
assert manifest["metadata"]["annotations"]["ci.ddupan.top/queued-job-id"] == "47"
spec = manifest["spec"]
assert spec["restartPolicy"] == "Never"
container = spec["containers"][0]
assert container["securityContext"] == {"privileged": True}
env = {item["name"]: item for item in container["env"]}
assert env["GITEA_RUNNER_LABELS"]["value"] == "self-hosted:host,pod:host"
assert env["GITEA_RUNNER_EPHEMERAL"]["value"] == "1"
assert env["GITEA_RUNNER_ONCE"]["value"] == "1"
assert all(item["name"] != "DOCKER_HOST" for item in container["env"])
volumes = {item["name"]: item for item in spec["volumes"]}
assert volumes["spire-agent-socket"]["csi"]["driver"] == "csi.spiffe.io"
assert "runner-config" not in volumes
assert "docker" not in volumes