154 lines
5.6 KiB
Python
154 lines
5.6 KiB
Python
import sys
|
|
from unittest.mock import MagicMock
|
|
|
|
from gateway.config import PlatformConfig
|
|
from gateway.platforms.api_server import APIServerAdapter
|
|
from tests.gateway._plugin_adapter_loader import load_plugin_adapter
|
|
|
|
|
|
def _ensure_telegram_mock():
|
|
if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"):
|
|
return
|
|
|
|
telegram_mod = MagicMock()
|
|
telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None)
|
|
telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2"
|
|
telegram_mod.constants.ChatType.GROUP = "group"
|
|
telegram_mod.constants.ChatType.SUPERGROUP = "supergroup"
|
|
telegram_mod.constants.ChatType.CHANNEL = "channel"
|
|
telegram_mod.constants.ChatType.PRIVATE = "private"
|
|
|
|
for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"):
|
|
sys.modules.setdefault(name, telegram_mod)
|
|
|
|
|
|
_ensure_telegram_mock()
|
|
|
|
from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402
|
|
|
|
_line = load_plugin_adapter("line")
|
|
LineAdapter = _line.LineAdapter
|
|
|
|
|
|
class _FakeStore:
|
|
def __init__(self, *, allow: bool, live_email: str | None = "dk96@bremen.com.tw"):
|
|
self.allow = allow
|
|
self.live_email = live_email
|
|
self.calls = []
|
|
|
|
def bound_email_matches_source(self, source, email, *, explicit_user_requested=False):
|
|
self.calls.append((source, email, explicit_user_requested))
|
|
return self.allow or explicit_user_requested
|
|
|
|
def get_verified_email_for_source(self, source):
|
|
return self.live_email
|
|
|
|
|
|
def test_telegram_guard_accepts_live_bound_email(monkeypatch):
|
|
store = _FakeStore(allow=True)
|
|
monkeypatch.setattr("plugins.platforms.telegram.adapter.GatewayUserStore", lambda: store)
|
|
adapter = TelegramAdapter(PlatformConfig(enabled=True, token="***", extra={}))
|
|
|
|
result = adapter._verified_email_target_if_bound("601504103", "dk96@bremen.com.tw")
|
|
|
|
assert result == "dk96@bremen.com.tw"
|
|
assert store.calls[0][1] == "dk96@bremen.com.tw"
|
|
assert store.calls[0][2] is False
|
|
|
|
|
|
def test_telegram_guard_blocks_mismatched_system_handoff(monkeypatch):
|
|
store = _FakeStore(allow=False, live_email="dk96@bremen.com.tw")
|
|
monkeypatch.setattr("plugins.platforms.telegram.adapter.GatewayUserStore", lambda: store)
|
|
adapter = TelegramAdapter(PlatformConfig(enabled=True, token="***", extra={}))
|
|
|
|
result = adapter._verified_email_target_if_bound("601504103", "wrong@bremen.com.tw")
|
|
|
|
assert result is None
|
|
assert store.calls[0][2] is False
|
|
|
|
|
|
def test_telegram_guard_allows_explicit_other_email(monkeypatch):
|
|
store = _FakeStore(allow=False, live_email="dk96@bremen.com.tw")
|
|
monkeypatch.setattr("plugins.platforms.telegram.adapter.GatewayUserStore", lambda: store)
|
|
adapter = TelegramAdapter(PlatformConfig(enabled=True, token="***", extra={}))
|
|
|
|
result = adapter._verified_email_target_if_bound(
|
|
"601504103",
|
|
"other@bremen.com.tw",
|
|
explicit_user_requested=True,
|
|
)
|
|
|
|
assert result == "other@bremen.com.tw"
|
|
assert store.calls[0][2] is True
|
|
|
|
|
|
def test_api_server_guard_accepts_live_bound_email(monkeypatch):
|
|
store = _FakeStore(allow=True)
|
|
monkeypatch.setattr("gateway.platforms.api_server.GatewayUserStore", lambda: store)
|
|
adapter = APIServerAdapter(PlatformConfig(enabled=True, extra={"key": "test-key"}))
|
|
|
|
result = adapter._verified_email_target_if_bound("openwebui-user", "dk96@bremen.com.tw")
|
|
|
|
assert result == "dk96@bremen.com.tw"
|
|
assert store.calls[0][1] == "dk96@bremen.com.tw"
|
|
assert store.calls[0][2] is False
|
|
|
|
|
|
def test_api_server_guard_blocks_mismatched_system_handoff(monkeypatch):
|
|
store = _FakeStore(allow=False, live_email="dk96@bremen.com.tw")
|
|
monkeypatch.setattr("gateway.platforms.api_server.GatewayUserStore", lambda: store)
|
|
adapter = APIServerAdapter(PlatformConfig(enabled=True, extra={"key": "test-key"}))
|
|
|
|
result = adapter._verified_email_target_if_bound("openwebui-user", "wrong@bremen.com.tw")
|
|
|
|
assert result is None
|
|
assert store.calls[0][2] is False
|
|
|
|
|
|
def test_api_server_guard_allows_explicit_other_email(monkeypatch):
|
|
store = _FakeStore(allow=False, live_email="dk96@bremen.com.tw")
|
|
monkeypatch.setattr("gateway.platforms.api_server.GatewayUserStore", lambda: store)
|
|
adapter = APIServerAdapter(PlatformConfig(enabled=True, extra={"key": "test-key"}))
|
|
|
|
result = adapter._verified_email_target_if_bound(
|
|
"openwebui-user",
|
|
"other@bremen.com.tw",
|
|
explicit_user_requested=True,
|
|
)
|
|
|
|
assert result == "other@bremen.com.tw"
|
|
assert store.calls[0][2] is True
|
|
|
|
|
|
def test_line_guard_accepts_live_bound_email(monkeypatch):
|
|
store = _FakeStore(allow=True)
|
|
monkeypatch.setattr("plugin_adapter_line.GatewayUserStore", lambda: store)
|
|
adapter = LineAdapter(
|
|
PlatformConfig(
|
|
enabled=True,
|
|
extra={"channel_access_token": "tok", "channel_secret": "sec"},
|
|
)
|
|
)
|
|
|
|
result = adapter._verified_email_target_if_bound("Uchat", "dk96@bremen.com.tw")
|
|
|
|
assert result == "dk96@bremen.com.tw"
|
|
assert store.calls[0][1] == "dk96@bremen.com.tw"
|
|
assert store.calls[0][2] is False
|
|
|
|
|
|
def test_line_guard_blocks_mismatched_system_handoff(monkeypatch):
|
|
store = _FakeStore(allow=False, live_email="dk96@bremen.com.tw")
|
|
monkeypatch.setattr("plugin_adapter_line.GatewayUserStore", lambda: store)
|
|
adapter = LineAdapter(
|
|
PlatformConfig(
|
|
enabled=True,
|
|
extra={"channel_access_token": "tok", "channel_secret": "sec"},
|
|
)
|
|
)
|
|
|
|
result = adapter._verified_email_target_if_bound("Uchat", "wrong@bremen.com.tw")
|
|
|
|
assert result is None
|
|
assert store.calls[0][2] is False
|