coreci
da15bda6d2
---ci--- phase: 3 milestone: v0.2 status: complete requirements: covered: [REQ-21, REQ-22, REQ-23] partial: [] ---/ci---
548 lines
22 KiB
Python
548 lines
22 KiB
Python
"""v0.2 acceptance matrix (PLAN P3-01; REQ-21/REQ-22/REQ-23 acceptance).
|
|
|
|
Consolidates the milestone's acceptance as thin, high-value assertions that
|
|
the deliverable holds end-to-end. The deep per-feature logic stays in its own
|
|
files — test_build_cli, test_cli_packaging, test_publish_image,
|
|
test_release_upload, test_dockerfile_consistency — which run green in the
|
|
same cumulative suite (``python3 -m unittest discover -s tests``; suite
|
|
greenness is the runner's job, asserted by nobody in here). This file does
|
|
NOT re-run their cases; the stub roundtrips are smoke-asserted instead via
|
|
module presence + runnable TestSuite sizes (defaultTestLoader).
|
|
|
|
The matrix (binding order):
|
|
1. boot parity through the artifact — the built ``.pyz`` is booted as a real
|
|
subprocess child (``[sys.executable, artifact]``, never a bare
|
|
"python3") with an ephemeral bind + temp state, and v0.1 acceptance
|
|
parity must hold through the packaged path: admin pass printed exactly
|
|
once, ``broker.db`` + ``broker.audit.log`` 0600 in the state dir,
|
|
``/ui/login`` + ``/ui/dashboard`` 200, ``/static/coreci_broker_client.py``
|
|
byte-exact vs the repo file, one admin-login + agent-register +
|
|
state-put/get + secret-read flow (v0.1 request patterns), second boot
|
|
silent (asserted only after terminate + full drain — G-010);
|
|
2. reproducibility — two consecutive same-environment builds produce
|
|
byte-identical artifacts (identical sha256, G-009);
|
|
3. self-containment — the archive contains ONLY the staging allowlist
|
|
(broker + client + UI pages + shim + ``_cli_version.py``): no ``.env``,
|
|
no ``dist/``, no state, no tests;
|
|
4. stub publish roundtrip + stub release upload green — their test modules
|
|
import cleanly and expose runnable TestSuites of the expected size;
|
|
5. broker zero-delta — ``coreci_broker.py`` unchanged at its v0.1.5 count
|
|
of 1390 lines (AC14 under the D-016..D-018 cap of 1400) and carrying no
|
|
v0.2 packaging references (``_cli_version`` / the shim / the shim's
|
|
materialized-asset dir — frozen means untouched); plus the AC14
|
|
no-touch clause machine-checked against the ``v0.1.5`` tag when the
|
|
working tree carries it (skipped otherwise);
|
|
6. Dockerfile/.dockerignore consistency — module present with its full
|
|
TestCase set (it runs in the same suite).
|
|
|
|
The build happens ONCE per class (setUpClass) and every artifact/boot is
|
|
reused across methods where safe; the boot tests dominate and the whole
|
|
file is budgeted to stay well under 30s. Every child is terminated +
|
|
drained exactly once (idempotent ``stop`` via ``addCleanup``); every temp
|
|
dir is ``addCleanup``-registered. Builds use the ``scripts/build_cli.py``
|
|
API (never a shell path) into ``TemporaryDirectory`` dist dirs — no
|
|
committed ``dist/``, no state, no live-forge contact.
|
|
"""
|
|
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import select
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
import urllib.error
|
|
import urllib.request
|
|
import uuid as uuidlib
|
|
import zipfile
|
|
|
|
REPO = os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir))
|
|
TESTS_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
BUILD_CLI_PATH = os.path.join(REPO, "scripts", "build_cli.py")
|
|
BROKER_SOURCE_PATH = os.path.join(REPO, "coreci_broker.py")
|
|
REPO_CLIENT_PATH = os.path.join(REPO, "coreci_broker_client.py")
|
|
VERSION = "v0.2.3" # the milestone release this acceptance gates (P3-05 ship)
|
|
|
|
ADMIN_USER = "admin"
|
|
ADMIN_PASS_PREFIX = "BROKER_ADMIN_PASS="
|
|
LISTENING_PREFIX = "coreci-broker listening on http://"
|
|
STATE_FILE_NAMES = (".env", "broker.db", "broker.audit.log")
|
|
|
|
LOGIN_PATH = "/api/broker/auth/login"
|
|
APPS_PATH = "/api/broker/app_registrations"
|
|
USERS_PATH = "/api/broker/user_allowlist"
|
|
REGISTER_PATH = "/api/broker/agents/register"
|
|
AGENT_STATE_PATH = "/api/broker/agents/proxmox/state"
|
|
SECRETS_PATH = "/api/broker/secrets/coreci/proxmox"
|
|
CLIENT_STATIC_PATH = "/static/coreci_broker_client.py"
|
|
UI_LOGIN_PATH = "/ui/login"
|
|
UI_DASHBOARD_PATH = "/ui/dashboard"
|
|
|
|
PROXMOX_PATH = "coreci/proxmox"
|
|
PROXMOX_SECRET = {"url": "https://pve.coreci.dev:8006", "token": "pve-tok"}
|
|
|
|
BOOT_DEADLINE_SECONDS = 30.0
|
|
POLL_SECONDS = 0.1
|
|
DRAIN_TIMEOUT_SECONDS = 20.0
|
|
HTTP_TIMEOUT_SECONDS = 10
|
|
GIT_TIMEOUT_SECONDS = 30
|
|
|
|
# Frozen-broker sources (PLAN: no task edits these in v0.2) — used only by
|
|
# the AC14 no-touch clause check, never modified.
|
|
FROZEN_SOURCES = ("coreci_broker.py", "coreci_broker_client.py",
|
|
"coreci_broker_ui", "reference_tools")
|
|
# v0.2 packaging artifacts the frozen broker must know nothing about.
|
|
FROZEN_MARKER_STRINGS = ("_cli_version", "cli/__main__", "coreci_broker_assets")
|
|
|
|
# The v0.2 test surface consolidated here — presence/count floors only
|
|
# (the roundtrips themselves run green in their own files, same suite).
|
|
STUB_MODULE_MIN_CASES = {
|
|
"test_publish_image.py": 9, # stub publish roundtrip
|
|
"test_release_upload.py": 15, # stub release upload + env-secrets
|
|
"test_dockerfile_consistency.py": 7,
|
|
}
|
|
|
|
|
|
def _load_build_cli():
|
|
spec = importlib.util.spec_from_file_location(
|
|
"acceptance_build_cli", BUILD_CLI_PATH)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
build_cli = _load_build_cli()
|
|
|
|
|
|
def _sha256(path):
|
|
digest = hashlib.sha256()
|
|
with open(path, "rb") as handle:
|
|
for chunk in iter(lambda: handle.read(65536), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
class _BrokerChild:
|
|
"""A subprocess child booted from the built zipapp artifact.
|
|
|
|
The merged stdout/stderr pipe is pumped via ``select`` + raw ``os.read``
|
|
under an explicit deadline until the broker announces its ephemeral
|
|
listen address. ``stop()`` is the G-010 kill/drain protocol:
|
|
``terminate()`` then a full ``communicate()`` drain (kill fallback on
|
|
drain timeout). It is idempotent and registered with ``addCleanup`` at
|
|
construction, so no child leaks even when an assertion trips mid-test.
|
|
Silence and pass-once assertions are made ONLY on ``full_output()``
|
|
after ``stop()``.
|
|
"""
|
|
|
|
def __init__(self, testcase, artifact, env, cwd):
|
|
self.testcase = testcase
|
|
self.artifact = artifact
|
|
self.proc = subprocess.Popen(
|
|
[sys.executable, artifact],
|
|
env=env, cwd=cwd,
|
|
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
|
)
|
|
self.lines = []
|
|
self.partial = b""
|
|
self.drained = b""
|
|
self.base_url = None
|
|
self._eof = False
|
|
self._stopped = False
|
|
testcase.addCleanup(self.stop)
|
|
self._await_listening(BOOT_DEADLINE_SECONDS)
|
|
|
|
def _pump(self, timeout):
|
|
if self._eof:
|
|
return False
|
|
fd = self.proc.stdout.fileno()
|
|
ready, _, _ = select.select([fd], [], [], timeout)
|
|
if not ready:
|
|
return True
|
|
try:
|
|
chunk = os.read(fd, 4096)
|
|
except OSError:
|
|
chunk = b""
|
|
if not chunk:
|
|
self._eof = True
|
|
return False
|
|
self.partial += chunk
|
|
while b"\n" in self.partial:
|
|
raw, self.partial = self.partial.split(b"\n", 1)
|
|
self.lines.append(raw.decode("utf-8", "replace"))
|
|
return True
|
|
|
|
def _await_listening(self, deadline_seconds):
|
|
deadline = time.monotonic() + deadline_seconds
|
|
cursor = 0
|
|
while time.monotonic() < deadline:
|
|
alive = self._pump(POLL_SECONDS)
|
|
while cursor < len(self.lines):
|
|
line = self.lines[cursor]
|
|
cursor += 1
|
|
if line.startswith(LISTENING_PREFIX):
|
|
self.base_url = line.split(" on ", 1)[1].strip()
|
|
return
|
|
if not alive:
|
|
break
|
|
self.stop()
|
|
self.testcase.fail(
|
|
"packaged broker did not announce %r within %.0fs "
|
|
"(artifact=%s exit=%s)\n--- captured child output ---\n%s"
|
|
% (LISTENING_PREFIX, deadline_seconds, self.artifact,
|
|
self.proc.returncode, self.full_output())
|
|
)
|
|
|
|
def stop(self):
|
|
if self._stopped:
|
|
return
|
|
self._stopped = True
|
|
if self.proc.poll() is None:
|
|
self.proc.terminate()
|
|
try:
|
|
remainder, _ = self.proc.communicate(timeout=DRAIN_TIMEOUT_SECONDS)
|
|
except subprocess.TimeoutExpired:
|
|
self.proc.kill()
|
|
remainder, _ = self.proc.communicate()
|
|
self.drained = remainder or b""
|
|
|
|
def full_output(self):
|
|
text = "\n".join(self.lines)
|
|
tail = self.partial + self.drained
|
|
if tail:
|
|
text += "\n" + tail.decode("utf-8", "replace")
|
|
return text
|
|
|
|
def printed_admin_pass(self):
|
|
values = [
|
|
line.split("=", 1)[1]
|
|
for line in self.lines
|
|
if line.startswith(ADMIN_PASS_PREFIX)
|
|
]
|
|
self.testcase.assertEqual(
|
|
1, len(values),
|
|
"expected exactly one %s line, got: %r" % (ADMIN_PASS_PREFIX, values)
|
|
)
|
|
return values[0]
|
|
|
|
|
|
class AcceptanceV02Tests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.build_root = tempfile.TemporaryDirectory(prefix="coreci-acc-v02-")
|
|
cls.addClassCleanup(cls.build_root.cleanup)
|
|
cls.artifact = build_cli.build(
|
|
VERSION, repo=REPO, dist_dir=cls.build_root.name
|
|
)
|
|
cls.artifact_sha256 = _sha256(cls.artifact)
|
|
|
|
# -- harness helpers --------------------------------------------------
|
|
|
|
def _temp_dir(self):
|
|
handle = tempfile.TemporaryDirectory(prefix="coreci-acc-v02-")
|
|
self.addCleanup(handle.cleanup)
|
|
return handle.name
|
|
|
|
def _child_env(self, **overrides):
|
|
"""A child env with no inherited BROKER_* keys, then overrides."""
|
|
env = {
|
|
key: value
|
|
for key, value in os.environ.items()
|
|
if not key.startswith("BROKER_")
|
|
}
|
|
env.update(overrides)
|
|
return env
|
|
|
|
def _state_env(self, state_dir):
|
|
"""Ephemeral bind + temp state + a seeded file creds source."""
|
|
creds_path = os.path.join(state_dir, "creds.json")
|
|
with open(creds_path, "w", encoding="utf-8") as handle:
|
|
json.dump({PROXMOX_PATH: PROXMOX_SECRET}, handle)
|
|
return self._child_env(
|
|
BROKER_BIND="127.0.0.1:0",
|
|
BROKER_ENV_PATH=os.path.join(state_dir, ".env"),
|
|
BROKER_DB_PATH=os.path.join(state_dir, "broker.db"),
|
|
BROKER_AUDIT_LOG_PATH=os.path.join(state_dir, "broker.audit.log"),
|
|
BROKER_CREDS_SOURCE="file",
|
|
BROKER_CREDS_FILE=creds_path,
|
|
)
|
|
|
|
def _boot(self, state_dir):
|
|
"""Boot the artifact against state_dir; CWD is a separate temp dir."""
|
|
return _BrokerChild(
|
|
self, self.artifact, self._state_env(state_dir), self._temp_dir()
|
|
)
|
|
|
|
def _request(self, base, method, path, payload=None, headers=None):
|
|
data = None if payload is None else json.dumps(payload).encode("utf-8")
|
|
request = urllib.request.Request(base + path, data=data, method=method)
|
|
for key, value in (headers or {}).items():
|
|
request.add_header(key, value)
|
|
try:
|
|
with urllib.request.urlopen(
|
|
request, timeout=HTTP_TIMEOUT_SECONDS
|
|
) as response:
|
|
return response.status, response.read(), response.headers
|
|
except urllib.error.HTTPError as error:
|
|
return error.code, error.read(), error.headers
|
|
|
|
def _request_json(self, base, method, path, payload=None, headers=None):
|
|
status, body, response_headers = self._request(
|
|
base, method, path, payload, headers
|
|
)
|
|
return status, json.loads(body.decode("utf-8")), response_headers
|
|
|
|
def _admin_login(self, base, password):
|
|
"""v0.1 pattern: POST login, assert the broker_session cookie."""
|
|
status, body, headers = self._request_json(
|
|
base, "POST", LOGIN_PATH,
|
|
{"username": ADMIN_USER, "password": password},
|
|
)
|
|
self.assertEqual(200, status)
|
|
self.assertEqual({"ok": True, "username": ADMIN_USER}, body)
|
|
raw_cookie = headers.get("Set-Cookie")
|
|
self.assertTrue(raw_cookie)
|
|
self.assertTrue(raw_cookie.startswith("broker_session="))
|
|
self.assertIn("HttpOnly", raw_cookie)
|
|
self.assertIn("SameSite=Lax", raw_cookie)
|
|
return {"Cookie": raw_cookie.split(";", 1)[0]}
|
|
|
|
def _load_test_module(self, file_name):
|
|
"""Import a sibling test module cleanly (by path, any invocation)."""
|
|
path = os.path.join(TESTS_DIR, file_name)
|
|
stem = os.path.splitext(file_name)[0]
|
|
spec = importlib.util.spec_from_file_location(
|
|
"acceptance_probe_" + stem, path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
def _runnable_case_count(self, module):
|
|
"""Flatten defaultTestLoader's suite, counting runnable TestCases."""
|
|
suite = unittest.defaultTestLoader.loadTestsFromModule(module)
|
|
stack, count = [suite], 0
|
|
while stack:
|
|
item = stack.pop()
|
|
if isinstance(item, unittest.TestSuite):
|
|
stack.extend(item)
|
|
else:
|
|
self.assertIsInstance(item, unittest.TestCase)
|
|
count += 1
|
|
return count
|
|
|
|
# -- matrix 1: boot parity through the artifact ----------------------
|
|
|
|
def test_boot_parity_through_artifact_second_boot_silent(self):
|
|
state_dir = self._temp_dir()
|
|
child = self._boot(state_dir)
|
|
base = child.base_url
|
|
printed_pass = child.printed_admin_pass()
|
|
|
|
# liveness + full UI served from the packaged artifact
|
|
status, body, _ = self._request_json(base, "GET", "/healthz")
|
|
self.assertEqual(200, status)
|
|
self.assertEqual({"ok": True}, body)
|
|
for ui_path in (UI_LOGIN_PATH, UI_DASHBOARD_PATH):
|
|
status, body, headers = self._request(base, "GET", ui_path)
|
|
self.assertEqual(200, status, ui_path)
|
|
self.assertTrue(
|
|
headers.get("Content-Type", "").startswith("text/html"),
|
|
ui_path,
|
|
)
|
|
# the tool-author surface: client module byte-exact vs the repo file
|
|
status, served, _ = self._request(base, "GET", CLIENT_STATIC_PATH)
|
|
self.assertEqual(200, status)
|
|
with open(REPO_CLIENT_PATH, "rb") as handle:
|
|
self.assertEqual(handle.read(), served)
|
|
|
|
# one consolidated v0.1 request-pattern flow:
|
|
# admin login -> app + user allowlist -> agent register -> state
|
|
# put/get -> secret read
|
|
cookie = self._admin_login(base, printed_pass)
|
|
app_uuid = str(uuidlib.uuid4())
|
|
status, body, _ = self._request_json(
|
|
base, "POST", APPS_PATH,
|
|
{"uuid": app_uuid, "name": "acceptance-tool"}, headers=cookie,
|
|
)
|
|
self.assertEqual(201, status)
|
|
self.assertEqual({"uuid": app_uuid, "status": "active"}, body)
|
|
status, body, _ = self._request_json(
|
|
base, "POST", USERS_PATH,
|
|
{"user_id": "ceo", "email": "ceo@coreci.dev"}, headers=cookie,
|
|
)
|
|
self.assertEqual(201, status)
|
|
status, body, _ = self._request_json(
|
|
base, "POST", REGISTER_PATH,
|
|
{"agent_id": "proxmox", "app_registration_uuid": app_uuid,
|
|
"user_id": "ceo", "tags": ["coreci"]},
|
|
)
|
|
self.assertEqual(201, status)
|
|
app_header = {"Authorization": "App %s %s" % (app_uuid, body["token"])}
|
|
state_document = {"milestone": "v0.2", "matrix": "acceptance"}
|
|
status, body, _ = self._request_json(
|
|
base, "PUT", AGENT_STATE_PATH, state_document, headers=app_header
|
|
)
|
|
self.assertEqual(200, status)
|
|
self.assertEqual({"ok": True}, body)
|
|
status, body, _ = self._request_json(
|
|
base, "GET", AGENT_STATE_PATH, headers=app_header
|
|
)
|
|
self.assertEqual(200, status)
|
|
self.assertEqual(state_document, body)
|
|
status, body, _ = self._request_json(
|
|
base, "GET", SECRETS_PATH, headers=app_header
|
|
)
|
|
self.assertEqual(200, status)
|
|
self.assertEqual({"path", "data", "version", "updated_at"}, set(body))
|
|
self.assertEqual(PROXMOX_PATH, body["path"])
|
|
self.assertEqual(PROXMOX_SECRET, body["data"])
|
|
|
|
# v0.1 state-file parity: 0600 in the state dir
|
|
for name in STATE_FILE_NAMES:
|
|
path = os.path.join(state_dir, name)
|
|
self.assertTrue(os.path.isfile(path), path)
|
|
self.assertEqual(0o600, os.stat(path).st_mode & 0o777, path)
|
|
|
|
# pass printed exactly once + pepper never printed — asserted on
|
|
# the FULLY drained output after terminate (G-010)
|
|
pepper = None
|
|
with open(os.path.join(state_dir, ".env"), "r", encoding="utf-8") as handle:
|
|
for line in handle:
|
|
if line.startswith("BROKER_INSTANCE_PEPPER="):
|
|
pepper = line.split("=", 1)[1].strip()
|
|
self.assertIsNotNone(pepper)
|
|
child.stop()
|
|
first_output = child.full_output()
|
|
self.assertEqual(1, first_output.count(ADMIN_PASS_PREFIX), first_output)
|
|
self.assertEqual(1, first_output.count(printed_pass))
|
|
self.assertNotIn(pepper, first_output)
|
|
|
|
# SECOND boot on the same state dir: silent after terminate + drain
|
|
second = self._boot(state_dir)
|
|
# the SAME pass still logs in: state persisted across the reboot
|
|
self._admin_login(second.base_url, printed_pass)
|
|
second.stop()
|
|
second_output = second.full_output()
|
|
self.assertIn(LISTENING_PREFIX, second_output)
|
|
self.assertNotIn(ADMIN_PASS_PREFIX, second_output)
|
|
self.assertNotIn(printed_pass, second_output)
|
|
|
|
# -- matrix 2: reproducibility ----------------------------------------
|
|
|
|
def test_reproducibility_two_consecutive_builds_identical_sha256(self):
|
|
digests = []
|
|
for _ in range(2):
|
|
with tempfile.TemporaryDirectory(prefix="coreci-acc-repro-") as dist_dir:
|
|
artifact = build_cli.build(
|
|
VERSION, repo=REPO, dist_dir=dist_dir
|
|
)
|
|
digests.append(_sha256(artifact))
|
|
self.assertNotEqual(digests[0], "0" * 64)
|
|
self.assertEqual(digests[0], digests[1])
|
|
# and the shared setUpClass build matches too (same environment)
|
|
self.assertEqual(self.artifact_sha256, digests[0])
|
|
print("v0.2 acceptance two-build sha256: {}".format(digests[0]))
|
|
|
|
# -- matrix 3: self-containment ---------------------------------------
|
|
|
|
def test_self_containment_archive_is_exactly_the_allowlist(self):
|
|
ui_root = os.path.join(REPO, "coreci_broker_ui")
|
|
expected = {
|
|
"__main__.py",
|
|
"_cli_version.py",
|
|
"coreci_broker.py",
|
|
"coreci_broker_client.py",
|
|
"coreci_broker_ui/",
|
|
}
|
|
expected.update(
|
|
"coreci_broker_ui/%s" % name
|
|
for name in os.listdir(ui_root)
|
|
if os.path.isfile(os.path.join(ui_root, name))
|
|
)
|
|
with zipfile.ZipFile(self.artifact) as archive:
|
|
names = set(archive.namelist())
|
|
# ONLY the allowlist members — nothing more, nothing less
|
|
self.assertEqual(expected, names)
|
|
# no .env, no dist/, no state, no tests, no bytecode can ride along
|
|
for name in names:
|
|
for forbidden in (".env", "dist", "broker.db", "broker.audit.log",
|
|
"test", "__pycache__", ".pyc"):
|
|
self.assertNotIn(forbidden, name, name)
|
|
# per-release artifact name (D-020) for the milestone release
|
|
self.assertEqual(
|
|
build_cli.artifact_name(VERSION),
|
|
os.path.basename(self.artifact),
|
|
)
|
|
|
|
# -- matrix 4: stub publish roundtrip + stub release upload green ------
|
|
|
|
def test_stub_publish_roundtrip_module_green_in_suite(self):
|
|
module = self._load_test_module("test_publish_image.py")
|
|
self.assertTrue(hasattr(module, "PublishImageTests"))
|
|
count = self._runnable_case_count(module)
|
|
self.assertGreaterEqual(
|
|
count, STUB_MODULE_MIN_CASES["test_publish_image.py"]
|
|
)
|
|
|
|
def test_stub_release_upload_module_green_in_suite(self):
|
|
module = self._load_test_module("test_release_upload.py")
|
|
self.assertTrue(hasattr(module, "ReleaseUploadTests"))
|
|
self.assertTrue(hasattr(module, "EnvSecretsLoaderTests"))
|
|
count = self._runnable_case_count(module)
|
|
self.assertGreaterEqual(
|
|
count, STUB_MODULE_MIN_CASES["test_release_upload.py"]
|
|
)
|
|
|
|
# -- matrix 6: Dockerfile/.dockerignore consistency ---------------------
|
|
|
|
def test_dockerfile_consistency_module_green_in_suite(self):
|
|
module = self._load_test_module("test_dockerfile_consistency.py")
|
|
self.assertTrue(hasattr(module, "DockerfileConsistencyTests"))
|
|
count = self._runnable_case_count(module)
|
|
self.assertGreaterEqual(
|
|
count, STUB_MODULE_MIN_CASES["test_dockerfile_consistency.py"]
|
|
)
|
|
|
|
# -- matrix 5: broker zero-delta (AC14, frozen means untouched) --------
|
|
|
|
def test_broker_zero_delta_1390_lines_no_packaging_references(self):
|
|
with open(BROKER_SOURCE_PATH, "r", encoding="utf-8") as handle:
|
|
line_count = sum(1 for _ in handle)
|
|
# exact v0.1.5 count, under the D-016..D-018 cap of 1400 (AC14)
|
|
self.assertEqual(1390, line_count)
|
|
self.assertLess(line_count, 1400)
|
|
with open(BROKER_SOURCE_PATH, "r", encoding="utf-8") as handle:
|
|
source = handle.read()
|
|
# frozen means untouched: no v0.2 packaging reference leaked into it
|
|
for marker in FROZEN_MARKER_STRINGS:
|
|
self.assertNotIn(marker, source)
|
|
|
|
def test_broker_untouched_by_any_commit_since_v0_1_5(self):
|
|
"""The AC14 no-touch clause, machine-checked when the tag exists."""
|
|
try:
|
|
verify = subprocess.run(
|
|
["git", "rev-parse", "-q", "--verify", "v0.1.5^{commit}"],
|
|
cwd=REPO, capture_output=True, text=True,
|
|
timeout=GIT_TIMEOUT_SECONDS,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
self.skipTest("git unavailable in this environment")
|
|
if verify.returncode != 0:
|
|
self.skipTest("v0.1.5 tag not present in this working tree")
|
|
log = subprocess.run(
|
|
["git", "log", "--oneline", "v0.1.5..HEAD", "--"] + list(FROZEN_SOURCES),
|
|
cwd=REPO, capture_output=True, text=True,
|
|
timeout=GIT_TIMEOUT_SECONDS,
|
|
)
|
|
self.assertEqual(0, log.returncode, log.stderr)
|
|
self.assertEqual(
|
|
"", log.stdout.strip(),
|
|
"v0.2 commits touched frozen broker sources:\n" + log.stdout
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |