"""TDD: ``muse sign propose`` subcommand. ``muse sign propose`` produces an Ed25519 signature over the canonical PROPOSE message, ready to submit to ``muse hub proposal create`` or the REST API. Canonical PROPOSE message (UTF-8, LF endings): PROPOSE repo_id: sha256: from_branch: to_branch: author: created_at: Acceptance criteria ------------------- T1 run_propose() emits JSON with proposer_signature, proposer_public_key, proposer_timestamp, canonical_message, handle, repo_id, from_branch, to_branch, author fields — all present and correctly typed. T2 proposer_public_key matches the signing identity's public key prefixed with 'ed25519:'. T3 The signature in proposer_signature verifies against proposer_public_key over the canonical_message bytes. T4 canonical_propose_message() is deterministic — same inputs always produce identical bytes. T5 canonical_propose_message() without proposal_id omits the proposal_id line. T6 canonical_propose_message() with proposal_id includes it as the second line. T7 run_propose() with --json=False prints human-readable text (no JSON). T8 run_propose() uses the created_at timestamp from proposer_timestamp (not an arbitrary server time). T9 The subcommand is registered under ``muse sign propose`` in the argparse tree (registration smoke test). """ from __future__ import annotations import argparse import json import sys import unittest import unittest.mock from collections.abc import Callable from datetime import datetime, timezone from io import StringIO from typing import TYPE_CHECKING from unittest.mock import patch from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from muse.core.types import b64url_decode, b64url_encode, decode_sig, decode_pubkey if TYPE_CHECKING: from muse.core.transport import SigningIdentity # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_signing(handle: str = "gabriel") -> "SigningIdentity": from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from muse.core.transport import SigningIdentity return SigningIdentity(handle=handle, private_key=Ed25519PrivateKey.generate()) def _base_args(**kwargs: _ArgVal) -> argparse.Namespace: defaults = dict( repo_id="sha256:" + "a" * 64, from_branch="feat/my-thing", to_branch="dev", hub="https://localhost:1337", agent_id=None, timestamp=None, json_out=True, ) defaults.update(kwargs) return argparse.Namespace(**defaults) # --------------------------------------------------------------------------- # T4, T5, T6 — canonical_propose_message (unit, no signing) # --------------------------------------------------------------------------- class TestCanonicalProposeMessage(unittest.TestCase): def _import(self) -> Callable[..., bytes]: from muse.cli.commands.sign import canonical_propose_message return canonical_propose_message # type: ignore[return-value] def test_deterministic(self) -> None: """T4 — same inputs always produce identical bytes.""" fn = self._import() created_at = datetime(2026, 5, 8, 19, 30, 34, tzinfo=timezone.utc) kwargs = dict( repo_id="sha256:" + "b" * 64, from_branch="feat/x", to_branch="dev", author="gabriel", created_at=created_at, ) assert fn(**kwargs) == fn(**kwargs) def test_without_proposal_id(self) -> None: """T5 — no proposal_id line when omitted.""" fn = self._import() created_at = datetime(2026, 5, 8, 19, 30, 34, tzinfo=timezone.utc) msg = fn( repo_id="sha256:" + "b" * 64, from_branch="feat/x", to_branch="dev", author="gabriel", created_at=created_at, ).decode() lines = msg.splitlines() assert lines[0] == "PROPOSE" assert not any(l.startswith("proposal_id:") for l in lines) assert lines[1].startswith("repo_id:") def test_with_proposal_id(self) -> None: """T6 — proposal_id is the second line when provided.""" fn = self._import() created_at = datetime(2026, 5, 8, 19, 30, 34, tzinfo=timezone.utc) pid = "sha256:" + "c" * 64 msg = fn( repo_id="sha256:" + "b" * 64, from_branch="feat/x", to_branch="dev", author="gabriel", created_at=created_at, proposal_id=pid, ).decode() lines = msg.splitlines() assert lines[0] == "PROPOSE" assert lines[1] == f"proposal_id: {pid}" assert lines[2].startswith("repo_id:") def test_format_contains_all_fields(self) -> None: """All required fields appear in the message.""" fn = self._import() created_at = datetime(2026, 5, 8, 19, 30, 34, tzinfo=timezone.utc) msg = fn( repo_id="sha256:" + "b" * 64, from_branch="feat/identity-v2", to_branch="dev", author="gabriel", created_at=created_at, ).decode() assert "PROPOSE" in msg assert "from_branch: feat/identity-v2" in msg assert "to_branch: dev" in msg assert "author: gabriel" in msg assert "created_at: 2026-05-08T19:30:34+00:00" in msg # --------------------------------------------------------------------------- # T1, T2, T3, T7, T8 — run_propose # --------------------------------------------------------------------------- _ArgVal = str | bool | int | None # argparse namespace field values type _JsonResponse = dict[str, str | int | float | bool | None] class TestRunPropose(unittest.TestCase): def setUp(self) -> None: self.signing = _make_signing("gabriel") def _run(self, **kwargs: _ArgVal) -> _JsonResponse: from muse.cli.commands.sign import run_propose args = _base_args(**kwargs) buf = StringIO() with patch("muse.cli.commands.sign._load_signing", return_value=self.signing): with patch("sys.stdout", buf): run_propose(args) return json.loads(buf.getvalue()) def test_json_fields_present(self) -> None: """T1 — all required fields present in JSON output.""" out = self._run() for field in ( "proposer_signature", "proposer_public_key", "proposer_timestamp", "canonical_message", "handle", "repo_id", "from_branch", "to_branch", "author", ): assert field in out, f"missing field: {field}" def test_public_key_matches_signing_identity(self) -> None: """T2 — proposer_public_key encodes the signing identity's public key.""" out = self._run() pub_key_str = out["proposer_public_key"] assert pub_key_str.startswith("ed25519:") algo, raw = decode_pubkey(pub_key_str) assert algo == "ed25519" expected_raw = self.signing.private_key.public_key().public_bytes_raw() assert raw == expected_raw def test_signature_verifies(self) -> None: """T3 — signature in proposer_signature verifies over canonical_message.""" out = self._run() algo, sig_bytes = decode_sig(out["proposer_signature"]) assert algo == "ed25519" _, key_bytes = decode_pubkey(out["proposer_public_key"]) message = out["canonical_message"].encode("utf-8") pub_key = Ed25519PublicKey.from_public_bytes(key_bytes) # raises InvalidSignature on failure pub_key.verify(sig_bytes, message) def test_human_readable_output(self) -> None: """T7 — non-JSON mode prints readable lines, not raw JSON.""" from muse.cli.commands.sign import run_propose args = _base_args(json_out=False) buf = StringIO() with patch("muse.cli.commands.sign._load_signing", return_value=self.signing): with patch("sys.stderr", buf): run_propose(args) text = buf.getvalue() assert "PROPOSER" in text or "proposer" in text.lower() assert "ed25519:" in text def test_timestamp_used_in_message(self) -> None: """T8 — proposer_timestamp appears verbatim in canonical_message.""" out = self._run() ts = out["proposer_timestamp"] assert ts in out["canonical_message"] def test_passthrough_fields(self) -> None: """repo_id, from_branch, to_branch, author echo back in JSON.""" repo = "sha256:" + "d" * 64 out = self._run(repo_id=repo, from_branch="bugfix/x", to_branch="main") assert out["repo_id"] == repo assert out["from_branch"] == "bugfix/x" assert out["to_branch"] == "main" assert out["author"] == "gabriel" # --------------------------------------------------------------------------- # T9 — argparse registration # --------------------------------------------------------------------------- class TestRegistration(unittest.TestCase): def test_propose_subcommand_registered(self) -> None: """T9 — 'propose' appears in the sign subcommand tree.""" import muse.cli.commands.sign as sign_mod p = argparse.ArgumentParser() sub = p.add_subparsers() sign_mod.register(sub) # Parse a minimal propose invocation — should not error args = p.parse_args([ "sign", "propose", "--repo-id", "sha256:" + "a" * 64, "--from-branch", "feat/x", "--to-branch", "dev", "--hub", "https://localhost:1337", ]) assert hasattr(args, "func") assert args.func is sign_mod.run_propose