diff --git a/templates/index.html b/templates/index.html
index 9d23815..acf2d20 100644
--- a/templates/index.html
+++ b/templates/index.html
@@ -181,6 +181,7 @@
📝 注册设置
+
diff --git a/tests/test_cloud_mail_service.py b/tests/test_cloud_mail_service.py
new file mode 100644
index 0000000..f6c8ee7
--- /dev/null
+++ b/tests/test_cloud_mail_service.py
@@ -0,0 +1,177 @@
+from datetime import datetime, timezone
+
+from src.services.cloud_mail import CloudMailService
+
+
+class FakeResponse:
+ def __init__(self, status_code=200, payload=None, text=""):
+ self.status_code = status_code
+ self._payload = payload
+ self.text = text
+ self.headers = {}
+
+ def json(self):
+ if self._payload is None:
+ raise ValueError("no json payload")
+ return self._payload
+
+
+class FakeHTTPClient:
+ def __init__(self, responses):
+ self.responses = list(responses)
+ self.calls = []
+
+ def request(self, method, url, **kwargs):
+ self.calls.append({
+ "method": method,
+ "url": url,
+ "kwargs": kwargs,
+ })
+ if not self.responses:
+ raise AssertionError(f"未准备响应: {method} {url}")
+ return self.responses.pop(0)
+
+
+def _to_timestamp(value: str) -> float:
+ normalized = value.replace(" ", "T")
+ return datetime.fromisoformat(normalized.replace("Z", "+00:00")).astimezone(timezone.utc).timestamp()
+
+
+def test_cloud_mail_creates_address_via_public_api():
+ service = CloudMailService({
+ "base_url": "https://mail.example.com",
+ "admin_email": "admin@example.com",
+ "admin_password": "admin-secret",
+ "default_domain": "mail.example.com",
+ })
+ service.http_client = FakeHTTPClient([
+ FakeResponse(
+ payload={
+ "code": 200,
+ "message": "success",
+ "data": {"token": "public-token"},
+ }
+ ),
+ FakeResponse(
+ payload={
+ "code": 200,
+ "message": "success",
+ "data": None,
+ }
+ ),
+ ])
+
+ result = service.create_email()
+
+ assert result["email"].endswith("@mail.example.com")
+ assert result["service_id"] == result["email"]
+ assert result["id"] == result["email"]
+ assert result["password"]
+
+ first_call = service.http_client.calls[0]
+ second_call = service.http_client.calls[1]
+
+ assert first_call["method"] == "POST"
+ assert first_call["url"] == "https://mail.example.com/api/public/genToken"
+ assert first_call["kwargs"]["json"] == {
+ "email": "admin@example.com",
+ "password": "admin-secret",
+ }
+
+ assert second_call["method"] == "POST"
+ assert second_call["url"] == "https://mail.example.com/api/public/addUser"
+ assert second_call["kwargs"]["headers"]["Authorization"] == "public-token"
+ assert second_call["kwargs"]["json"]["list"][0]["email"] == result["email"]
+
+
+def test_cloud_mail_extracts_openai_verification_code_from_public_email_list():
+ service = CloudMailService({
+ "base_url": "https://mail.example.com",
+ "admin_email": "admin@example.com",
+ "admin_password": "admin-secret",
+ "default_domain": "mail.example.com",
+ })
+ service.http_client = FakeHTTPClient([
+ FakeResponse(
+ payload={
+ "code": 200,
+ "message": "success",
+ "data": {"token": "public-token"},
+ }
+ ),
+ FakeResponse(
+ payload={
+ "code": 200,
+ "message": "success",
+ "data": [
+ {
+ "emailId": 1,
+ "sendEmail": "noreply@openai.com",
+ "sendName": "OpenAI",
+ "subject": "Your OpenAI verification code",
+ "text": "Your OpenAI verification code is 654321",
+ "content": "",
+ }
+ ],
+ }
+ ),
+ ])
+
+ code = service.get_verification_code(
+ email="tester@mail.example.com",
+ timeout=1,
+ )
+
+ assert code == "654321"
+
+
+def test_cloud_mail_ignores_messages_received_before_otp_sent_at():
+ service = CloudMailService({
+ "base_url": "https://mail.example.com",
+ "admin_email": "admin@example.com",
+ "admin_password": "admin-secret",
+ "default_domain": "mail.example.com",
+ })
+ service.http_client = FakeHTTPClient([
+ FakeResponse(
+ payload={
+ "code": 200,
+ "message": "success",
+ "data": {"token": "public-token"},
+ }
+ ),
+ FakeResponse(
+ payload={
+ "code": 200,
+ "message": "success",
+ "data": [
+ {
+ "emailId": 1,
+ "sendEmail": "noreply@openai.com",
+ "sendName": "OpenAI",
+ "subject": "Old code",
+ "text": "111111",
+ "content": "",
+ "createTime": "2026-03-23 10:00:00",
+ },
+ {
+ "emailId": 2,
+ "sendEmail": "noreply@openai.com",
+ "sendName": "OpenAI",
+ "subject": "New code",
+ "text": "222222",
+ "content": "",
+ "createTime": "2026-03-23 10:00:05",
+ },
+ ],
+ }
+ ),
+ ])
+
+ code = service.get_verification_code(
+ email="tester@mail.example.com",
+ timeout=1,
+ otp_sent_at=_to_timestamp("2026-03-23T10:00:02Z"),
+ )
+
+ assert code == "222222"
diff --git a/tests/test_email_service_cloudmail_routes.py b/tests/test_email_service_cloudmail_routes.py
new file mode 100644
index 0000000..29ca2ba
--- /dev/null
+++ b/tests/test_email_service_cloudmail_routes.py
@@ -0,0 +1,146 @@
+import asyncio
+from contextlib import contextmanager
+from pathlib import Path
+
+from src.config.constants import EmailServiceType
+from src.database.models import Base, EmailService
+from src.database.session import DatabaseSessionManager
+from src.services.base import EmailServiceFactory
+from src.web.routes import email as email_routes
+from src.web.routes import registration as registration_routes
+
+
+class DummySettings:
+ custom_domain_base_url = ""
+ custom_domain_api_key = None
+ tempmail_base_url = "https://api.tempmail.lol/v2"
+ tempmail_timeout = 30
+ tempmail_max_retries = 3
+
+
+def test_cloud_mail_service_registered():
+ service_type = EmailServiceType("cloud_mail")
+ service_class = EmailServiceFactory.get_service_class(service_type)
+ assert service_class is not None
+ assert service_class.__name__ == "CloudMailService"
+
+
+def test_email_service_types_include_cloud_mail():
+ result = asyncio.run(email_routes.get_service_types())
+ cloud_mail_type = next(item for item in result["types"] if item["value"] == "cloud_mail")
+
+ assert cloud_mail_type["label"] == "Cloud Mail"
+ field_names = [field["name"] for field in cloud_mail_type["config_fields"]]
+ assert "base_url" in field_names
+ assert "admin_email" in field_names
+ assert "admin_password" in field_names
+ assert "default_domain" in field_names
+
+
+def test_filter_sensitive_config_marks_cloud_mail_admin_password():
+ filtered = email_routes.filter_sensitive_config({
+ "base_url": "https://mail.example.com",
+ "admin_email": "admin@example.com",
+ "admin_password": "admin-secret",
+ "default_domain": "mail.example.com",
+ })
+
+ assert filtered["base_url"] == "https://mail.example.com"
+ assert filtered["admin_email"] == "admin@example.com"
+ assert filtered["default_domain"] == "mail.example.com"
+ assert filtered["has_admin_password"] is True
+ assert "admin_password" not in filtered
+
+
+def test_registration_available_services_include_cloud_mail(monkeypatch):
+ runtime_dir = Path("tests_runtime")
+ runtime_dir.mkdir(exist_ok=True)
+ db_path = runtime_dir / "cloudmail_routes.db"
+ if db_path.exists():
+ db_path.unlink()
+
+ manager = DatabaseSessionManager(f"sqlite:///{db_path}")
+ Base.metadata.create_all(bind=manager.engine)
+
+ with manager.session_scope() as session:
+ session.add(
+ EmailService(
+ service_type="cloud_mail",
+ name="Cloud Mail 主服务",
+ config={
+ "base_url": "https://mail.example.com",
+ "admin_email": "admin@example.com",
+ "admin_password": "admin-secret",
+ "default_domain": "mail.example.com",
+ },
+ enabled=True,
+ priority=0,
+ )
+ )
+
+ @contextmanager
+ def fake_get_db():
+ session = manager.SessionLocal()
+ try:
+ yield session
+ finally:
+ session.close()
+
+ monkeypatch.setattr(registration_routes, "get_db", fake_get_db)
+
+ import src.config.settings as settings_module
+
+ monkeypatch.setattr(settings_module, "get_settings", lambda: DummySettings())
+ monkeypatch.setattr(registration_routes, "get_settings", lambda: DummySettings())
+
+ result = asyncio.run(registration_routes.get_available_email_services())
+
+ assert result["cloud_mail"]["available"] is True
+ assert result["cloud_mail"]["count"] == 1
+ assert result["cloud_mail"]["services"][0]["name"] == "Cloud Mail 主服务"
+ assert result["cloud_mail"]["services"][0]["type"] == "cloud_mail"
+ assert result["cloud_mail"]["services"][0]["default_domain"] == "mail.example.com"
+
+
+def test_build_email_service_candidates_supports_cloud_mail(monkeypatch):
+ runtime_dir = Path("tests_runtime")
+ runtime_dir.mkdir(exist_ok=True)
+ db_path = runtime_dir / "cloudmail_candidates.db"
+ if db_path.exists():
+ db_path.unlink()
+
+ manager = DatabaseSessionManager(f"sqlite:///{db_path}")
+ Base.metadata.create_all(bind=manager.engine)
+
+ with manager.session_scope() as session:
+ session.add(
+ EmailService(
+ service_type="cloud_mail",
+ name="Cloud Mail 主服务",
+ config={
+ "base_url": "https://mail.example.com",
+ "admin_email": "admin@example.com",
+ "admin_password": "admin-secret",
+ "default_domain": "mail.example.com",
+ },
+ enabled=True,
+ priority=0,
+ )
+ )
+
+ registration_routes.email_service_circuit_breakers.clear()
+ monkeypatch.setattr(registration_routes, "get_settings", lambda: DummySettings())
+
+ with manager.session_scope() as session:
+ candidates = registration_routes._build_email_service_candidates(
+ db=session,
+ service_type=EmailServiceType("cloud_mail"),
+ actual_proxy_url=None,
+ email_service_id=None,
+ email_service_config=None,
+ )
+
+ assert len(candidates) == 1
+ assert candidates[0]["service_type"] == EmailServiceType("cloud_mail")
+ assert candidates[0]["config"]["base_url"] == "https://mail.example.com"
+ assert candidates[0]["db_service"].name == "Cloud Mail 主服务"