实现一次性 Pod Runner 后端

tests/test_pod_worker.py

@@ -0,0 +1,43 @@
+import pytest
+
+from gitea_microvm_runner.models import RunnerRequest
+from gitea_microvm_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():
+    assert pod_worker.identity_path("panxiao81/example", "publish image") == (
+        "panxiao81/example/publish%20image"
+    )
+    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