285 lines
13 KiB
Python
285 lines
13 KiB
Python
"""Release gates and package transitions bind the intended immutable artifacts."""
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from scripts.releases.stable import (
|
|
check_claim, ensure_final_tag, plan_transitions, read_manifest, require_stable_identity,
|
|
require_success, retarget_release, validate_candidates,
|
|
)
|
|
|
|
BASE = "https://releases.example"
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def candidates(tag, commit, digest):
|
|
packages = []
|
|
native_version = f"2026.5761.{100 + int(tag.rsplit('.', 1)[1])}.0"
|
|
for platform in ("windows", "macos"):
|
|
for arch in ("x64", "arm64"):
|
|
packages.append({
|
|
"platform": platform, "arch": arch, "tag": tag, "commit": commit,
|
|
"identity": "test.application",
|
|
"version": native_version if platform == "windows" else tag[1:],
|
|
**({"publisher": "CN=Test", "applicationId": "App"} if platform == "windows" else {"teamId": "ABCDEFGHIJ"}),
|
|
"artifact": {"sha256": digest,
|
|
"url": f"{BASE}/releases/tag/{tag}/{arch}" + (".msixbundle" if platform == "windows" else ".zip")},
|
|
})
|
|
return {"schema": 2, "tag": tag, "commit": commit, "packages": packages,
|
|
"smoke_results": {name: {"result": "success"} for name in (
|
|
"smoke-darwin", "smoke-win32", "smoke-win32-universal")}}
|
|
|
|
|
|
def test_gate_requires_every_success_including_real_cli(tmp_path):
|
|
required = ["ci", "docker", "acceptance", "publication"]
|
|
success = {name: {"result": "success"} for name in required}
|
|
require_success(success, required)
|
|
with pytest.raises(ValueError, match="required-job list"):
|
|
require_success(success, [])
|
|
for name in required:
|
|
for result in ("failure", "cancelled", "skipped", None):
|
|
needs = copy.deepcopy(success)
|
|
if result:
|
|
needs[name]["result"] = result
|
|
else:
|
|
del needs[name]
|
|
with pytest.raises(ValueError, match=name):
|
|
require_success(needs, required)
|
|
summary = tmp_path / "summary.md"
|
|
env = {**os.environ, "RELEASE_NEEDS": json.dumps(success), "GITHUB_STEP_SUMMARY": str(summary), "PYTHONPATH": str(ROOT)}
|
|
argv = [sys.executable, "-m", "scripts.releases.stable", "gate", *required]
|
|
assert subprocess.run(argv, cwd=tmp_path, env=env, capture_output=True).returncode == 0
|
|
empty = subprocess.run(argv[:4], cwd=tmp_path, env=env, capture_output=True, text=True, encoding="utf-8")
|
|
assert empty.returncode != 0
|
|
assert "required-job list" in empty.stderr
|
|
env["RELEASE_NEEDS"] = json.dumps({**success, "publication": {"result": "cancelled"}})
|
|
result = subprocess.run(argv, cwd=tmp_path, env=env, capture_output=True, text=True, encoding="utf-8")
|
|
assert result.returncode != 0
|
|
assert "publication=cancelled" in result.stderr
|
|
|
|
|
|
def test_transitions_bind_all_arches_identity_version_and_archive():
|
|
old = candidates("v1.2.3", "a" * 40, "1" * 64)
|
|
old["schema"] = 1
|
|
del old["smoke_results"]
|
|
with pytest.raises(ValueError, match="Legacy candidate"):
|
|
validate_candidates(old, old["tag"], old["commit"], BASE)
|
|
new = candidates("v1.2.4", "b" * 40, "2" * 64)
|
|
require_stable_identity(new["tag"], new["commit"])
|
|
for tag in ("v1.2.4+canary.20260907T143420Z", "v1.2.4-rc"):
|
|
with pytest.raises(ValueError):
|
|
require_stable_identity(tag, new["commit"])
|
|
transitions = plan_transitions(old, new, BASE)
|
|
assert {row["target"] for row in transitions} == {"windows-x64", "windows-arm64", "macos-x64", "macos-arm64"}
|
|
assert all(row["transition"]["new"]["commit"] == new["commit"] for row in transitions)
|
|
missing = copy.deepcopy(new)
|
|
missing["packages"].pop()
|
|
with pytest.raises(ValueError, match="both architectures"):
|
|
plan_transitions(old, missing, BASE)
|
|
with pytest.raises(ValueError, match="identity"):
|
|
validate_candidates(new, new["tag"], old["commit"], BASE)
|
|
for key, value in [("commit", old["commit"]), ("identity", "different"), ("publisher", "CN=Other"), ("version", "9.9.9.0")]:
|
|
changed = copy.deepcopy(new)
|
|
changed["packages"][0][key] = value
|
|
with pytest.raises(ValueError):
|
|
plan_transitions(old, changed, BASE)
|
|
mutable = copy.deepcopy(new)
|
|
mutable["packages"][0]["artifact"]["url"] = f"{BASE}/releases/win32/stable/current.msixbundle"
|
|
with pytest.raises(ValueError, match="immutable"):
|
|
plan_transitions(old, mutable, BASE)
|
|
for suffix in ("../other.zip", "%2e%2e/other.zip", "%252e%252e/other.zip"):
|
|
traversal = copy.deepcopy(new)
|
|
traversal["packages"][0]["artifact"]["url"] = f"{BASE}/releases/tag/{new['tag']}/{suffix}"
|
|
with pytest.raises(ValueError, match="path encoding"):
|
|
plan_transitions(old, traversal, BASE)
|
|
with pytest.raises(ValueError, match="increase"):
|
|
plan_transitions(new, candidates("v1.2.3", "a" * 40, "1" * 64), BASE)
|
|
|
|
|
|
@pytest.fixture
|
|
def https_origin(tmp_path, monkeypatch):
|
|
import datetime
|
|
import ipaddress
|
|
import ssl
|
|
import threading
|
|
import urllib.request
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
from cryptography import x509
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
from cryptography.x509.oid import NameOID
|
|
|
|
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "localhost")])
|
|
cert = (x509.CertificateBuilder().subject_name(name).issuer_name(name)
|
|
.public_key(key.public_key()).serial_number(x509.random_serial_number())
|
|
.not_valid_before(datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc))
|
|
.not_valid_after(datetime.datetime(2099, 1, 1, tzinfo=datetime.timezone.utc))
|
|
.add_extension(x509.SubjectAlternativeName([
|
|
x509.DNSName("localhost"), x509.IPAddress(ipaddress.ip_address("127.0.0.1")),
|
|
]), critical=False).sign(key, hashes.SHA256()))
|
|
cert_file, key_file = tmp_path / "cert.pem", tmp_path / "key.pem"
|
|
cert_file.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
|
|
key_file.write_bytes(key.private_bytes(serialization.Encoding.PEM,
|
|
serialization.PrivateFormat.PKCS8, serialization.NoEncryption()))
|
|
requests = []
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
requests.append(self.path)
|
|
if self.path in ("/same", "/cross"):
|
|
self.send_response(302)
|
|
host = "127.0.0.1" if self.path == "/same" else "localhost"
|
|
self.send_header("Location", f"https://{host}:{self.server.server_port}/manifest")
|
|
self.end_headers()
|
|
else:
|
|
item = self.server.store.get(self.path.lstrip('/'))
|
|
data = item[0] if item else b'not found'
|
|
self.send_response(200 if item else 404)
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
server.store = {'manifest': (b'{"schema":1}', '"e"')}
|
|
server.requests = requests
|
|
server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
|
server_context.load_cert_chain(cert_file, key_file)
|
|
server.socket = server_context.wrap_socket(server.socket, server_side=True)
|
|
client_context = ssl.create_default_context(cafile=str(cert_file))
|
|
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}),
|
|
urllib.request.HTTPSHandler(context=client_context)).open
|
|
base = f"https://127.0.0.1:{server.server_port}"
|
|
server.base, server.opener = base, opener
|
|
monkeypatch.setenv('SSL_CERT_FILE', str(cert_file))
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
yield server
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(timeout=5)
|
|
|
|
|
|
def test_manifest_origin_checks_with_real_https(https_origin):
|
|
server = https_origin
|
|
base, opener = server.base, server.opener
|
|
digest = hashlib.sha256(b'{"schema":1}').hexdigest()
|
|
assert read_manifest(f'{base}/same', digest, expected_origin=base, opener=opener) == {'schema': 1}
|
|
with pytest.raises(ValueError, match='digest'):
|
|
read_manifest(f'{base}/manifest', 'f' * 64, opener=opener)
|
|
with pytest.raises(ValueError, match='origin'):
|
|
read_manifest(f'{base}/cross', opener=opener)
|
|
server.requests.clear()
|
|
with pytest.raises(ValueError, match='origin'):
|
|
read_manifest(f'https://localhost:{server.server_port}/manifest', expected_origin=base, opener=opener)
|
|
assert server.requests == []
|
|
|
|
|
|
def test_claim_object_movement_and_lightweight_tags_fail_closed(tmp_path, monkeypatch):
|
|
commit = "a" * 40
|
|
claim_object = "b" * 40
|
|
ref = "refs/tags/v1.2.3-rc"
|
|
env = {"RELEASE_CLAIM_TAG": "v1.2.3-rc", "RELEASE_CLAIM_OBJECT": claim_object,
|
|
"GITHUB_SHA": commit, "GITHUB_REF": ref}
|
|
|
|
def git(argv):
|
|
if argv[1] == "ls-remote":
|
|
return f"{claim_object}\t{ref}\n{commit}\t{ref}^{{}}"
|
|
if argv[1] == "cat-file":
|
|
return "tag"
|
|
if argv[1] == "rev-parse":
|
|
return claim_object if argv[-1] == ref else commit
|
|
if argv[1] == "tag":
|
|
return json.dumps({
|
|
"schema": 1, "version": "1.2.3", "commit": commit,
|
|
"autopublish": False,
|
|
})
|
|
return ""
|
|
|
|
assert check_claim(env, git) == {
|
|
"claim_tag": "v1.2.3-rc", "claim_object": claim_object,
|
|
"tag": "v1.2.3", "version": "1.2.3", "commit": commit,
|
|
"autopublish": False,
|
|
}
|
|
with pytest.raises(ValueError, match="moved"):
|
|
check_claim(env, lambda argv: f"{'c' * 40}\t{ref}\n{commit}\t{ref}^{{}}"
|
|
if argv[1] == "ls-remote" else git(argv))
|
|
with pytest.raises(ValueError, match="annotated"):
|
|
check_claim(env, lambda argv: "commit" if argv[1] == "cat-file" else git(argv))
|
|
|
|
repo = tmp_path / "repo"
|
|
remote = tmp_path / "remote.git"
|
|
repo.mkdir()
|
|
monkeypatch.chdir(repo)
|
|
subprocess.run(["git", "init", "-b", "main"], check=True, capture_output=True)
|
|
subprocess.run(["git", "init", "--bare", str(remote)], check=True, capture_output=True)
|
|
subprocess.run(["git", "config", "user.name", "fixture"], check=True)
|
|
subprocess.run(["git", "config", "user.email", "fixture@example.invalid"], check=True)
|
|
(repo / "input").write_text("first", encoding="utf-8")
|
|
subprocess.run(["git", "add", "input"], check=True)
|
|
subprocess.run(["git", "commit", "-m", "first"], check=True, capture_output=True)
|
|
actual = subprocess.check_output(["git", "rev-parse", "HEAD"], text=True, encoding="utf-8").strip()
|
|
subprocess.run(["git", "remote", "add", "origin", str(remote)], check=True)
|
|
metadata = json.dumps({
|
|
"schema": 1, "version": "1.2.3", "commit": actual,
|
|
"autopublish": False,
|
|
}, sort_keys=True, separators=(",", ":"))
|
|
subprocess.run(["git", "tag", "-a", "v1.2.3-rc", "-m", metadata], check=True)
|
|
subprocess.run(["git", "push", "origin", "main", "v1.2.3-rc"], check=True, capture_output=True)
|
|
env.update({"GITHUB_SHA": actual, "RELEASE_CLAIM_OBJECT": subprocess.check_output(
|
|
["git", "rev-parse", ref], text=True, encoding="utf-8").strip()})
|
|
claim = check_claim(env)
|
|
assert claim["commit"] == actual
|
|
final_object = ensure_final_tag(
|
|
"v1.2.3", actual, claim,
|
|
candidate_manifest_sha256="c" * 64,
|
|
docker_manifest_digest="sha256:" + "d" * 64,
|
|
)
|
|
remote_final = subprocess.check_output(
|
|
["git", "ls-remote", "origin", "refs/tags/v1.2.3", "refs/tags/v1.2.3^{}"],
|
|
text=True, encoding="utf-8",
|
|
)
|
|
assert f"{final_object}\trefs/tags/v1.2.3" in remote_final
|
|
assert f"{actual}\trefs/tags/v1.2.3^{{}}" in remote_final
|
|
subprocess.run(["git", "--git-dir", str(remote), "update-ref", "-d", ref], check=True)
|
|
with pytest.raises(ValueError, match="moved"):
|
|
check_claim(env)
|
|
|
|
|
|
@pytest.mark.parametrize("publish", [False, True])
|
|
def test_retarget_release_preserves_the_database_id_and_explicit_draft_policy(publish):
|
|
commit = "a" * 40
|
|
calls = []
|
|
patched = False
|
|
|
|
def gh(argv):
|
|
nonlocal patched
|
|
calls.append(argv)
|
|
if argv[1:3] == ["api", "--method"]:
|
|
patched = True
|
|
return "{}"
|
|
if argv[1:3] == ["api", "repos/example/project/releases/42"]:
|
|
return json.dumps({
|
|
"id": 42, "tag_name": "v1.2.3" if patched else "v1.2.3-rc",
|
|
"target_commitish": commit, "prerelease": False,
|
|
"draft": not publish if patched else True,
|
|
})
|
|
return "{}"
|
|
|
|
retarget_release("example/project", 42, "v1.2.3", commit, publish=publish, run=gh)
|
|
|
|
assert ["--field", f"draft={str(not publish).lower()}"] == calls[1][-2:]
|
|
assert calls[2] == ["gh", "api", "repos/example/project/releases/42"]
|