broker/tests/test_release_upload.py

"""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()