From ee6a98398fc77559e20ceffdfdad039cce23d1b2 Mon Sep 17 00:00:00 2001 From: Xiaoxia AI Date: Sun, 21 Jun 2026 12:50:33 +0800 Subject: [PATCH] chore(release): add external service smoke checks --- docs/PRODUCTION-RELEASE-CHECKLIST.md | 2 + scripts/smoke_external_services.py | 129 +++++++++++++++++++++ tests/unit/test_smoke_external_services.py | 96 +++++++++++++++ 3 files changed, 227 insertions(+) create mode 100644 scripts/smoke_external_services.py create mode 100644 tests/unit/test_smoke_external_services.py diff --git a/docs/PRODUCTION-RELEASE-CHECKLIST.md b/docs/PRODUCTION-RELEASE-CHECKLIST.md index 4e0ee3596..c1b17b69b 100644 --- a/docs/PRODUCTION-RELEASE-CHECKLIST.md +++ b/docs/PRODUCTION-RELEASE-CHECKLIST.md @@ -23,6 +23,8 @@ - `GENERATED_FILES_HOST_DIR` 生产环境不得指向 staging 目录。 - 本地校验文件：`python scripts/validate_release_env.py /var/lib/xiaoxia-saas-production/.env --strict-external`。 - 容器内校验已注入环境：`docker exec xiaoxia-api-production python /app/scripts/validate_release_env.py --from-environ --strict-external`。 +- 外部服务 smoke：`docker exec xiaoxia-api-production python /app/scripts/smoke_external_services.py --strict`。 +- SMTP 真发信 smoke：`docker exec xiaoxia-api-production python /app/scripts/smoke_external_services.py --strict --send-email-to <测试邮箱>`。 - 如开启邮件/session： - `ENABLE_EMAIL_DELIVERY=true` 前先验证 SMTP 凭证。 - `ENABLE_REDIS_SESSIONS=true` 前先验证 Redis 连通性。 diff --git a/scripts/smoke_external_services.py b/scripts/smoke_external_services.py new file mode 100644 index 000000000..c8787828d --- /dev/null +++ b/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()) diff --git a/tests/unit/test_smoke_external_services.py b/tests/unit/test_smoke_external_services.py new file mode 100644 index 000000000..92400bac6 --- /dev/null +++ b/tests/unit/test_smoke_external_services.py @@ -0,0 +1,96 @@ +import types + +import scripts.smoke_external_services as smoke + + +def test_external_smoke_skips_disabled_services(monkeypatch, capsys): + for key in [ + "ENABLE_REDIS_SESSIONS", + "ENABLE_EMAIL_DELIVERY", + "OSS_ENDPOINT", + "OSS_ACCESS_KEY_ID", + "OSS_ACCESS_KEY_SECRET", + "OSS_BUCKET_NAME", + ]: + monkeypatch.delenv(key, raising=False) + + smoke.smoke_redis(strict=False) + smoke.smoke_smtp(strict=False, send_email_to=None) + smoke.smoke_oss(strict=False) + + output = capsys.readouterr().out + assert "SKIP redis sessions" in output + assert "SKIP smtp" in output + assert "SKIP oss" in output + + +def test_external_smoke_requires_oss_in_strict_mode(monkeypatch): + monkeypatch.delenv("OSS_ENDPOINT", raising=False) + + try: + smoke.smoke_oss(strict=True) + except RuntimeError as error: + assert "missing required env: OSS_ENDPOINT" in str(error) + else: + raise AssertionError("strict OSS smoke should fail without credentials") + + +def test_external_smoke_redis_round_trip(monkeypatch, capsys): + class FakeRedis: + def __init__(self): + self.values = {} + + def ping(self): + return True + + def setex(self, key, seconds, value): + self.values[key] = value + + def get(self, key): + return self.values[key] + + def delete(self, key): + self.values.pop(key, None) + + fake_client = FakeRedis() + monkeypatch.setenv("ENABLE_REDIS_SESSIONS", "true") + monkeypatch.setenv("REDIS_URL", "redis://redis:6379/0") + monkeypatch.setattr(smoke.redis.Redis, "from_url", lambda *args, **kwargs: fake_client) + + smoke.smoke_redis(strict=False) + + assert "OK redis sessions" in capsys.readouterr().out + + +def test_external_smoke_smtp_connect_login(monkeypatch, capsys): + actions = [] + + class FakeSMTP: + def __init__(self, host, port, timeout): + actions.append((host, port, timeout)) + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def starttls(self): + actions.append("tls") + + def login(self, user, password): + actions.append(("login", user, password)) + + monkeypatch.setenv("ENABLE_EMAIL_DELIVERY", "true") + monkeypatch.setenv("SMTP_HOST", "smtp.local") + monkeypatch.setenv("SMTP_PORT", "587") + monkeypatch.setenv("SMTP_USER", "mailer") + monkeypatch.setenv("SMTP_PASSWORD", "secret") + monkeypatch.setenv("SMTP_FROM_EMAIL", "noreply@example.test") + monkeypatch.setenv("SMTP_USE_TLS", "true") + monkeypatch.setattr(smoke.smtplib, "SMTP", FakeSMTP) + + smoke.smoke_smtp(strict=False, send_email_to=None) + + assert actions == [("smtp.local", 587, 10), "tls", ("login", "mailer", "secret")] + assert "OK smtp connect/login" in capsys.readouterr().out