runv-server

server tooling for runv.club
Log | Files | Refs | README

smoke_test_email_aliases.py (11834B)

      1 #!/usr/bin/env python3
      2 """
      3 Smoke test dos aliases de email runv.club (Linux).
      4 
      5 Por defeito usa diretório temporário. Modo --direct chama a biblioteca in-process
      6 (útil em dev/WSL sem sudo). Na VPS use sudo sem --direct para testar os bins.
      7 
      8   sudo python3 scripts/admin/smoke_test_email_aliases.py --user MEMBRO
      9 """
     10 
     11 from __future__ import annotations
     12 
     13 import argparse
     14 import json
     15 import os
     16 import subprocess
     17 import sys
     18 import tempfile
     19 from collections.abc import Callable
     20 from contextlib import contextmanager
     21 from pathlib import Path
     22 
     23 REPO_ROOT = Path(__file__).resolve().parents[2]
     24 BIN_EMAIL = REPO_ROOT / "tools" / "bin" / "runv-email-alias"
     25 BIN_ADMIN = REPO_ROOT / "tools" / "bin" / "runv-admin-email-alias"
     26 
     27 
     28 def fail(msg: str) -> None:
     29     print(f"FAIL: {msg}", file=sys.stderr)
     30     raise SystemExit(1)
     31 
     32 
     33 def ok(msg: str) -> None:
     34     print(f"OK: {msg}")
     35 
     36 
     37 @contextmanager
     38 def push_env(updates: dict[str, str]):
     39     old: dict[str, str | None] = {}
     40     for key, val in updates.items():
     41         old[key] = os.environ.get(key)
     42         os.environ[key] = val
     43     try:
     44         yield
     45     finally:
     46         for key, prev in old.items():
     47             if prev is None:
     48                 os.environ.pop(key, None)
     49             else:
     50                 os.environ[key] = prev
     51 
     52 
     53 def run_cmd(
     54     cmd: list[str],
     55     *,
     56     env: dict[str, str],
     57     as_root: bool = False,
     58     as_user: str | None = None,
     59     check: bool = True,
     60 ) -> subprocess.CompletedProcess[str]:
     61     full = list(cmd)
     62     if as_user:
     63         full = ["sudo", "-n", "-u", as_user] + full
     64     elif as_root and os.geteuid() != 0:
     65         full = ["sudo", "-n"] + full
     66     # Fundir com os.environ: sudo -u não propaga um env mínimo de 3 chaves.
     67     child_env = {**os.environ, **env}
     68     proc = subprocess.run(
     69         full,
     70         env=child_env,
     71         capture_output=True,
     72         text=True,
     73         cwd=str(REPO_ROOT),
     74         timeout=60,
     75     )
     76     if check and proc.returncode != 0:
     77         fail(
     78             f"{' '.join(full)}\nstdout={proc.stdout!r}\nstderr={proc.stderr!r}"
     79         )
     80     return proc
     81 
     82 
     83 def expect_exit(fn: Callable[[], object]) -> bool:
     84     try:
     85         fn()
     86     except SystemExit:
     87         return True
     88     return False
     89 
     90 
     91 def _prepare_smoke_temp_layout(base: Path) -> None:
     92     """Espelha permissões de produção para subprocess como membro runv-members."""
     93     import grp
     94 
     95     try:
     96         member_gid = grp.getgrnam("runv-members").gr_gid
     97     except KeyError:
     98         fail("grupo runv-members ausente; execute setup_email_aliases.py primeiro")
     99 
    100     queue = base / "queue"
    101     aliases = base / "email-aliases.json"
    102     lock = base / "email-aliases.lock"
    103 
    104     queue.mkdir(parents=True, exist_ok=True)
    105     for sub in ("approved", "rejected", "cancelled"):
    106         (queue / sub).mkdir(parents=True, exist_ok=True)
    107 
    108     aliases.write_text("{}\n", encoding="utf-8")
    109     lock.touch()
    110 
    111     os.chmod(base, 0o755)
    112     for path in (queue, queue / "approved", queue / "rejected", queue / "cancelled"):
    113         os.chown(path, 0, member_gid)
    114         os.chmod(path, 0o2770)
    115     os.chown(aliases, 0, member_gid)
    116     os.chmod(aliases, 0o640)
    117     os.chown(lock, 0, member_gid)
    118     os.chmod(lock, 0o660)
    119 
    120 
    121 def main() -> int:
    122     if sys.platform == "win32":
    123         print(
    124             "Este smoke test requer Linux (pwd/grp/fcntl). "
    125             "Execute na VPS runv ou em WSL.",
    126             file=sys.stderr,
    127         )
    128         return 2
    129 
    130     p = argparse.ArgumentParser(description="Smoke test aliases de email")
    131     p.add_argument(
    132         "--production",
    133         action="store_true",
    134         help="usar paths em /var/lib/runv (cuidado: altera estado real)",
    135     )
    136     p.add_argument(
    137         "--direct",
    138         action="store_true",
    139         help="chamar runv_email_aliases in-process (temp dir; sem testar bins admin root)",
    140     )
    141     p.add_argument(
    142         "--user",
    143         default="",
    144         help="username Unix para pedidos",
    145     )
    146     args = p.parse_args()
    147     if args.production and args.direct:
    148         fail("--production e --direct são incompatíveis")
    149 
    150     username = args.user.strip() or os.environ.get("SUDO_USER", "").strip()
    151     if not username:
    152         username = os.environ.get("USER", "").strip()
    153     if not username:
    154         fail("defina --user ou execute com USER/SUDO_USER definido")
    155     if os.geteuid() == 0 and not args.user and not os.environ.get("SUDO_USER"):
    156         fail("como root directo, use --user MEMBRO (não reservado)")
    157 
    158     sys.path.insert(0, str(REPO_ROOT / "tools" / "lib"))
    159     import runv_email_aliases as ea  # noqa: E402
    160 
    161     try:
    162         ea.validate_alias_username(username)
    163     except SystemExit:
    164         fail(f"username {username!r} inválido ou reservado para alias")
    165 
    166     if args.production:
    167         queue = Path("/var/lib/runv/email-alias-queue")
    168         aliases = Path("/var/lib/runv/email-aliases.json")
    169         lock = Path("/var/lib/runv/email-aliases.lock")
    170         tmp_ctx = None
    171     else:
    172         tmp_ctx = tempfile.TemporaryDirectory(prefix="runv-email-smoke-")
    173         base = Path(tmp_ctx.name)
    174         queue = base / "queue"
    175         aliases = base / "email-aliases.json"
    176         lock = base / "email-aliases.lock"
    177         _prepare_smoke_temp_layout(base)
    178 
    179     env = {
    180         "RUNV_EMAIL_ALIAS_QUEUE_DIR": str(queue),
    181         "RUNV_EMAIL_ALIASES_PATH": str(aliases),
    182         "RUNV_EMAIL_ALIASES_LOCK_PATH": str(lock),
    183     }
    184 
    185     use_direct = bool(args.direct) or (tmp_ctx is not None and os.geteuid() != 0)
    186     if use_direct and not args.direct:
    187         print(
    188             "aviso: sem root; a usar modo direct (biblioteca). "
    189             "Na VPS: sudo python3 scripts/admin/smoke_test_email_aliases.py --user MEMBRO",
    190             file=sys.stderr,
    191         )
    192 
    193     with push_env(env):
    194         if use_direct:
    195             _run_direct(username, queue, aliases)
    196         else:
    197             _run_subprocess(env, username, queue, aliases)
    198 
    199     if tmp_ctx is not None:
    200         tmp_ctx.cleanup()
    201 
    202     print("\nSmoke test aliases de email: PASS")
    203     return 0
    204 
    205 
    206 def _run_direct(username: str, queue: Path, aliases: Path) -> None:
    207     import runv_email_aliases as ea
    208 
    209     for dest in ("foo", "x@runv.club"):
    210         if not expect_exit(lambda d=dest: ea.validate_destination_email(d)):
    211             fail(f"validate {dest!r} deveria falhar")
    212     ok("validações de destino inválido rejeitadas")
    213 
    214     dest_ok = "smoke-alias-test@example.org"
    215     ea.create_pending_request(username, dest_ok)
    216     ok("request criado")
    217 
    218     if ea.find_pending_for_user(username) is None:
    219         fail("pending não encontrado")
    220     if not expect_exit(lambda: ea.create_pending_request(username, dest_ok)):
    221         fail("segundo request deveria falhar")
    222     ok("duplo pending bloqueado")
    223 
    224     entry = ea.approve_pending(username, "smoke-test")
    225     ok("approve")
    226 
    227     rows = ea.list_active_aliases()
    228     if not any(r[2] == dest_ok for r in rows):
    229         fail(f"list_active sem {dest_ok!r}")
    230     ok("list")
    231 
    232     if ea.get_active_alias(username) is None:
    233         fail("alias activo ausente após approve")
    234     ok("status active")
    235 
    236     data = json.loads(aliases.read_text(encoding="utf-8"))
    237     created_at = data[username].get("created_at")
    238     if not any((queue / "approved").glob("*.json")):
    239         fail("nenhum pedido em approved/")
    240     ok("pedido arquivado em approved/")
    241 
    242     dest2 = "smoke-alias-test2@example.org"
    243     ea.create_pending_request(username, dest2)
    244     ea.approve_pending(username, "smoke-test")
    245     data2 = json.loads(aliases.read_text(encoding="utf-8"))
    246     if data2[username].get("destination") != dest2:
    247         fail("destino não actualizado")
    248     if data2[username].get("created_at") != created_at:
    249         fail("created_at não preservado")
    250     ok("alteração de destino com created_at preservado")
    251 
    252     dest3 = "smoke-cancel@example.org"
    253     ea.create_pending_request(username, dest3)
    254     if ea.cancel_latest_pending(username) is None:
    255         fail("cancel falhou")
    256     ok("cancel")
    257 
    258     dest4 = "smoke-reject@example.org"
    259     ea.create_pending_request(username, dest4)
    260     ea.reject_pending(username, "smoke-test", "smoke test")
    261     if not any((queue / "rejected").glob("*.json")):
    262         fail("reject não arquivou")
    263     ok("reject")
    264 
    265     if not expect_exit(lambda: ea.approve_pending("entre", "smoke-test")):
    266         fail("approve entre deveria falhar")
    267     ok("username reservado rejeitado no approve")
    268 
    269 
    270 def _run_subprocess(
    271     env: dict[str, str],
    272     username: str,
    273     queue: Path,
    274     aliases: Path,
    275 ) -> None:
    276     py = sys.executable
    277     email_bin = str(BIN_EMAIL) if BIN_EMAIL.is_file() else "runv-email-alias"
    278     admin_bin = str(BIN_ADMIN) if BIN_ADMIN.is_file() else "runv-admin-email-alias"
    279     member_user = username if os.geteuid() == 0 else None
    280 
    281     for dest in ("foo", "x@runv.club"):
    282         proc = run_cmd(
    283             [py, email_bin, "request", dest],
    284             env=env,
    285             as_user=member_user,
    286             check=False,
    287         )
    288         if proc.returncode == 0:
    289             fail(f"request {dest!r} deveria falhar")
    290     ok("validações de destino inválido rejeitadas")
    291 
    292     dest_ok = "smoke-alias-test@example.org"
    293     run_cmd([py, email_bin, "request", dest_ok], env=env, as_user=member_user)
    294     ok("request criado")
    295 
    296     proc = run_cmd([py, email_bin, "status"], env=env, as_user=member_user)
    297     if "pending" not in proc.stdout.lower():
    298         fail(f"status sem pending: {proc.stdout!r}")
    299 
    300     proc2 = run_cmd(
    301         [py, email_bin, "request", dest_ok],
    302         env=env,
    303         as_user=member_user,
    304         check=False,
    305     )
    306     if proc2.returncode == 0:
    307         fail("segundo request deveria falhar")
    308     ok("duplo pending bloqueado")
    309 
    310     run_cmd([py, admin_bin, "approve", username], env=env, as_root=True)
    311     ok("approve")
    312 
    313     proc = run_cmd([py, admin_bin, "list"], env=env, as_root=True)
    314     if dest_ok not in proc.stdout:
    315         fail(f"list não mostra destino: {proc.stdout!r}")
    316     ok("list")
    317 
    318     proc = run_cmd([py, email_bin, "status"], env=env, as_user=member_user)
    319     if "active" not in proc.stdout.lower():
    320         fail(f"status sem active: {proc.stdout!r}")
    321     ok("status active")
    322 
    323     data = json.loads(aliases.read_text(encoding="utf-8"))
    324     entry = data.get(username)
    325     if not entry or entry.get("status") != "active":
    326         fail(f"email-aliases.json inválido: {data!r}")
    327     if not any((queue / "approved").glob("*.json")):
    328         fail("nenhum pedido em approved/")
    329     ok("pedido arquivado em approved/")
    330 
    331     dest2 = "smoke-alias-test2@example.org"
    332     run_cmd([py, email_bin, "request", dest2], env=env, as_user=member_user)
    333     run_cmd([py, admin_bin, "approve", username], env=env, as_root=True)
    334     data2 = json.loads(aliases.read_text(encoding="utf-8"))
    335     if data2[username].get("destination") != dest2:
    336         fail("destino não actualizado")
    337     if data2[username].get("created_at") != entry.get("created_at"):
    338         fail("created_at não preservado")
    339     ok("alteração de destino")
    340 
    341     dest3 = "smoke-cancel@example.org"
    342     run_cmd([py, email_bin, "request", dest3], env=env, as_user=member_user)
    343     run_cmd([py, email_bin, "cancel"], env=env, as_user=member_user)
    344     if not any((queue / "cancelled").glob("*.json")):
    345         fail("cancel não arquivou")
    346     ok("cancel")
    347 
    348     dest4 = "smoke-reject@example.org"
    349     run_cmd([py, email_bin, "request", dest4], env=env, as_user=member_user)
    350     run_cmd(
    351         [py, admin_bin, "reject", username, "--reason", "smoke test"],
    352         env=env,
    353         as_root=True,
    354     )
    355     if not any((queue / "rejected").glob("*.json")):
    356         fail("reject não arquivou")
    357     ok("reject")
    358 
    359     proc = run_cmd(
    360         [py, admin_bin, "approve", "entre"],
    361         env=env,
    362         as_root=True,
    363         check=False,
    364     )
    365     if proc.returncode == 0:
    366         fail("approve entre deveria falhar")
    367     ok("username reservado rejeitado")
    368 
    369 
    370 if __name__ == "__main__":
    371     raise SystemExit(main())