File size: 3,607 Bytes
89419ea | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | from contextlib import contextmanager
from pathlib import Path
from src.database.models import Base, EmailService
from src.database.session import DatabaseSessionManager
from src.services.temp_mail import TempMailService
from src.web.routes import registration as registration_routes
class DummySettings:
custom_domain_base_url = ""
custom_domain_api_key = None
@contextmanager
def _session_context(manager):
session = manager.SessionLocal()
try:
yield session
finally:
session.close()
def _run_coroutine_without_awaits(coro):
try:
coro.send(None)
except StopIteration as exc:
return exc.value
raise AssertionError("coroutine unexpectedly awaited")
def test_registration_available_services_exposes_temp_mail_domain_pool(monkeypatch):
runtime_dir = Path("tests_runtime")
runtime_dir.mkdir(exist_ok=True)
db_path = runtime_dir / "registration_available_temp_mail.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="temp_mail",
name="Worker temp mail",
config={"domains": ["alpha.example.com", "beta.example.com"]},
enabled=True,
priority=0,
)
)
monkeypatch.setattr(registration_routes, "get_db", lambda: _session_context(manager))
monkeypatch.setattr(registration_routes, "get_settings", lambda: DummySettings())
payload = _run_coroutine_without_awaits(registration_routes.get_available_email_services())
service = payload["temp_mail"]["services"][0]
assert payload["temp_mail"]["available"] is True
assert payload["temp_mail"]["count"] == 1
assert service["name"] == "Worker temp mail"
assert service["domain_count"] == 2
assert service["domains_preview"] == ["alpha.example.com", "beta.example.com"]
assert service["domain_source"] == "config_domains"
def test_registration_available_services_uses_worker_domain_pool_when_config_pool_missing(monkeypatch):
runtime_dir = Path("tests_runtime")
runtime_dir.mkdir(exist_ok=True)
db_path = runtime_dir / "registration_available_temp_mail_worker.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="temp_mail",
name="Worker-backed temp mail",
config={
"base_url": "https://apmail.889110.xyz",
"admin_password": "admin888",
"domain": "fallback.example.com",
},
enabled=True,
priority=0,
)
)
monkeypatch.setattr(registration_routes, "get_db", lambda: _session_context(manager))
monkeypatch.setattr(registration_routes, "get_settings", lambda: DummySettings())
monkeypatch.setattr(TempMailService, "_fetch_domains_from_worker", lambda self: {"domains": ["worker-a.example.com", "worker-b.example.com"]})
payload = _run_coroutine_without_awaits(registration_routes.get_available_email_services())
service = payload["temp_mail"]["services"][0]
assert service["domain_source"] == "worker_api"
assert service["domains_preview"] == ["worker-a.example.com", "worker-b.example.com"]
|