"""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(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--strict", action="store_true", help="Fail when disabled/unconfigured services are missing.")
    parser.add_argument(
        "--skip-smtp", action="store_true", help="Skip SMTP smoke when email delivery is intentionally disabled."
    )
    parser.add_argument(
        "--skip-redis", action="store_true", help="Skip Redis session smoke when sessions are intentionally disabled."
    )
    parser.add_argument(
        "--skip-oss", action="store_true", help="Skip OSS smoke when object storage is intentionally disabled."
    )
    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(argv)

    if args.skip_redis:
        print("SKIP redis sessions: explicitly skipped")
    else:
        smoke_redis(args.strict)
    if args.skip_smtp:
        print("SKIP smtp: explicitly skipped")
    else:
        smoke_smtp(args.strict, args.send_email_to)
    if args.skip_oss:
        print("SKIP oss: explicitly skipped")
    else:
        smoke_oss(args.strict)
    print("OK external service smoke complete")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())