chore(release): add external service smoke checks

scripts/smoke_external_services.py

@@ -0,0 +1,129 @@
+"""Smoke test external production services without printing secrets."""
+
+from __future__ import annotations
+
+import argparse
+import os
+import smtplib
+import tempfile
+import time
+from pathlib import Path
+
+import redis
+
+try:
+    import oss2
+except ImportError:  # pragma: no cover
+    oss2 = None
+
+
+def is_enabled(value: str | None) -> bool:
+    return str(value or "").lower() in {"1", "true", "yes", "on"}
+
+
+def require_env(name: str) -> str:
+    value = os.getenv(name)
+    if not value:
+        raise RuntimeError(f"missing required env: {name}")
+    return value
+
+
+def smoke_redis(strict: bool) -> None:
+    enabled = is_enabled(os.getenv("ENABLE_REDIS_SESSIONS"))
+    if not strict and not enabled:
+        print("SKIP redis sessions: ENABLE_REDIS_SESSIONS is disabled")
+        return
+
+    client = redis.Redis.from_url(require_env("REDIS_URL"), decode_responses=True, socket_timeout=5)
+    if client.ping() is not True:
+        raise RuntimeError("redis ping failed")
+    key = f"xiaoxia:smoke:{int(time.time())}"
+    client.setex(key, 30, "ok")
+    if client.get(key) != "ok":
+        raise RuntimeError("redis set/get smoke failed")
+    client.delete(key)
+    print("OK redis sessions")
+
+
+def smoke_smtp(strict: bool, send_email_to: str | None) -> None:
+    enabled = is_enabled(os.getenv("ENABLE_EMAIL_DELIVERY"))
+    if not strict and not enabled:
+        print("SKIP smtp: ENABLE_EMAIL_DELIVERY is disabled")
+        return
+
+    host = require_env("SMTP_HOST")
+    port = int(require_env("SMTP_PORT"))
+    user = os.getenv("SMTP_USER", "")
+    password = os.getenv("SMTP_PASSWORD", "")
+    from_email = require_env("SMTP_FROM_EMAIL")
+    use_tls = is_enabled(os.getenv("SMTP_USE_TLS", "true"))
+
+    with smtplib.SMTP(host, port, timeout=10) as server:
+        if use_tls:
+            server.starttls()
+        if user or password:
+            server.login(user, password)
+        if send_email_to:
+            server.sendmail(
+                from_email,
+                [send_email_to],
+                "Subject: Xiaoxia SaaS external service smoke\n\nOK: SMTP smoke test.",
+            )
+    action = "send" if send_email_to else "connect/login"
+    print(f"OK smtp {action}")
+
+
+def smoke_oss(strict: bool) -> None:
+    endpoint = os.getenv("OSS_ENDPOINT", "")
+    access_key_id = os.getenv("OSS_ACCESS_KEY_ID", "")
+    access_key_secret = os.getenv("OSS_ACCESS_KEY_SECRET", "")
+    bucket_name = os.getenv("OSS_BUCKET_NAME", "")
+    if not strict and not (endpoint and access_key_id and access_key_secret and bucket_name):
+        print("SKIP oss: OSS credentials are not fully configured")
+        return
+    for name, value in {
+        "OSS_ENDPOINT": endpoint,
+        "OSS_ACCESS_KEY_ID": access_key_id,
+        "OSS_ACCESS_KEY_SECRET": access_key_secret,
+        "OSS_BUCKET_NAME": bucket_name,
+    }.items():
+        if not value:
+            raise RuntimeError(f"missing required env: {name}")
+
+    if oss2 is None:
+        raise RuntimeError("oss2 is required for OSS smoke")
+
+    auth = oss2.Auth(access_key_id, access_key_secret)
+    bucket = oss2.Bucket(auth, endpoint, bucket_name)
+    key = f"smoke/external-services-{int(time.time())}.txt"
+    with tempfile.NamedTemporaryFile("w", delete=False, encoding="utf-8") as handle:
+        handle.write("xiaoxia oss smoke ok\n")
+        local_path = Path(handle.name)
+    try:
+        bucket.put_object_from_file(key, str(local_path), headers={"Content-Type": "text/plain"})
+        if not bucket.object_exists(key):
+            raise RuntimeError("oss uploaded object does not exist")
+        bucket.get_object(key).read()
+        bucket.delete_object(key)
+    finally:
+        local_path.unlink(missing_ok=True)
+    print("OK oss upload/download/delete")
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--strict", action="store_true", help="Fail when disabled/unconfigured services are missing.")
+    parser.add_argument(
+        "--send-email-to", help="Actually send a test email to this address. Otherwise SMTP only connects/logs in."
+    )
+    args = parser.parse_args()
+
+    smoke_redis(args.strict)
+    smoke_smtp(args.strict, args.send_email_to)
+    smoke_oss(args.strict)
+    print("OK external service smoke complete")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())