331 lines
11 KiB
Python
331 lines
11 KiB
Python
"""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
|
|
self.telemetry_calls = []
|
|
|
|
|
|
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 == "/memory/search":
|
|
body = json.loads(request.content)
|
|
assert body["agent_id"] == "market:catalog", body
|
|
score = 0.9 if body["query"] == LISTING_ID else 0.8
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"results": [
|
|
{
|
|
"id": "mem-1",
|
|
"score": score,
|
|
"metadata": {"listing_id": LISTING_ID, "listing": LISTING},
|
|
}
|
|
],
|
|
"routing": {},
|
|
},
|
|
)
|
|
|
|
if host == "catalog.test" and path == "/memory/store":
|
|
body = json.loads(request.content)
|
|
assert body["agent_id"] == "market:evidence", body
|
|
state.telemetry_calls.append(body)
|
|
return httpx.Response(200, json={"id": "mem-telemetry-1"})
|
|
|
|
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
|
|
assert len(fake_backend.telemetry_calls) == 1
|
|
telemetry = fake_backend.telemetry_calls[0]
|
|
assert telemetry["metadata"]["outcome"] == "applied"
|
|
assert telemetry["metadata"]["listing_id"] == LISTING_ID
|
|
|
|
|
|
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
|