"""erp-compose deploy / teardown (0.9.12) with the dodil CLI FAKED: every command is recorded, no secret ever
appears in an argv, a re-run makes no new resource, roles are merged (never replaced), teardown leaves nothing."""
import json
import os
import pathlib
import stat
import subprocess
import pytest
from erp_platform import deploy, gate
from erp_platform.deploy import Deploy, Dodil, Teardown
SA_SECRET, SC_SECRET = "sa-SECRET-9f8e7d", "sc-SECRET-1a2b3c"
ORG, APP = "acme", "shop"
DRN = f"drn:dodil:k3:uk-lon-1:org-uuid:bucket/{APP}"
class FakeCLI:
"""The dodil CLI's state for one org, answering the commands deploy/teardown send (JSON, like -o json)."""
def __init__(self, pool_roles=None):
self.buckets, self.pools, self.sas, self.repos, self.apps = {}, {}, {}, {}, {}
self.roles = dict(pool_roles or {})
self.settings = {"allow_signup": True, "redirect_uris": ["https://claude.ai/api/mcp/auth_callback"]}
self.clients, self.git_secrets, self.sa_roles, self.registry = {}, {}, {}, set()
self.pushed = None
self.calls = []
self.n = 0
def __call__(self, argv, input=None, capture_output=True, text=True, timeout=None):
self.calls.append((argv, input))
args = argv[1:]
assert args[-2:] == ["-o", "json"] and args[-4] in ("--org", "--org-name") and args[-3] == ORG
a = args[:-4]
try:
out = self.answer(a, input)
except KeyError as e:
return subprocess.CompletedProcess(argv, 1, "", f"Error: HTTP 404: {e} not found")
if isinstance(out, tuple):
return subprocess.CompletedProcess(argv, out[0], "", out[1])
return subprocess.CompletedProcess(argv, 0, json.dumps(out), "")
def answer(self, a, stdin):
k = tuple(a[:3])
if k == ("data", "bucket", "get"):
return self.buckets[a[3]]
if k == ("data", "bucket", "create"):
self.buckets[a[3]] = {"name": a[3], "drn": DRN}
self.sas[f"k3-tables-sa-{a[3]}-0001@{ORG}.sa"] = "uuid-tables"
return {}
if k == ("data", "bucket", "delete"):
del self.buckets[a[3]]
return {}
if k == ("data", "bucket", "list"):
return {"buckets": list(self.buckets.values())}
if k == ("appid", "pool", "get"):
self.pools[a[3]]
return {"pool": {"name": a[3], "settingsJson": json.dumps(self.settings)}}
if k == ("appid", "pool", "create"):
self.pools[a[3]] = True
return {}
if k == ("appid", "pool", "update"):
assert a[4] == "--settings" and a[5].startswith("@")
self.settings = json.loads(pathlib.Path(a[5][1:]).read_text())
return {}
if k == ("appid", "pool", "delete"):
assert "--yes" in a
del self.pools[a[3]]
self.clients.clear()
return {}
if k == ("appid", "pool", "list"):
return {"pools": [{"name": p} for p in self.pools]}
if k == ("appid", "roles", "get"):
return self.roles
if k == ("appid", "roles", "set"):
assert a[4] == "--file" and a[5].startswith("@")
self.roles = json.loads(pathlib.Path(a[5][1:]).read_text())
return {}
if k == ("appid", "service-client", "list"):
return {"serviceClients": [{"clientId": c} for c in self.clients]}
if k == ("appid", "service-client", "create"):
self.clients[a[a.index("--client-id") + 1]] = [a[i + 1] for i, x in enumerate(a) if x == "--role"]
return {"serviceClient": {"clientId": a[5], "clientSecret": SC_SECRET},
"tokenEndpoint": "https://appid.example/token"}
if k == ("appid", "service-client", "delete"):
del self.clients[a[4]]
return {}
if k == ("auth", "service-account", "list"):
needle = a[a.index("--filter") + 1]
return {"serviceAccounts": [{"serviceAccountId": i, "serviceAccountUuid": u} for i, u in self.sas.items()
if needle in i]}
if k == ("auth", "service-account", "create"):
self.n += 1
sid = f"cli-{a[3]}@{ORG}.sa"
self.sas[sid] = f"uuid-{self.n}"
return {"serviceAccountId": sid, "uuid": f"uuid-{self.n}", "client_id": sid, "secret": SA_SECRET}
if k == ("auth", "service-account", "roles"):
return {"roles": self.sa_roles.get(a[3], [])}
if k == ("auth", "service-account", "grant-role"):
self.sa_roles.setdefault(a[3], []).append({"service": a[4], "role": a[5], "resources": [a[7]]})
return {}
if k == ("auth", "service-account", "delete"):
assert "--yes" in a
sid = next(i for i, u in self.sas.items() if u == a[3])
del self.sas[sid]
return {}
if k == ("git", "repo", "get"):
return self.repos[a[3]]
if k == ("git", "repo", "create"):
assert a[a.index("--visibility") + 1] == "private"
self.repos[a[3]] = {"name": a[3], "visibility": "private",
"http_url": f"https://git.example/{ORG}/{a[3]}.git"}
return {}
if k == ("git", "repo", "delete"):
del self.repos[a[3]]
return {}
if k == ("git", "repo", "list"):
return {"repos": list(self.repos.values())}
if k == ("git", "secret", "list"):
return {"secrets": [{"name": n} for n in self.git_secrets]}
if k == ("git", "secret", "set"):
assert stdin, "a secret value comes on stdin"
self.git_secrets[a[4]] = stdin
return {}
if k == ("git", "ci", "init"):
return (1, "unknown command")
if k == ("git", "ci", "verify"):
return {"ok": True}
if k == ("git", "check", "list"):
if self.pushed != a[4]:
return {"checks": [], "rollup": "unspecified"}
self.apps.setdefault(APP, {"name": APP, "public_urls": [f"{APP}-{ORG}-8000.ignite.dodil.cloud"],
"deployment_state": "deployed", "scaling": {"reserved_capacity": 0}})
self.registry.add(APP)
return {"checks": [{"name": n, "state": "passed", "check_id": "c"} for n in
("build/build", "ci/signoff", "dodil-ci/deploy-config", "dodil-ci/deploy")]}
if k == ("ignite", "app", "get"):
return {"data": self.apps[a[3]]}
if k == ("ignite", "app", "update"):
self.apps[a[3]]["scaling"]["reserved_capacity"] = int(a[a.index("--reserved") + 1])
return {}
if k == ("ignite", "app", "delete"):
del self.apps[a[3]]
return {}
if k == ("ignite", "app", "list"):
return {"data": list(self.apps.values())}
if k == ("ignite", "secret", "list"):
return {"secrets": []}
if k == ("registry", "repo", "list"):
return {"items": [{"name": n} for n in self.registry]}
if k == ("registry", "repo", "delete"):
self.registry.discard(a[3])
return {}
raise AssertionError(f"unexpected dodil command: {a}")
def argvs(self):
return [" ".join(c[0][1:]) for c in self.calls]
class FakeHTTP:
class R:
def __init__(self, code, body=None, headers=None):
self.status_code, self._b, self.headers = code, body, headers or {}
self.text = json.dumps(body) if body is not None else ""
def json(self):
return self._b
def __init__(self):
self.seen = []
def get(self, url):
self.seen.append(("GET", url))
return self.R(200, {"ok": True})
def post(self, url, json=None, data=None, headers=None):
self.seen.append(("POST", url, (json or {}).get("method")))
if data:
assert data["client_secret"] == SC_SECRET
return self.R(200, {"access_token": "tok"})
m = json["method"]
if m == "initialize":
return self.R(200, {"jsonrpc": "2.0", "id": 1, "result": {}}, {"mcp-session-id": "s1"})
if m == "notifications/initialized":
return self.R(202)
return self.R(200, {"jsonrpc": "2.0", "id": json["id"], "result": {"tools": [{"name": "a"}, {"name": "b"}]}})
@pytest.fixture
def comp(tmp_path, monkeypatch):
root = tmp_path / "shop"
(root / "config").mkdir(parents=True)
(root / "compose.yaml").write_text("name: shop\n")
(root / "requirements.txt").write_text("erp-platform @ git+https://git.dodil.io/dodil/erp-platform.git@v0.9.11\n")
(root / "config" / "uk-wholesale-sample.calls.yaml").write_text("calls: []\n")
(root / "config" / "uk-wholesale.calls.yaml").write_text("calls: []\n")
monkeypatch.setattr(gate, "check", lambda *a, **k: [{"module": "ar", "process": "ar", "state": "signed",
"detail": "ok"}])
monkeypatch.setattr(Deploy, "_catalog", lambda self: {"manager": ["*:*:*"], "erp_machine": ["gl:journal:post"]})
for k in ("GIT_AUTHOR_NAME", "GIT_COMMITTER_NAME"):
monkeypatch.setenv(k, "t")
for k in ("GIT_AUTHOR_EMAIL", "GIT_COMMITTER_EMAIL"):
monkeypatch.setenv(k, "t@t")
return root
def make(root, cli, tmp_path, **kw):
local, out = [], []
remote = {"sha": None}
def local_run(argv, env, cwd):
local.append((argv[3:], {k: env.get(k) for k in ("BUCKET", "DODIL_SERVICE_ACCOUNT_SECRET")}))
return 0, "ok"
def git_run(args, cwd):
if "ls-remote" in args:
return 0, f"{cli.pushed}\trefs/heads/main\n" if cli.pushed else ""
if "push" in args:
assert "credential.helper=" in args and any(a.endswith("git credential-helper") for a in args)
remote["sha"] = subprocess.run(["git", "rev-parse", "HEAD"], cwd=cwd, capture_output=True,
text=True).stdout.strip()
cli.pushed = remote["sha"]
return 0, ""
p = subprocess.run(["git", *args], cwd=cwd, capture_output=True, text=True)
return p.returncode, p.stdout + p.stderr
d = Deploy(root / "compose.yaml", ORG, APP, dodil=Dodil(ORG, bin="dodil", runner=cli),
secrets_dir=tmp_path / "secrets", say=out.append, local_run=local_run, git_run=git_run,
http=FakeHTTP(), sleep=lambda s: None, **kw)
return d, local, out
def no_secret_in_argv(cli):
for argv, _ in cli.calls:
for s in (SA_SECRET, SC_SECRET):
assert all(s not in a for a in argv), argv
def test_refuses_unless_signed(comp, tmp_path, monkeypatch):
monkeypatch.setattr(gate, "check", lambda *a, **k: [{"module": "ar", "process": "ar", "state": "unsigned",
"detail": "no receipt"}])
cli = FakeCLI()
d, local, out = make(comp, cli, tmp_path)
assert d.run() == 1
assert cli.calls == [] and local == []
assert any("signoff record --all" in line for line in out)
def test_deploy_end_to_end_then_idempotent_rerun(comp, tmp_path):
cli = FakeCLI(pool_roles={"sales": ["so:order:read"], "manager": ["pj:*:read"]})
d, local, out = make(comp, cli, tmp_path, warm=True)
assert d.run() == 0, "\n".join(out)
no_secret_in_argv(cli)
# secrets: on stdin to the repo's CI store, in migrate's env, in 0600 files outside the repo — never in the repo
assert cli.git_secrets == {"DODIL_SERVICE_ACCOUNT_ID": f"cli-{APP}-app@{ORG}.sa",
"DODIL_SERVICE_ACCOUNT_SECRET": SA_SECRET}
assert local[0] == (["migrate", str(comp / "compose.yaml")], {"BUCKET": APP, "DODIL_SERVICE_ACCOUNT_SECRET":
SA_SECRET})
assert [pathlib.Path(x[0][2]).name for x in local[1:]] == ["uk-wholesale.calls.yaml",
"uk-wholesale-sample.calls.yaml"]
for f in ("sa.json", "sc.json"):
assert stat.S_IMODE(os.stat(tmp_path / "secrets" / f).st_mode) == 0o600
for p in comp.rglob("*"):
if p.is_file() and ".git/" not in str(p):
text = p.read_text(errors="ignore")
assert SA_SECRET not in text and SC_SECRET not in text, p
# least privilege: k3.editor on THIS bucket only; the pool's roles merged, never replaced
assert cli.sa_roles["uuid-1"] == [{"service": "k3-authorization-service", "role": "k3.editor",
"resources": [f"resource={DRN}"]}]
assert cli.roles["sales"] == ["so:order:read"] and cli.roles["manager"] == ["pj:*:read", "*:*:*"]
assert cli.clients[f"{APP}-mcp-check"] == ["erp_machine", "manager"]
assert cli.settings["allow_signup"] is False and cli.settings["cimd_enabled"] is True
assert cli.settings["redirect_uris"] == ["https://claude.ai/api/mcp/auth_callback",
f"https://{APP}-{ORG}-8000.ignite.dodil.cloud/.dodil/auth/callback"]
# the files: CD waits for both checks, the build is `dodil git ci init`'s, never --code
dy = (comp / ".dodil/deploy.yaml").read_text()
assert "require_checks: [build/build, ci/signoff]" in dy and "${{ secrets.DODIL_SERVICE_ACCOUNT_SECRET }}" in dy
assert "reserved" not in dy.split("version: 1")[1]
assert "name: build" in (comp / ".github/workflows/build.yml").read_text()
assert "erp-compose signoff verify --all" in (comp / ".github/workflows/ci.yml").read_text()
assert "FROM python:3.12-slim" in (comp / "Dockerfile").read_text()
assert not any("--code" in a for a in cli.argvs())
assert cli.apps[APP]["scaling"]["reserved_capacity"] == 1
inv = json.loads((comp / ".dodil/inventory.json").read_text())
assert {"bucket", "bucket_tables_sa", "pool", "service_account", "service_client", "repo", "app",
"registry"} <= set(inv["resources"])
assert inv["mcp_tools"] == 2
tracked = subprocess.run(["git", "ls-files"], cwd=comp, capture_output=True, text=True).stdout.split()
assert ".dodil/inventory.json" not in tracked and ".dodil/deploy.yaml" in tracked
# re-run: nothing new is made, nothing is pushed again, migrate is not repeated
n = len(cli.calls)
d2, local2, out2 = make(comp, cli, tmp_path, warm=True)
assert d2.run() == 0, "\n".join(out2)
again = [" ".join(c[0][1:]) for c in cli.calls[n:]]
assert not [a for a in again if any(w in a.split()[:4] for w in ("create", "set", "grant-role", "rotate",
"update", "delete"))], again
assert local2 == []
assert any("already" in line for line in out2)
no_secret_in_argv(cli)
def test_secret_in_a_file_is_never_committed(comp, tmp_path):
cli = FakeCLI()
d, _, out = make(comp, cli, tmp_path)
(comp / "notes.txt").write_text(f"the secret is {SA_SECRET}\n")
assert d.run() == 1
assert any("notes.txt holds a secret" in line for line in out)
assert cli.pushed is None
no_secret_in_argv(cli)
def test_argv_guard_refuses_a_secret():
d = Dodil(ORG, bin="dodil", runner=lambda *a, **k: pytest.fail("ran"))
d.keep_secret("s3cr3t")
with pytest.raises(deploy.Refused):
d.run("git", "secret", "set", "r", "X=s3cr3t")
def test_teardown_deletes_everything_and_lists_none(comp, tmp_path):
cli = FakeCLI()
d, _, out = make(comp, cli, tmp_path)
assert d.run() == 0, "\n".join(out)
said = []
t = Teardown(comp / "compose.yaml", ORG, dodil=Dodil(ORG, bin="dodil", runner=cli),
secrets_dir=tmp_path / "secrets", say=said.append, sleep=lambda s: None)
assert t.run() == 0, "\n".join(said)
assert not (cli.buckets or cli.pools or cli.sas or cli.repos or cli.apps or cli.registry)
assert not (tmp_path / "secrets").exists()
assert all(line.endswith(": none") for line in said[said.index("\nlisted after teardown:") + 1:])
no_secret_in_argv(cli)
def test_teardown_keep_repo(comp, tmp_path):
cli = FakeCLI()
d, _, _ = make(comp, cli, tmp_path)
assert d.run() == 0
t = Teardown(comp / "compose.yaml", ORG, keep_repo=True, dodil=Dodil(ORG, bin="dodil", runner=cli),
secrets_dir=tmp_path / "secrets", say=lambda s: None, sleep=lambda s: None)
assert t.run() == 0
assert list(cli.repos) == [APP] and not cli.buckets
def test_a_customers_own_file_is_kept(comp, tmp_path):
(comp / "Dockerfile").write_text("FROM mine\n")
d, _, out = make(comp, FakeCLI(), tmp_path)
assert d.run() == 0
assert (comp / "Dockerfile").read_text() == "FROM mine\n"
assert any("Dockerfile: kept" in line for line in out)
def test_a_master_data_brick_has_nothing_to_sign(tmp_path):
"""0.9.12: init lists business-partner (process_view "none: …") in modules:; the gate must not ask for its
signature (record --all died on 'no workflow has transitions')."""
(tmp_path / "compose.yaml").write_text(
"modules: [business-partner, ar]\nbricks:\n"
' - { name: business-partner, brick: "x:BP", version: 0.4.0, process_view: "none: master data only" }\n'
' - { name: ar, brick: "x:AR", version: 0.1.0 }\n')
assert gate.modules(tmp_path) == ["ar"]