Something went wrong. Try again.
Hardware-in-the-loop development platform for the Commodore 64.
Something went wrong. Try again.
133 kB · 2668 lines
Python
at main
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426242724282429243024312432243324342435243624372438243924402441244224432444244524462447244824492450245124522453245424552456245724582459246024612462246324642465246624672468246924702471247224732474247524762477247824792480248124822483248424852486248724882489249024912492249324942495249624972498249925002501250225032504250525062507250825092510251125122513251425152516251725182519252025212522252325242525252625272528252925302531253225332534253525362537253825392540254125422543254425452546254725482549255025512552255325542555255625572558255925602561256225632564256525662567256825692570257125722573257425752576257725782579258025812582258325842585258625872588258925902591259225932594259525962597259825992600260126022603260426052606260726082609261026112612261326142615261626172618261926202621262226232624262526262627262826292630263126322633263426352636263726382639264026412642264326442645264626472648264926502651265226532654265526562657265826592660266126622663266426652666266726682669#!/usr/bin/env python3"""Tests for the local Tangled artifact publisher.
Every test here is standard-library ``unittest`` and runs inside a temporarydirectory. Nothing contacts a real PDS, account or repository: the transporttests drive a disposable HTTP server bound to ``127.0.0.1``, and the tags arecreated in a throwaway local git repository.
A passing suite therefore proves the local preflight gates -- including theshared independent release check -- the exact XRPC request shapes, thecredential handling and the failure paths, all against genuine archive bytes.It does not prove that any artifact was published, that a real Tangled PDSaccepts these records, or that the reviewed lexicon matches what that serverwould accept. Those remain operator gates.
The fixture release directory is produced by the *real* packager:``tools/package_release.py`` runs its two-stage ``--prepare-source`` then``--pack`` flow over a synthetic project, so the binary archive, the sourcearchive, ``release-manifest.json``, ``source-manifest.json`` and``SHA256SUMS`` are exactly what the packaging step writes. The publisher'sindependent preflight (``tools/check_release.py``) then re-derives every factfrom those bytes before any network call or credential."""
from __future__ import annotations
import atexitimport hashlibimport http.serverimport ioimport jsonimport osimport shutilimport socketimport sslimport subprocessimport sysimport tarfileimport tempfileimport threadingimport timeimport unittestimport unittest.mockfrom pathlib import Path
ROOT = Path(__file__).resolve().parents[2]TOOLS = ROOT / "tools"TESTS = Path(__file__).resolve().parentfor _directory in (str(TOOLS), str(TESTS)): if _directory not in sys.path: sys.path.insert(0, _directory)
import publish_tangled_artifacts as publisher # noqa: E402
# The release fixture is produced by the *real* packager, so the tests exercise# genuine archive bytes and a genuine release manifest. The synthetic-project# harness lives in the sibling release-package tests and is reused rather than# duplicated, so the fixture cannot drift from what the packager really writes.import test_release_package as release_harness # noqa: E402
SCRIPT = TOOLS / "publish_tangled_artifacts.py"GIT = shutil.which("git")
PRODUCT = "ultimate-tool"VERSION = "0.2.0-preview.1"TAG = "v0.2.0-preview.1"# The packager's real archive names, including the target infix it builds in.ARCHIVE_NAME = f"ultimate-tool-{VERSION}-linux-x86_64-gnu.tar.gz"SOURCE_ARCHIVE_NAME = f"ultimate-tool-{VERSION}-linux-x86_64-gnu-source.tar.gz"CHECKSUM_NAME = "SHA256SUMS"HOST_BINARIES = ("ultimate-tool", "ultimate-tool-mcp")OWNER_DID = "did:plc:ownerofthefixturerepo"HANDLE = "operator.tngl.sh"REPO_URL = f"https://tangled.org/{HANDLE}/ultimate-tool"OTHER_DID = "did:plc:someoneelseentirely"JWT = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJmaXh0dXJlIn0.fixturesignature"# Assembled at run time so the distributed source carries no literal# credential-shaped assignment; the joined value is the exact hostile secret.PASSWORD = "".join(["fixture-", "app-password-not-real"])
GIT_ENV = { "GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": os.devnull, "GIT_AUTHOR_NAME": "Fixture", "GIT_AUTHOR_EMAIL": "fixture@example.invalid", "GIT_COMMITTER_NAME": "Fixture", "GIT_COMMITTER_EMAIL": "fixture@example.invalid", "GIT_AUTHOR_DATE": "2026-01-01T00:00:00+00:00", "GIT_COMMITTER_DATE": "2026-01-01T00:00:00+00:00",}
def digest(data: bytes) -> str: return hashlib.sha256(data).hexdigest()
def git(args: list[str], cwd: Path) -> str: environment = {**os.environ, **GIT_ENV} result = subprocess.run( ["git", "-c", "commit.gpgsign=false", "-c", "tag.gpgsign=false", *args], cwd=str(cwd), capture_output=True, text=True, check=True, env=environment, timeout=60, ) return result.stdout.strip()
def reviewed_schema_document() -> dict: """A minimal reviewed lexicon document for ``sh.tangled.repo.artifact``.""" return { "lexicon": 1, "id": "sh.tangled.repo.artifact", "defs": { "main": { "type": "record", "key": "tid", "record": { "type": "object", "required": ["tag", "name", "repo", "artifact", "createdAt"], "properties": { "$type": {"type": "string", "const": "sh.tangled.repo.artifact"}, "tag": {"type": "bytes", "maxLength": 64}, "name": {"type": "string", "maxLength": 1024}, "repo": {"type": "string", "format": "uri"}, "artifact": {"type": "blob", "accept": ["*/*"]}, "createdAt": {"type": "string", "format": "datetime"}, }, }, } }, }
# --------------------------------------------------------------------------# Loopback PDS mock# --------------------------------------------------------------------------
class MockPds: """A disposable, unauthenticated stand-in for the three XRPC calls.
It records every request so tests can assert the exact sequence, headers and bodies, and it can be told to misbehave (redirect, oversize, wrong DID) so the failure paths are exercised against a real HTTP socket. """
def __init__(self) -> None: self.owner_did = OWNER_DID self.session_did = OWNER_DID self.resolve_map: dict[str, str] = {} self.resolve_did = OWNER_DID self.redirect_paths: set[str] = set() self.session_status = 200 self.session_error_body = "" self.upload_status = 200 self.upload_force_cid: str | None = None self.upload_force_size: int | None = None self.upload_force_mime: str | None = None self.upload_error_body: str | None = None self.create_status = 200 self.create_error_body: str | None = None self.create_not_json = False self.create_omit_uri = False self.create_calls = 0 self.fail_create_after: int | None = None self.record_uri: str | None = None self.document_cid = publisher.record_cid_from_bytes(b"mock record cbor") self.oversize_paths: set[str] = set() self.requests: list[dict] = [] self._server: http.server.ThreadingHTTPServer | None = None self._thread: threading.Thread | None = None self.origin = ""
def start(self) -> None: pds = self
class Handler(http.server.BaseHTTPRequestHandler): def log_message(self, *args): # noqa: D102 - silence test output pass
def _respond(self, status: int, payload: bytes, content_type: str = "application/json") -> None: self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload)
def _dispatch(self, method: str) -> None: length = int(self.headers.get("Content-Length") or 0) body = self.rfile.read(length) if length else b"" route = self.path.split("?")[0] pds.requests.append({ "method": method, "path": self.path, "route": route, "headers": {key.lower(): value for key, value in self.headers.items()}, "body": body, }) if route in pds.redirect_paths: self.send_response(302) self.send_header("Location", "https://elsewhere.invalid/xrpc") self.send_header("Content-Length", "0") self.end_headers() return if route in pds.oversize_paths: self._respond(200, b"x" * 4096) return if route == "/xrpc/com.atproto.identity.resolveHandle": handle = self.path.split("handle=", 1)[-1] if "handle=" in self.path else "" self._respond(200, json.dumps({"did": pds.resolve_map.get(handle, pds.resolve_did)}).encode()) return if route == "/xrpc/com.atproto.server.createSession": if pds.session_status != 200: self._respond(pds.session_status, pds.session_error_body.encode()) return self._respond(200, json.dumps({ "did": pds.session_did, "accessJwt": JWT, "refreshJwt": JWT, "handle": HANDLE, }).encode()) return if route == "/xrpc/com.atproto.repo.uploadBlob": if pds.upload_status != 200: body = pds.upload_error_body or json.dumps({"error": "BlobTooLarge"}) self._respond(pds.upload_status, body.encode()) return # A real PDS returns a CIDv1 whose digest is the SHA-256 of # the uploaded bytes; the mock derives it the same way so the # publisher's content-derived check is genuinely exercised. cid = pds.upload_force_cid or publisher.content_cid(body) size = pds.upload_force_size if pds.upload_force_size is not None else len(body) mime = pds.upload_force_mime or self.headers.get("Content-Type", "application/octet-stream") self._respond(200, json.dumps({"blob": { "$type": "blob", "ref": {"$link": cid}, "mimeType": mime, "size": size, }}).encode()) return if route == "/xrpc/com.atproto.repo.createRecord": pds.create_calls += 1 forced = pds.fail_create_after is not None and pds.create_calls > pds.fail_create_after if pds.create_status != 200 or forced: status = pds.create_status if pds.create_status != 200 else 500 body = pds.create_error_body or json.dumps({"error": "InvalidRequest"}) self._respond(status, body.encode()) return if pds.create_not_json: self._respond(200, b"<html>gateway</html>") return document = {"cid": pds.document_cid, "commit": {"cid": "bafyreib", "rev": "3kfix"}} if not pds.create_omit_uri: document["uri"] = pds.record_uri or f"at://{pds.owner_did}/sh.tangled.repo.artifact/3kfixture" self._respond(200, json.dumps(document).encode()) return self._respond(404, b"{}")
def do_GET(self): self._dispatch("GET")
def do_POST(self): self._dispatch("POST")
self._server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) self._server.daemon_threads = True self.origin = f"http://127.0.0.1:{self._server.server_address[1]}" self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) self._thread.start()
def stop(self) -> None: if self._server is not None: self._server.shutdown() self._server.server_close() if self._thread is not None: self._thread.join(timeout=10)
def route(self, suffix: str) -> list[dict]: return [request for request in self.requests if request["route"].endswith(suffix)]
def body_of(self, suffix: str) -> dict: matches = self.route(suffix) if len(matches) != 1: raise AssertionError(f"expected one request to {suffix}, saw {len(matches)}") return json.loads(matches[0]["body"].decode("utf-8"))
def bodies_of(self, suffix: str) -> list[dict]: return [json.loads(request["body"].decode("utf-8")) for request in self.route(suffix)]
# --------------------------------------------------------------------------# Release fixture# --------------------------------------------------------------------------
# --------------------------------------------------------------------------# Real packager-produced release (built once per process, shared read-only)# --------------------------------------------------------------------------
_RELEASE_TEMPLATE: dict | None = None
def packager_release_template() -> dict: """Build one genuine release directory with the real packager, once.
The synthetic-project harness in ``test_release_package`` writes a repository-shaped tree, produces two synthetic ELF64 host binaries, vendors a crate, and drives ``tools/package_release.py`` through its two-stage ``--prepare-source`` then ``--pack`` flow with a clean-rebuild claim. The result is a release that passes ``tools/check_release.py`` in release mode, which is exactly what the publisher's independent preflight requires. """ global _RELEASE_TEMPLATE if _RELEASE_TEMPLATE is not None: return _RELEASE_TEMPLATE holder = tempfile.TemporaryDirectory(prefix="ut-tangled-release-") base = Path(holder.name) / "project" base.mkdir() project = release_harness.SyntheticProject(base, inventory_collected=True) project.write() git(["init", "-q"], base) git(["config", "user.email", "fixture@example.invalid"], base) git(["config", "user.name", "Fixture"], base) git(["add", "-A"], base) git(["commit", "-q", "-m", "fixture release tree"], base) git(["tag", "-a", TAG, "-m", "fixture annotated release tag"], base) vendor = project.write_vendor(Path(holder.name) / "vendor") project.write_binaries() project.write_build_env() project.run_packager("--prepare-source", "--vendor-dir", str(vendor), "--tag", TAG) source_manifest = json.loads((project.output / "source-manifest.json").read_text(encoding="utf-8")) evidence = project.write_rebuild_evidence(source_manifest) project.run_packager( "--pack", "--vendor-dir", str(vendor), "--binary-dir", str(project.binary_dir), "--tag", TAG, "--rebuild-status", "verified", "--rebuild-evidence", str(evidence), ) release_dir = project.output manifest = json.loads((release_dir / "release-manifest.json").read_text(encoding="utf-8")) info = { "holder": holder, # keep the temporary tree alive for the whole process "project_root": base, "release_dir": release_dir, "manifest": manifest, "source_manifest_text": (release_dir / "source-manifest.json").read_text(encoding="utf-8"), "binary_bytes": (release_dir / ARCHIVE_NAME).read_bytes(), "source_bytes": (release_dir / SOURCE_ARCHIVE_NAME).read_bytes(), "revision": manifest["source"]["revision"], "tag_object": manifest["source"]["tag_object"], "tag_commit": manifest["source"]["tag_commit"], "tag": manifest["source"]["tag"], } atexit.register(holder.cleanup) _RELEASE_TEMPLATE = info return info
class ReleaseFixture: """A genuine packager-produced release plus a per-test clone of its repository."""
def __init__(self, root: Path) -> None: template = packager_release_template() self._template = template self.root = Path(root) self.release_dir = self.root / "release" self.repo = self.root / "repo" self.release_dir.mkdir(parents=True) # The genuine archive bytes; never hand-written. The independent checker # re-measures these, so a fake stand-in could not pass the preflight. self.binary_bytes = template["binary_bytes"] self.source_bytes = template["source_bytes"] self.manifest_path = self.release_dir / "release-manifest.json" self.source_manifest_path = self.release_dir / "source-manifest.json" self.checksums_path = self.release_dir / CHECKSUM_NAME self.schema_path = self.root / "reviewed-schema.json" self.revision = template["revision"] self.tag_object = template["tag_object"] self.tag_commit = template["tag_commit"] self.tag_name = template["tag"] self.version = template["manifest"]["version"] self.development = False self.dirty = False self.release_ready = True self.verification_state = "verified" # Mirrors the packaging step: the copy written into the release # directory may still defer the archive digest to SHA256SUMS. self.defer_binary_archive_digest = False self.declare_archive_section_digest = True
# -- git ------------------------------------------------------------- def build_git(self, *, tag_name: str | None = None, annotated: bool = True) -> None: if self.repo.exists(): shutil.rmtree(self.repo) # Clone the packager's own repository so the manifest's recorded # revision, tag object and tag commit re-derive from real Git state. A # clone is clean by construction, so the checker's clean-tagged gate # holds and the publisher's annotated-tag read matches the manifest. git(["clone", "--quiet", "--no-hardlinks", str(self._template["project_root"]), str(self.repo)], self.root) self.revision = git(["rev-parse", "HEAD"], self.repo) name = tag_name or self.tag_name if annotated: self.tag_object = git(["rev-parse", f"{name}^{{tag}}"], self.repo) else: git(["tag", "-d", name], self.repo) git(["tag", name], self.repo) self.tag_object = self.revision self.tag_commit = self.revision self.tag_name = name
def add_tag(self, name: str) -> None: """Create an additional annotated tag at HEAD and adopt it as the release tag.""" git(["tag", "-a", name, "-m", f"fixture tag {name}"], self.repo) self.tag_name = name self.tag_object = git(["rev-parse", f"{name}^{{tag}}"], self.repo) self.tag_commit = self.revision
# -- release directory ----------------------------------------------- def manifest_document(self) -> dict: """The publisher's view of the release manifest.
It starts from the genuine manifest ``tools/package_release.py`` wrote and applies only the documented fixture knobs, so a mutation test changes exactly one fact and the default document is faithful to real packager output (which the independent checker re-derives from bytes). """ document = json.loads(json.dumps(self._template["manifest"])) document["development"] = self.development document["source"]["dirty"] = self.dirty document["source"]["revision"] = self.revision document["source"]["tag"] = self.tag_name document["source"]["tag_object"] = self.tag_object document["source"]["tag_commit"] = self.tag_commit document["release_ready"] = self.release_ready if document.get("verification") and self.verification_state != "verified": document["verification"][0]["state"] = self.verification_state
binary_sha = None if self.defer_binary_archive_digest else digest(self.binary_bytes) source_sha = digest(self.source_bytes) for entry in document["artifacts"]: if entry["kind"] == publisher.PUBLISHED_ARTIFACT_KIND: if binary_sha is None: entry.pop("sha256", None) entry["self_reference"] = ( "the hash of an archive cannot be recorded inside itself; see SHA256SUMS" ) else: entry["sha256"] = binary_sha elif entry["kind"] == publisher.SOURCE_ARTIFACT_KIND: entry["sha256"] = source_sha document["archive"]["sha256"] = binary_sha if self.declare_archive_section_digest else None document["source_archive"]["sha256"] = source_sha if not self.release_ready: document["publication"] = {"status": "not_publishable", "blocking": ["no clean extracted-source rebuild"]} return document
def write_checksums(self) -> None: lines = [] for name in (ARCHIVE_NAME, SOURCE_ARCHIVE_NAME, "release-manifest.json", "source-manifest.json"): path = self.release_dir / name if path.exists(): lines.append(f"{digest(path.read_bytes())} {name}") self.checksums_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
def write_manifest(self, mutate=None) -> Path: document = self.manifest_document() if mutate is not None: mutate(document) self.manifest_path.write_text(json.dumps(document, indent=2) + "\n", encoding="utf-8") return self.manifest_path
def write_schema(self, mutate=None) -> Path: document = reviewed_schema_document() if mutate is not None: mutate(document) self.schema_path.write_text(json.dumps(document, indent=2) + "\n", encoding="utf-8") return self.schema_path
def write_release(self, mutate=None) -> "ReleaseFixture": """Write the whole release directory the way the packaging step does.
Every auxiliary file the packager wrote (rebuild evidence and anything else the shared checker reads) is seeded from the template release, then the archives, manifest and ``SHA256SUMS`` are (re)written. """ self.release_dir.mkdir(parents=True, exist_ok=True) template_release = Path(self._template["release_dir"]) for item in sorted(template_release.iterdir()): if item.is_file() and not item.is_symlink(): shutil.copyfile(item, self.release_dir / item.name) (self.release_dir / ARCHIVE_NAME).write_bytes(self.binary_bytes) (self.release_dir / SOURCE_ARCHIVE_NAME).write_bytes(self.source_bytes) self.source_manifest_path.write_text(self._template["source_manifest_text"], encoding="utf-8") self.write_manifest(mutate) self.write_checksums() self.write_schema() return self
def prepare(self) -> "ReleaseFixture": self.build_git() return self.write_release()
# -- invocation ------------------------------------------------------ def cli_args(self, *, publish: bool = False, repo: str | None = None, owner_did: str | None = None, identifier: str | None = None, tag: str | None = None, pds_origin: str | None = "https://tngl.sh", reviewed_schema: bool = False, max_blob_bytes: int | None = None, mock_origin: str | None = None, allow_loopback: bool = False, receipt: Path | None = None, extra: list[str] | None = None) -> list[str]: args = [ sys.executable, str(SCRIPT), "--json", "--release-manifest", str(self.manifest_path), "--repo", repo if repo is not None else REPO_URL, "--owner-did", owner_did if owner_did is not None else OWNER_DID, "--identifier", identifier if identifier is not None else HANDLE, "--tag", tag if tag is not None else self.tag_name, "--repo-dir", str(self.repo), ] if pds_origin is not None: args += ["--pds-origin", pds_origin] if publish: args.append("--publish") if reviewed_schema: args += ["--reviewed-schema", str(self.schema_path)] if max_blob_bytes is not None: args += ["--max-blob-bytes", str(max_blob_bytes)] if mock_origin is not None: args += ["--mock-endpoint", mock_origin] if allow_loopback: args.append("--allow-loopback-http") if receipt is not None: args += ["--receipt", str(receipt)] if extra: args += extra return args
def cli_env(self, **extra) -> dict: environment = { key: value for key, value in os.environ.items() if key not in (publisher.DEFAULT_PASSWORD_ENV, publisher.MOCK_ENV_GUARD) } environment.update({key: str(value) for key, value in extra.items()}) return environment
def run_cli(self, args: list[str], *, env: dict | None = None, timeout: int = 90) -> subprocess.CompletedProcess: return subprocess.run( args, capture_output=True, text=True, timeout=timeout, check=False, env=env if env is not None else self.cli_env(), stdin=subprocess.DEVNULL, )
class FixtureTestCase(unittest.TestCase): def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory() self.addCleanup(self._tmp.cleanup) self.root = Path(self._tmp.name) self.fixture = ReleaseFixture(self.root / "project")
def envelope(self, completed: subprocess.CompletedProcess) -> dict: """Parse the single JSON envelope a caller sees on stdout.""" try: return json.loads(completed.stdout) except json.JSONDecodeError as exc: # pragma: no cover - surfaced as a failure raise AssertionError(f"no JSON envelope on stdout ({exc}); stderr={completed.stderr!r}") from exc
def error_code(self, completed: subprocess.CompletedProcess) -> str: envelope = self.envelope(completed) self.assertEqual(envelope["record_kind"], "publish_error", envelope) return envelope["error"]["code"]
def error_message(self, completed: subprocess.CompletedProcess) -> str: return self.envelope(completed)["error"]["message"]
def assert_fails(self, completed: subprocess.CompletedProcess, code: str, exit_code: int = publisher.EXIT_PREFLIGHT): self.assertEqual(completed.returncode, exit_code, f"stdout={completed.stdout!r} stderr={completed.stderr!r}") self.assertEqual(self.error_code(completed), code)
# --------------------------------------------------------------------------# Manifest schema# --------------------------------------------------------------------------
@unittest.skipUnless(GIT, "git is not available")class ManifestSchemaTests(FixtureTestCase): def setUp(self) -> None: super().setUp() self.fixture.build_git()
def parse(self, document: dict) -> dict: return publisher.parse_release_manifest(document, path=self.fixture.manifest_path)
def good_document(self) -> dict: return self.fixture.manifest_document()
def with_mutation(self, mutate) -> dict: document = self.good_document() mutate(document) return document
def artifact(self, document: dict, kind: str) -> dict: return next(entry for entry in document["artifacts"] if entry["kind"] == kind)
def assert_manifest_invalid(self, document: dict, code: str = "release_manifest_invalid") -> None: with self.assertRaises(publisher.PublishError) as caught: self.parse(document) self.assertEqual(caught.exception.code, code)
def test_accepts_the_packager_schema(self) -> None: facts = self.parse(self.good_document()) self.assertEqual(facts["product"], PRODUCT) self.assertEqual(facts["version"], VERSION) self.assertEqual(facts["target"]["target_triple"], "x86_64-unknown-linux-gnu") self.assertEqual(facts["tag"], TAG) self.assertEqual(facts["tag_type"], "annotated") self.assertEqual(facts["tag_commit"], self.fixture.tag_commit) self.assertEqual(facts["artifacts"][publisher.PUBLISHED_ARTIFACT_KIND]["name"], ARCHIVE_NAME) self.assertEqual(facts["artifacts"][publisher.SOURCE_ARTIFACT_KIND]["name"], SOURCE_ARCHIVE_NAME) self.assertEqual(len(facts["artifacts"]["host_binaries"]), len(HOST_BINARIES)) self.assertTrue(facts["release_ready"]) self.assertEqual(facts["verification"][0]["state"], "verified") self.assertEqual(facts["archive_sha256"], digest(self.fixture.binary_bytes))
def test_accepts_deferred_archive_digest(self) -> None: self.fixture.defer_binary_archive_digest = True self.fixture.declare_archive_section_digest = False facts = self.parse(self.good_document()) self.assertIsNone(facts["artifacts"][publisher.PUBLISHED_ARTIFACT_KIND]["sha256"]) self.assertIsNone(facts["archive_sha256"])
def test_accepts_a_full_packager_shaped_manifest(self) -> None: """Field-for-field acceptance of what ``tools/package_release.py`` writes.""" def enrich(document: dict) -> None: document.update({ "product_title": "Ultimate Tool", "release_type": "preview", "development_label": None, "created_at": "2026-10-11T00:00:00+00:00", "build": { "command": ["make", "dist"], "environment_status": "verified", "environment_problems": [], "environment": {}, "expected_environment": {}, "compiler_reproducibility": "not_claimed", }, "release_policy": { "reserved_preview_version": VERSION, "release_type": "preview", "license": "GPL-3.0-or-later", "license_file": "LICENSE", "required_checks": ["binary_elf_target"], "release_ready_requires": "every required check verified", "supported_versions": [VERSION], }, "source_completeness": {"status": "verified", "clean_rebuild": "verified"}, "inventory": [{"name": "rust", "kind": "host", "license": "MIT"}], "privacy": {"files_scanned": 12, "blocking_findings": 0, "review_required_findings": 0}, "runtime_requirements": [{"name": "python3", "requirement": "Python 3"}], "generator": {"tool": "tools/package_release.py", "schema": 1}, "publication": { "status": "preflight_required", "blocking": [], "publisher_interface": { "consumer": "tools/publish_tangled_artifacts.py", "fields": ["version", "source.tag"], "note": "reject when release_ready is false", }, }, }) document["archive"].update({ "size": len(self.fixture.binary_bytes), "members": 12, "tar_safety": "verified", "determinism": "sorted members, mtime 0", "recorded_in": "SHA256SUMS", "uncompressed_bytes": 4096, "member_kinds": {"file": 12}, }) document["source_archive"].update({ "prefix": f"ultimate-tool-{VERSION}-src", "size": len(self.fixture.source_bytes), "members": 40, "tar_safety": "verified", }) for entry in document["artifacts"]: if entry["kind"] == publisher.HOST_BINARY_KIND: entry["elf"] = { "class": "ELF64", "machine": "Advanced Micro Devices X86-64", "type": "DYN", "linkage": "dynamic", "interpreter": "/lib64/ld-linux-x86-64.so.2", } document["verification"].append({ "check": "vendor_sources_present", "state": "not_applicable", "detail": "no vendored sources in this tree", })
facts = self.parse(self.with_mutation(enrich)) self.assertTrue(facts["release_ready"]) self.assertEqual(facts["verification"][-1]["state"], "not_applicable")
def test_rejects_an_unknown_verification_state(self) -> None: self.assert_manifest_invalid( self.with_mutation(lambda doc: doc["verification"][0].update({"state": "probably-fine"})) )
def test_optional_sections_may_be_absent(self) -> None: document = self.with_mutation(lambda doc: [doc.pop(key) for key in ("baseline", "support_boundary", "source_archive", "archive")]) facts = self.parse(document) self.assertIsNone(facts["baseline"]) self.assertIsNone(facts["support_boundary"])
def test_rejects_unknown_top_level_field(self) -> None: self.assert_manifest_invalid(self.with_mutation(lambda doc: doc.update({"release": {}})))
def test_rejects_unsupported_schema_version(self) -> None: self.assert_manifest_invalid(self.with_mutation(lambda doc: doc.update({"schema_version": 2})))
def test_rejects_wrong_record_kind(self) -> None: self.assert_manifest_invalid(self.with_mutation(lambda doc: doc.update({"record_kind": "release"})))
def test_rejects_missing_required_section(self) -> None: self.assert_manifest_invalid(self.with_mutation(lambda doc: doc.pop("target")))
def test_rejects_development_flag(self) -> None: self.assert_manifest_invalid(self.with_mutation(lambda doc: doc.update({"development": True})), "development_manifest")
def test_rejects_dirty_source(self) -> None: self.assert_manifest_invalid( self.with_mutation(lambda doc: doc["source"].update({"dirty": True})), "development_manifest", )
def test_rejects_lightweight_tag_type(self) -> None: self.assert_manifest_invalid( self.with_mutation(lambda doc: doc["source"].update({"tag_type": None})), "development_manifest", )
def test_rejects_untagged_source(self) -> None: for field in ("tag", "tag_object", "tag_commit"): with self.subTest(field=field): document = self.with_mutation(lambda doc, field=field: doc["source"].update({field: None})) self.assert_manifest_invalid(document, "development_manifest")
def test_rejects_non_hex_object_ids(self) -> None: for field in ("revision", "tag_object", "tag_commit"): with self.subTest(field=field): document = self.with_mutation(lambda doc, field=field: doc["source"].update({field: "not-a-hash"})) self.assert_manifest_invalid(document)
def test_rejects_release_not_ready(self) -> None: def mutate(document: dict) -> None: document["release_ready"] = False document["publication"] = { "status": "not_publishable", "blocking": ["no clean extracted-source rebuild is recorded"], }
document = self.with_mutation(mutate) with self.assertRaises(publisher.PublishError) as caught: self.parse(document) self.assertEqual(caught.exception.code, "release_not_ready") self.assertIn("clean extracted-source rebuild", caught.exception.message)
def test_rejects_pending_verification_state(self) -> None: document = self.with_mutation( lambda doc: doc["verification"].append({"check": "privacy_sentinels", "state": "pending", "detail": "-"}) ) with self.assertRaises(publisher.PublishError) as caught: self.parse(document) self.assertEqual(caught.exception.code, "verification_incomplete") self.assertIn("privacy_sentinels=pending", caught.exception.message)
def test_rejects_not_verified_state(self) -> None: document = self.with_mutation( lambda doc: doc["verification"][0].update({"state": "not_verified"}) ) with self.assertRaises(publisher.PublishError) as caught: self.parse(document) self.assertEqual(caught.exception.code, "verification_incomplete")
def test_rejects_malformed_verification_entry(self) -> None: self.assert_manifest_invalid(self.with_mutation(lambda doc: doc["verification"].append({"state": "verified"})))
def test_rejects_duplicate_binary_archive(self) -> None: document = self.with_mutation( lambda doc: doc["artifacts"].append(dict(self.artifact(doc, publisher.PUBLISHED_ARTIFACT_KIND))) ) with self.assertRaises(publisher.PublishError) as caught: self.parse(document) self.assertEqual(caught.exception.code, "release_manifest_invalid") self.assertIn("binary-archive", caught.exception.message)
def test_rejects_absent_source_archive(self) -> None: document = self.with_mutation( lambda doc: doc.update({ "artifacts": [entry for entry in doc["artifacts"] if entry["kind"] != publisher.SOURCE_ARTIFACT_KIND], }) ) with self.assertRaises(publisher.PublishError) as caught: self.parse(document) self.assertEqual(caught.exception.code, "release_manifest_invalid") self.assertIn("source-archive", caught.exception.message)
def test_rejects_absent_host_binary(self) -> None: document = self.with_mutation( lambda doc: doc.update({ "artifacts": [entry for entry in doc["artifacts"] if entry["kind"] != publisher.HOST_BINARY_KIND], }) ) with self.assertRaises(publisher.PublishError) as caught: self.parse(document) self.assertEqual(caught.exception.code, "release_manifest_invalid") self.assertIn("host-binary", caught.exception.message)
def test_rejects_unknown_artifact_kind(self) -> None: self.assert_manifest_invalid( self.with_mutation(lambda doc: self.artifact(doc, publisher.HOST_BINARY_KIND).update({"kind": "binary"})) )
def test_rejects_artifact_entry_without_path(self) -> None: self.assert_manifest_invalid( self.with_mutation(lambda doc: self.artifact(doc, publisher.HOST_BINARY_KIND).pop("path")) )
def test_rejects_uppercase_digest(self) -> None: self.assert_manifest_invalid( self.with_mutation(lambda doc: self.artifact(doc, publisher.PUBLISHED_ARTIFACT_KIND).update( {"sha256": digest(self.fixture.binary_bytes).upper()} )) )
def test_rejects_artifact_name_with_separator(self) -> None: self.assert_manifest_invalid( self.with_mutation(lambda doc: self.artifact(doc, publisher.PUBLISHED_ARTIFACT_KIND).update( {"name": "dist/release.tar.gz"} )) )
def test_rejects_absolute_and_traversing_paths(self) -> None: for path in ("/etc/passwd", "../escape.tar.gz", "a/../../escape.tar.gz", "a\\escape"): with self.subTest(path=path): document = self.with_mutation( lambda doc, path=path: self.artifact(doc, publisher.PUBLISHED_ARTIFACT_KIND).update({"path": path}) ) self.assert_manifest_invalid(document)
def test_rejects_negative_size(self) -> None: self.assert_manifest_invalid( self.with_mutation(lambda doc: self.artifact(doc, publisher.PUBLISHED_ARTIFACT_KIND).update({"size": -1})) )
def test_rejects_wrong_type_for_a_parsed_section(self) -> None: self.assert_manifest_invalid(self.with_mutation(lambda doc: doc.update({"archive": [1, 2, 3]}))) self.assert_manifest_invalid(self.with_mutation(lambda doc: doc.update({"publication": "none"})))
def test_accepts_list_shaped_optional_sections(self) -> None: """`inventory` and `runtime_requirements` are lists in the real manifest.""" document = self.with_mutation(lambda doc: doc.update({ "inventory": [{"name": "rust"}], "runtime_requirements": [{"name": "python3"}], })) self.assertTrue(self.parse(document)["release_ready"])
def test_rejects_target_missing_a_field(self) -> None: self.assert_manifest_invalid(self.with_mutation(lambda doc: doc["target"].pop("target_triple")))
def test_rejects_bad_declared_archive_section_digest(self) -> None: self.assert_manifest_invalid(self.with_mutation(lambda doc: doc["archive"].update({"sha256": "nope"})))
def test_rejects_the_earlier_draft_schema(self) -> None: """The draft shape this tool briefly assumed must not be accepted.""" document = { "schema_version": 1, "version": VERSION, "target": "linux-x86_64", "source": {"revision": "0" * 40, "dirty": False, "tag": TAG, "tag_object": "0" * 40}, "artifacts": [{"name": ARCHIVE_NAME, "path": ARCHIVE_NAME, "sha256": "0" * 64, "size": 1, "kind": "binary"}], } self.assert_manifest_invalid(document)
# --------------------------------------------------------------------------# Checksums, origins, repo references, tag encoding# --------------------------------------------------------------------------
class ChecksumParsingTests(FixtureTestCase): def test_parses_coreutils_format(self) -> None: entries = publisher.parse_checksums( f"{'a' * 64} one.tar.gz\n{'b' * 64} sub/two.tar.gz\n".encode(), where="SHA256SUMS", ) self.assertEqual(entries, {"one.tar.gz": "a" * 64, "sub/two.tar.gz": "b" * 64})
def test_rejects_malformed_line(self) -> None: with self.assertRaises(publisher.PublishError) as caught: publisher.parse_checksums(b"not a checksum line\n", where="SHA256SUMS") self.assertEqual(caught.exception.code, "checksums_invalid")
def test_rejects_duplicate_entry(self) -> None: line = f"{'a' * 64} one.tar.gz\n" with self.assertRaises(publisher.PublishError) as caught: publisher.parse_checksums((line + line).encode(), where="SHA256SUMS") self.assertEqual(caught.exception.code, "checksums_invalid")
def test_rejects_absolute_names(self) -> None: with self.assertRaises(publisher.PublishError) as caught: publisher.parse_checksums(f"{'a' * 64} /etc/passwd\n".encode(), where="SHA256SUMS") self.assertEqual(caught.exception.code, "release_manifest_invalid")
def test_covers_the_archives_under_basenames(self) -> None: entries = {ARCHIVE_NAME: digest(self.fixture.binary_bytes)} self.assertTrue(publisher.checksum_lists( entries, target=self.fixture.release_dir / ARCHIVE_NAME, base_dir=self.fixture.release_dir, digest=digest(self.fixture.binary_bytes), ))
def test_reports_missing_artifact_coverage(self) -> None: entries = {ARCHIVE_NAME: digest(self.fixture.binary_bytes)} self.assertFalse(publisher.checksum_lists( entries, target=self.fixture.release_dir / SOURCE_ARCHIVE_NAME, base_dir=self.fixture.release_dir, digest=digest(self.fixture.source_bytes), ))
class OriginTests(unittest.TestCase): def test_accepts_https_root(self) -> None: origin = publisher.parse_origin("https://tngl.sh", mock=False) self.assertEqual(origin["origin"], "https://tngl.sh") self.assertEqual(origin["scheme"], "https")
def test_rejects_plain_http_for_publishing(self) -> None: with self.assertRaises(publisher.PublishError) as caught: publisher.parse_origin("http://tngl.sh", mock=False) self.assertEqual(caught.exception.code, "insecure_origin")
def test_rejects_http_lan_and_public_hosts_in_mock_mode(self) -> None: for origin in ("http://10.23.42.5:6555", "http://pds.example.com:8080"): with self.subTest(origin=origin): with self.assertRaises(publisher.PublishError) as caught: publisher.parse_origin(origin, mock=True) self.assertEqual(caught.exception.code, "mock_endpoint_invalid")
def test_mock_mode_requires_explicit_port(self) -> None: with self.assertRaises(publisher.PublishError) as caught: publisher.parse_origin("http://127.0.0.1", mock=True) self.assertEqual(caught.exception.code, "mock_endpoint_invalid")
def test_accepts_explicit_loopback_mock(self) -> None: self.assertEqual(publisher.parse_origin("http://127.0.0.1:8080", mock=True)["origin"], "http://127.0.0.1:8080")
def test_rejects_url_credentials(self) -> None: with self.assertRaises(publisher.PublishError) as caught: publisher.parse_origin("https://user:secret@tngl.sh", mock=False) self.assertEqual(caught.exception.code, "origin_credentials")
def test_rejects_path(self) -> None: with self.assertRaises(publisher.PublishError) as caught: publisher.parse_origin("https://tngl.sh/xrpc", mock=False) self.assertEqual(caught.exception.code, "origin_not_root")
def test_rejects_query_and_fragment(self) -> None: for origin in ("https://tngl.sh?x=1", "https://tngl.sh#frag"): with self.subTest(origin=origin): with self.assertRaises(publisher.PublishError) as caught: publisher.parse_origin(origin, mock=False) self.assertEqual(caught.exception.code, "origin_not_root")
class RepoReferenceTests(unittest.TestCase): def test_accepts_https_tangled_url(self) -> None: reference = publisher.parse_repo_reference(REPO_URL, owner_did=OWNER_DID) self.assertEqual(reference["kind"], "web") self.assertEqual(reference["deferred"], HANDLE)
def test_accepts_https_url_owned_by_a_did(self) -> None: reference = publisher.parse_repo_reference(f"https://tangled.org/{OWNER_DID}/ultimate-tool", owner_did=OWNER_DID) self.assertIsNone(reference["deferred"]) self.assertTrue(reference["owner_is_did"])
def test_accepts_canonical_at_uri(self) -> None: value = f"at://{OWNER_DID}/sh.tangled.repo/3kfixture" reference = publisher.parse_repo_reference(value, owner_did=OWNER_DID) self.assertEqual(reference["kind"], "at_uri") self.assertEqual(reference["value"], value)
def test_rejects_at_uri_with_non_did_repo(self) -> None: with self.assertRaises(publisher.PublishError) as caught: publisher.parse_repo_reference("at://operator.tngl.sh/sh.tangled.repo/3k", owner_did=OWNER_DID) self.assertEqual(caught.exception.code, "repo_reference_invalid")
def test_rejects_at_uri_owner_mismatch(self) -> None: with self.assertRaises(publisher.PublishError) as caught: publisher.parse_repo_reference(f"at://{OTHER_DID}/sh.tangled.repo/3k", owner_did=OWNER_DID) self.assertEqual(caught.exception.code, "repo_owner_mismatch")
def test_rejects_non_tangled_host_and_scheme(self) -> None: for value in ("https://github.com/owner/repo", "http://tangled.org/owner/repo", "https://tangled.org/owner"): with self.subTest(value=value): with self.assertRaises(publisher.PublishError) as caught: publisher.parse_repo_reference(value, owner_did=OWNER_DID) self.assertEqual(caught.exception.code, "repo_reference_invalid")
def test_rejects_query_fragment_and_credentials(self) -> None: for value in (f"{REPO_URL}?ref=main", f"{REPO_URL}#readme", "https://user:secret@tangled.org/owner/repo"): with self.subTest(value=value): with self.assertRaises(publisher.PublishError) as caught: publisher.parse_repo_reference(value, owner_did=OWNER_DID) self.assertIn(caught.exception.code, {"repo_reference_invalid", "origin_credentials"})
class TagEncodingTests(unittest.TestCase): def test_encodes_object_hash_as_unpadded_base64(self) -> None: encoded = publisher.tag_object_bytes_base64("00" * 20) self.assertEqual(encoded, "AAAAAAAAAAAAAAAAAAAAAAAAAAA") self.assertNotIn("=", encoded) self.assertEqual(len(encoded), 27)
def test_encodes_sha256_object_hash(self) -> None: self.assertEqual(len(publisher.tag_object_bytes_base64("ab" * 32)), 43)
def test_rejects_wrong_length_and_non_hex(self) -> None: for value in ("00" * 8, "zz" * 20, "0" * 41): with self.subTest(value=value): with self.assertRaises(publisher.PublishError) as caught: publisher.tag_object_bytes_base64(value) self.assertEqual(caught.exception.code, "tag_object_mismatch")
# --------------------------------------------------------------------------# Reviewed schema gate# --------------------------------------------------------------------------
class ReviewedSchemaTests(FixtureTestCase): def test_accepts_reviewed_lexicon_and_records_raw_hash(self) -> None: path = self.fixture.write_schema() schema = publisher.load_reviewed_schema(path, expected_sha256=None) self.assertEqual(schema["id"], "sh.tangled.repo.artifact") self.assertEqual(schema["sha256"], digest(path.read_bytes())) self.assertEqual(schema["required"], ["tag", "name", "repo", "artifact", "createdAt"])
def test_pins_the_reviewed_bytes(self) -> None: path = self.fixture.write_schema() expected = digest(path.read_bytes()) self.assertEqual(publisher.load_reviewed_schema(path, expected_sha256=expected)["sha256"], expected) with self.assertRaises(publisher.PublishError) as caught: publisher.load_reviewed_schema(path, expected_sha256="0" * 64) self.assertEqual(caught.exception.code, "reviewed_schema_mismatch")
def test_rejects_lexicon_missing_a_published_field(self) -> None: def drop_artifact(document: dict) -> None: document["defs"]["main"]["record"]["properties"].pop("artifact")
with self.assertRaises(publisher.PublishError) as caught: publisher.load_reviewed_schema(self.fixture.write_schema(drop_artifact), expected_sha256=None) self.assertEqual(caught.exception.code, "reviewed_schema_invalid")
def test_rejects_wrong_lexicon_id_and_non_lexicon(self) -> None: for mutate in (lambda doc: doc.__setitem__("id", "sh.tangled.repo"), lambda doc: doc.pop("lexicon")): with self.subTest(mutate=mutate): with self.assertRaises(publisher.PublishError) as caught: publisher.load_reviewed_schema(self.fixture.write_schema(mutate), expected_sha256=None) self.assertEqual(caught.exception.code, "reviewed_schema_invalid")
def test_rejects_malformed_json(self) -> None: self.fixture.schema_path.write_text("{not json", encoding="utf-8") with self.assertRaises(publisher.PublishError) as caught: publisher.load_reviewed_schema(self.fixture.schema_path, expected_sha256=None) self.assertEqual(caught.exception.code, "reviewed_schema_invalid")
# --------------------------------------------------------------------------# Git tag handling# --------------------------------------------------------------------------
@unittest.skipUnless(GIT, "git is not available")class AnnotatedTagTests(FixtureTestCase): def test_reads_annotated_tag_object_and_target(self) -> None: self.fixture.build_git() parsed = publisher.read_annotated_tag(self.fixture.repo, TAG) self.assertEqual(parsed["object"], self.fixture.tag_object) self.assertEqual(parsed["target"], self.fixture.revision) self.assertEqual(parsed["target_type"], "commit") self.assertEqual(len(publisher.tag_object_bytes_base64(parsed["object"])), 27)
def test_rejects_lightweight_tag(self) -> None: self.fixture.build_git(annotated=False) with self.assertRaises(publisher.PublishError) as caught: publisher.read_annotated_tag(self.fixture.repo, TAG) self.assertEqual(caught.exception.code, "tag_not_annotated")
def test_reports_missing_tag(self) -> None: self.fixture.build_git() with self.assertRaises(publisher.PublishError) as caught: publisher.read_annotated_tag(self.fixture.repo, "v9.9.9") self.assertEqual(caught.exception.code, "git_failed")
def test_rejects_unsafe_tag_names_without_running_git(self) -> None: self.fixture.build_git() for name in ("--help", "bad name", "refs/tags/../x", "tag@{0}"): with self.subTest(name=name): with self.assertRaises(publisher.PublishError) as caught: publisher.read_annotated_tag(self.fixture.repo, name) self.assertEqual(caught.exception.code, "invalid_tag_name")
# --------------------------------------------------------------------------# Dry run (no network, no credential)# --------------------------------------------------------------------------
@unittest.skipUnless(GIT, "git is not available")class DryRunTests(FixtureTestCase): def setUp(self) -> None: super().setUp() self.fixture.prepare() self.pds = MockPds() self.pds.start() self.addCleanup(self.pds.stop)
def run_dry(self, **kwargs) -> subprocess.CompletedProcess: defaults = dict(mock_origin=self.pds.origin, allow_loopback=True, pds_origin=None) defaults.update(kwargs) return self.fixture.run_cli( self.fixture.cli_args(**defaults), env=self.fixture.cli_env(**{publisher.MOCK_ENV_GUARD: "1"}), )
def test_dry_run_succeeds_and_makes_no_request(self) -> None: completed = self.run_dry() self.assertEqual(completed.returncode, 0, completed.stderr) envelope = self.envelope(completed) self.assertEqual(envelope["record_kind"], "publish_result") self.assertEqual(envelope["data"]["status"], "dry_run_ok") self.assertEqual(self.pds.requests, []) data = envelope["data"] self.assertFalse(data["would_create"]["validate"]) # The complete published set: binary archive, source archive, SHA256SUMS # and the release manifest, each with the name the file really has. roles = [entry["role"] for entry in data["objects"]] self.assertEqual(roles, ["binary-archive", "source-archive", "checksums", "release-manifest"]) records = data["would_create"]["records"] self.assertEqual(len(records), 4) record = records[0] self.assertEqual(record["name"], ARCHIVE_NAME) self.assertEqual(record["repo"], REPO_URL) self.assertEqual(record["artifact"]["$type"], "blob") self.assertEqual(publisher.parse_blob_cid(record["artifact"]["ref"]["$link"]), digest(self.fixture.binary_bytes)) self.assertEqual(len(record["tag"]["$bytes"]), 27) self.assertNotIn("=", record["tag"]["$bytes"]) self.assertEqual(records[2]["name"], CHECKSUM_NAME) self.assertEqual(records[3]["name"], "release-manifest.json") self.assertEqual(data["artifact"]["sha256"], digest(self.fixture.binary_bytes))
def test_dry_run_reports_unverified_blob_bound(self) -> None: checks = {check["name"]: check for check in self.envelope(self.run_dry())["data"]["checks"]} self.assertEqual(checks["blob_bound"]["status"], "not_checked") self.assertEqual(checks[publisher.PUBLISHED_ARTIFACT_KIND]["status"], "ok") self.assertEqual(checks["host_binaries"]["status"], "ok")
def test_dry_run_needs_no_credential_and_does_not_prompt(self) -> None: self.assertEqual(self.run_dry(extra=["--no-prompt"]).returncode, 0)
def test_dry_run_accepts_reviewed_schema_and_declared_bound(self) -> None: completed = self.run_dry(reviewed_schema=True, max_blob_bytes=1 << 20) self.assertEqual(completed.returncode, 0, completed.stderr) checks = {check["name"]: check for check in self.envelope(completed)["data"]["checks"]} self.assertEqual(checks["reviewed_schema"]["status"], "ok") self.assertEqual(checks["blob_bound"]["status"], "ok")
def test_host_binary_entries_are_not_read_from_disk(self) -> None: """`bin/<name>` lives inside the archive, so the dry run must not require it.""" self.assertFalse((self.fixture.release_dir / f"bin/{HOST_BINARIES[0]}").exists()) completed = self.run_dry() self.assertEqual(completed.returncode, 0, completed.stderr)
def test_accepts_a_manifest_that_defers_the_archive_digest(self) -> None: self.fixture.defer_binary_archive_digest = True self.fixture.declare_archive_section_digest = False self.fixture.write_release() completed = self.run_dry() self.assertEqual(completed.returncode, 0, completed.stderr) data = self.envelope(completed)["data"] self.assertEqual(data["artifact"]["sha256"], digest(self.fixture.binary_bytes))
def test_rejects_development_manifest(self) -> None: self.fixture.development = True self.fixture.write_release() self.assert_fails(self.fixture.run_cli(self.fixture.cli_args()), "development_manifest")
def test_rejects_artifact_hash_mismatch(self) -> None: (self.fixture.release_dir / ARCHIVE_NAME).write_bytes(b"tampered") self.assert_fails(self.fixture.run_cli(self.fixture.cli_args()), "artifact_hash_mismatch")
def test_rejects_archive_section_digest_mismatch(self) -> None: self.fixture.write_release(lambda doc: doc["archive"].update({"sha256": "a" * 64})) self.assert_fails(self.fixture.run_cli(self.fixture.cli_args()), "artifact_hash_mismatch")
def test_rejects_artifact_size_mismatch(self) -> None: self.fixture.write_release( lambda doc: next(entry for entry in doc["artifacts"] if entry["kind"] == publisher.PUBLISHED_ARTIFACT_KIND).update({"size": 1}) ) completed = self.fixture.run_cli(self.fixture.cli_args()) self.assert_fails(completed, "artifact_size_mismatch")
def test_rejects_missing_artifact(self) -> None: (self.fixture.release_dir / SOURCE_ARCHIVE_NAME).unlink() self.fixture.write_checksums() self.assert_fails(self.fixture.run_cli(self.fixture.cli_args()), "artifact_missing")
def test_rejects_missing_checksum_file(self) -> None: self.fixture.checksums_path.unlink() self.assert_fails(self.fixture.run_cli(self.fixture.cli_args()), "artifact_missing")
def test_rejects_checksum_digest_mismatch(self) -> None: self.fixture.checksums_path.write_text(f"{'a' * 64} {ARCHIVE_NAME}\n", encoding="utf-8") self.assert_fails(self.fixture.run_cli(self.fixture.cli_args()), "checksums_digest_mismatch")
def test_rejects_archive_absent_from_checksum_file(self) -> None: self.fixture.checksums_path.write_text( f"{digest(self.fixture.binary_bytes)} {ARCHIVE_NAME}\n", encoding="utf-8", ) self.assert_fails(self.fixture.run_cli(self.fixture.cli_args()), "checksums_artifact_not_listed")
def test_rejects_checksum_entry_whose_file_is_missing(self) -> None: sums = self.fixture.checksums_path.read_text(encoding="utf-8") + f"{'b' * 64} ghost.tar.gz\n" self.fixture.checksums_path.write_text(sums, encoding="utf-8") self.assert_fails(self.fixture.run_cli(self.fixture.cli_args()), "file_unreadable")
def test_rejects_tampered_checksum_file_itself(self) -> None: """SHA256SUMS covers the manifest, so a stale sums file is caught.""" self.fixture.write_manifest(lambda doc: doc.update({"created_at": "2026-01-01T00:00:00+00:00"})) self.assert_fails(self.fixture.run_cli(self.fixture.cli_args()), "checksums_digest_mismatch")
def test_rejects_symlinked_artifact(self) -> None: target = self.fixture.release_dir / ARCHIVE_NAME real = self.fixture.release_dir / "elsewhere.tar.gz" real.write_bytes(target.read_bytes()) target.unlink() target.symlink_to(real) self.assert_fails(self.fixture.run_cli(self.fixture.cli_args()), "unsafe_path")
def test_rejects_artifact_under_a_symlinked_parent(self) -> None: payload = self.fixture.release_dir / "payload" payload.mkdir() (payload / "x.tar.gz").write_bytes(self.fixture.source_bytes) (self.fixture.release_dir / "link").symlink_to(payload) self.fixture.write_release( lambda doc: next(entry for entry in doc["artifacts"] if entry["kind"] == publisher.SOURCE_ARTIFACT_KIND).update({"path": "link/x.tar.gz"}) ) self.assert_fails(self.fixture.run_cli(self.fixture.cli_args()), "unsafe_path")
def test_rejects_tag_disagreements(self) -> None: self.assert_fails(self.fixture.run_cli(self.fixture.cli_args(tag="v0.2.0-preview.2")), "tag_mismatch") self.fixture.write_release(lambda doc: doc["source"].update({"tag_object": "0" * 40})) self.assert_fails(self.fixture.run_cli(self.fixture.cli_args()), "tag_object_mismatch")
def test_rejects_a_tag_object_that_no_longer_matches_the_repository(self) -> None: """A tag recreated after packaging leaves the manifest's commit stale.""" self.fixture.add_tag("v0.2.0-preview.1-recreated") self.fixture.tag_name = TAG self.fixture.tag_object = git(["rev-parse", f"{TAG}^{{tag}}"], self.fixture.repo) self.fixture.write_release(lambda doc: doc["source"].update({"tag_commit": "1" * 40, "revision": "1" * 40})) self.assert_fails(self.fixture.run_cli(self.fixture.cli_args()), "tag_commit_mismatch")
def test_rejects_tag_not_agreeing_with_version(self) -> None: self.fixture.add_tag("release-0.2.0-preview.1") self.fixture.write_release() completed = self.fixture.run_cli(self.fixture.cli_args(tag="release-0.2.0-preview.1")) self.assert_fails(completed, "tag_version_mismatch")
def test_rejects_blob_above_the_declared_bound(self) -> None: self.assert_fails(self.run_dry(max_blob_bytes=4), "blob_too_large")
def test_publish_requires_reviewed_schema_and_blob_bound(self) -> None: self.assert_fails(self.fixture.run_cli(self.fixture.cli_args(publish=True)), "reviewed_schema_required") self.assert_fails( self.fixture.run_cli(self.fixture.cli_args(publish=True, reviewed_schema=True)), "blob_limit_required", )
def test_publish_gates_fail_before_any_network_or_secret(self) -> None: """An unusable origin fails in preflight, before a credential is read.""" args = self.fixture.cli_args(publish=True, reviewed_schema=True, max_blob_bytes=1 << 20, pds_origin="ftp://tngl.sh") completed = self.fixture.run_cli( args, env=self.fixture.cli_env(**{publisher.DEFAULT_PASSWORD_ENV: PASSWORD}), ) self.assert_fails(completed, "insecure_origin") self.assertNotIn(PASSWORD, completed.stdout + completed.stderr)
def test_mock_endpoint_and_pds_origin_together_are_rejected(self) -> None: args = self.fixture.cli_args(mock_origin=self.pds.origin, allow_loopback=True) completed = self.fixture.run_cli(args, env=self.fixture.cli_env(**{publisher.MOCK_ENV_GUARD: "1"})) self.assert_fails(completed, "usage", publisher.EXIT_USAGE) self.assertEqual(self.pds.requests, [])
class LocalPreflightFailureTests(FixtureTestCase): """Preflight failures must not reach a socket or a credential prompt."""
def setUp(self) -> None: super().setUp() self.fixture.prepare() self.pds = MockPds() self.pds.start() self.addCleanup(self.pds.stop)
def test_no_secret_is_sent_when_local_preflight_fails(self) -> None: (self.fixture.release_dir / ARCHIVE_NAME).write_bytes(b"tampered") args = self.fixture.cli_args(publish=True, reviewed_schema=True, max_blob_bytes=1 << 20, pds_origin=None, mock_origin=self.pds.origin, allow_loopback=True) completed = self.fixture.run_cli( args, env=self.fixture.cli_env(**{publisher.MOCK_ENV_GUARD: "1", publisher.DEFAULT_PASSWORD_ENV: PASSWORD}), ) self.assert_fails(completed, "artifact_hash_mismatch") self.assertEqual(self.pds.requests, []) self.assertNotIn(PASSWORD, completed.stdout + completed.stderr)
def test_preflight_failure_is_not_a_secret_error(self) -> None: args = self.fixture.cli_args(publish=True, reviewed_schema=True, max_blob_bytes=1, pds_origin=None, mock_origin=self.pds.origin, allow_loopback=True) completed = self.fixture.run_cli(args, env=self.fixture.cli_env(**{publisher.MOCK_ENV_GUARD: "1"})) self.assert_fails(completed, "blob_too_large") self.assertEqual(self.pds.requests, [])
class FifoRejectionTests(FixtureTestCase): """A FIFO artifact must fail fast instead of blocking a bounded read."""
def test_fifo_artifact_is_rejected_without_blocking(self) -> None: self.fixture.prepare() target = self.fixture.release_dir / ARCHIVE_NAME target.unlink() os.mkfifo(target) completed = self.fixture.run_cli(self.fixture.cli_args(), timeout=30) self.assert_fails(completed, "unsafe_file")
# --------------------------------------------------------------------------# Credential and mock gating# --------------------------------------------------------------------------
class GatingTests(FixtureTestCase): def setUp(self) -> None: super().setUp() self.fixture.prepare() self.pds = MockPds() self.pds.start() self.addCleanup(self.pds.stop)
def test_mock_endpoint_requires_the_loopback_flag(self) -> None: args = self.fixture.cli_args(mock_origin=self.pds.origin) completed = self.fixture.run_cli(args, env=self.fixture.cli_env(**{publisher.MOCK_ENV_GUARD: "1"})) self.assert_fails(completed, "mock_not_allowed", publisher.EXIT_REFUSED)
def test_mock_endpoint_requires_the_environment_guard(self) -> None: args = self.fixture.cli_args(mock_origin=self.pds.origin, allow_loopback=True) self.assert_fails(self.fixture.run_cli(args), "mock_not_allowed", publisher.EXIT_REFUSED)
def test_loopback_flag_without_mock_endpoint_is_rejected(self) -> None: args = self.fixture.cli_args(allow_loopback=True) completed = self.fixture.run_cli(args, env=self.fixture.cli_env(**{publisher.MOCK_ENV_GUARD: "1"})) self.assert_fails(completed, "mock_not_allowed", publisher.EXIT_REFUSED)
def test_real_publishing_requires_an_origin(self) -> None: self.assert_fails(self.fixture.run_cli(self.fixture.cli_args(pds_origin=None)), "usage", publisher.EXIT_USAGE)
def test_secret_is_required_without_a_prompt(self) -> None: args = self.fixture.cli_args(publish=True, reviewed_schema=True, max_blob_bytes=1 << 20, identifier=OWNER_DID, repo=f"https://tangled.org/{OWNER_DID}/ultimate-tool", pds_origin=None, mock_origin=self.pds.origin, allow_loopback=True, extra=["--no-prompt"]) completed = self.fixture.run_cli(args, env=self.fixture.cli_env(**{publisher.MOCK_ENV_GUARD: "1"})) self.assert_fails(completed, "secret_unavailable", publisher.EXIT_REMOTE) self.assertEqual(self.pds.requests, [])
def test_non_tty_stdin_never_reads_a_password_from_a_pipe(self) -> None: args = self.fixture.cli_args(publish=True, reviewed_schema=True, max_blob_bytes=1 << 20, identifier=OWNER_DID, repo=f"https://tangled.org/{OWNER_DID}/ultimate-tool", pds_origin=None, mock_origin=self.pds.origin, allow_loopback=True) completed = self.fixture.run_cli(args, env=self.fixture.cli_env(**{publisher.MOCK_ENV_GUARD: "1"})) self.assert_fails(completed, "secret_unavailable", publisher.EXIT_REMOTE) self.assertIn("terminal", self.error_message(completed))
# --------------------------------------------------------------------------# Publishing against the loopback mock# --------------------------------------------------------------------------
@unittest.skipUnless(GIT, "git is not available")class PublishMockTests(FixtureTestCase): def setUp(self) -> None: super().setUp() self.fixture.prepare() self.pds = MockPds() self.pds.start() self.addCleanup(self.pds.stop)
def publish(self, **kwargs) -> subprocess.CompletedProcess: defaults = dict(publish=True, reviewed_schema=True, max_blob_bytes=1 << 20, mock_origin=self.pds.origin, allow_loopback=True, pds_origin=None) defaults.update(kwargs) environment = self.fixture.cli_env(**{ publisher.MOCK_ENV_GUARD: "1", publisher.DEFAULT_PASSWORD_ENV: PASSWORD, }) return self.fixture.run_cli(self.fixture.cli_args(**defaults), env=environment)
def test_publishes_the_expected_xrpc_sequence(self) -> None: completed = self.publish() self.assertEqual(completed.returncode, 0, completed.stderr) self.assertEqual(self.envelope(completed)["data"]["status"], "published") # Two owner resolutions, one session, then an upload and a createRecord # for each of the four published objects -- in a fixed order. self.assertEqual([request["route"] for request in self.pds.requests], [ "/xrpc/com.atproto.identity.resolveHandle", "/xrpc/com.atproto.identity.resolveHandle", "/xrpc/com.atproto.server.createSession", "/xrpc/com.atproto.repo.uploadBlob", "/xrpc/com.atproto.repo.createRecord", "/xrpc/com.atproto.repo.uploadBlob", "/xrpc/com.atproto.repo.createRecord", "/xrpc/com.atproto.repo.uploadBlob", "/xrpc/com.atproto.repo.createRecord", "/xrpc/com.atproto.repo.uploadBlob", "/xrpc/com.atproto.repo.createRecord", ])
def test_session_request_carries_identifier_and_password(self) -> None: self.publish() self.assertEqual(self.pds.body_of("/xrpc/com.atproto.server.createSession"), {"identifier": HANDLE, "password": PASSWORD})
def test_blob_upload_body_is_the_exact_archive(self) -> None: self.publish() uploads = self.pds.route("/xrpc/com.atproto.repo.uploadBlob") # Binary archive, source archive, SHA256SUMS and release manifest. self.assertEqual(len(uploads), 4) self.assertEqual(uploads[0]["body"], self.fixture.binary_bytes) self.assertEqual(uploads[0]["headers"]["content-length"], str(len(self.fixture.binary_bytes))) self.assertEqual(uploads[0]["headers"]["authorization"], f"Bearer {JWT}") self.assertEqual(uploads[0]["headers"]["content-type"], "application/octet-stream") self.assertEqual(uploads[1]["body"], self.fixture.source_bytes) self.assertEqual(uploads[0]["body"], (self.fixture.release_dir / ARCHIVE_NAME).read_bytes())
def test_record_is_bound_to_owner_tag_and_repo(self) -> None: self.publish() request = self.pds.bodies_of("/xrpc/com.atproto.repo.createRecord")[0] self.assertEqual(request["repo"], OWNER_DID) self.assertEqual(request["collection"], "sh.tangled.repo.artifact") self.assertFalse(request["validate"]) record = request["record"] self.assertEqual(record["$type"], "sh.tangled.repo.artifact") self.assertEqual(record["name"], ARCHIVE_NAME) self.assertEqual(record["repo"], REPO_URL) self.assertEqual(record["tag"], {"$bytes": publisher.tag_object_bytes_base64(self.fixture.tag_object)}) self.assertEqual(record["artifact"]["$type"], "blob") self.assertEqual(publisher.parse_blob_cid(record["artifact"]["ref"]["$link"]), digest(self.fixture.binary_bytes)) self.assertEqual(record["artifact"]["size"], len(self.fixture.binary_bytes)) self.assertRegex(record["createdAt"], r"^[0-9]{4}-[0-9]{2}-[0-9]{2}T") # Every record is bound to the same annotated tag object. for other in self.pds.bodies_of("/xrpc/com.atproto.repo.createRecord")[1:]: self.assertEqual(other["record"]["tag"], record["tag"])
def test_publishing_a_deferred_digest_manifest_binds_the_computed_hash(self) -> None: self.fixture.defer_binary_archive_digest = True self.fixture.declare_archive_section_digest = False self.fixture.write_release() receipt = self.root / "deferred.json" completed = self.publish(receipt=receipt) self.assertEqual(completed.returncode, 0, completed.stderr) document = json.loads(receipt.read_text(encoding="utf-8")) self.assertEqual(document["artifact"]["sha256"], digest(self.fixture.binary_bytes))
def test_receipt_is_written_private_and_secret_free(self) -> None: receipt = self.root / "receipt.json" completed = self.publish(receipt=receipt) self.assertEqual(completed.returncode, 0, completed.stderr) document = json.loads(receipt.read_text(encoding="utf-8")) self.assertEqual(document["status"], "published") self.assertEqual(document["owner_did"], OWNER_DID) self.assertEqual([entry["role"] for entry in document["objects"]], ["binary-archive", "source-archive", "checksums", "release-manifest"]) for entry in document["objects"]: self.assertEqual(entry["status"], "published") self.assertEqual(entry["record"]["rkey"], "3kfixture") self.assertEqual(document["tag_bytes"], publisher.tag_object_bytes_base64(self.fixture.tag_object)) self.assertEqual(document["reviewed_schema"]["id"], "sh.tangled.repo.artifact") self.assertEqual(document["release"]["tag_commit"], self.fixture.tag_commit) self.assertEqual(document["release"]["verification_states"]["binary_elf_target"], "verified") self.assertEqual(document["source_archive"]["sha256"], digest(self.fixture.source_bytes)) self.assertEqual(document["checksums"]["name"], CHECKSUM_NAME) self.assertEqual(sorted(entry["path"] for entry in document["host_binaries"]), [f"bin/{name}" for name in sorted(HOST_BINARIES)]) self.assertEqual(receipt.stat().st_mode & 0o777, 0o600) text = receipt.read_text(encoding="utf-8") self.assertNotIn(PASSWORD, text) self.assertNotIn(JWT, text) self.assertNotIn(PASSWORD, completed.stdout + completed.stderr) self.assertNotIn(JWT, completed.stdout + completed.stderr)
def test_handle_identifier_must_resolve_to_the_owner(self) -> None: self.pds.resolve_did = OTHER_DID completed = self.publish() self.assert_fails(completed, "owner_mismatch", publisher.EXIT_PREFLIGHT) self.assertEqual(self.pds.route("/xrpc/com.atproto.server.createSession"), [])
def test_repo_owner_handle_must_resolve_to_the_owner(self) -> None: self.pds.resolve_map = {HANDLE: OTHER_DID} self.assert_fails(self.publish(identifier=OWNER_DID), "repo_owner_mismatch", publisher.EXIT_PREFLIGHT)
def test_authenticated_did_must_match_the_owner_before_upload(self) -> None: self.pds.session_did = OTHER_DID completed = self.publish(identifier=OWNER_DID) self.assert_fails(completed, "owner_mismatch", publisher.EXIT_REMOTE) self.assertEqual(self.pds.route("/xrpc/com.atproto.repo.uploadBlob"), [])
def test_redirect_is_never_followed(self) -> None: self.pds.redirect_paths = {"/xrpc/com.atproto.server.createSession"} completed = self.publish(identifier=OWNER_DID) self.assert_fails(completed, "remote_redirect", publisher.EXIT_REMOTE) self.assertEqual(self.pds.route("/xrpc/com.atproto.repo.uploadBlob"), []) self.assertEqual(self.pds.route("/xrpc/com.atproto.repo.createRecord"), [])
def test_oversized_response_is_refused(self) -> None: self.pds.oversize_paths = {"/xrpc/com.atproto.server.createSession"} completed = self.publish(identifier=OWNER_DID, extra=["--max-response-bytes", "64"]) self.assert_fails(completed, "remote_response_too_large", publisher.EXIT_REMOTE)
def test_remote_error_body_is_redacted(self) -> None: self.pds.session_status = 500 self.pds.session_error_body = f"upstream rejected credentials {PASSWORD} token {JWT}" completed = self.publish(identifier=OWNER_DID) self.assert_fails(completed, "remote_error", publisher.EXIT_REMOTE) combined = completed.stdout + completed.stderr self.assertNotIn(PASSWORD, combined) self.assertNotIn(JWT, combined) self.assertNotIn("upstream rejected", combined) # The message is a stable operation + status; the remote body is never echoed. self.assertEqual(json.loads(completed.stdout)["error"]["message"], "createSession was rejected with HTTP 500")
def test_create_record_is_never_retried_and_reports_partial_state(self) -> None: self.pds.create_status = 500 completed = self.publish() self.assert_fails(completed, "publish_ambiguous", publisher.EXIT_REMOTE) self.assertEqual(len(self.pds.route("/xrpc/com.atproto.repo.uploadBlob")), 1) self.assertEqual(len(self.pds.route("/xrpc/com.atproto.repo.createRecord")), 1) self.assertIn("Reconcile manually", completed.stdout)
def test_create_record_without_a_uri_is_ambiguous(self) -> None: self.pds.create_omit_uri = True completed = self.publish() self.assert_fails(completed, "publish_ambiguous", publisher.EXIT_REMOTE) self.assertEqual(len(self.pds.route("/xrpc/com.atproto.repo.createRecord")), 1)
def test_partial_receipt_is_written_for_reconciliation(self) -> None: self.pds.create_status = 500 receipt = self.root / "partial.json" self.assert_fails(self.publish(receipt=receipt), "publish_ambiguous", publisher.EXIT_REMOTE) document = json.loads(receipt.read_text(encoding="utf-8")) self.assertEqual(document["status"], "partial") self.assertIn(publisher.content_cid(self.fixture.binary_bytes), document["ambiguity"]) self.assertEqual(document["objects"][0]["status"], "record_rejected") self.assertNotIn(PASSWORD, json.dumps(document))
def test_upload_failure_is_reported_and_never_retried(self) -> None: self.pds.upload_status = 500 completed = self.publish() self.assert_fails(completed, "remote_error", publisher.EXIT_REMOTE) self.assertEqual(len(self.pds.route("/xrpc/com.atproto.repo.uploadBlob")), 1) self.assertEqual(self.pds.route("/xrpc/com.atproto.repo.createRecord"), [])
def test_did_identifier_skips_handle_resolution(self) -> None: completed = self.publish(identifier=OWNER_DID, repo=f"https://tangled.org/{OWNER_DID}/ultimate-tool") self.assertEqual(completed.returncode, 0, completed.stderr) self.assertEqual(self.pds.route("/xrpc/com.atproto.identity.resolveHandle"), [])
def test_at_uri_repo_reference_is_used_verbatim(self) -> None: value = f"at://{OWNER_DID}/sh.tangled.repo/3kfixture" completed = self.publish(repo=value, identifier=OWNER_DID) self.assertEqual(completed.returncode, 0, completed.stderr) self.assertEqual(self.pds.bodies_of("/xrpc/com.atproto.repo.createRecord")[0]["record"]["repo"], value)
def test_reviewed_schema_pin_is_enforced(self) -> None: completed = self.publish(extra=["--reviewed-schema-sha256", "0" * 64]) self.assert_fails(completed, "reviewed_schema_mismatch") self.assertEqual(self.pds.requests, [])
def test_artifact_name_override_is_reported(self) -> None: completed = self.publish(extra=["--artifact-name", "renamed.tar.gz"]) self.assertEqual(completed.returncode, 0, completed.stderr) records = self.pds.bodies_of("/xrpc/com.atproto.repo.createRecord") self.assertEqual(records[0]["record"]["name"], "renamed.tar.gz") # Only the binary record is renamed; the other objects keep their names. self.assertEqual(records[1]["record"]["name"], SOURCE_ARCHIVE_NAME) self.assertEqual(records[2]["record"]["name"], CHECKSUM_NAME) self.assertEqual(records[3]["record"]["name"], "release-manifest.json") checks = {check["name"]: check for check in self.envelope(completed)["data"]["checks"]} self.assertEqual(checks["artifact_name_override"]["status"], "not_checked")
class ReceiptSafetyTests(unittest.TestCase): def test_refuses_a_receipt_containing_a_credential(self) -> None: with self.assertRaises(publisher.PublishError) as caught: publisher.assert_receipt_is_secret_free({"leak": PASSWORD}, (PASSWORD,)) self.assertEqual(caught.exception.code, "receipt_unsafe")
def test_refuses_a_receipt_containing_a_session_token(self) -> None: with self.assertRaises(publisher.PublishError) as caught: publisher.assert_receipt_is_secret_free({"token": JWT}, ()) self.assertEqual(caught.exception.code, "receipt_unsafe")
def test_redaction_summarizes_and_bounds_remote_text(self) -> None: redacted = publisher.redact(f"failure {PASSWORD} {JWT} " + "y" * 1000, (PASSWORD,)) self.assertNotIn(PASSWORD, redacted) self.assertNotIn(JWT, redacted) self.assertLessEqual(len(redacted), 400)
# --------------------------------------------------------------------------# Independent release preflight (the shared checker)# --------------------------------------------------------------------------
class IndependentReleasePreflightTests(FixtureTestCase): """The publisher's last preflight step is the shared independent checker."""
def test_the_shared_checker_is_loaded_from_beside_the_tool(self) -> None: module = publisher.load_release_checker() self.assertTrue(callable(module.validate_release)) self.assertEqual(Path(module.__file__).resolve(), (TOOLS / "check_release.py").resolve())
def test_the_fixture_is_a_genuine_packager_archive(self) -> None: """The release under test holds real tar members, not hand-written bytes.""" self.fixture.prepare() archive = self.fixture.release_dir / ARCHIVE_NAME prefix = f"ultimate-tool-{VERSION}-linux-x86_64-gnu" with tarfile.open(archive, "r:gz") as tar: names = set(tar.getnames()) self.assertIn(f"{prefix}/bin/ultimate-tool", names) self.assertIn(f"{prefix}/bin/ultimate-tool-mcp", names) self.assertIn(f"{prefix}/LICENSE", names) self.assertEqual(archive.read_bytes()[:2], b"\x1f\x8b") # a real gzip stream
def test_a_genuine_release_passes_the_gate(self) -> None: self.fixture.prepare() verdict = publisher.release_preflight(self.fixture.release_dir, source_root=self.fixture.repo) self.assertTrue(verdict["ready"], verdict.get("blockers")) states = {check["check"]: check["state"] for check in verdict["checks"]} for name in ("checksums", "mandatory_artifacts", "archive_safety", "in_archive_manifest", "binary_remeasured", "source_archive_contents"): with self.subTest(check=name): self.assertEqual(states.get(name), "verified")
def test_a_tampered_release_is_refused(self) -> None: self.fixture.prepare() (self.fixture.release_dir / ARCHIVE_NAME).write_bytes(b"tampered archive bytes") verdict = publisher.release_preflight(self.fixture.release_dir, source_root=self.fixture.repo) self.assertFalse(verdict["ready"]) self.assertTrue(verdict["blockers"])
def test_the_gate_asks_the_checker_for_a_ready_release(self) -> None: self.fixture.prepare() seen: dict = {}
class RecordingChecker: def validate_release(self, directory, source_root=None, require_ready=False): seen.update(directory=str(directory), source_root=str(source_root), require_ready=require_ready) return {"ready": True, "verdict": "passed", "checks": [], "blockers": [], "counts": {}}
verdict = publisher.release_preflight( self.fixture.release_dir, source_root=self.fixture.repo, checker=RecordingChecker(), ) self.assertTrue(verdict["ready"]) self.assertEqual(seen["directory"], str(self.fixture.release_dir)) self.assertEqual(seen["source_root"], str(self.fixture.repo)) self.assertIs(seen["require_ready"], True)
def test_a_checker_without_a_verdict_is_a_clean_failure(self) -> None: class BrokenChecker: def validate_release(self, directory, source_root=None, require_ready=False): return {}
with self.assertRaises(publisher.PublishError) as caught: publisher.release_preflight(self.fixture.release_dir, checker=BrokenChecker()) self.assertEqual(caught.exception.code, "release_check_failed")
class ReleasePreflightEnforcementTests(FixtureTestCase): """The gate cannot be reached past, and runs before identity or secret."""
def options(self, **kwargs): return publisher.build_parser().parse_args(self.fixture.cli_args(**kwargs)[2:])
def test_publication_is_refused_before_identity_or_secret_when_the_gate_fails(self) -> None: self.fixture.prepare() options = self.options(publish=True, reviewed_schema=True, max_blob_bytes=1 << 20) failing = {"ready": False, "verdict": "failed", "checks": [], "blockers": ["injected failure"], "counts": {}} with unittest.mock.patch.object(publisher, "release_preflight", return_value=failing) as gate, \ unittest.mock.patch.object(publisher, "resolve_identity") as resolve, \ unittest.mock.patch.object(publisher, "acquire_secret") as secret: with self.assertRaises(publisher.PublishError) as caught: publisher.run(options) self.assertEqual(caught.exception.code, "release_preflight_failed") self.assertEqual(caught.exception.exit_code, publisher.EXIT_PREFLIGHT) gate.assert_called_once() resolve.assert_not_called() secret.assert_not_called()
def test_publication_proceeds_past_the_gate_only_once_it_passes(self) -> None: self.fixture.prepare() options = self.options(publish=True, reviewed_schema=True, max_blob_bytes=1 << 20) order: list[str] = [] passing = {"ready": True, "verdict": "passed", "checks": [], "blockers": [], "counts": {}} with unittest.mock.patch.object(publisher, "release_preflight", return_value=passing), \ unittest.mock.patch.object( publisher, "resolve_identity", side_effect=lambda *a, **k: order.append("identity") or {"identifier": HANDLE, "did": OWNER_DID}, ), \ unittest.mock.patch.object( publisher, "acquire_secret", side_effect=publisher.PublishError("secret_unavailable", "stop here", exit_code=publisher.EXIT_REMOTE), ): with self.assertRaises(publisher.PublishError): publisher.run(options) self.assertEqual(order, ["identity"])
# --------------------------------------------------------------------------# Hardening: the complete object set, canonical inputs, staging, blob identity,# lexicon constraints, receipts, redaction and deadlines# --------------------------------------------------------------------------
class ObjectSetAndCanonicalInputTests(FixtureTestCase): def run_options(self, **extra: str): argv = ["--release-manifest", str(self.fixture.manifest_path), "--repo", REPO_URL, "--owner-did", OWNER_DID, "--identifier", HANDLE, "--tag", TAG, "--repo-dir", str(self.fixture.repo), "--pds-origin", "https://tngl.sh"] for key, value in extra.items(): argv += ["--" + key.replace("_", "-"), value] return publisher.build_parser().parse_args(argv)
def test_the_dry_run_plans_the_complete_published_set(self) -> None: self.fixture.prepare() result = publisher.run(self.run_options()) names = [entry["name"] for entry in result["objects"]] self.assertEqual(names, [ARCHIVE_NAME, SOURCE_ARCHIVE_NAME, CHECKSUM_NAME, "release-manifest.json"]) self.assertEqual([entry["role"] for entry in result["objects"]], ["binary-archive", "source-archive", "checksums", "release-manifest"]) # Every planned record is bound to the annotated tag object. tags = {json.dumps(record["tag"]) for record in result["would_create"]["records"]} self.assertEqual(tags, {json.dumps({"$bytes": result["would_create"]["records"][0]["tag"]["$bytes"]})})
def test_a_noncanonical_checksums_path_is_refused_before_any_secret(self) -> None: self.fixture.prepare() elsewhere = self.fixture.root / "elsewhere-sums.txt" elsewhere.write_text(self.fixture.checksums_path.read_text(encoding="utf-8"), encoding="utf-8") with self.assertRaises(publisher.PublishError) as caught: publisher.run(self.run_options(checksums=str(elsewhere))) self.assertEqual(caught.exception.code, "noncanonical_checksums")
def test_a_noncanonical_manifest_path_is_refused(self) -> None: self.fixture.prepare() alias = self.fixture.release_dir / "copy-of-manifest.json" alias.write_text(self.fixture.manifest_path.read_text(encoding="utf-8"), encoding="utf-8") options = publisher.build_parser().parse_args( ["--release-manifest", str(alias), "--repo", REPO_URL, "--owner-did", OWNER_DID, "--identifier", HANDLE, "--tag", TAG, "--repo-dir", str(self.fixture.repo), "--pds-origin", "https://tngl.sh"] ) with self.assertRaises(publisher.PublishError) as caught: publisher.run(options) self.assertEqual(caught.exception.code, "noncanonical_manifest")
def test_a_symlinked_checksum_file_is_refused(self) -> None: self.fixture.prepare() real = self.fixture.checksums_path.read_bytes() target = self.fixture.release_dir / "real-sums" target.write_bytes(real) self.fixture.checksums_path.unlink() self.fixture.checksums_path.symlink_to(target) with self.assertRaises(publisher.PublishError) as caught: publisher.run(self.run_options()) self.assertEqual(caught.exception.code, "unsafe_path")
def test_staging_selects_only_reviewed_files(self) -> None: self.fixture.prepare() # A private file dropped into the release directory must never be staged. (self.fixture.release_dir / "private-notes.txt").write_text("operator secret\n", encoding="utf-8") _holder, staging, _staged = publisher.stage_release( self.fixture.release_dir, publisher.required_publish_objects( publisher.load_release_manifest(self.fixture.manifest_path), checksums_name=CHECKSUM_NAME, ), file_names=(ARCHIVE_NAME, SOURCE_ARCHIVE_NAME, CHECKSUM_NAME, "release-manifest.json", "source-manifest.json"), ) self.assertFalse((staging / "private-notes.txt").exists()) self.assertTrue((staging / CHECKSUM_NAME).is_file())
def test_the_evidence_file_cannot_escape_the_release(self) -> None: with self.assertRaises(publisher.PublishError) as caught: publisher.relative_safe_path("../escape.json", field="staged release file") self.assertEqual(caught.exception.code, "release_manifest_invalid")
def test_the_dry_run_validates_every_record_against_the_lexicon(self) -> None: self.fixture.prepare() # A lexicon that only accepts octet-stream blobs must reject the JSON # manifest record (and the text/plain checksum record) before any upload. document = reviewed_schema_document() document["defs"]["main"]["record"]["properties"]["artifact"] = { "type": "blob", "accept": ["application/octet-stream"], } self.fixture.schema_path.write_text(json.dumps(document), encoding="utf-8") completed = self.fixture.run_cli(self.fixture.cli_args(reviewed_schema=True)) self.assert_fails(completed, "record_schema_mismatch")
def test_every_object_is_bounded_before_any_secret(self) -> None: self.fixture.prepare() # --max-blob-bytes applies to every object, not just the binary archive. manifest_size = (self.fixture.release_dir / "release-manifest.json").stat().st_size with self.assertRaises(publisher.PublishError) as caught: publisher.run(self.run_options(**{"max_blob_bytes": str(manifest_size - 1)})) self.assertEqual(caught.exception.code, "blob_too_large")
class BlobIdentityTests(FixtureTestCase): """The uploaded blob reference must be content-derived before a record is made."""
def setUp(self) -> None: super().setUp() self.fixture.prepare() self.pds = MockPds() self.pds.start() self.addCleanup(self.pds.stop)
def publish(self, **kwargs) -> subprocess.CompletedProcess: defaults = dict(publish=True, reviewed_schema=True, max_blob_bytes=1 << 20, mock_origin=self.pds.origin, allow_loopback=True, pds_origin=None) defaults.update(kwargs) environment = self.fixture.cli_env(**{ publisher.MOCK_ENV_GUARD: "1", publisher.DEFAULT_PASSWORD_ENV: PASSWORD, }) return self.fixture.run_cli(self.fixture.cli_args(**defaults), env=environment)
def test_a_blob_cid_for_other_bytes_is_refused(self) -> None: self.pds.upload_force_cid = publisher.content_cid(b"other bytes entirely") completed = self.publish() self.assert_fails(completed, "blob_content_mismatch", publisher.EXIT_REMOTE) self.assertEqual(self.pds.route("/xrpc/com.atproto.repo.createRecord"), [])
def test_a_lying_blob_size_is_refused(self) -> None: self.pds.upload_force_size = 1 completed = self.publish() self.assert_fails(completed, "blob_size_mismatch", publisher.EXIT_REMOTE) self.assertEqual(self.pds.route("/xrpc/com.atproto.repo.createRecord"), [])
def test_a_lying_blob_mime_is_refused(self) -> None: self.pds.upload_force_mime = "text/html" completed = self.publish() self.assert_fails(completed, "blob_mime_mismatch", publisher.EXIT_REMOTE) self.assertEqual(self.pds.route("/xrpc/com.atproto.repo.createRecord"), [])
def test_a_partial_result_is_structured_even_without_a_receipt(self) -> None: self.pds.create_status = 500 completed = self.publish() # no --receipt self.assertEqual(completed.returncode, publisher.EXIT_REMOTE) envelope = json.loads(completed.stdout) partial = envelope["error"]["partial"] self.assertFalse(partial["receipt_written"]) self.assertEqual(partial["workflow"][0]["status"], "record_rejected") self.assertEqual(partial["owner_did"], OWNER_DID) self.assertNotIn(PASSWORD, completed.stdout)
class ReceiptFileSafetyTests(FixtureTestCase): def setUp(self) -> None: super().setUp() self.fixture.prepare() self.pds = MockPds() self.pds.start() self.addCleanup(self.pds.stop)
def publish(self, receipt) -> subprocess.CompletedProcess: args = self.fixture.cli_args(publish=True, reviewed_schema=True, max_blob_bytes=1 << 20, mock_origin=self.pds.origin, allow_loopback=True, pds_origin=None, receipt=receipt) environment = self.fixture.cli_env(**{ publisher.MOCK_ENV_GUARD: "1", publisher.DEFAULT_PASSWORD_ENV: PASSWORD, }) return self.fixture.run_cli(args, env=environment)
def test_an_existing_receipt_path_is_never_overwritten(self) -> None: receipt = self.root / "receipt.json" receipt.write_text("operator sentinel\n", encoding="utf-8") self.assert_fails(self.publish(receipt), "receipt_unsafe") self.assertEqual(receipt.read_text(encoding="utf-8"), "operator sentinel\n") self.assertEqual(self.pds.requests, [])
def test_a_fifo_receipt_path_is_refused(self) -> None: receipt = self.root / "receipt.fifo" os.mkfifo(receipt) self.assert_fails(self.publish(receipt), "receipt_unsafe") self.assertEqual(self.pds.requests, [])
class LexiconConstraintTests(FixtureTestCase): def schema(self, mutate=None) -> dict: document = reviewed_schema_document() if mutate is not None: mutate(document) path = self.root / "schema.json" path.write_text(json.dumps(document), encoding="utf-8") return publisher.load_reviewed_schema(path, expected_sha256=None)
def base_record(self) -> dict: return { "$type": "sh.tangled.repo.artifact", "tag": {"$bytes": publisher.tag_object_bytes_base64("ab" * 20)}, "name": "ultimate-tool.tar.gz", "repo": REPO_URL, "artifact": {"$type": "blob", "ref": {"$link": publisher.content_cid(b"x")}, "mimeType": "application/octet-stream", "size": 1}, "createdAt": "2026-01-01T00:00:00+00:00", }
def test_unknown_field_types_fail_closed(self) -> None: with self.assertRaises(publisher.PublishError) as caught: self.schema(lambda doc: doc["defs"]["main"]["record"]["properties"].update( {"tags": {"type": "array", "items": {"type": "string"}}})) self.assertEqual(caught.exception.code, "reviewed_schema_unsupported")
def test_extra_required_fields_fail_closed(self) -> None: with self.assertRaises(publisher.PublishError) as caught: self.schema(lambda doc: doc["defs"]["main"]["record"]["required"].append("licence")) self.assertIn(caught.exception.code, ("reviewed_schema_invalid", "reviewed_schema_unsupported"))
def test_wrong_datetime_is_refused(self) -> None: schema = self.schema() record = self.base_record() record["createdAt"] = "yesterday" with self.assertRaises(publisher.PublishError) as caught: publisher.validate_record_against_schema(record, schema) self.assertEqual(caught.exception.code, "record_schema_mismatch")
def test_over_long_tag_is_refused(self) -> None: schema = self.schema(lambda doc: doc["defs"]["main"]["record"]["properties"].update( {"tag": {"type": "bytes", "maxLength": 4}})) record = self.base_record() with self.assertRaises(publisher.PublishError) as caught: publisher.validate_record_against_schema(record, schema) self.assertEqual(caught.exception.code, "record_schema_mismatch")
def test_blob_max_size_is_enforced(self) -> None: schema = self.schema(lambda doc: doc["defs"]["main"]["record"]["properties"].update( {"artifact": {"type": "blob", "accept": ["*/*"], "maxSize": 1}})) record = self.base_record() record["artifact"]["size"] = 100 with self.assertRaises(publisher.PublishError) as caught: publisher.validate_record_against_schema(record, schema) self.assertEqual(caught.exception.code, "record_schema_mismatch")
def test_blob_mime_accept_is_enforced(self) -> None: schema = self.schema(lambda doc: doc["defs"]["main"]["record"]["properties"].update( {"artifact": {"type": "blob", "accept": ["application/octet-stream"]}})) record = self.base_record() record["artifact"]["mimeType"] = "text/plain" with self.assertRaises(publisher.PublishError) as caught: publisher.validate_record_against_schema(record, schema) self.assertEqual(caught.exception.code, "record_schema_mismatch")
def test_a_valid_record_passes_all_constraints(self) -> None: publisher.validate_record_against_schema(self.base_record(), self.schema())
def test_an_unknown_constraint_key_fails_closed(self) -> None: with self.assertRaises(publisher.PublishError) as caught: self.schema(lambda doc: doc["defs"]["main"]["record"]["properties"].update( {"name": {"type": "string", "maxGraphemes": 3}})) self.assertEqual(caught.exception.code, "reviewed_schema_unsupported")
def test_an_unsupported_format_fails_closed(self) -> None: with self.assertRaises(publisher.PublishError) as caught: self.schema(lambda doc: doc["defs"]["main"]["record"]["properties"].update( {"createdAt": {"type": "string", "format": "cron"}})) self.assertEqual(caught.exception.code, "reviewed_schema_unsupported")
def test_an_impossible_calendar_date_is_refused(self) -> None: schema = self.schema() record = self.base_record() record["createdAt"] = "2026-13-45T00:00:00+00:00" # matches the regex, not the calendar with self.assertRaises(publisher.PublishError) as caught: publisher.validate_record_against_schema(record, schema) self.assertEqual(caught.exception.code, "record_schema_mismatch")
class TimeoutAndRedactionTests(FixtureTestCase): def test_non_finite_timeouts_are_rejected(self) -> None: for flag, value in (("--timeout", "nan"), ("--timeout", "inf"), ("--total-timeout", "nan")): with self.subTest(flag=flag, value=value): options = publisher.build_parser().parse_args( ["--release-manifest", "x", "--repo", "x", "--owner-did", "x", "--identifier", "x", "--tag", "x", flag, value] ) with self.assertRaises(publisher.PublishError) as caught: publisher.run(options) self.assertEqual(caught.exception.code, "usage")
def test_the_total_deadline_bounds_a_request(self) -> None: transport = publisher.HttpsTransport( {"scheme": "https", "host": "tngl.sh", "port": None, "origin": "https://tngl.sh"}, timeout=5.0, deadline=time.monotonic() - 1.0, ) with self.assertRaises(publisher.PublishError) as caught: transport.request("GET", "/xrpc/probe") self.assertEqual(caught.exception.code, "remote_timeout")
def test_the_access_token_is_redacted(self) -> None: token = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ4In0.sig" self.assertEqual(publisher.redact(f"rejected {token}", (token,)), "rejected <redacted>")
class ErrorRedactionTests(FixtureTestCase): """Remote error bodies never leak a token, a password or a DID."""
def setUp(self) -> None: super().setUp() self.fixture.prepare() self.pds = MockPds() self.pds.start() self.addCleanup(self.pds.stop)
def publish(self) -> subprocess.CompletedProcess: args = self.fixture.cli_args(publish=True, reviewed_schema=True, max_blob_bytes=1 << 20, mock_origin=self.pds.origin, allow_loopback=True, pds_origin=None) environment = self.fixture.cli_env(**{ publisher.MOCK_ENV_GUARD: "1", publisher.DEFAULT_PASSWORD_ENV: PASSWORD, }) return self.fixture.run_cli(args, env=environment)
def test_upload_error_body_is_redacted(self) -> None: self.pds.upload_status = 500 self.pds.upload_error_body = f"rejected password {PASSWORD} token {JWT} owner {OWNER_DID}" completed = self.publish() self.assert_fails(completed, "remote_error", publisher.EXIT_REMOTE) message = json.loads(completed.stdout)["error"]["message"] # The remote body is never echoed; the message is a stable operation + status. self.assertEqual(message, "uploadBlob was rejected with HTTP 500") self.assertNotIn(PASSWORD, message) self.assertNotIn(JWT, message) self.assertNotIn(OWNER_DID, message) self.assertNotIn(self.pds.upload_error_body, completed.stdout + completed.stderr)
def test_create_error_body_is_redacted_in_the_partial(self) -> None: self.pds.create_status = 500 self.pds.create_error_body = f"token {JWT} not permitted for {OWNER_DID}" completed = self.publish() self.assert_fails(completed, "publish_ambiguous", publisher.EXIT_REMOTE) combined = completed.stdout + completed.stderr self.assertNotIn(PASSWORD, combined) self.assertNotIn(JWT, combined) self.assertNotIn("not permitted for", combined) envelope = json.loads(completed.stdout) self.assertEqual(envelope["error"]["partial"]["workflow"][0]["status"], "record_rejected")
def test_a_non_json_create_response_is_a_structured_partial(self) -> None: self.pds.create_not_json = True completed = self.publish() self.assert_fails(completed, "remote_invalid_response", publisher.EXIT_REMOTE) envelope = json.loads(completed.stdout) self.assertEqual(envelope["error"]["partial"]["workflow"][0]["status"], "creating") self.assertEqual(len(self.pds.route("/xrpc/com.atproto.repo.createRecord")), 1)
class UploadStreamTests(unittest.TestCase): """The uploaded bytes are proven against the reviewed snapshot."""
class _CaptureTransport: def __init__(self) -> None: self.data = b""
def request(self, method, path, *, body=None, headers=None, secrets=()): data = b"" while True: chunk = body.read(65536) if not chunk: break data += chunk self.data = data blob = {"$type": "blob", "ref": {"$link": publisher.content_cid(data)}, "mimeType": (headers or {}).get("Content-Type"), "size": len(data)} return publisher.HttpResponse(200, {}, json.dumps({"blob": blob}).encode())
def _write(self, data: bytes) -> Path: holder = tempfile.TemporaryDirectory() self.addCleanup(holder.cleanup) path = Path(holder.name) / "object.bin" path.write_bytes(data) return path
def test_a_shrunk_stream_is_detected(self) -> None: payload = b"abcdefgh" path = self._write(payload) client = publisher.AtProtoClient(self._CaptureTransport()) with self.assertRaises(publisher.PublishError) as caught: client.upload_blob(path, size=len(payload) + 10, sha256=publisher.sha256_hex(payload), content_type="application/octet-stream", access_jwt="token") self.assertEqual(caught.exception.code, "upload_tampered")
def test_a_changed_digest_is_detected(self) -> None: path = self._write(b"abcdefgh") client = publisher.AtProtoClient(self._CaptureTransport()) with self.assertRaises(publisher.PublishError) as caught: client.upload_blob(path, size=8, sha256=publisher.sha256_hex(b"different!"), content_type="application/octet-stream", access_jwt="token") self.assertEqual(caught.exception.code, "upload_tampered")
def test_a_correct_stream_is_accepted(self) -> None: payload = b"reviewed bytes" path = self._write(payload) transport = self._CaptureTransport() client = publisher.AtProtoClient(transport) blob = client.upload_blob(path, size=len(payload), sha256=publisher.sha256_hex(payload), content_type="application/octet-stream", access_jwt="token") self.assertEqual(publisher.parse_blob_cid(blob["ref"]["$link"]), publisher.sha256_hex(payload)) self.assertEqual(transport.data, payload)
def test_a_blob_without_a_mime_type_is_refused(self) -> None: class NoMimeTransport(UploadStreamTests._CaptureTransport): def request(self, method, path, *, body=None, headers=None, secrets=()): response = super().request(method, path, body=body, headers=headers, secrets=secrets) document = json.loads(response.body) document["blob"].pop("mimeType", None) return publisher.HttpResponse(200, {}, json.dumps(document).encode())
path = self._write(b"x") client = publisher.AtProtoClient(NoMimeTransport()) with self.assertRaises(publisher.PublishError) as caught: client.upload_blob(path, size=1, sha256=publisher.sha256_hex(b"x"), content_type="application/octet-stream", access_jwt="token") self.assertEqual(caught.exception.code, "blob_mime_mismatch")
class GitEnvironmentTests(unittest.TestCase): def test_the_git_helper_never_inherits_credentials(self) -> None: captured: dict = {}
class Result: returncode = 0 stdout = b"" stderr = b""
def fake_run(argv, **kwargs): captured["env"] = dict(kwargs.get("env", {})) return Result()
original = publisher.subprocess.run publisher.subprocess.run = fake_run saved = {name: os.environ.get(name) for name in ("C64U_PASSWORD", publisher.DEFAULT_PASSWORD_ENV)} os.environ["C64U_PASSWORD"] = "git-secret-one" os.environ[publisher.DEFAULT_PASSWORD_ENV] = "git-secret-two" try: publisher.run_git(Path("."), ["rev-parse", "HEAD"]) finally: publisher.subprocess.run = original for name, value in saved.items(): if value is None: os.environ.pop(name, None) else: os.environ[name] = value self.assertNotIn("C64U_PASSWORD", captured["env"]) self.assertNotIn(publisher.DEFAULT_PASSWORD_ENV, captured["env"])
class TrickleDeadlineTests(unittest.TestCase): """A slow trickle is bounded by the total deadline, not just socket inactivity."""
def _trickle_server(self, *, chunk_delay: float, chunks: int, http10: bool = False) -> int: server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind(("127.0.0.1", 0)) server.listen(1) port = server.getsockname()[1]
def serve() -> None: try: conn, _ = server.accept() except OSError: return try: conn.recv(65536) version = "HTTP/1.0" if http10 else "HTTP/1.1" close = "Connection: close\r\n" if http10 else "" conn.sendall(f"{version} 200 OK\r\n{close}Content-Length: {chunks}\r\n\r\n".encode()) for _ in range(chunks): time.sleep(chunk_delay) conn.sendall(b"x") except OSError: pass finally: try: conn.close() except OSError: pass server.close()
threading.Thread(target=serve, daemon=True).start() return port
def test_a_slow_trickle_is_bounded_by_the_total_deadline(self) -> None: port = self._trickle_server(chunk_delay=0.3, chunks=10) transport = publisher.HttpsTransport( {"scheme": "http", "host": "127.0.0.1", "port": port, "origin": f"http://127.0.0.1:{port}"}, timeout=5.0, deadline=time.monotonic() + 0.6, ) start = time.monotonic() with self.assertRaises(publisher.PublishError) as caught: transport.request("GET", "/xrpc/probe") elapsed = time.monotonic() - start self.assertEqual(caught.exception.code, "remote_timeout") self.assertLess(elapsed, 2.0) # bounded well under the trickle's own 3s duration
def test_a_connection_close_trickle_is_still_bounded(self) -> None: # HTTP/1.0 + Connection: close is where http.client drops its own socket # reference; the watchdog holds the real socket, so the deadline still holds. port = self._trickle_server(chunk_delay=0.3, chunks=10, http10=True) transport = publisher.HttpsTransport( {"scheme": "http", "host": "127.0.0.1", "port": port, "origin": f"http://127.0.0.1:{port}"}, timeout=5.0, deadline=time.monotonic() + 0.6, ) start = time.monotonic() with self.assertRaises(publisher.PublishError) as caught: transport.request("GET", "/xrpc/probe") self.assertEqual(caught.exception.code, "remote_timeout") self.assertLess(time.monotonic() - start, 2.0)
def test_a_stalled_dns_lookup_is_bounded(self) -> None: import subprocess as subprocess_module
def fake_run(*args, **kwargs): raise subprocess_module.TimeoutExpired(cmd=args[0] if args else "resolver", timeout=kwargs.get("timeout"))
original = publisher.subprocess.run publisher.subprocess.run = fake_run try: with self.assertRaises(publisher.PublishError) as caught: publisher.resolve_addresses("stalled.invalid", 443, time.monotonic() + 0.3) finally: publisher.subprocess.run = original self.assertEqual(caught.exception.code, "remote_timeout")
def _stall_server(self, prefix: bytes, *, read: bool = True) -> int: server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind(("127.0.0.1", 0)) server.listen(1) port = server.getsockname()[1]
def serve() -> None: try: conn, _ = server.accept() except OSError: return try: if read: conn.recv(65536) if prefix: conn.sendall(prefix) time.sleep(5.0) except OSError: pass finally: try: conn.close() except OSError: pass server.close()
threading.Thread(target=serve, daemon=True).start() return port
def test_a_stalled_response_header_is_bounded(self) -> None: port = self._stall_server(b"HTTP/1.1 200 OK\r\n") transport = publisher.HttpsTransport( {"scheme": "http", "host": "127.0.0.1", "port": port, "origin": f"http://127.0.0.1:{port}"}, timeout=5.0, deadline=time.monotonic() + 0.6, ) start = time.monotonic() with self.assertRaises(publisher.PublishError) as caught: transport.request("GET", "/xrpc/probe") self.assertEqual(caught.exception.code, "remote_timeout") self.assertLess(time.monotonic() - start, 2.0)
def test_a_slow_upload_is_bounded(self) -> None: port = self._stall_server(b"", read=False) # the server never reads the body transport = publisher.HttpsTransport( {"scheme": "http", "host": "127.0.0.1", "port": port, "origin": f"http://127.0.0.1:{port}"}, timeout=5.0, deadline=time.monotonic() + 0.6, ) start = time.monotonic() with self.assertRaises(publisher.PublishError) as caught: transport.request("POST", "/xrpc/com.atproto.repo.uploadBlob", body=b"x" * 4_000_000, headers={"Content-Length": "4000000"}) self.assertEqual(caught.exception.code, "remote_timeout") self.assertLess(time.monotonic() - start, 2.0)
def test_the_transport_resolves_the_host_exactly_once(self) -> None: calls: list[str] = []
def fake_resolve(host, port, deadline): calls.append(host) return [(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("127.0.0.1", port))]
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind(("127.0.0.1", 0)) server.listen(1) port = server.getsockname()[1]
def serve() -> None: conn, _ = server.accept() conn.recv(65536) conn.sendall(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok") conn.close() server.close()
threading.Thread(target=serve, daemon=True).start() original = publisher.resolve_addresses publisher.resolve_addresses = fake_resolve try: transport = publisher.HttpsTransport( {"scheme": "http", "host": "resolved.invalid", "port": port, "origin": f"http://resolved.invalid:{port}"}, timeout=5.0, ) response = transport.request("GET", "/xrpc/probe") finally: publisher.resolve_addresses = original self.assertEqual(response.status, 200) self.assertEqual(calls, ["resolved.invalid"]) # one lookup, consumed as a sockaddr
def test_an_exhausted_resolution_budget_is_refused(self) -> None: with self.assertRaises(publisher.PublishError) as caught: publisher.resolve_addresses("example.invalid", 80, time.monotonic() - 1.0) self.assertEqual(caught.exception.code, "remote_timeout")
class RemoteStateSanitizationTests(FixtureTestCase): """Remote-derived URIs/CIDs are validated and scrubbed before any output."""
def setUp(self) -> None: super().setUp() self.fixture.prepare() self.pds = MockPds() self.pds.start() self.addCleanup(self.pds.stop)
def publish(self, **kwargs) -> subprocess.CompletedProcess: defaults = dict(publish=True, reviewed_schema=True, max_blob_bytes=1 << 20, mock_origin=self.pds.origin, allow_loopback=True, pds_origin=None) defaults.update(kwargs) environment = self.fixture.cli_env(**{ publisher.MOCK_ENV_GUARD: "1", publisher.DEFAULT_PASSWORD_ENV: PASSWORD, }) return self.fixture.run_cli(self.fixture.cli_args(**defaults), env=environment)
def test_all_four_objects_publish_without_a_receipt_and_leak_nothing(self) -> None: completed = self.publish() self.assertEqual(completed.returncode, 0, completed.stderr) combined = completed.stdout + completed.stderr self.assertNotIn(JWT, combined) self.assertNotIn(PASSWORD, combined) self.assertEqual(len(self.pds.route("/xrpc/com.atproto.repo.uploadBlob")), 4) self.assertEqual(len(self.pds.route("/xrpc/com.atproto.repo.createRecord")), 4)
def test_a_record_uri_for_another_owner_is_refused_and_state_kept(self) -> None: self.pds.record_uri = f"at://{OTHER_DID}/sh.tangled.repo.artifact/3kfixture" completed = self.publish() self.assert_fails(completed, "publish_ambiguous", publisher.EXIT_REMOTE) partial = json.loads(completed.stdout)["error"]["partial"] # Every object's state is preserved, not discarded. self.assertEqual(len(partial["workflow"]), 4) self.assertEqual(partial["workflow"][0]["status"], "record_ambiguous") self.assertEqual(partial["workflow"][1]["status"], "pending") self.assertNotIn(OTHER_DID, json.dumps(partial))
def test_a_token_echoed_in_a_record_uri_is_refused_not_published(self) -> None: # A returned URI that echoes a credential is rejected, never scrubbed # into a claimed success. self.pds.record_uri = f"at://{OWNER_DID}/sh.tangled.repo.artifact/{JWT}" completed = self.publish() self.assert_fails(completed, "publish_ambiguous", publisher.EXIT_REMOTE) partial = json.loads(completed.stdout)["error"]["partial"] self.assertEqual(partial["workflow"][0]["status"], "record_ambiguous") self.assertIsNone(partial["workflow"][0]["record"]) self.assertNotIn(JWT, completed.stdout + completed.stderr)
def test_a_record_cid_that_is_not_a_dag_cbor_cid_is_refused(self) -> None: # A raw blob CID (0x55) is not a valid record CID (dag-cbor 0x71). self.pds.document_cid = publisher.content_cid(b"a raw blob, not a record") completed = self.publish() self.assert_fails(completed, "publish_ambiguous", publisher.EXIT_REMOTE)
def test_a_malformed_record_cid_is_refused(self) -> None: self.pds.document_cid = "bafyreinotarealcidentifier" completed = self.publish() self.assert_fails(completed, "publish_ambiguous", publisher.EXIT_REMOTE)
def test_a_token_echoed_after_a_first_success_is_sanitized_in_the_partial(self) -> None: # The first object publishes; the next fails, and the retained state is # still scrubbed of the token the error body echoed. self.pds.fail_create_after = 1 self.pds.create_error_body = f"token {JWT}" completed = self.publish() self.assert_fails(completed, "publish_ambiguous", publisher.EXIT_REMOTE) partial = json.loads(completed.stdout)["error"]["partial"] self.assertEqual(partial["workflow"][0]["status"], "published") self.assertEqual(partial["workflow"][1]["status"], "record_rejected") self.assertNotIn(JWT, completed.stdout + completed.stderr) self.assertNotIn(PASSWORD, completed.stdout + completed.stderr)
class ReceiptPersistTests(unittest.TestCase): def test_no_receipt_path_is_reported_as_not_written(self) -> None: self.assertEqual(publisher.persist_receipt(None, {"a": 1}), (False, None))
def test_a_broken_descriptor_is_reported_not_raised(self) -> None: fd = os.open(os.devnull, os.O_WRONLY) os.close(fd) # every subsequent write fails written, error = publisher.persist_receipt(fd, {"a": 1}) self.assertFalse(written) self.assertIsInstance(error, str)
class TlsDeadlineTests(unittest.TestCase): """The deadline covers TCP, the TLS handshake, send and read, on the live socket."""
class _FakeSocket: def __init__(self, *, connect_delay: float = 0.0): self.timeouts: list[float] = [] self._connect_delay = connect_delay self.closed = False self.shutdown_called = False
def settimeout(self, value): self.timeouts.append(value)
def setsockopt(self, *args): pass
def connect(self, sockaddr): time.sleep(self._connect_delay)
def bind(self, address): pass
def shutdown(self, how): self.shutdown_called = True
def close(self): self.closed = True
def do_handshake(self): raise ssl.SSLError("handshake stalled")
class _DistinctContext: """wrap_socket returns a DISTINCT socket, as the real SSL wrap does."""
def __init__(self): self.check_hostname = True self.verify_mode = ssl.CERT_REQUIRED self.server_hostname = None self.wrapped = None
def wrap_socket(self, sock, server_hostname=None, do_handshake_on_connect=True): self.server_hostname = server_hostname self.wrapped = TlsDeadlineTests._FakeSocket() return self.wrapped
def test_the_tls_wrap_and_handshake_use_the_refreshed_remaining_budget(self) -> None: raw = self._FakeSocket(connect_delay=0.5) context = self._DistinctContext() original = publisher.socket.socket publisher.socket.socket = lambda *args, **kwargs: raw try: connection = publisher._SockaddrHTTPSConnection( "origin.example", 443, resolved=[(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("127.0.0.1", 443))], deadline=time.monotonic() + 0.6, timeout=5.0, context=context, ) start = time.monotonic() with self.assertRaises(ssl.SSLError): connection.connect() elapsed = time.monotonic() - start finally: publisher.socket.socket = original # TCP (0.5) refreshed at the raw socket, and the wrapped socket is bounded # to the *remaining* budget immediately before the explicit handshake. self.assertGreaterEqual(len(raw.timeouts), 1) self.assertLessEqual(raw.timeouts[-1], 0.6) self.assertLess(context.wrapped.timeouts[-1], raw.timeouts[0]) self.assertLessEqual(context.wrapped.timeouts[-1], 0.2) self.assertLess(elapsed, 2.0) self.assertEqual(context.server_hostname, "origin.example") self.assertTrue(context.check_hostname) self.assertEqual(context.verify_mode, ssl.CERT_REQUIRED)
def test_tls_ownership_transfers_to_the_wrapped_socket(self) -> None: raw = self._FakeSocket() context = self._DistinctContext() recorded: dict = {}
class RecordingWatchdog: def arm(self, sock): recorded["armed"] = sock
def replace_socket(self, sock): recorded["replaced"] = sock
@property def armed(self): return True
original = publisher.socket.socket publisher.socket.socket = lambda *args, **kwargs: raw try: connection = publisher._SockaddrHTTPSConnection( "origin.example", 443, resolved=[(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("127.0.0.1", 443))], deadline=time.monotonic() + 5.0, watchdog=RecordingWatchdog(), timeout=5.0, context=context, ) with self.assertRaises(ssl.SSLError): connection.connect() finally: publisher.socket.socket = original self.assertIs(recorded["armed"], raw) # armed with the raw socket before TCP self.assertIs(recorded["replaced"], context.wrapped) # ownership moved to the live SSLSocket self.assertIsNot(recorded["replaced"], raw)
def test_replace_socket_moves_the_deadline_to_the_new_socket(self) -> None: watchdog = publisher._DeadlineWatchdog(deadline=time.monotonic() + 0.05) raw = self._FakeSocket() wrapped = self._FakeSocket() watchdog.arm(raw) watchdog.replace_socket(wrapped) self.assertTrue(watchdog.armed) self.assertIs(watchdog.owned_socket, wrapped) time.sleep(0.2) # the single timer fires on the live (replaced) socket self.assertTrue(wrapped.shutdown_called) self.assertFalse(raw.shutdown_called) watchdog.close()
def test_a_failed_address_does_not_start_a_second_timer(self) -> None: watchdog = publisher._DeadlineWatchdog(deadline=time.monotonic() + 5.0) first = self._FakeSocket() second = self._FakeSocket() watchdog.arm(first) timer = watchdog._timer watchdog.arm(second) # a second address attempt reuses the one timer self.assertIs(watchdog._timer, timer) self.assertIs(watchdog.owned_socket, second) watchdog.close()
def test_an_expired_replace_is_refused_before_the_handshake(self) -> None: watchdog = publisher._DeadlineWatchdog(deadline=time.monotonic() - 1.0) wrapped = self._FakeSocket() with self.assertRaises(publisher.PublishError) as caught: watchdog.replace_socket(wrapped) self.assertEqual(caught.exception.code, "remote_timeout") self.assertTrue(wrapped.shutdown_called) watchdog.close()
def test_close_joins_the_timer_and_leaves_no_leftover(self) -> None: watchdog = publisher._DeadlineWatchdog(deadline=time.monotonic() + 30.0) sock = self._FakeSocket() watchdog.arm(sock) timer = watchdog._timer self.assertIsNotNone(timer) self.assertTrue(timer.is_alive()) watchdog.close() # Cancelled *and joined*: no timer thread survives the request. self.assertFalse(timer.is_alive()) self.assertIsNone(watchdog._timer) self.assertFalse(watchdog.armed) self.assertFalse(sock.shutdown_called) # closed before the deadline, so no fire
class ResolverEnvTests(unittest.TestCase): def test_the_resolver_child_gets_no_credential_environment(self) -> None: captured: dict = {}
class Result: returncode = 0 stdout = b'[[2, 1, 6, "", ["127.0.0.1", 80]]]'
def fake_run(*args, **kwargs): captured["env"] = dict(kwargs.get("env", {})) return Result()
original = publisher.subprocess.run saved = {name: os.environ.get(name) for name in ("C64U_PASSWORD", publisher.DEFAULT_PASSWORD_ENV)} os.environ["C64U_PASSWORD"] = "resolver-secret-one" os.environ[publisher.DEFAULT_PASSWORD_ENV] = "resolver-secret-two" publisher.subprocess.run = fake_run try: addresses = publisher.resolve_addresses("example.invalid", 80, time.monotonic() + 5.0) finally: publisher.subprocess.run = original for name, value in saved.items(): if value is None: os.environ.pop(name, None) else: os.environ[name] = value self.assertEqual(len(addresses), 1) self.assertNotIn("C64U_PASSWORD", captured["env"]) self.assertNotIn(publisher.DEFAULT_PASSWORD_ENV, captured["env"]) self.assertIn("LANG", captured["env"])
# --------------------------------------------------------------------------# Transport plumbing# --------------------------------------------------------------------------
class TransportTests(unittest.TestCase): def test_exact_reader_never_exceeds_the_declared_size(self) -> None: reader = publisher._ExactReader(io.BytesIO(b"0123456789"), 4) self.assertEqual(reader.read(2), b"01") self.assertEqual(reader.read(100), b"23") self.assertEqual(reader.read(100), b"")
def test_record_rkey_parsing(self) -> None: self.assertEqual(publisher.record_rkey(f"at://{OWNER_DID}/sh.tangled.repo.artifact/3kabc"), "3kabc") self.assertIsNone(publisher.record_rkey("not-a-uri"))
class SocketSafetyTests(unittest.TestCase): """The transport must be the only thing that can open a socket."""
def test_https_transport_uses_verified_default_context(self) -> None: import ssl
created = {}
class FakeConnection: def __init__(self, host, port, **kwargs): created["context"] = kwargs.get("context") self.sock = None
def connect(self): raise socket.timeout("stop here")
def close(self): pass
original_socket = publisher.resolve_addresses original_connection = publisher._SockaddrHTTPSConnection publisher.resolve_addresses = lambda host, port, deadline: [ (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("127.0.0.1", 443)) ] publisher._SockaddrHTTPSConnection = FakeConnection try: transport = publisher.HttpsTransport({"scheme": "https", "host": "tngl.sh", "port": None, "origin": "https://tngl.sh"}, timeout=1.0) with self.assertRaises(publisher.PublishError) as caught: transport.request("GET", "/xrpc/probe") self.assertEqual(caught.exception.code, "remote_timeout") finally: publisher.resolve_addresses = original_socket publisher._SockaddrHTTPSConnection = original_connection self.assertIsInstance(created["context"], ssl.SSLContext) self.assertTrue(created["context"].check_hostname) self.assertEqual(created["context"].verify_mode, ssl.CERT_REQUIRED)
if __name__ == "__main__": unittest.main()