File size: 5,305 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 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 | from contextlib import contextmanager
from pathlib import Path
import src.config.settings as settings_module
from src.core import email_service_catalog
from src.database.models import Base, CpaService, EmailService, Sub2ApiService, TeamManagerService
from src.database.session import DatabaseSessionManager
class DummySettings:
external_api_enabled = True
@contextmanager
def _session_context(manager):
session = manager.SessionLocal()
try:
yield session
finally:
session.close()
def _make_db(name: str):
runtime_dir = Path("tests_runtime")
runtime_dir.mkdir(exist_ok=True)
db_path = runtime_dir / name
if db_path.exists():
db_path.unlink()
manager = DatabaseSessionManager(f"sqlite:///{db_path}")
Base.metadata.create_all(bind=manager.engine)
return manager
def test_build_external_capabilities_does_not_expose_legacy_tempmail_placeholder(monkeypatch):
manager = _make_db("email_service_catalog_no_legacy_tempmail.db")
monkeypatch.setattr(email_service_catalog, "get_db", lambda: _session_context(manager))
monkeypatch.setattr(email_service_catalog, "get_settings", lambda: DummySettings())
payload = email_service_catalog.build_external_capabilities()
assert payload["email_types"] == []
def test_build_external_capabilities_survives_temp_mail_configs_without_worker_credentials(monkeypatch):
manager = _make_db("email_service_catalog_capabilities.db")
with manager.session_scope() as session:
session.add_all([
EmailService(
service_type="temp_mail",
name="Temp pool",
config={"domains": ["alpha.example.com", "beta.example.com"]},
enabled=True,
priority=0,
),
CpaService(name="CPA", api_url="https://cpa.test", api_token="token", enabled=True, priority=0),
Sub2ApiService(name="Sub2", api_url="https://sub2.test", api_key="key", enabled=True, priority=0),
TeamManagerService(name="TM", api_url="https://tm.test", api_key="key", enabled=True, priority=0),
])
monkeypatch.setattr(email_service_catalog, "get_db", lambda: _session_context(manager))
monkeypatch.setattr(email_service_catalog, "get_settings", lambda: DummySettings())
payload = email_service_catalog.build_external_capabilities()
assert all(item["type"] != "tempmail" for item in payload["email_types"])
temp_mail = next(item for item in payload["email_types"] if item["type"] == "temp_mail")
assert temp_mail["available"] is True
assert temp_mail["count"] == 1
assert temp_mail["services"][0]["domain_count"] == 2
assert temp_mail["services"][0]["domains_preview"] == ["alpha.example.com", "beta.example.com"]
assert temp_mail["services"][0]["domain_source"] == "config_domains"
settings = payload["settings"]
assert settings["external_api_enabled"] is True
assert {provider["provider"] for provider in payload["upload_providers"]} == {"cpa", "sub2api", "tm"}
def test_build_external_capabilities_uses_worker_domain_summary_when_credentials_exist(monkeypatch):
manager = _make_db("email_service_catalog_worker_domains.db")
with manager.session_scope() as session:
session.add(
EmailService(
service_type="temp_mail",
name="Temp worker",
config={
"base_url": "https://apmail.889110.xyz",
"admin_password": "admin888",
"domain": "fallback.example.com",
},
enabled=True,
priority=0,
)
)
monkeypatch.setattr(email_service_catalog, "get_db", lambda: _session_context(manager))
monkeypatch.setattr(email_service_catalog, "get_settings", lambda: DummySettings())
monkeypatch.setattr(email_service_catalog.TempMailService, "_fetch_domains_from_worker", lambda self: {"domains": ["w1.example.com", "w2.example.com"]})
payload = email_service_catalog.build_external_capabilities()
temp_mail = next(item for item in payload["email_types"] if item["type"] == "temp_mail")
assert temp_mail["services"][0]["domain_source"] == "worker_api"
assert temp_mail["services"][0]["domains_preview"] == ["w1.example.com", "w2.example.com"]
def test_build_external_capabilities_temp_mail_gracefully_handles_empty_summary(monkeypatch):
manager = _make_db("email_service_catalog_empty_domains.db")
with manager.session_scope() as session:
session.add(
EmailService(
service_type="temp_mail",
name="Temp empty",
config={},
enabled=True,
priority=0,
)
)
monkeypatch.setattr(email_service_catalog, "get_db", lambda: _session_context(manager))
monkeypatch.setattr(email_service_catalog, "get_settings", lambda: DummySettings())
payload = email_service_catalog.build_external_capabilities()
temp_mail = next(item for item in payload["email_types"] if item["type"] == "temp_mail")
assert temp_mail["services"][0]["domain_source"] == "none"
assert temp_mail["services"][0]["domain_count"] == 0
assert temp_mail["services"][0]["domains_preview"] == []
|