m2-market/cli/tests/test_publish.py
m2 (AI Agent) 82393ccdc5 T027: publish command (validate -> branch -> release -> PR)
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-02 03:38:59 +02:00

161 lines
6 KiB
Python

"""Inline verification for the publish CLI command (contracts/cli.md, registry-layout.md).
Covers: (1) invalid bundle -> schema errors printed, exit 5, before any registry
call; (2) valid bundle -> happy-path branch/commit/release/PR sequence against
a MockTransport fake of the Forgejo API.
"""
from __future__ import annotations
import json
import httpx
import pytest
from click.testing import CliRunner
import m2_market.cli as cli_mod
LISTING_ID = "lst_mm-pdf-report"
SOLUTION_ID = "sol_mm-pdf-report"
INSTALL_REF = "lst_mm-pdf-report-v1.0.0"
SELLER = "sdjs-operator"
VALID_SOLUTION = {
"schema_version": "m2.solution.v1",
"solution_id": SOLUTION_ID,
"name": "MM PDF Branded Report",
"summary": "Generate branded PDF reports from Markdown",
"intent": "Produce publish-ready branded PDF output without manual design work",
"deployment": {
"recipe_ref": "recipes/mm-pdf-report/recipe.yaml",
"entrypoint": "skills/mm-pdf/render.py",
"verify_command": "python skills/mm-pdf/render.py --self-test",
},
"applicability": {"image_classes": ["desktop-agent"]},
"tenant_scope": "m2-core",
"evidence": [{"source": "session:2026-05-11-mm-pdf-branded-report"}],
"content_hash": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"price": {"amount": 500, "currency": "m2cr", "model": "fixed"},
"seller": SELLER,
"license": {"terms": "internal", "major_version_coverage": True},
"permissions": [{"scope": "filesystem:write", "path": "/tmp/mm-pdf-output"}],
}
VALID_LISTING = {
"schema_version": "m2.listing.v1",
"listing_id": LISTING_ID,
"solution_id": SOLUTION_ID,
"solution_version": "1.0.0",
"inventory_type": "solution",
"name": "MM PDF Branded Report",
"summary": "Generate branded PDF reports from Markdown",
"category": "reporting",
"price": {"amount": 500, "currency": "m2cr", "model": "fixed"},
"seller": SELLER,
"evidence_summary": "Generated branded PDF report; operator confirmed styling matched brand deck.",
"tenant_visibility": ["*"],
"stats": {"installs": 0, "proposals_shown": 0, "proposals_accepted": 0},
"status": "draft",
"install_ref": INSTALL_REF,
}
def _write_bundle(bundle_dir, solution=None, listing=None):
bundle_dir.mkdir(parents=True, exist_ok=True)
(bundle_dir / "solution.json").write_text(json.dumps(solution if solution is not None else VALID_SOLUTION))
(bundle_dir / "listing.json").write_text(json.dumps(listing if listing is not None else VALID_LISTING))
payload_dir = bundle_dir / "payload"
payload_dir.mkdir(exist_ok=True)
(payload_dir / "report.md").write_text("# hello\n")
(bundle_dir / "recipe.yaml").write_text("targets: []\nchecks: []\n")
return bundle_dir
@pytest.fixture
def config_env(tmp_path, monkeypatch):
config_path = tmp_path / "config.toml"
config_path.write_text(
"""
operator_id = "buyer1"
tenant = "test-tenant"
ledger_url = "https://ledger.test"
ledger_api_key = "lk"
memory_api_url = "https://catalog.test"
memory_api_key = "mk"
forgejo_url = "https://registry.test"
forgejo_token = "rt"
"""
)
monkeypatch.setenv("M2_MARKET_CONFIG", str(config_path))
monkeypatch.setenv("M2_MARKET_SCHEMAS_DIR", str(_repo_schemas_dir()))
return config_path
def _repo_schemas_dir():
from pathlib import Path
return Path(__file__).resolve().parents[2] / "schemas"
def test_invalid_bundle_fails_validation_before_any_network_call(tmp_path, config_env, monkeypatch):
def fail_if_called(*args, **kwargs):
raise AssertionError("registry must not be contacted when validation fails")
monkeypatch.setattr(httpx.Client, "post", fail_if_called)
bad_solution = dict(VALID_SOLUTION)
bad_solution["evidence"] = [] # violates evidence minItems (Principle III evidence-backed listings)
bundle_dir = _write_bundle(tmp_path / "bundle", solution=bad_solution)
runner = CliRunner()
result = runner.invoke(cli_mod.main, ["publish", str(bundle_dir)])
assert result.exit_code == cli_mod.EXIT_VALIDATION, result.output
assert "evidence" in result.output
def test_publish_happy_path_opens_pr(tmp_path, config_env, monkeypatch):
calls = {"branch": 0, "files": [], "release": 0, "asset": 0, "pr": 0}
def handler(request: httpx.Request) -> httpx.Response:
path = request.url.path
if path.endswith("/branches") and request.method == "POST":
calls["branch"] += 1
return httpx.Response(201, json={"name": f"listing/{LISTING_ID}"})
if "/contents/" in path and request.method == "POST":
calls["files"].append(path)
return httpx.Response(201, json={"content": {"path": path}})
if path.endswith("/releases") and request.method == "POST":
calls["release"] += 1
return httpx.Response(201, json={"id": 42, "tag_name": INSTALL_REF})
if path.endswith("/releases/42/assets") and request.method == "POST":
calls["asset"] += 1
return httpx.Response(201, json={"name": f"{INSTALL_REF}.tar.gz"})
if path.endswith("/pulls") and request.method == "POST":
calls["pr"] += 1
return httpx.Response(
201,
json={"html_url": f"https://registry.test/m2/market-registry/pulls/7"},
)
raise AssertionError(f"unexpected request: {request.method} {request.url}")
orig_init = httpx.Client.__init__
def patched_init(self, *args, **kwargs):
kwargs["transport"] = httpx.MockTransport(handler)
orig_init(self, *args, **kwargs)
monkeypatch.setattr(httpx.Client, "__init__", patched_init)
bundle_dir = _write_bundle(tmp_path / "bundle")
runner = CliRunner()
result = runner.invoke(cli_mod.main, ["publish", str(bundle_dir)])
assert result.exit_code == 0, result.output
assert "https://registry.test/m2/market-registry/pulls/7" in result.output
assert calls["branch"] == 1
assert len(calls["files"]) == 2
assert calls["release"] == 1
assert calls["asset"] == 1
assert calls["pr"] == 1