test_cmd_sign_propose.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 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 collections.abc import Callable |
| 42 | from datetime import datetime, timezone |
| 43 | from io import StringIO |
| 44 | from typing import TYPE_CHECKING |
| 45 | from unittest.mock import patch |
| 46 | |
| 47 | from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey |
| 48 | from muse.core.types import b64url_decode, b64url_encode, decode_sig, decode_pubkey |
| 49 | |
| 50 | if TYPE_CHECKING: |
| 51 | from muse.core.transport import SigningIdentity |
| 52 | |
| 53 | |
| 54 | # --------------------------------------------------------------------------- |
| 55 | # Helpers |
| 56 | # --------------------------------------------------------------------------- |
| 57 | |
| 58 | def _make_signing(handle: str = "gabriel") -> "SigningIdentity": |
| 59 | from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey |
| 60 | from muse.core.transport import SigningIdentity |
| 61 | return SigningIdentity(handle=handle, private_key=Ed25519PrivateKey.generate()) |
| 62 | |
| 63 | |
| 64 | def _base_args(**kwargs: _ArgVal) -> argparse.Namespace: |
| 65 | defaults = dict( |
| 66 | repo_id="sha256:" + "a" * 64, |
| 67 | from_branch="feat/my-thing", |
| 68 | to_branch="dev", |
| 69 | hub="https://localhost:1337", |
| 70 | agent_id=None, |
| 71 | timestamp=None, |
| 72 | json_out=True, |
| 73 | ) |
| 74 | defaults.update(kwargs) |
| 75 | return argparse.Namespace(**defaults) |
| 76 | |
| 77 | |
| 78 | # --------------------------------------------------------------------------- |
| 79 | # T4, T5, T6 — canonical_propose_message (unit, no signing) |
| 80 | # --------------------------------------------------------------------------- |
| 81 | |
| 82 | class TestCanonicalProposeMessage(unittest.TestCase): |
| 83 | def _import(self) -> Callable[..., bytes]: |
| 84 | from muse.cli.commands.sign import canonical_propose_message |
| 85 | return canonical_propose_message # type: ignore[return-value] |
| 86 | |
| 87 | def test_deterministic(self) -> None: |
| 88 | """T4 — same inputs always produce identical bytes.""" |
| 89 | fn = self._import() |
| 90 | created_at = datetime(2026, 5, 8, 19, 30, 34, tzinfo=timezone.utc) |
| 91 | kwargs = dict( |
| 92 | repo_id="sha256:" + "b" * 64, |
| 93 | from_branch="feat/x", |
| 94 | to_branch="dev", |
| 95 | author="gabriel", |
| 96 | created_at=created_at, |
| 97 | ) |
| 98 | assert fn(**kwargs) == fn(**kwargs) |
| 99 | |
| 100 | def test_without_proposal_id(self) -> None: |
| 101 | """T5 — no proposal_id line when omitted.""" |
| 102 | fn = self._import() |
| 103 | created_at = datetime(2026, 5, 8, 19, 30, 34, tzinfo=timezone.utc) |
| 104 | msg = fn( |
| 105 | repo_id="sha256:" + "b" * 64, |
| 106 | from_branch="feat/x", |
| 107 | to_branch="dev", |
| 108 | author="gabriel", |
| 109 | created_at=created_at, |
| 110 | ).decode() |
| 111 | lines = msg.splitlines() |
| 112 | assert lines[0] == "PROPOSE" |
| 113 | assert not any(l.startswith("proposal_id:") for l in lines) |
| 114 | assert lines[1].startswith("repo_id:") |
| 115 | |
| 116 | def test_with_proposal_id(self) -> None: |
| 117 | """T6 — proposal_id is the second line when provided.""" |
| 118 | fn = self._import() |
| 119 | created_at = datetime(2026, 5, 8, 19, 30, 34, tzinfo=timezone.utc) |
| 120 | pid = "sha256:" + "c" * 64 |
| 121 | msg = fn( |
| 122 | repo_id="sha256:" + "b" * 64, |
| 123 | from_branch="feat/x", |
| 124 | to_branch="dev", |
| 125 | author="gabriel", |
| 126 | created_at=created_at, |
| 127 | proposal_id=pid, |
| 128 | ).decode() |
| 129 | lines = msg.splitlines() |
| 130 | assert lines[0] == "PROPOSE" |
| 131 | assert lines[1] == f"proposal_id: {pid}" |
| 132 | assert lines[2].startswith("repo_id:") |
| 133 | |
| 134 | def test_format_contains_all_fields(self) -> None: |
| 135 | """All required fields appear in the message.""" |
| 136 | fn = self._import() |
| 137 | created_at = datetime(2026, 5, 8, 19, 30, 34, tzinfo=timezone.utc) |
| 138 | msg = fn( |
| 139 | repo_id="sha256:" + "b" * 64, |
| 140 | from_branch="feat/identity-v2", |
| 141 | to_branch="dev", |
| 142 | author="gabriel", |
| 143 | created_at=created_at, |
| 144 | ).decode() |
| 145 | assert "PROPOSE" in msg |
| 146 | assert "from_branch: feat/identity-v2" in msg |
| 147 | assert "to_branch: dev" in msg |
| 148 | assert "author: gabriel" in msg |
| 149 | assert "created_at: 2026-05-08T19:30:34+00:00" in msg |
| 150 | |
| 151 | |
| 152 | # --------------------------------------------------------------------------- |
| 153 | # T1, T2, T3, T7, T8 — run_propose |
| 154 | # --------------------------------------------------------------------------- |
| 155 | |
| 156 | _ArgVal = str | bool | int | None # argparse namespace field values |
| 157 | type _JsonResponse = dict[str, str | int | float | bool | None] |
| 158 | |
| 159 | |
| 160 | class TestRunPropose(unittest.TestCase): |
| 161 | def setUp(self) -> None: |
| 162 | self.signing = _make_signing("gabriel") |
| 163 | |
| 164 | def _run(self, **kwargs: _ArgVal) -> _JsonResponse: |
| 165 | from muse.cli.commands.sign import run_propose |
| 166 | args = _base_args(**kwargs) |
| 167 | buf = StringIO() |
| 168 | with patch("muse.cli.commands.sign._load_signing", return_value=self.signing): |
| 169 | with patch("sys.stdout", buf): |
| 170 | run_propose(args) |
| 171 | return json.loads(buf.getvalue()) |
| 172 | |
| 173 | def test_json_fields_present(self) -> None: |
| 174 | """T1 — all required fields present in JSON output.""" |
| 175 | out = self._run() |
| 176 | for field in ( |
| 177 | "proposer_signature", "proposer_public_key", "proposer_timestamp", |
| 178 | "canonical_message", "handle", "repo_id", "from_branch", "to_branch", "author", |
| 179 | ): |
| 180 | assert field in out, f"missing field: {field}" |
| 181 | |
| 182 | def test_public_key_matches_signing_identity(self) -> None: |
| 183 | """T2 — proposer_public_key encodes the signing identity's public key.""" |
| 184 | out = self._run() |
| 185 | pub_key_str = out["proposer_public_key"] |
| 186 | assert pub_key_str.startswith("ed25519:") |
| 187 | algo, raw = decode_pubkey(pub_key_str) |
| 188 | assert algo == "ed25519" |
| 189 | expected_raw = self.signing.private_key.public_key().public_bytes_raw() |
| 190 | assert raw == expected_raw |
| 191 | |
| 192 | def test_signature_verifies(self) -> None: |
| 193 | """T3 — signature in proposer_signature verifies over canonical_message.""" |
| 194 | out = self._run() |
| 195 | algo, sig_bytes = decode_sig(out["proposer_signature"]) |
| 196 | assert algo == "ed25519" |
| 197 | _, key_bytes = decode_pubkey(out["proposer_public_key"]) |
| 198 | message = out["canonical_message"].encode("utf-8") |
| 199 | pub_key = Ed25519PublicKey.from_public_bytes(key_bytes) |
| 200 | # raises InvalidSignature on failure |
| 201 | pub_key.verify(sig_bytes, message) |
| 202 | |
| 203 | def test_human_readable_output(self) -> None: |
| 204 | """T7 — non-JSON mode prints readable lines, not raw JSON.""" |
| 205 | from muse.cli.commands.sign import run_propose |
| 206 | args = _base_args(json_out=False) |
| 207 | buf = StringIO() |
| 208 | with patch("muse.cli.commands.sign._load_signing", return_value=self.signing): |
| 209 | with patch("sys.stderr", buf): |
| 210 | run_propose(args) |
| 211 | text = buf.getvalue() |
| 212 | assert "PROPOSER" in text or "proposer" in text.lower() |
| 213 | assert "ed25519:" in text |
| 214 | |
| 215 | def test_timestamp_used_in_message(self) -> None: |
| 216 | """T8 — proposer_timestamp appears verbatim in canonical_message.""" |
| 217 | out = self._run() |
| 218 | ts = out["proposer_timestamp"] |
| 219 | assert ts in out["canonical_message"] |
| 220 | |
| 221 | def test_passthrough_fields(self) -> None: |
| 222 | """repo_id, from_branch, to_branch, author echo back in JSON.""" |
| 223 | repo = "sha256:" + "d" * 64 |
| 224 | out = self._run(repo_id=repo, from_branch="bugfix/x", to_branch="main") |
| 225 | assert out["repo_id"] == repo |
| 226 | assert out["from_branch"] == "bugfix/x" |
| 227 | assert out["to_branch"] == "main" |
| 228 | assert out["author"] == "gabriel" |
| 229 | |
| 230 | |
| 231 | # --------------------------------------------------------------------------- |
| 232 | # T9 — argparse registration |
| 233 | # --------------------------------------------------------------------------- |
| 234 | |
| 235 | class TestRegistration(unittest.TestCase): |
| 236 | def test_propose_subcommand_registered(self) -> None: |
| 237 | """T9 — 'propose' appears in the sign subcommand tree.""" |
| 238 | import muse.cli.commands.sign as sign_mod |
| 239 | p = argparse.ArgumentParser() |
| 240 | sub = p.add_subparsers() |
| 241 | sign_mod.register(sub) |
| 242 | # Parse a minimal propose invocation — should not error |
| 243 | args = p.parse_args([ |
| 244 | "sign", "propose", |
| 245 | "--repo-id", "sha256:" + "a" * 64, |
| 246 | "--from-branch", "feat/x", |
| 247 | "--to-branch", "dev", |
| 248 | "--hub", "https://localhost:1337", |
| 249 | ]) |
| 250 | assert hasattr(args, "func") |
| 251 | assert args.func is sign_mod.run_propose |