"""Shared mailu helpers — headless mailbox provisioning via the admin container's `flask mailu` CLI. mailu's admin container ships the management CLI (`flask mailu domain|user|alias|config-export ...`), the canonical headless way to provision mail objects. We exec it via the harness (service="admin"). The primary MAIL_DOMAIN is auto-created at boot; `flask mailu domain ... || true` is idempotent. """ from __future__ import annotations import json import os import sys sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "runner")) from harness import lifecycle # noqa: E402 def flask_mailu(domain: str, shell_cmd: str) -> str: """Run a `flask mailu ...` shell command in the admin container; return stdout.""" return lifecycle.exec_in_app(domain, ["sh", "-c", shell_cmd], service="admin") def ensure_domain(domain: str, mail_domain: str) -> None: # idempotent: the primary domain is auto-created; tolerate "already exists" flask_mailu(domain, f"flask mailu domain {mail_domain} || true") def create_user(domain: str, local: str, mail_domain: str, password: str) -> str: email = f"{local}@{mail_domain}" flask_mailu(domain, f"flask mailu user {local} {mail_domain} '{password}'") return email def config_export(domain: str) -> dict: """`flask mailu config-export --json` → parsed dict (keys: domain/user/alias/relay).""" out = flask_mailu(domain, "flask mailu config-export --json") # Be robust to any leading banner: extract the JSON object. start = out.find("{") end = out.rfind("}") assert start != -1 and end != -1, f"config-export produced no JSON: {out[:200]!r}" return json.loads(out[start : end + 1]) def user_emails(cfg: dict) -> list[str]: return [u.get("email") for u in cfg.get("user", [])]