test_cmd_sign_propose.py
python
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d
feat(pack): delta-encode snapshots in MPackBundle wire format
Sonnet 4.6
minor
⚠ breaking
121 days ago
| 1 | """TDD: ``muse sign propose`` subcommand. |
| 2 | |
| 3 | ``muse sign propose`` produces an Ed25519 signature over the canonical PROPOSE |
| 4 | message, ready to submit to ``muse hub proposal create`` or the REST API. |
| 5 | |
| 6 | Canonical PROPOSE message (UTF-8, LF endings): |
| 7 | PROPOSE |
| 8 | repo_id: sha256:<hex> |
| 9 | from_branch: <name> |
| 10 | to_branch: <name> |
| 11 | author: <handle> |
| 12 | created_at: <ISO-8601 UTC with offset> |
| 13 | |
| 14 | Acceptance criteria |
| 15 | ------------------- |
| 16 | T1 run_propose() emits JSON with proposer_signature, proposer_public_key, |
| 17 | proposer_timestamp, canonical_message, handle, repo_id, from_branch, |
| 18 | to_branch, author fields — all present and correctly typed. |
| 19 | T2 proposer_public_key matches the signing identity's public key prefixed |
| 20 | with 'ed25519:'. |
| 21 | T3 The signature in proposer_signature verifies against proposer_public_key |
| 22 | over the canonical_message bytes. |
| 23 | T4 canonical_propose_message() is deterministic — same inputs always produce |
| 24 | identical bytes. |
| 25 | T5 canonical_propose_message() without proposal_id omits the proposal_id line. |
| 26 | T6 canonical_propose_message() with proposal_id includes it as the second line. |
| 27 | T7 run_propose() with --json=False prints human-readable text (no JSON). |
| 28 | T8 run_propose() uses the created_at timestamp from proposer_timestamp (not |
| 29 | an arbitrary server time). |
| 30 | T9 The subcommand is registered under ``muse sign propose`` in the argparse |
| 31 | tree (registration smoke test). |
| 32 | """ |
| 33 | |
| 34 | from __future__ import annotations |
| 35 | |
| 36 | import argparse |
| 37 | import json |
| 38 | import sys |
| 39 | import unittest |
| 40 | import unittest.mock |
| 41 | from datetime import datetime, timezone |
| 42 | from io import StringIO |
| 43 | from typing import TYPE_CHECKING |
| 44 | from unittest.mock import patch |
| 45 | |
| 46 | from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey |
| 47 | from muse.core.types import b64url_decode, b64url_encode, decode_sig, decode_pubkey |
| 48 | |
| 49 | if TYPE_CHECKING: |
| 50 | from muse.core.transport import SigningIdentity |
| 51 | |
| 52 | |
| 53 | # --------------------------------------------------------------------------- |
| 54 | # Helpers |
| 55 | # --------------------------------------------------------------------------- |
| 56 | |
| 57 | def _make_signing(handle: str = "gabriel") -> "SigningIdentity": |
| 58 | from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey |
| 59 | from muse.core.transport import SigningIdentity |
| 60 | return SigningIdentity(handle=handle, private_key=Ed25519PrivateKey.generate()) |
| 61 | |
| 62 | |
| 63 | def _base_args(**kwargs) -> argparse.Namespace: |
| 64 | defaults = dict( |
| 65 | repo_id="sha256:" + "a" * 64, |
| 66 | from_branch="feat/my-thing", |
| 67 | to_branch="dev", |
| 68 | hub="https://localhost:1337", |
| 69 | agent_id=None, |
| 70 | timestamp=None, |
| 71 | json_out=True, |
| 72 | ) |
| 73 | defaults.update(kwargs) |
| 74 | return argparse.Namespace(**defaults) |
| 75 | |
| 76 | |
| 77 | # --------------------------------------------------------------------------- |
| 78 | # T4, T5, T6 — canonical_propose_message (unit, no signing) |
| 79 | # --------------------------------------------------------------------------- |
| 80 | |
| 81 | class TestCanonicalProposeMessage(unittest.TestCase): |
| 82 | def _import(self): |
| 83 | from muse.cli.commands.sign import canonical_propose_message |
| 84 | return canonical_propose_message |
| 85 | |
| 86 | def test_deterministic(self): |
| 87 | """T4 — same inputs always produce identical bytes.""" |
| 88 | fn = self._import() |
| 89 | created_at = datetime(2026, 5, 8, 19, 30, 34, tzinfo=timezone.utc) |
| 90 | kwargs = dict( |
| 91 | repo_id="sha256:" + "b" * 64, |
| 92 | from_branch="feat/x", |
| 93 | to_branch="dev", |
| 94 | author="gabriel", |
| 95 | created_at=created_at, |
| 96 | ) |
| 97 | assert fn(**kwargs) == fn(**kwargs) |
| 98 | |
| 99 | def test_without_proposal_id(self): |
| 100 | """T5 — no proposal_id line when omitted.""" |
| 101 | fn = self._import() |
| 102 | created_at = datetime(2026, 5, 8, 19, 30, 34, tzinfo=timezone.utc) |
| 103 | msg = fn( |
| 104 | repo_id="sha256:" + "b" * 64, |
| 105 | from_branch="feat/x", |
| 106 | to_branch="dev", |
| 107 | author="gabriel", |
| 108 | created_at=created_at, |
| 109 | ).decode() |
| 110 | lines = msg.splitlines() |
| 111 | assert lines[0] == "PROPOSE" |
| 112 | assert not any(l.startswith("proposal_id:") for l in lines) |
| 113 | assert lines[1].startswith("repo_id:") |
| 114 | |
| 115 | def test_with_proposal_id(self): |
| 116 | """T6 — proposal_id is the second line when provided.""" |
| 117 | fn = self._import() |
| 118 | created_at = datetime(2026, 5, 8, 19, 30, 34, tzinfo=timezone.utc) |
| 119 | pid = "sha256:" + "c" * 64 |
| 120 | msg = fn( |
| 121 | repo_id="sha256:" + "b" * 64, |
| 122 | from_branch="feat/x", |
| 123 | to_branch="dev", |
| 124 | author="gabriel", |
| 125 | created_at=created_at, |
| 126 | proposal_id=pid, |
| 127 | ).decode() |
| 128 | lines = msg.splitlines() |
| 129 | assert lines[0] == "PROPOSE" |
| 130 | assert lines[1] == f"proposal_id: {pid}" |
| 131 | assert lines[2].startswith("repo_id:") |
| 132 | |
| 133 | def test_format_contains_all_fields(self): |
| 134 | """All required fields appear in the message.""" |
| 135 | fn = self._import() |
| 136 | created_at = datetime(2026, 5, 8, 19, 30, 34, tzinfo=timezone.utc) |
| 137 | msg = fn( |
| 138 | repo_id="sha256:" + "b" * 64, |
| 139 | from_branch="feat/identity-v2", |
| 140 | to_branch="dev", |
| 141 | author="gabriel", |
| 142 | created_at=created_at, |
| 143 | ).decode() |
| 144 | assert "PROPOSE" in msg |
| 145 | assert "from_branch: feat/identity-v2" in msg |
| 146 | assert "to_branch: dev" in msg |
| 147 | assert "author: gabriel" in msg |
| 148 | assert "created_at: 2026-05-08T19:30:34+00:00" in msg |
| 149 | |
| 150 | |
| 151 | # --------------------------------------------------------------------------- |
| 152 | # T1, T2, T3, T7, T8 — run_propose |
| 153 | # --------------------------------------------------------------------------- |
| 154 | |
| 155 | type _JsonResponse = dict[str, str | int | float | bool | None] |
| 156 | |
| 157 | |
| 158 | class TestRunPropose(unittest.TestCase): |
| 159 | def setUp(self): |
| 160 | self.signing = _make_signing("gabriel") |
| 161 | |
| 162 | def _run(self, **kwargs) -> _JsonResponse: |
| 163 | from muse.cli.commands.sign import run_propose |
| 164 | args = _base_args(**kwargs) |
| 165 | buf = StringIO() |
| 166 | with patch("muse.cli.commands.sign._load_signing", return_value=self.signing): |
| 167 | with patch("sys.stdout", buf): |
| 168 | run_propose(args) |
| 169 | return json.loads(buf.getvalue()) |
| 170 | |
| 171 | def test_json_fields_present(self): |
| 172 | """T1 — all required fields present in JSON output.""" |
| 173 | out = self._run() |
| 174 | for field in ( |
| 175 | "proposer_signature", "proposer_public_key", "proposer_timestamp", |
| 176 | "canonical_message", "handle", "repo_id", "from_branch", "to_branch", "author", |
| 177 | ): |
| 178 | assert field in out, f"missing field: {field}" |
| 179 | |
| 180 | def test_public_key_matches_signing_identity(self): |
| 181 | """T2 — proposer_public_key encodes the signing identity's public key.""" |
| 182 | out = self._run() |
| 183 | pub_key_str = out["proposer_public_key"] |
| 184 | assert pub_key_str.startswith("ed25519:") |
| 185 | algo, raw = decode_pubkey(pub_key_str) |
| 186 | assert algo == "ed25519" |
| 187 | expected_raw = self.signing.private_key.public_key().public_bytes_raw() |
| 188 | assert raw == expected_raw |
| 189 | |
| 190 | def test_signature_verifies(self): |
| 191 | """T3 — signature in proposer_signature verifies over canonical_message.""" |
| 192 | out = self._run() |
| 193 | algo, sig_bytes = decode_sig(out["proposer_signature"]) |
| 194 | assert algo == "ed25519" |
| 195 | _, key_bytes = decode_pubkey(out["proposer_public_key"]) |
| 196 | message = out["canonical_message"].encode("utf-8") |
| 197 | pub_key = Ed25519PublicKey.from_public_bytes(key_bytes) |
| 198 | # raises InvalidSignature on failure |
| 199 | pub_key.verify(sig_bytes, message) |
| 200 | |
| 201 | def test_human_readable_output(self): |
| 202 | """T7 — non-JSON mode prints readable lines, not raw JSON.""" |
| 203 | from muse.cli.commands.sign import run_propose |
| 204 | args = _base_args(json_out=False) |
| 205 | buf = StringIO() |
| 206 | with patch("muse.cli.commands.sign._load_signing", return_value=self.signing): |
| 207 | with patch("sys.stderr", buf): |
| 208 | run_propose(args) |
| 209 | text = buf.getvalue() |
| 210 | assert "PROPOSER" in text or "proposer" in text.lower() |
| 211 | assert "ed25519:" in text |
| 212 | |
| 213 | def test_timestamp_used_in_message(self): |
| 214 | """T8 — proposer_timestamp appears verbatim in canonical_message.""" |
| 215 | out = self._run() |
| 216 | ts = out["proposer_timestamp"] |
| 217 | assert ts in out["canonical_message"] |
| 218 | |
| 219 | def test_passthrough_fields(self): |
| 220 | """repo_id, from_branch, to_branch, author echo back in JSON.""" |
| 221 | repo = "sha256:" + "d" * 64 |
| 222 | out = self._run(repo_id=repo, from_branch="bugfix/x", to_branch="main") |
| 223 | assert out["repo_id"] == repo |
| 224 | assert out["from_branch"] == "bugfix/x" |
| 225 | assert out["to_branch"] == "main" |
| 226 | assert out["author"] == "gabriel" |
| 227 | |
| 228 | |
| 229 | # --------------------------------------------------------------------------- |
| 230 | # T9 — argparse registration |
| 231 | # --------------------------------------------------------------------------- |
| 232 | |
| 233 | class TestRegistration(unittest.TestCase): |
| 234 | def test_propose_subcommand_registered(self): |
| 235 | """T9 — 'propose' appears in the sign subcommand tree.""" |
| 236 | import muse.cli.commands.sign as sign_mod |
| 237 | p = argparse.ArgumentParser() |
| 238 | sub = p.add_subparsers() |
| 239 | sign_mod.register(sub) |
| 240 | # Parse a minimal propose invocation — should not error |
| 241 | args = p.parse_args([ |
| 242 | "sign", "propose", |
| 243 | "--repo-id", "sha256:" + "a" * 64, |
| 244 | "--from-branch", "feat/x", |
| 245 | "--to-branch", "dev", |
| 246 | "--hub", "https://localhost:1337", |
| 247 | ]) |
| 248 | assert hasattr(args, "func") |
| 249 | assert args.func is sign_mod.run_propose |
File History
1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d
feat(pack): delta-encode snapshots in MPackBundle wire format
Sonnet 4.6
minor
⚠
121 days ago