File size: 2,560 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 | from src.services.temp_mail_domain_provider import resolve_temp_mail_domains, summarize_temp_mail_domains
from src.core.email_service_catalog import build_temp_mail_service_entry
def test_resolve_prefers_config_domains_and_skips_worker_fetch():
called = {"value": False}
def _fetcher():
called["value"] = True
return {"domains": ["ignored.example.com"]}
domains = resolve_temp_mail_domains(
{
"domains": ["a.example.com", "a.example.com", " b.example.com ", ""],
"domain": "fallback.example.com",
},
fetch_domains=_fetcher,
)
assert domains == ["a.example.com", "b.example.com"]
assert called["value"] is False
def test_resolve_uses_worker_domains_when_config_pool_missing():
domains = resolve_temp_mail_domains(
{"domain": "fallback.example.com"},
fetch_domains=lambda: {"domains": ["w1.example.com", "w2.example.com", "w1.example.com"]},
)
assert domains == ["w1.example.com", "w2.example.com"]
def test_resolve_falls_back_to_single_domain_when_worker_unavailable():
def _boom():
raise RuntimeError("worker down")
domains = resolve_temp_mail_domains(
{"domain": "fallback.example.com"},
fetch_domains=_boom,
)
assert domains == ["fallback.example.com"]
def test_summarize_returns_preview_and_source():
summary = summarize_temp_mail_domains(
{"domain": "fallback.example.com"},
fetch_domains=lambda: {"domains": ["d1.example.com", "d2.example.com", "d3.example.com", "d4.example.com"]},
preview_limit=3,
)
assert summary["domain_source"] == "worker_api"
assert summary["domain_count"] == 4
assert summary["domains_preview"] == ["d1.example.com", "d2.example.com", "d3.example.com"]
assert summary["domain"] == "d1.example.com"
def test_catalog_entry_exposes_temp_mail_domain_summary_without_secrets():
class Service:
id = 12
name = "mail-worker"
priority = 2
config = {
"domain": "fallback.example.com",
"admin_password": "admin888",
}
entry = build_temp_mail_service_entry(
Service(),
fetch_domains=lambda: {"domains": ["d1.example.com", "d2.example.com"]},
preview_limit=1,
)
assert entry["id"] == 12
assert entry["type"] == "temp_mail"
assert entry["domain_count"] == 2
assert entry["domains_preview"] == ["d1.example.com"]
assert entry["domain_source"] == "worker_api"
assert "admin_password" not in entry
|