coreci
da15bda6d2
---ci--- phase: 3 milestone: v0.2 status: complete requirements: covered: [REQ-21, REQ-22, REQ-23] partial: [] ---/ci---
511 lines
21 KiB
Python
511 lines
21 KiB
Python
"""Stub-gitea release-attacher tests (REQ-22, RESEARCH §2.4).
|
|
|
|
Drives scripts/release.py against an in-process stub gitea
|
|
(``_StubGiteaHandler`` on ``127.0.0.1:0`` — the v0.1 ``_StubOwuiHandler``
|
|
precedent; fake token, no live-forge writes) plus the sanctioned
|
|
secrets loader scripts/_env_secrets.py: create-or-get in both orders,
|
|
byte-exact multipart shape, sha256-from-file discipline (G-017),
|
|
idempotent re-attach, bounded-retry exhaustion -> RELEASE_PENDING marker,
|
|
and file-only token resolution asserted before the first HTTP call.
|
|
"""
|
|
|
|
import contextlib
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
import urllib.parse
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
REPO = pathlib.Path(__file__).resolve().parent.parent
|
|
RELEASE_PATH = REPO / "scripts" / "release.py"
|
|
ENV_SECRETS_PATH = REPO / "scripts" / "_env_secrets.py"
|
|
|
|
sys.path.insert(0, str(REPO))
|
|
|
|
import coreci_broker # repo root on sys.path (parser-parity assertion)
|
|
|
|
|
|
def _load_module(name, path):
|
|
spec = importlib.util.spec_from_file_location(name, str(path))
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
release = _load_module("release", RELEASE_PATH)
|
|
env_secrets = _load_module("env_secrets", ENV_SECRETS_PATH)
|
|
|
|
OWNER = "coreci"
|
|
REPO_NAME = "broker"
|
|
TAG = "v0.2.1"
|
|
TOKEN = "testtoken123"
|
|
ARTIFACT_NAME = "coreci-broker-v0.2.1.pyz"
|
|
|
|
_RELEASE_BY_TAG_RE = re.compile(
|
|
r"^/api/v1/repos/([^/]+)/([^/]+)/releases/tags/([^/]+)$")
|
|
_RELEASES_RE = re.compile(r"^/api/v1/repos/([^/]+)/([^/]+)/releases$")
|
|
_ASSETS_RE = re.compile(r"^/api/v1/repos/([^/]+)/([^/]+)/releases/(\d+)/assets$")
|
|
|
|
|
|
def _header_params(value):
|
|
"""'a; b=\"c\"' -> {'b': 'c'} — Content-Type/Disposition params."""
|
|
params = {}
|
|
for item in value.split(";")[1:]:
|
|
key, _, val = item.strip().partition("=")
|
|
if key:
|
|
params[key.strip()] = val.strip().strip('"')
|
|
return params
|
|
|
|
|
|
def _split_multipart(content_type, body):
|
|
"""Minimal multipart parser: boundary from the Content-Type header."""
|
|
boundary = _header_params(content_type).get("boundary", "")
|
|
parts = []
|
|
if boundary:
|
|
for element in body.split(("--" + boundary).encode("utf-8"))[1:]:
|
|
if element.startswith(b"--"):
|
|
continue # closing delimiter
|
|
if element.startswith(b"\r\n"):
|
|
element = element[2:]
|
|
header_blob, separator, content = element.partition(b"\r\n\r\n")
|
|
if not separator:
|
|
continue
|
|
if content.endswith(b"\r\n"):
|
|
content = content[:-2] # framing CRLF before the next delimiter
|
|
part_headers = {}
|
|
for line in header_blob.decode("utf-8", "replace").split("\r\n"):
|
|
name, _, value = line.partition(":")
|
|
if name:
|
|
part_headers[name.strip().lower()] = value.strip()
|
|
parts.append((part_headers, content))
|
|
return boundary, parts
|
|
|
|
|
|
class _StubGiteaHandler(BaseHTTPRequestHandler):
|
|
"""Scriptable gitea stub (RESEARCH §2.1 routes); records everything."""
|
|
|
|
state = None # scriptable state dict (class attr, bound per server)
|
|
requests = None # every request: method, path, headers, body (+extras)
|
|
|
|
def log_message(self, format, *args):
|
|
pass
|
|
|
|
def _record(self):
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|
body = self.rfile.read(length) if length else b""
|
|
record = {
|
|
"method": self.command,
|
|
"path": self.path,
|
|
"headers": {key.lower(): value
|
|
for key, value in self.headers.items()},
|
|
"body": body,
|
|
}
|
|
self.requests.append(record)
|
|
return record
|
|
|
|
def _send_json(self, status, payload):
|
|
data = json.dumps(payload).encode("utf-8")
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
def do_GET(self):
|
|
self._record()
|
|
parsed = urllib.parse.urlsplit(self.path)
|
|
if _RELEASE_BY_TAG_RE.match(parsed.path):
|
|
if self.state.get("release_exists"):
|
|
self._send_json(200, {"id": self.state.get("release_id", 970)})
|
|
else:
|
|
self._send_json(404, {"message": "Release not found"})
|
|
return
|
|
if _ASSETS_RE.match(parsed.path):
|
|
self._send_json(200, list(self.state.get("assets", [])))
|
|
return
|
|
self._send_json(404, {"message": "Not Found"})
|
|
|
|
def do_POST(self):
|
|
record = self._record()
|
|
parsed = urllib.parse.urlsplit(self.path)
|
|
if _RELEASES_RE.match(parsed.path):
|
|
self.state.setdefault("create_calls", []).append(record)
|
|
status = self.state.get("create_status", 201)
|
|
if status >= 400:
|
|
self._send_json(
|
|
status,
|
|
{"message": self.state.get("create_error", "create failed")})
|
|
else:
|
|
self._send_json(status, {"id": self.state.get("release_id", 970)})
|
|
return
|
|
if _ASSETS_RE.match(parsed.path):
|
|
self.state.setdefault("attach_calls", []).append(record)
|
|
status = self.state.get("attach_status", 201)
|
|
if status >= 400:
|
|
self._send_json(
|
|
status,
|
|
{"message": self.state.get("attach_error", "attach failed")})
|
|
return
|
|
self._describe_attachment(record, parsed)
|
|
self._send_json(201, {
|
|
"id": 2,
|
|
"name": record["attach"]["query_name"],
|
|
"size": record["attach"]["size"],
|
|
})
|
|
return
|
|
self._send_json(404, {"message": "Not Found"})
|
|
|
|
def _describe_attachment(self, record, parsed):
|
|
boundary, parts = _split_multipart(
|
|
record["headers"].get("content-type", ""), record["body"])
|
|
field_name = filename = part_type = None
|
|
content = b""
|
|
if len(parts) == 1:
|
|
part_headers, content = parts[0]
|
|
disposition = _header_params(
|
|
part_headers.get("content-disposition", ""))
|
|
field_name = disposition.get("name")
|
|
filename = disposition.get("filename")
|
|
part_type = part_headers.get("content-type")
|
|
expected = self.state.get("expected_asset_bytes")
|
|
record["attach"] = {
|
|
"boundary": boundary,
|
|
"field_name": field_name,
|
|
"filename": filename,
|
|
"part_content_type": part_type,
|
|
"query_name":
|
|
(urllib.parse.parse_qs(parsed.query).get("name") or [None])[0],
|
|
"content": content,
|
|
"size": len(content),
|
|
"content_matches_expected":
|
|
None if expected is None else content == expected,
|
|
}
|
|
|
|
|
|
class StubGiteaServer:
|
|
"""Bound stub server (v0.1 StubOwuiServer pattern)."""
|
|
|
|
def __init__(self, state):
|
|
self.state = state
|
|
self.requests = []
|
|
handler = type("BoundStubHandler", (_StubGiteaHandler,), {
|
|
"state": state,
|
|
"requests": self.requests,
|
|
})
|
|
self.server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
|
|
threading.Thread(target=self.server.serve_forever, daemon=True).start()
|
|
|
|
@property
|
|
def base_url(self):
|
|
return "http://127.0.0.1:%d" % self.server.server_address[1]
|
|
|
|
def close(self):
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
|
|
|
|
class ReleaseUploadTests(unittest.TestCase):
|
|
def setUp(self):
|
|
tmp = tempfile.TemporaryDirectory(prefix="coreci-release-test-")
|
|
self.addCleanup(tmp.cleanup)
|
|
self.tmp = tmp.name
|
|
self.artifact_path = os.path.join(self.tmp, ARTIFACT_NAME)
|
|
# Includes every byte value (CR/LF inside) — byte-exact multipart
|
|
# transport must survive content that looks like framing.
|
|
self.artifact_bytes = (
|
|
b"#!/usr/bin/env python3\nstub-zipapp-payload:" + bytes(range(256)) * 2)
|
|
with open(self.artifact_path, "wb") as handle:
|
|
handle.write(self.artifact_bytes)
|
|
self.env_root = os.path.join(self.tmp, "env-root")
|
|
os.makedirs(self.env_root)
|
|
secrets_fd = os.open(
|
|
os.path.join(self.env_root, ".env.secrets"),
|
|
os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
|
with os.fdopen(secrets_fd, "w", encoding="utf-8") as handle:
|
|
handle.write("GITEA_TOKEN={}\n".format(TOKEN))
|
|
self.state = {
|
|
"release_exists": True,
|
|
"release_id": 970,
|
|
"assets": [],
|
|
"create_status": 201,
|
|
"attach_status": 201,
|
|
"expected_asset_bytes": self.artifact_bytes,
|
|
}
|
|
self.stub = StubGiteaServer(self.state)
|
|
self.addCleanup(self.stub.close)
|
|
|
|
def _argv(self, **overrides):
|
|
argv = [
|
|
TAG, self.artifact_path,
|
|
"--base-url", self.stub.base_url,
|
|
"--owner", OWNER, "--repo", REPO_NAME,
|
|
"--env-root", self.env_root,
|
|
]
|
|
for key, value in overrides.items():
|
|
argv.extend(["--" + key.replace("_", "-"), value])
|
|
return argv
|
|
|
|
def _run(self, **overrides):
|
|
stdout, stderr = io.StringIO(), io.StringIO()
|
|
with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
|
|
code = release.main(self._argv(**overrides))
|
|
return code, stdout.getvalue(), stderr.getvalue()
|
|
|
|
def _paths(self, method):
|
|
return [record["path"] for record in self.stub.requests
|
|
if record["method"] == method]
|
|
|
|
def test_existing_release_attaches_exact_multipart_shape(self):
|
|
code, out, err = self._run()
|
|
self.assertEqual(0, code)
|
|
self.assertEqual(
|
|
["/api/v1/repos/{}/{}/releases/tags/{}".format(OWNER, REPO_NAME, TAG),
|
|
"/api/v1/repos/{}/{}/releases/970/assets".format(OWNER, REPO_NAME)],
|
|
self._paths("GET"))
|
|
self.assertEqual(
|
|
["/api/v1/repos/{}/{}/releases/970/assets?name={}".format(
|
|
OWNER, REPO_NAME, ARTIFACT_NAME)],
|
|
self._paths("POST"))
|
|
self.assertEqual([], self.state.get("create_calls", []))
|
|
attach = self.stub.requests[-1]
|
|
shape = attach["attach"]
|
|
self.assertEqual("attachment", shape["field_name"])
|
|
self.assertEqual(ARTIFACT_NAME, shape["filename"])
|
|
self.assertEqual(ARTIFACT_NAME, shape["query_name"])
|
|
self.assertEqual("application/octet-stream", shape["part_content_type"])
|
|
self.assertEqual(self.artifact_bytes, shape["content"])
|
|
self.assertTrue(shape["content_matches_expected"])
|
|
self.assertIn("multipart/form-data; boundary=",
|
|
attach["headers"]["content-type"])
|
|
self.assertTrue(self.stub.requests)
|
|
for record in self.stub.requests:
|
|
self.assertEqual("token " + TOKEN, record["headers"]["authorization"])
|
|
|
|
def test_missing_release_creates_with_notes_then_attaches(self):
|
|
self.state["release_exists"] = False
|
|
code, out, err = self._run()
|
|
self.assertEqual(0, code)
|
|
create_calls = self.state.get("create_calls", [])
|
|
self.assertEqual(1, len(create_calls))
|
|
create = create_calls[0]
|
|
self.assertEqual(
|
|
"/api/v1/repos/{}/{}/releases".format(OWNER, REPO_NAME), create["path"])
|
|
self.assertEqual("application/json", create["headers"]["content-type"])
|
|
payload = json.loads(create["body"])
|
|
self.assertEqual(TAG, payload["tag_name"])
|
|
self.assertEqual(TAG + ": coreci-broker release", payload["name"])
|
|
self.assertIs(False, payload["draft"])
|
|
self.assertIs(False, payload["prerelease"])
|
|
for section in ("## Summary", "## Requirements", "## Verification",
|
|
"## CLI install + usage", "## Docker usage"):
|
|
self.assertIn(section, payload["body"])
|
|
self.assertEqual(1, len(self.state.get("attach_calls", [])))
|
|
|
|
def test_matching_asset_is_idempotent_skip(self):
|
|
self.state["assets"] = [{
|
|
"id": 1, "name": ARTIFACT_NAME, "size": len(self.artifact_bytes),
|
|
}]
|
|
code, out, err = self._run()
|
|
self.assertEqual(0, code)
|
|
self.assertEqual([], self._paths("POST"))
|
|
self.assertEqual(
|
|
["/api/v1/repos/{}/{}/releases/tags/{}".format(OWNER, REPO_NAME, TAG),
|
|
"/api/v1/repos/{}/{}/releases/970/assets".format(OWNER, REPO_NAME)],
|
|
self._paths("GET"))
|
|
self.assertEqual([], self.state.get("attach_calls", []))
|
|
self.assertEqual([], self.state.get("create_calls", []))
|
|
|
|
def test_same_name_but_wrong_size_reattaches(self):
|
|
self.state["assets"] = [{"id": 1, "name": ARTIFACT_NAME, "size": 1}]
|
|
code, out, err = self._run()
|
|
self.assertEqual(0, code)
|
|
self.assertEqual(1, len(self.state.get("attach_calls", [])))
|
|
|
|
def test_attach_500_exhausts_retries_and_marks_pending(self):
|
|
self.state["attach_status"] = 500
|
|
completed = subprocess.run(
|
|
[sys.executable, str(RELEASE_PATH)] + self._argv(),
|
|
capture_output=True, text=True, timeout=120,
|
|
)
|
|
self.assertEqual(1, completed.returncode)
|
|
self.assertIn(
|
|
"RELEASE_PENDING {} {}".format(TAG, self.artifact_path),
|
|
completed.stdout)
|
|
self.assertEqual(3, len(self.state.get("attach_calls", [])))
|
|
self.assertEqual([], self.state.get("create_calls", []))
|
|
self.assertIn("500", completed.stderr)
|
|
self.assertNotIn(TOKEN, completed.stdout)
|
|
self.assertNotIn(TOKEN, completed.stderr)
|
|
|
|
def test_permanent_attach_error_reports_gitea_body(self):
|
|
self.state["attach_status"] = 400
|
|
self.state["attach_error"] = "attachment too large"
|
|
code, out, err = self._run()
|
|
self.assertEqual(1, code)
|
|
self.assertEqual(1, len(self.state.get("attach_calls", []))) # 4xx: no retry
|
|
self.assertIn("attachment too large", err)
|
|
self.assertIn(
|
|
"RELEASE_PENDING {} {}".format(TAG, self.artifact_path), out)
|
|
|
|
def test_sha256_comes_from_file_and_cli_has_no_sha_argument(self):
|
|
before = len(self.stub.requests)
|
|
with contextlib.redirect_stderr(io.StringIO()):
|
|
with self.assertRaises(SystemExit) as raised:
|
|
release.main(self._argv() + ["--sha", "deadbeef"])
|
|
self.assertEqual(2, raised.exception.code)
|
|
with contextlib.redirect_stderr(io.StringIO()):
|
|
with self.assertRaises(SystemExit) as raised:
|
|
release.main(self._argv() + ["0deadbeef"])
|
|
self.assertEqual(2, raised.exception.code)
|
|
self.assertEqual(before, len(self.stub.requests)) # refused pre-HTTP
|
|
self.state["release_exists"] = False # force create so notes hit the wire
|
|
code, out, err = self._run()
|
|
self.assertEqual(0, code)
|
|
notes = json.loads(self.state["create_calls"][0]["body"])["body"]
|
|
expected = hashlib.sha256(self.artifact_bytes).hexdigest()
|
|
self.assertIn("sha256:" + expected, notes)
|
|
self.assertNotIn("deadbeef", notes)
|
|
|
|
def test_token_resolves_from_env_file_on_every_request(self):
|
|
code, out, err = self._run()
|
|
self.assertEqual(0, code)
|
|
self.assertTrue(self.stub.requests)
|
|
for record in self.stub.requests:
|
|
self.assertEqual("token " + TOKEN, record["headers"]["authorization"])
|
|
# Structural discipline: the loader + attacher never touch the
|
|
# process environment, and the token value is never printed.
|
|
for path in (ENV_SECRETS_PATH, RELEASE_PATH):
|
|
source = path.read_text(encoding="utf-8")
|
|
self.assertNotIn("os.environ", source)
|
|
self.assertNotIn("os.getenv", source)
|
|
self.assertNotIn(TOKEN, out)
|
|
self.assertNotIn(TOKEN, err)
|
|
|
|
def test_missing_or_empty_token_fails_before_any_http(self):
|
|
empty_root = os.path.join(self.tmp, "empty-env-root")
|
|
os.makedirs(empty_root)
|
|
code, out, err = self._run(env_root=empty_root)
|
|
self.assertEqual(2, code)
|
|
self.assertEqual([], self.stub.requests) # no HTTP attempted
|
|
self.assertIn("GITEA_TOKEN", err)
|
|
self.assertNotIn("RELEASE_PENDING", out)
|
|
with open(os.path.join(empty_root, ".env.secrets"), "w",
|
|
encoding="utf-8") as handle:
|
|
handle.write("GITEA_TOKEN=\n")
|
|
code, out, err = self._run(env_root=empty_root)
|
|
self.assertEqual(2, code)
|
|
self.assertEqual([], self.stub.requests)
|
|
self.assertIn("GITEA_TOKEN", err)
|
|
|
|
def test_notes_sections_sha_line_and_docker_usage(self):
|
|
self.state["release_exists"] = False
|
|
code, out, err = self._run()
|
|
self.assertEqual(0, code)
|
|
notes = json.loads(self.state["create_calls"][0]["body"])["body"]
|
|
for section in ("## Summary", "## Requirements", "## Verification",
|
|
"## CLI install + usage", "## Docker usage"):
|
|
self.assertIn(section, notes)
|
|
expected = hashlib.sha256(self.artifact_bytes).hexdigest()
|
|
self.assertIn("sha256:" + expected, notes)
|
|
self.assertIn("chmod +x " + ARTIFACT_NAME, notes)
|
|
self.assertIn("--version", notes)
|
|
self.assertIn("README", notes)
|
|
self.assertIn("BROKER_ENV_PATH", notes)
|
|
self.assertIn("BROKER_ADMIN_PASS", notes)
|
|
self.assertIn("docker pull ", notes)
|
|
self.assertIn("/{}/{}:{}".format(OWNER, REPO_NAME, TAG), notes)
|
|
self.assertIn("-v broker-state:/state", notes)
|
|
|
|
|
|
class EnvSecretsLoaderTests(unittest.TestCase):
|
|
def setUp(self):
|
|
tmp = tempfile.TemporaryDirectory(prefix="coreci-envsecrets-")
|
|
self.addCleanup(tmp.cleanup)
|
|
self.root = tmp.name
|
|
|
|
def _write(self, relative, content):
|
|
path = os.path.join(self.root, relative)
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
with open(path, "w", encoding="utf-8") as handle:
|
|
handle.write(content)
|
|
return path
|
|
|
|
def test_parse_semantics_mirror_broker_parser(self):
|
|
fixture = (
|
|
"# comment\n"
|
|
"\n"
|
|
"GITEA_TOKEN='quoted-single'\n"
|
|
"OTHER_TOKEN=\"quoted double\"\n"
|
|
"SPACED = spaced value \n"
|
|
"PLAIN=unquoted\n"
|
|
"EMPTY=\n"
|
|
"ONE_QUOTE=\"\n"
|
|
"NO_EQUALS_SIGN\n"
|
|
"=\n"
|
|
)
|
|
path = self._write(".env.secrets", fixture)
|
|
self.assertEqual(
|
|
coreci_broker._parse_env_file(path),
|
|
env_secrets._parse_env_file(path))
|
|
values = env_secrets.load(self.root)
|
|
self.assertEqual("quoted-single", values["GITEA_TOKEN"])
|
|
self.assertEqual("quoted double", values["OTHER_TOKEN"])
|
|
self.assertEqual("spaced value", values["SPACED"])
|
|
self.assertEqual("unquoted", values["PLAIN"])
|
|
self.assertEqual("", values["EMPTY"])
|
|
self.assertEqual('"', values["ONE_QUOTE"])
|
|
self.assertNotIn("NO_EQUALS_SIGN", values)
|
|
|
|
def test_later_files_win_across_locations_and_patterns(self):
|
|
ladder = (
|
|
(".env", "GITEA_TOKEN=step1\n"),
|
|
(".env.secrets", "GITEA_TOKEN=step2\n"),
|
|
(".env.local", "GITEA_TOKEN=step3\n"), # root .env.* glob
|
|
(os.path.join(".ciagent", ".env"), "GITEA_TOKEN=step4\n"),
|
|
(os.path.join(".ciagent", ".env.secrets"), "GITEA_TOKEN=step5\n"),
|
|
(os.path.join(".ciagent", ".env.staging"), "GITEA_TOKEN=step6\n"),
|
|
)
|
|
for relative, content in ladder:
|
|
self._write(relative, content)
|
|
self.assertEqual(
|
|
content.partition("=")[2].strip(),
|
|
env_secrets.load(self.root)["GITEA_TOKEN"])
|
|
|
|
def test_missing_or_empty_token_raises_clear_error(self):
|
|
with self.assertRaises(env_secrets.SecretsError) as caught:
|
|
env_secrets.get_token(root=self.root)
|
|
self.assertIn("GITEA_TOKEN", str(caught.exception))
|
|
self._write(os.path.join(".ciagent", ".env.secrets"), "GITEA_TOKEN=\n")
|
|
with self.assertRaises(env_secrets.SecretsError):
|
|
env_secrets.get_token(root=self.root)
|
|
self._write(os.path.join(".ciagent", ".env.secrets"),
|
|
'GITEA_TOKEN=" "\n') # whitespace-only is empty too
|
|
with self.assertRaises(env_secrets.SecretsError):
|
|
env_secrets.get_token(root=self.root)
|
|
self._write(os.path.join(".ciagent", ".env.secrets"),
|
|
"GITEA_TOKEN='tok'\n")
|
|
self.assertEqual("tok", env_secrets.get_token(root=self.root))
|
|
|
|
def test_custom_variable_name(self):
|
|
self._write(".env", "OTHER_TOKEN=tok\n")
|
|
self.assertEqual("tok", env_secrets.get_token("OTHER_TOKEN", root=self.root))
|
|
with self.assertRaises(env_secrets.SecretsError) as caught:
|
|
env_secrets.get_token("GITEA_TOKEN", root=self.root)
|
|
self.assertIn("GITEA_TOKEN", str(caught.exception))
|
|
|
|
def test_default_root_is_repo_root(self):
|
|
self.assertEqual(
|
|
os.path.realpath(str(REPO)), os.path.realpath(env_secrets.default_root()))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |