Genarrative/scripts/test_gitea_cache_export.py

"""master 产物的增量归档与 Gitea v4 上传协议回归。"""

import hashlib
import base64
from contextlib import redirect_stdout
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import importlib.util
import io
import json
import os
from pathlib import Path
import tarfile
import tempfile
import threading
import unittest
from unittest.mock import patch
import urllib.parse
import zipfile

from gitea_cache_upload_cleanup import decode_owner


spec = importlib.util.spec_from_file_location("cache_export", Path(__file__).with_name("export-gitea-rust-cache.py"))
exporter = importlib.util.module_from_spec(spec)
spec.loader.exec_module(exporter)


class ExportTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        (self.root / "objects").mkdir()

    def object(self, key, value, mtime):
        path = self.root / "objects" / key[0] / key[1] / key
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_bytes(value)
        os.utime(path, ns=(mtime * 10**9, mtime * 10**9))
        return path

    def unpack(self, baseline, limit=1024):
        destination = self.root / "snapshot.zip"
        result = exporter.pack_snapshot(self.root, destination, {"job": "fixture"}, baseline, limit)
        with zipfile.ZipFile(destination) as bundle:
            self.assertEqual(bundle.namelist(), ["snapshot.tar"])
            with tarfile.open(fileobj=io.BytesIO(bundle.read("snapshot.tar"))) as archive:
                files = {member.name: archive.extractfile(member).read() for member in archive}
        return result, json.loads(files.pop("manifest.json")), files

    def test_delta_omits_hit_bytes_preserves_recency_and_exports_new_keys(self):
        inherited = self.object("a" * 64, b"old", 10)
        baseline_path = self.root / "baseline.json"
        exporter.save_baseline(self.root, baseline_path)
        baseline = json.loads(baseline_path.read_text())
        # sccache 对命中对象更新 mtime，不能因此重复上传已有对象。
        os.utime(inherited, ns=(20 * 10**9, 20 * 10**9))
        self.object("b" * 64, b"new", 30)
        result, manifest, files = self.unpack(baseline)
        key = "objects/b/b/" + "b" * 64
        self.assertEqual(result, (1, 3))
        self.assertEqual(files, {key: b"new"})
        self.assertEqual(manifest["mode"], "delta")
        self.assertEqual(manifest["touched"], [{"path": "objects/a/a/" + "a" * 64, "mtime_ns": 20 * 10**9}])
        self.assertEqual(manifest["objects"][0]["sha256"], hashlib.sha256(b"new").hexdigest())

    def test_empty_delta_is_valid_and_changed_size_is_exported(self):
        path = self.object("a" * 64, b"old", 10)
        baseline = {"a/a/" + "a" * 64: {"size": 3, "mtime_ns": 10 * 10**9}}
        self.assertEqual(self.unpack(baseline)[0], (0, 0))
        path.write_bytes(b"replacement")
        self.assertEqual(self.unpack(baseline)[0], (1, 11))

    def test_capacity_prefers_newest_objects_and_excludes_unrelated_files(self):
        self.object("a" * 64, b"old", 10)
        self.object("b" * 64, b"new", 20)
        (self.root / "objects" / "credentials").write_bytes(b"secret")
        (self.root / "sccache").write_bytes(b"binary")
        result, _, files = self.unpack({}, limit=3)
        self.assertEqual(result, (1, 3))
        self.assertEqual(list(files.values()), [b"new"])

    @unittest.skipUnless(os.name == "posix", "symlink fixture requires POSIX")
    def test_symlink_objects_and_subdirectories_are_not_exported(self):
        outside = self.root / "secret"
        outside.write_bytes(b"secret")
        linked = self.object("a" * 64, b"old", 10)
        linked.unlink()
        linked.symlink_to(outside)
        (self.root / "objects" / "b").symlink_to(self.root, target_is_directory=True)
        self.assertEqual(self.unpack({})[0], (0, 0))

    def test_non_master_and_unclean_shutdown_do_not_scan_or_upload(self):
        with patch.object(exporter, "scan_objects", side_effect=AssertionError("must not scan")), \
                patch.object(exporter, "upload_snapshot", side_effect=AssertionError("must not upload")):
            exporter.export({"GITHUB_EVENT_NAME": "pull_request", "GITHUB_REF": "refs/heads/master"})
            exporter.export({"GITHUB_EVENT_NAME": "push", "GITHUB_REF": "refs/heads/feature"})
            exporter.export({"GITHUB_EVENT_NAME": "push", "GITHUB_REF": "refs/heads/master"})

    def test_internal_repository_route_bypasses_rpc_only_gateway(self):
        env = {
            "GITHUB_REPOSITORY": "owner/repo",
            "GITHUB_SERVER_URL": "http://fetch-gate:8080",
            "GENARRATIVE_GITEA_REPOSITORY_URL": "http://gitea:3000/sub/owner/repo.git",
        }
        self.assertEqual(exporter.artifact_base_url(env), "http://gitea:3000/sub")
        env["GENARRATIVE_GITEA_REPOSITORY_URL"] = "http://token@gitea:3000/owner/repo.git"
        with self.assertRaises(ValueError):
            exporter.artifact_base_url(env)

    def test_native_v4_chunks_finalize_with_verified_size_and_hash(self):
        calls = []

        class Handler(BaseHTTPRequestHandler):
            def log_message(self, *args):
                pass

            def do_POST(self):
                body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
                calls.append((self.command, self.path, dict(self.headers), body))
                payload = {"ok": True}
                if self.path.endswith("CreateArtifact"):
                    payload["signedUploadUrl"] = "https://external.invalid/twirp/github.actions.results.api.v1.ArtifactService/UploadArtifact?sig=private&taskID=4&artifactID=8"
                else:
                    payload["artifactId"] = "8"
                self.send_response(200)
                self.end_headers()
                self.wfile.write(json.dumps(payload).encode())

            def do_PUT(self):
                body = self.rfile.read(int(self.headers["Content-Length"]))
                calls.append((self.command, self.path, dict(self.headers), body))
                self.send_response(201)
                self.end_headers()
                self.wfile.write(b'"created"')

        server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        try:
            path = self.root / "test.zip"
            path.write_bytes(b"123456789")
            with patch.object(exporter, "CHUNK_BYTES", 4):
                exporter.upload_snapshot(path, "rust-cache-v1-backend-tests-attempt-1", 123, f"http://127.0.0.1:{server.server_port}", "ephemeral")
        finally:
            server.shutdown()
            server.server_close()
            thread.join()
        self.assertEqual(len(calls), 6)
        self.assertEqual(calls[0][3]["workflowRunBackendId"], "123")
        remaining = datetime.fromisoformat(calls[0][3]["expiresAt"]) - datetime.now(timezone.utc)
        self.assertGreater(remaining.total_seconds(), 7 * 24 * 3600)
        self.assertLessEqual(remaining.total_seconds(), 8 * 24 * 3600)
        blocks = calls[1:4]
        self.assertEqual(b"".join(call[3] for call in blocks), b"123456789")
        self.assertEqual(len({urllib.parse.parse_qs(urllib.parse.urlsplit(call[1]).query)["blockid"][0] for call in blocks}), 3)
        for call in blocks:
            block_id = urllib.parse.parse_qs(urllib.parse.urlsplit(call[1]).query)["blockid"][0]
            gitea_filename_suffix = base64.urlsafe_b64encode(block_id.encode()).decode()
            self.assertEqual(decode_owner(gitea_filename_suffix), ("backend-tests", 1))
        self.assertTrue(all("Authorization" not in call[2] for call in calls[1:5]))
        self.assertEqual(calls[-1][2]["Authorization"], "Bearer ephemeral")
        self.assertEqual(calls[-1][3]["size"], "9")
        self.assertEqual(calls[-1][3]["hash"], "sha256:" + hashlib.sha256(b"123456789").hexdigest())

    def test_upload_failure_never_finalizes_an_incomplete_artifact(self):
        path = self.root / "test.zip"
        path.write_bytes(b"data")
        with patch.object(exporter, "request", side_effect=[
            {"ok": True, "signedUploadUrl": "http://gitea/twirp/github.actions.results.api.v1.ArtifactService/UploadArtifact?sig=x"},
            RuntimeError("upload failed"),
        ]) as request:
            with self.assertRaisesRegex(RuntimeError, "upload failed"):
                exporter.upload_snapshot(path, "rust-cache-v1-backend-tests-attempt-1", 123, "http://gitea", "ephemeral")
            self.assertEqual(request.call_count, 2)

    def test_failed_optional_export_never_emits_host_completion_marker(self):
        (self.root / "export-baseline.json").write_text("{}")
        (self.root / "rustc.txt").write_text("rustc fixture\n")
        (self.root / "workspace.txt").write_text(str(Path.cwd().resolve()))
        (self.root / "source-commit.txt").write_text("a" * 40)
        self.object("b" * 64, b"new", 20)
        env = {
            "GITHUB_EVENT_NAME": "push", "GITHUB_REF": "refs/heads/master",
            "GENARRATIVE_CI_RUST_CACHE_EXPORT_READY": "1",
            "GITHUB_JOB": "backend-tests", "GITHUB_SHA": "b" * 40,
            "GITHUB_RUN_ID": "123", "GITHUB_RUN_ATTEMPT": "1",
            "GENARRATIVE_GITEA_TOKEN": "ephemeral",
            "GITHUB_REPOSITORY": "owner/repo",
            "GENARRATIVE_GITEA_REPOSITORY_URL": "http://gitea/owner/repo.git",
            "GENARRATIVE_CI_RUST_CACHE_ROOT": str(self.root),
        }
        output = io.StringIO()
        with patch.object(exporter.subprocess, "check_output", side_effect=["rustc fixture\n", "sccache 0.18.0\n"]), \
                patch.object(exporter, "upload_snapshot", side_effect=RuntimeError("upload failed")), \
                redirect_stdout(output):
            with self.assertRaisesRegex(RuntimeError, "upload failed"):
                exporter.export(env)
        self.assertNotIn("complete=true", output.getvalue())
        self.assertFalse((self.root / "export-baseline.json").exists())


if __name__ == "__main__":
    unittest.main()