"""T026 — CLI integration tests: REAL ledger (uvicorn on an ephemeral port, tmp sqlite), fixture catalog/registry, REAL temp bundle applied by the local adapter into a temp HOME. quickstart.md Scenarios B/C at test scale: funded install end-to-end, insufficient funds, idempotent re-run, apply-failure refund + rollback. """ from __future__ import annotations import hashlib import json import os import shutil import socket import tarfile import threading import time from pathlib import Path import httpx import pytest import uvicorn from click.testing import CliRunner SERVICE_KEY = "svc-test-key" ADMIN_KEY = "adm-test-key" LISTING_ID = "lst_mm_pdf_report" PRICE = 40 # --- real ledger server ------------------------------------------------------------- @pytest.fixture() def ledger_url(tmp_path_factory, monkeypatch): db = tmp_path_factory.mktemp("ledger") / "ledger.db" monkeypatch.setenv("LEDGER_DB_PATH", str(db)) monkeypatch.setenv("LEDGER_SERVICE_KEYS", SERVICE_KEY) monkeypatch.setenv("LEDGER_ADMIN_KEYS", ADMIN_KEY) # Import AFTER env is set so module-level defaults read the test values. from m2_ledger.api import app # noqa: PLC0415 with socket.socket() as s: s.bind(("127.0.0.1", 0)) port = s.getsockname()[1] server = uvicorn.Server(uvicorn.Config(app, host="127.0.0.1", port=port, log_level="error")) thread = threading.Thread(target=server.run, daemon=True) thread.start() url = f"http://127.0.0.1:{port}" for _ in range(100): try: if httpx.get(url + "/health", timeout=1).status_code == 200: break except httpx.HTTPError: time.sleep(0.05) else: pytest.fail("ledger server did not come up") yield url server.should_exit = True thread.join(timeout=5) def _ledger_get(url: str, path: str, key: str = SERVICE_KEY) -> dict: r = httpx.get(url + path, headers={"X-API-Key": key}) r.raise_for_status() return r.json() def _grant(url: str, operator: str, amount: int = 100) -> None: httpx.post( url + "/grant/starter", json={"operator_id": operator, "amount": amount}, headers={"X-API-Key": ADMIN_KEY}, ).raise_for_status() def _tx_count(url: str) -> int: return len(_ledger_get(url, "/tx")["transactions"]) # --- temp bundle + fixtures --------------------------------------------------------- def _build_bundle(root: Path, failing_check: bool = False) -> tuple[Path, str]: """Real bundle tarball; returns (tar_path, content_hash).""" src = root / "bundle-src" (src / "payload").mkdir(parents=True) (src / "payload" / "hello.txt").write_text("hello from m2-market test bundle\n") check_cmd = "false" if failing_check else "test -f ~/m2-market-test/hello.txt" (src / "recipe.yaml").write_text( "targets:\n" " - src: hello.txt\n" " dest: ~/m2-market-test/hello.txt\n" "checks:\n" f" - cmd: {check_cmd}\n" " expect_exit: 0\n" "entrypoint: cat ~/m2-market-test/hello.txt\n" ) (src / "solution.json").write_text(json.dumps({"solution_id": "sol_test"})) tar_path = root / "bundle.tar.gz" with tarfile.open(tar_path, "w:gz") as tar: tar.add(src, arcname="bundle") digest = hashlib.sha256(tar_path.read_bytes()).hexdigest() return tar_path, f"sha256:{digest}" @pytest.fixture() def env(tmp_path, monkeypatch, ledger_url): """Isolated HOME/state/config + fixture catalog/registry pointed at the real ledger.""" home = tmp_path / "home" home.mkdir() monkeypatch.setenv("HOME", str(home)) monkeypatch.setenv("M2_MARKET_STATE", str(tmp_path / "state.json")) bundle_path, content_hash = _build_bundle(tmp_path) config = tmp_path / "config.toml" config.write_text( f'operator_id = "m2bd"\n' f'tenant = "m2-core"\n' f'ledger_url = "{ledger_url}"\n' f'ledger_api_key = "{SERVICE_KEY}"\n' f'memory_api_url = "http://catalog.invalid"\n' f'memory_api_key = "k"\n' f'forgejo_url = "http://registry.invalid"\n' f'forgejo_token = "t"\n' f'apply_adapter = "local"\n' ) monkeypatch.setenv("M2_MARKET_CONFIG", str(config)) listing = { "listing_id": LISTING_ID, "name": "MM PDF Report", "summary": "branded pdf reports", "price": {"amount": PRICE, "currency": "m2cr", "model": "fixed"}, "seller": "sdjs-operator", "solution_version": "1.0.0", "install_ref": f"{LISTING_ID}-v1.0.0", } solution = { "solution_id": "sol_test", "content_hash": content_hash, "permissions": ["write ~/m2-market-test"], "deployment": {"entrypoint": "cat ~/m2-market-test/hello.txt"}, } from m2_market import cli as cli_mod # noqa: PLC0415 monkeypatch.setattr(cli_mod.CatalogClient, "get", lambda self, lid: dict(listing) if lid == LISTING_ID else None) monkeypatch.setattr(cli_mod.RegistryClient, "get_solution", lambda self, lid: dict(solution)) monkeypatch.setattr( cli_mod.RegistryClient, "download_bundle", lambda self, ref, dest: (Path(dest).mkdir(parents=True, exist_ok=True), shutil.copy(bundle_path, Path(dest) / "bundle.tar.gz"))[1] and Path(dest) / "bundle.tar.gz", ) return { "home": home, "ledger_url": ledger_url, "bundle_path": bundle_path, "listing": listing, "solution": solution, "tmp": tmp_path, "monkeypatch": monkeypatch, "cli_mod": cli_mod, } def _run(args, **kwargs): from m2_market.cli import main # noqa: PLC0415 return CliRunner().invoke(main, args, catch_exceptions=False, **kwargs) # --- scenarios ---------------------------------------------------------------------- def test_funded_install_end_to_end(env): _grant(env["ledger_url"], "m2bd", 100) result = _run(["install", LISTING_ID, "--yes"]) assert result.exit_code == 0, result.output # ledger rows exact: net + cut assert _ledger_get(env["ledger_url"], "/balance/m2bd")["balance"] == 100 - PRICE assert _ledger_get(env["ledger_url"], "/balance/sdjs-operator")["balance"] == 36 assert _ledger_get(env["ledger_url"], "/balance/platform")["balance"] == 4 # grant exists lic = httpx.get( env["ledger_url"] + f"/license/m2bd/{LISTING_ID}", headers={"X-API-Key": SERVICE_KEY}, ) assert lic.status_code == 200 # file placed into the temp HOME by the local adapter assert (env["home"] / "m2-market-test" / "hello.txt").exists() # state.json updated state = json.loads(Path(os.environ["M2_MARKET_STATE"]).read_text()) assert state["installs"][LISTING_ID]["status"] == "applied" # entrypoint hint printed assert "cat ~/m2-market-test/hello.txt" in result.output def test_insufficient_funds_exit_3_nothing_written(env): # no grant: wallet is empty before = _tx_count(env["ledger_url"]) result = _run(["install", LISTING_ID, "--yes"]) assert result.exit_code == 3 assert _tx_count(env["ledger_url"]) == before assert not (env["home"] / "m2-market-test").exists() def test_idempotent_rerun_no_second_debit(env): _grant(env["ledger_url"], "m2bd", 100) assert _run(["install", LISTING_ID, "--yes"]).exit_code == 0 txs_after_first = _tx_count(env["ledger_url"]) rerun = _run(["install", LISTING_ID, "--yes"]) assert rerun.exit_code == 0 assert "no-op" in rerun.output assert _tx_count(env["ledger_url"]) == txs_after_first assert _ledger_get(env["ledger_url"], "/balance/m2bd")["balance"] == 100 - PRICE def test_apply_failure_refunds_and_rolls_back(env): _grant(env["ledger_url"], "m2bd", 100) # swap in a bundle whose post-install check fails bad_bundle, bad_hash = _build_bundle(env["tmp"] / "bad", failing_check=True) env["solution"]["content_hash"] = bad_hash env["monkeypatch"].setattr( env["cli_mod"].RegistryClient, "download_bundle", lambda self, ref, dest: (Path(dest).mkdir(parents=True, exist_ok=True), shutil.copy(bad_bundle, Path(dest) / "b.tar.gz"))[1] and Path(dest) / "b.tar.gz", ) before = _ledger_get(env["ledger_url"], "/balance/m2bd")["balance"] result = _run(["install", LISTING_ID, "--yes"]) assert result.exit_code == 4 assert "refund" in result.output.lower() # refund compensates exactly assert _ledger_get(env["ledger_url"], "/balance/m2bd")["balance"] == before refunds = _ledger_get(env["ledger_url"], "/tx?reason=refund")["transactions"] assert refunds, "compensating refund rows expected" # rollback removed placed files assert not (env["home"] / "m2-market-test" / "hello.txt").exists()