Xiaoxia AI
130 lines
4.2 KiB
Python
130 lines
4.2 KiB
Python
"""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())
|