merge: cli-commands (US1 wave 2)
This commit is contained in:
commit
9bb441e4a7
3 changed files with 605 additions and 7 deletions
|
|
@ -2,12 +2,23 @@
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import socket
|
||||
import sys
|
||||
import tarfile
|
||||
import tempfile
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
|
||||
import click
|
||||
|
||||
from m2_market import __version__
|
||||
from m2_market.config import ConfigError, load_config
|
||||
from m2_market.apply import RailsNotLanded, get_adapter
|
||||
from m2_market.apply.base import InstallState
|
||||
from m2_market.catalog import CatalogClient
|
||||
from m2_market.config import Config, ConfigError, load_config
|
||||
from m2_market.ledger import InsufficientFunds, LedgerClient
|
||||
from m2_market.output import emit_json, format_price, render_kv, render_table, search_results_table
|
||||
from m2_market.registry import RegistryClient, RegistryError
|
||||
|
||||
# Exit-code map (contracts/cli.md).
|
||||
EXIT_OK = 0 # ok / no-op
|
||||
|
|
@ -46,7 +57,29 @@ def _not_implemented(ctx: click.Context, command: str) -> None:
|
|||
@click.pass_context
|
||||
def search(ctx: click.Context, query: str, listing_type: str | None, limit: int | None) -> None:
|
||||
"""Semantic catalog query, tenant-filtered."""
|
||||
_not_implemented(ctx, "search")
|
||||
config: Config = ctx.obj["config"]
|
||||
with CatalogClient(config.memory_api_url, config.memory_api_key, config.tenant) as catalog:
|
||||
items = catalog.search(query, limit=limit or 10)
|
||||
|
||||
if listing_type:
|
||||
items = [
|
||||
item
|
||||
for item in items
|
||||
if (item.get("listing") or item).get("inventory_type") == listing_type
|
||||
]
|
||||
|
||||
if ctx.obj["json"]:
|
||||
emit_json(items)
|
||||
else:
|
||||
search_results_table(items)
|
||||
|
||||
|
||||
def _fetch_solution(registry: RegistryClient, listing_id: str) -> dict:
|
||||
"""Best-effort solution lookup — permissions/entrypoint are a nice-to-have, not fatal."""
|
||||
try:
|
||||
return registry.get_solution(listing_id)
|
||||
except Exception:
|
||||
return {}
|
||||
|
||||
|
||||
@main.command()
|
||||
|
|
@ -54,16 +87,180 @@ def search(ctx: click.Context, query: str, listing_type: str | None, limit: int
|
|||
@click.pass_context
|
||||
def show(ctx: click.Context, listing_id: str) -> None:
|
||||
"""Full listing detail: evidence, price, permissions diff, seller, version."""
|
||||
_not_implemented(ctx, "show")
|
||||
config: Config = ctx.obj["config"]
|
||||
json_output = ctx.obj["json"]
|
||||
|
||||
with CatalogClient(config.memory_api_url, config.memory_api_key, config.tenant) as catalog:
|
||||
listing = catalog.get(listing_id)
|
||||
|
||||
if listing is None:
|
||||
click.echo(f"m2-market show: listing not found: {listing_id}", err=True)
|
||||
ctx.exit(EXIT_NOT_FOUND)
|
||||
return
|
||||
|
||||
with RegistryClient(config.forgejo_url, config.forgejo_token) as registry:
|
||||
solution = _fetch_solution(registry, listing_id)
|
||||
|
||||
permissions = solution.get("permissions") or []
|
||||
|
||||
if json_output:
|
||||
emit_json({"listing": listing, "solution": solution, "permissions": permissions})
|
||||
return
|
||||
|
||||
click.echo(f"{listing.get('name', listing_id)} ({listing_id})")
|
||||
click.echo(listing.get("summary", ""))
|
||||
click.echo()
|
||||
render_kv(
|
||||
[
|
||||
("evidence_summary", listing.get("evidence_summary", "-")),
|
||||
("price", format_price(listing.get("price"))),
|
||||
("seller", listing.get("seller", "-")),
|
||||
("version", listing.get("solution_version", "-")),
|
||||
("install_ref", listing.get("install_ref", "-")),
|
||||
]
|
||||
)
|
||||
if permissions:
|
||||
click.echo("\nPermissions:")
|
||||
for perm in permissions:
|
||||
click.echo(f" - {perm}")
|
||||
|
||||
|
||||
def _extract_bundle(bundle_path: Path, extract_dir: Path) -> Path:
|
||||
with tarfile.open(bundle_path) as tar:
|
||||
tar.extractall(extract_dir, filter="data")
|
||||
|
||||
entries = list(extract_dir.iterdir())
|
||||
if len(entries) == 1 and entries[0].is_dir():
|
||||
return entries[0]
|
||||
return extract_dir
|
||||
|
||||
|
||||
def _print_entrypoint_hint(json_output: bool, solution: dict, extra: dict | None = None) -> None:
|
||||
entrypoint = (solution.get("deployment") or {}).get("entrypoint")
|
||||
if json_output:
|
||||
payload = {"status": "ok", "entrypoint": entrypoint}
|
||||
if extra:
|
||||
payload.update(extra)
|
||||
emit_json(payload)
|
||||
elif entrypoint:
|
||||
click.echo(f"Installed. Run: {entrypoint}")
|
||||
else:
|
||||
click.echo("Installed.")
|
||||
|
||||
|
||||
@main.command()
|
||||
@click.argument("listing_id")
|
||||
@click.option("--yes", is_flag=True, default=False, help="Skip interactive confirmation.")
|
||||
@click.option("--adapter", "adapter_name", default=None, help="Override the configured apply adapter.")
|
||||
@click.pass_context
|
||||
def install(ctx: click.Context, listing_id: str, yes: bool) -> None:
|
||||
"""Resolve, pay, fetch, apply, and record a listing install."""
|
||||
_not_implemented(ctx, "install")
|
||||
def install(ctx: click.Context, listing_id: str, yes: bool, adapter_name: str | None) -> None:
|
||||
"""Resolve, pay, fetch, apply, and record a listing install (contracts/cli.md FR-005 sequence)."""
|
||||
config: Config = ctx.obj["config"]
|
||||
json_output = ctx.obj["json"]
|
||||
|
||||
state = InstallState()
|
||||
|
||||
with LedgerClient(config.ledger_url, config.ledger_api_key) as ledger:
|
||||
# Idempotent re-run guard (Story 1 AC-3): if we've already applied this
|
||||
# listing and a license grant still exists, no-op BEFORE debiting.
|
||||
cached = state.get(listing_id)
|
||||
if cached is not None and cached.get("status") == "applied":
|
||||
grant = ledger.license(config.operator_id, listing_id)
|
||||
if grant is not None:
|
||||
with RegistryClient(config.forgejo_url, config.forgejo_token) as registry:
|
||||
solution = _fetch_solution(registry, listing_id)
|
||||
if not json_output:
|
||||
click.echo(f"{listing_id} already installed — no-op.")
|
||||
_print_entrypoint_hint(json_output, solution, {"listing_id": listing_id, "noop": True})
|
||||
return
|
||||
|
||||
with CatalogClient(config.memory_api_url, config.memory_api_key, config.tenant) as catalog:
|
||||
listing = catalog.get(listing_id)
|
||||
|
||||
if listing is None:
|
||||
click.echo(f"m2-market install: listing not found: {listing_id}", err=True)
|
||||
ctx.exit(EXIT_NOT_FOUND)
|
||||
return
|
||||
|
||||
with RegistryClient(config.forgejo_url, config.forgejo_token) as registry:
|
||||
solution = _fetch_solution(registry, listing_id)
|
||||
price = listing.get("price") or {}
|
||||
permissions = solution.get("permissions") or []
|
||||
|
||||
if not json_output:
|
||||
click.echo(f"{listing.get('name', listing_id)} — {format_price(price)}")
|
||||
if permissions:
|
||||
click.echo("Permissions requested:")
|
||||
for perm in permissions:
|
||||
click.echo(f" - {perm}")
|
||||
|
||||
if not yes and not click.confirm("Proceed with install?", default=False):
|
||||
return
|
||||
|
||||
ref = f"{listing_id}:{socket.gethostname()}:{uuid.uuid4()}"
|
||||
seller = listing.get("seller") or solution.get("seller")
|
||||
amount = price.get("amount")
|
||||
|
||||
try:
|
||||
grant = ledger.install(buyer=config.operator_id, seller=seller, amount=amount, ref=ref)
|
||||
except InsufficientFunds as exc:
|
||||
click.echo(f"m2-market install: insufficient funds: {exc}", err=True)
|
||||
ctx.exit(EXIT_INSUFFICIENT_FUNDS)
|
||||
return
|
||||
|
||||
install_ref = listing.get("install_ref")
|
||||
content_hash = solution.get("content_hash")
|
||||
|
||||
with tempfile.TemporaryDirectory(prefix="m2-market-bundle-") as tmp_dir_str:
|
||||
tmp_dir = Path(tmp_dir_str)
|
||||
try:
|
||||
bundle_path = registry.download_bundle(install_ref, tmp_dir / "download")
|
||||
if content_hash and not RegistryClient.verify_bundle(bundle_path, content_hash):
|
||||
raise RegistryError(f"content hash mismatch for {install_ref}")
|
||||
bundle_root = _extract_bundle(bundle_path, tmp_dir / "extracted")
|
||||
except (RegistryError, OSError, tarfile.TarError) as exc:
|
||||
ledger.refund(ref)
|
||||
click.echo(
|
||||
f"m2-market install: bundle fetch/verify failed, refunded: {exc}",
|
||||
err=True,
|
||||
)
|
||||
ctx.exit(EXIT_APPLY_FAILED)
|
||||
return
|
||||
|
||||
adapter_choice = adapter_name or config.apply_adapter
|
||||
try:
|
||||
adapter = get_adapter(adapter_choice)
|
||||
changeset = adapter.plan(bundle_root, state)
|
||||
apply_result = adapter.apply(changeset)
|
||||
if not apply_result.ok:
|
||||
raise RuntimeError("; ".join(apply_result.errors) or "apply reported failure")
|
||||
if not adapter.verify(bundle_root, state):
|
||||
raise RuntimeError("post-install verification failed")
|
||||
except RailsNotLanded as exc:
|
||||
ledger.refund(ref)
|
||||
click.echo(f"m2-market install: {exc}", err=True)
|
||||
ctx.exit(EXIT_RAILS_NOT_LANDED)
|
||||
return
|
||||
except Exception as exc:
|
||||
ledger.refund(ref)
|
||||
click.echo(
|
||||
f"m2-market install: apply failed, refunded: {exc}\n"
|
||||
"Remediation: inspect the recipe's post-install checks, fix the "
|
||||
"underlying issue, then retry `m2-market install`.",
|
||||
err=True,
|
||||
)
|
||||
ctx.exit(EXIT_APPLY_FAILED)
|
||||
return
|
||||
|
||||
state.record_install(
|
||||
listing_id,
|
||||
listing.get("solution_version") or changeset.version,
|
||||
grant.get("grant_id", ""),
|
||||
changeset.changeset_hash,
|
||||
"applied",
|
||||
)
|
||||
|
||||
_print_entrypoint_hint(json_output, solution, {"listing_id": listing_id, "noop": False})
|
||||
|
||||
|
||||
@main.command()
|
||||
|
|
@ -71,7 +268,36 @@ def install(ctx: click.Context, listing_id: str, yes: bool) -> None:
|
|||
@click.pass_context
|
||||
def wallet(ctx: click.Context, operator: str | None) -> None:
|
||||
"""Balance + recent transactions from the ledger."""
|
||||
_not_implemented(ctx, "wallet")
|
||||
config: Config = ctx.obj["config"]
|
||||
json_output = ctx.obj["json"]
|
||||
operator_id = operator or config.operator_id
|
||||
|
||||
with LedgerClient(config.ledger_url, config.ledger_api_key) as ledger:
|
||||
balance = ledger.balance(operator_id)
|
||||
txs = ledger.transactions(operator_id=operator_id)
|
||||
|
||||
recent = list(reversed(txs))[:10]
|
||||
|
||||
if json_output:
|
||||
emit_json({"balance": balance, "transactions": recent})
|
||||
return
|
||||
|
||||
click.echo(f"Operator: {operator_id}")
|
||||
click.echo(f"Balance: {balance.get('balance')} (as of {balance.get('as_of')})")
|
||||
click.echo()
|
||||
headers = ["ts", "from", "to", "amount", "reason", "ref"]
|
||||
rows = [
|
||||
[
|
||||
str(tx.get("ts", "-")),
|
||||
str(tx.get("from", "-")),
|
||||
str(tx.get("to", "-")),
|
||||
str(tx.get("amount", "-")),
|
||||
str(tx.get("reason", "-")),
|
||||
str(tx.get("ref", "-")),
|
||||
]
|
||||
for tx in recent
|
||||
]
|
||||
render_table(headers, rows)
|
||||
|
||||
|
||||
@main.command()
|
||||
|
|
|
|||
64
cli/src/m2_market/output.py
Normal file
64
cli/src/m2_market/output.py
Normal file
|
|
@ -0,0 +1,64 @@
|
|||
"""Table/JSON rendering helpers for the m2-market CLI (contracts/cli.md)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json as _json
|
||||
|
||||
import click
|
||||
|
||||
|
||||
def emit_json(data: object) -> None:
|
||||
click.echo(_json.dumps(data, indent=2, sort_keys=True, default=str))
|
||||
|
||||
|
||||
def _col_widths(headers: list[str], rows: list[list[str]]) -> list[int]:
|
||||
widths = [len(h) for h in headers]
|
||||
for row in rows:
|
||||
for i, cell in enumerate(row):
|
||||
widths[i] = max(widths[i], len(cell))
|
||||
return widths
|
||||
|
||||
|
||||
def render_table(headers: list[str], rows: list[list[str]]) -> None:
|
||||
if not rows:
|
||||
click.echo("(no results)")
|
||||
return
|
||||
widths = _col_widths(headers, rows)
|
||||
line = " ".join(h.ljust(w) for h, w in zip(headers, widths))
|
||||
click.echo(line)
|
||||
click.echo(" ".join("-" * w for w in widths))
|
||||
for row in rows:
|
||||
click.echo(" ".join(cell.ljust(w) for cell, w in zip(row, widths)))
|
||||
|
||||
|
||||
def format_price(price: dict | None) -> str:
|
||||
if not price:
|
||||
return "-"
|
||||
amount = price.get("amount")
|
||||
currency = price.get("currency", "")
|
||||
return f"{amount} {currency}".strip()
|
||||
|
||||
|
||||
def search_results_table(items: list[dict]) -> None:
|
||||
headers = ["listing_id", "name", "summary", "price", "installs", "rating"]
|
||||
rows = []
|
||||
for item in items:
|
||||
listing = item.get("listing", item)
|
||||
stats = listing.get("stats") or {}
|
||||
rows.append(
|
||||
[
|
||||
str(item.get("listing_id", listing.get("listing_id", "-"))),
|
||||
str(listing.get("name", "-")),
|
||||
str(listing.get("summary", "-")),
|
||||
format_price(listing.get("price")),
|
||||
str(stats.get("installs", "-")),
|
||||
str(stats.get("rating", "-")),
|
||||
]
|
||||
)
|
||||
render_table(headers, rows)
|
||||
|
||||
|
||||
def render_kv(pairs: list[tuple[str, str]]) -> None:
|
||||
width = max((len(k) for k, _ in pairs), default=0)
|
||||
for key, value in pairs:
|
||||
click.echo(f"{key.ljust(width)} : {value}")
|
||||
308
cli/tests/test_cli_install.py
Normal file
308
cli/tests/test_cli_install.py
Normal file
|
|
@ -0,0 +1,308 @@
|
|||
"""Inline verification for the install/search/show/wallet CLI commands.
|
||||
|
||||
Wires httpx.MockTransport fakes for the ledger/registry/catalog clients (per
|
||||
contracts/ledger-api.md, contracts/registry-layout.md, contracts/catalog-index.md)
|
||||
and a fake ApplyAdapter (the T021 local adapter lands on a sibling branch and
|
||||
isn't merged into this worktree yet — the CLI must not depend on its internals).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import io
|
||||
import json
|
||||
import tarfile
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
import m2_market.cli as cli_mod
|
||||
from m2_market.apply.base import ApplyResult, Changeset
|
||||
|
||||
LISTING_ID = "lst_pdf_export"
|
||||
INSTALL_REF = "lst_pdf_export-v1.0.0"
|
||||
OPERATOR = "buyer1"
|
||||
SELLER = "seller1"
|
||||
|
||||
LISTING = {
|
||||
"listing_id": LISTING_ID,
|
||||
"solution_id": "sol_pdf_export",
|
||||
"solution_version": "1.0.0",
|
||||
"inventory_type": "solution",
|
||||
"name": "PDF Export",
|
||||
"summary": "Branded PDF reports.",
|
||||
"category": "productivity",
|
||||
"price": {"amount": 20, "currency": "m2cr", "model": "fixed"},
|
||||
"seller": SELLER,
|
||||
"evidence_summary": "Used on 12 sessions, saved 40 min avg.",
|
||||
"tenant_visibility": ["*"],
|
||||
"stats": {"installs": 3, "rating": 4.5, "proposals_shown": 5, "proposals_accepted": 3},
|
||||
"status": "published",
|
||||
"install_ref": INSTALL_REF,
|
||||
}
|
||||
|
||||
SOLUTION = {
|
||||
"schema_version": "m2.solution.v1",
|
||||
"solution_id": "sol_pdf_export",
|
||||
"name": "PDF Export",
|
||||
"summary": "Branded PDF reports.",
|
||||
"intent": "export pdfs",
|
||||
"deployment": {
|
||||
"recipe_ref": "recipe.yaml",
|
||||
"entrypoint": "m2-pdf-export --help",
|
||||
"verify_command": "true",
|
||||
},
|
||||
"applicability": {},
|
||||
"tenant_scope": "m2-core",
|
||||
"evidence": [{"source": "session-123"}],
|
||||
"content_hash": "",
|
||||
"price": {"amount": 20, "currency": "m2cr", "model": "fixed"},
|
||||
"seller": SELLER,
|
||||
"license": {"terms": "internal", "major_version_coverage": True},
|
||||
"permissions": ["fs:write:~/reports"],
|
||||
}
|
||||
|
||||
|
||||
def _make_bundle_bytes() -> bytes:
|
||||
buf = io.BytesIO()
|
||||
with tarfile.open(fileobj=buf, mode="w:gz") as tar:
|
||||
data = b"entrypoint stub\n"
|
||||
info = tarfile.TarInfo(name="bundle/recipe.yaml")
|
||||
info.size = len(data)
|
||||
tar.addfile(info, io.BytesIO(data))
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
BUNDLE_BYTES = _make_bundle_bytes()
|
||||
|
||||
|
||||
class FakeState:
|
||||
"""Records calls made against the fake ledger/registry backends."""
|
||||
|
||||
def __init__(self):
|
||||
self.install_calls = 0
|
||||
self.refund_calls = 0
|
||||
self.license_exists = False
|
||||
self.insufficient_funds = False
|
||||
|
||||
|
||||
def _b64_json(payload: dict) -> str:
|
||||
return base64.b64encode(json.dumps(payload).encode("utf-8")).decode("ascii")
|
||||
|
||||
|
||||
def make_handler(state: FakeState):
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
host = request.url.host
|
||||
path = request.url.path
|
||||
|
||||
if host == "ledger.test":
|
||||
if path == "/tx/install" and request.method == "POST":
|
||||
state.install_calls += 1
|
||||
if state.insufficient_funds:
|
||||
return httpx.Response(402, json={"error": "insufficient_funds", "balance": 0})
|
||||
return httpx.Response(
|
||||
200,
|
||||
json={
|
||||
"tx_ids": ["tx1", "tx2"],
|
||||
"grant": {"grant_id": "grant-1", "listing_id": LISTING_ID, "operator_id": OPERATOR},
|
||||
},
|
||||
)
|
||||
if path == "/tx/refund" and request.method == "POST":
|
||||
state.refund_calls += 1
|
||||
return httpx.Response(200, json={"tx_ids": ["tx-refund"]})
|
||||
if path.startswith("/license/"):
|
||||
if state.license_exists:
|
||||
return httpx.Response(200, json={"grant": {"grant_id": "grant-1"}})
|
||||
return httpx.Response(404, json={"error": "not_found"})
|
||||
if path.startswith("/balance/"):
|
||||
return httpx.Response(200, json={"operator_id": OPERATOR, "balance": 80, "as_of": "now"})
|
||||
if path == "/tx":
|
||||
return httpx.Response(
|
||||
200,
|
||||
json={
|
||||
"transactions": [
|
||||
{"ts": "t1", "from": OPERATOR, "to": SELLER, "amount": 18, "reason": "install", "ref": "r1"},
|
||||
{"ts": "t2", "from": OPERATOR, "to": "platform", "amount": 2, "reason": "install", "ref": "r1"},
|
||||
]
|
||||
},
|
||||
)
|
||||
|
||||
if host == "registry.test":
|
||||
if path.endswith("/contents/listings/lst_pdf_export/solution.json"):
|
||||
return httpx.Response(200, json={"content": _b64_json(SOLUTION)})
|
||||
if path.endswith(f"/releases/tags/{INSTALL_REF}"):
|
||||
return httpx.Response(
|
||||
200,
|
||||
json={
|
||||
"assets": [
|
||||
{
|
||||
"name": "bundle.tar.gz",
|
||||
"browser_download_url": "https://registry.test/download/bundle.tar.gz",
|
||||
}
|
||||
]
|
||||
},
|
||||
)
|
||||
if path == "/download/bundle.tar.gz":
|
||||
return httpx.Response(200, content=BUNDLE_BYTES)
|
||||
|
||||
if host == "catalog.test" and path == "/search":
|
||||
body = json.loads(request.content)
|
||||
if body["query_text"] == LISTING_ID:
|
||||
return httpx.Response(200, json={"items": [{"listing_id": LISTING_ID, "score": 0.9, "listing": LISTING}]})
|
||||
return httpx.Response(200, json={"items": [{"listing_id": LISTING_ID, "score": 0.8, "listing": LISTING}]})
|
||||
|
||||
raise AssertionError(f"unexpected request: {request.method} {request.url}")
|
||||
|
||||
return handler
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake_backend(monkeypatch):
|
||||
state = FakeState()
|
||||
handler = make_handler(state)
|
||||
|
||||
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)
|
||||
return state
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def config_env(tmp_path, monkeypatch):
|
||||
config_path = tmp_path / "config.toml"
|
||||
config_path.write_text(
|
||||
f"""
|
||||
operator_id = "{OPERATOR}"
|
||||
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"
|
||||
apply_adapter = "fake"
|
||||
"""
|
||||
)
|
||||
state_path = tmp_path / "state.json"
|
||||
monkeypatch.setenv("M2_MARKET_CONFIG", str(config_path))
|
||||
monkeypatch.setenv("M2_MARKET_STATE", str(state_path))
|
||||
return state_path
|
||||
|
||||
|
||||
class FakeAdapterCalls:
|
||||
def __init__(self):
|
||||
self.plan_calls = 0
|
||||
self.apply_calls = 0
|
||||
self.verify_calls = 0
|
||||
self.apply_should_fail = False
|
||||
|
||||
|
||||
def make_fake_get_adapter(calls: FakeAdapterCalls):
|
||||
class FakeAdapter:
|
||||
name = "fake"
|
||||
|
||||
def plan(self, bundle, state):
|
||||
calls.plan_calls += 1
|
||||
return Changeset(listing_id="sol_pdf_export", version="1.0.0", items=[], changeset_hash="sha256:abc")
|
||||
|
||||
def apply(self, changeset):
|
||||
calls.apply_calls += 1
|
||||
if calls.apply_should_fail:
|
||||
return ApplyResult(ok=False, applied=[], errors=["recipe check failed"])
|
||||
return ApplyResult(ok=True, applied=[], errors=[])
|
||||
|
||||
def verify(self, bundle, state):
|
||||
calls.verify_calls += 1
|
||||
return True
|
||||
|
||||
def rollback(self, changeset):
|
||||
pass
|
||||
|
||||
def get_adapter(name):
|
||||
return FakeAdapter()
|
||||
|
||||
return get_adapter
|
||||
|
||||
|
||||
def test_funded_install_happy_path(fake_backend, config_env, monkeypatch):
|
||||
calls = FakeAdapterCalls()
|
||||
monkeypatch.setattr(cli_mod, "get_adapter", make_fake_get_adapter(calls))
|
||||
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli_mod.main, ["install", LISTING_ID, "--yes"])
|
||||
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "m2-pdf-export --help" in result.output
|
||||
assert fake_backend.install_calls == 1
|
||||
assert fake_backend.refund_calls == 0
|
||||
assert calls.plan_calls == 1
|
||||
assert calls.apply_calls == 1
|
||||
assert calls.verify_calls == 1
|
||||
|
||||
|
||||
def test_insufficient_funds_no_adapter_call(fake_backend, config_env, monkeypatch):
|
||||
fake_backend.insufficient_funds = True
|
||||
calls = FakeAdapterCalls()
|
||||
monkeypatch.setattr(cli_mod, "get_adapter", make_fake_get_adapter(calls))
|
||||
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli_mod.main, ["install", LISTING_ID, "--yes"])
|
||||
|
||||
assert result.exit_code == cli_mod.EXIT_INSUFFICIENT_FUNDS, result.output
|
||||
assert calls.plan_calls == 0
|
||||
assert calls.apply_calls == 0
|
||||
assert fake_backend.refund_calls == 0
|
||||
|
||||
|
||||
def test_rerun_is_idempotent_noop_without_second_debit(fake_backend, config_env, monkeypatch):
|
||||
calls = FakeAdapterCalls()
|
||||
monkeypatch.setattr(cli_mod, "get_adapter", make_fake_get_adapter(calls))
|
||||
|
||||
runner = CliRunner()
|
||||
first = runner.invoke(cli_mod.main, ["install", LISTING_ID, "--yes"])
|
||||
assert first.exit_code == 0, first.output
|
||||
assert fake_backend.install_calls == 1
|
||||
|
||||
fake_backend.license_exists = True
|
||||
second = runner.invoke(cli_mod.main, ["install", LISTING_ID, "--yes"])
|
||||
|
||||
assert second.exit_code == 0, second.output
|
||||
assert fake_backend.install_calls == 1 # no second debit
|
||||
assert calls.plan_calls == 1 # adapter not invoked again
|
||||
|
||||
|
||||
def test_apply_failure_triggers_refund_and_exit_4(fake_backend, config_env, monkeypatch):
|
||||
calls = FakeAdapterCalls()
|
||||
calls.apply_should_fail = True
|
||||
monkeypatch.setattr(cli_mod, "get_adapter", make_fake_get_adapter(calls))
|
||||
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli_mod.main, ["install", LISTING_ID, "--yes"])
|
||||
|
||||
assert result.exit_code == cli_mod.EXIT_APPLY_FAILED, result.output
|
||||
assert fake_backend.install_calls == 1
|
||||
assert fake_backend.refund_calls == 1
|
||||
assert "refunded" in result.output
|
||||
|
||||
|
||||
def test_wallet_prints_balance_and_recent_tx(fake_backend, config_env):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli_mod.main, ["wallet"])
|
||||
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "80" in result.output
|
||||
assert "install" in result.output
|
||||
|
||||
|
||||
def test_search_renders_table(fake_backend, config_env):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli_mod.main, ["search", "pdf report"])
|
||||
|
||||
assert result.exit_code == 0, result.output
|
||||
assert LISTING_ID in result.output
|
||||
assert "PDF Export" in result.output
|
||||
Loading…
Reference in a new issue