"""Tests for the Microsoft Teams platform adapter plugin.""" import json import sys import types from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch import httpx import pytest from gateway.config import Platform, PlatformConfig, HomeChannel from plugins.teams_pipeline.models import TeamsMeetingRef, TeamsMeetingSummaryPayload from tests.gateway._plugin_adapter_loader import load_plugin_adapter # --------------------------------------------------------------------------- # SDK Mock — install in sys.modules before importing the adapter # --------------------------------------------------------------------------- def _ensure_teams_mock(): """Install a teams SDK mock in sys.modules if the real package isn't present.""" if "microsoft_teams" in sys.modules and hasattr(sys.modules["microsoft_teams"], "__file__"): return # Build the module hierarchy microsoft_teams = types.ModuleType("microsoft_teams") microsoft_teams_apps = types.ModuleType("microsoft_teams.apps") microsoft_teams_api = types.ModuleType("microsoft_teams.api") microsoft_teams_api_activities = types.ModuleType("microsoft_teams.api.activities") microsoft_teams_api_activities_typing = types.ModuleType("microsoft_teams.api.activities.typing") microsoft_teams_api_activities_invoke = types.ModuleType("microsoft_teams.api.activities.invoke") microsoft_teams_api_activities_invoke_adaptive_card = types.ModuleType( "microsoft_teams.api.activities.invoke.adaptive_card" ) microsoft_teams_common = types.ModuleType("microsoft_teams.common") microsoft_teams_common_http = types.ModuleType("microsoft_teams.common.http") microsoft_teams_common_http_client = types.ModuleType("microsoft_teams.common.http.client") microsoft_teams_api_models = types.ModuleType("microsoft_teams.api.models") microsoft_teams_api_models_adaptive_card = types.ModuleType("microsoft_teams.api.models.adaptive_card") microsoft_teams_api_models_invoke_response = types.ModuleType("microsoft_teams.api.models.invoke_response") microsoft_teams_cards = types.ModuleType("microsoft_teams.cards") microsoft_teams_apps_http = types.ModuleType("microsoft_teams.apps.http") microsoft_teams_apps_http_adapter = types.ModuleType("microsoft_teams.apps.http.adapter") # App class mock class MockApp: def __init__(self, **kwargs): self._client_id = kwargs.get("client_id") self.server = MagicMock() self.server.handle_request = AsyncMock(return_value={"status": 200, "body": None}) self.credentials = MagicMock() self.credentials.client_id = self._client_id @property def id(self): return self._client_id def on_message(self, func): self._message_handler = func return func def on_card_action(self, func): self._card_action_handler = func return func async def initialize(self): pass async def send(self, conversation_id, activity): result = MagicMock() result.id = "sent-activity-id" return result async def start(self, port=3978): pass async def stop(self): pass microsoft_teams_apps.App = MockApp microsoft_teams_apps.ActivityContext = MagicMock microsoft_teams_common_http_client.ClientOptions = MagicMock # MessageActivity mock microsoft_teams_api.MessageActivity = MagicMock microsoft_teams_api.ConversationReference = MagicMock microsoft_teams_api.MessageActivityInput = MagicMock microsoft_teams_api.Attachment = MagicMock # TypingActivityInput mock class MockTypingActivityInput: pass microsoft_teams_api_activities_typing.TypingActivityInput = MockTypingActivityInput # Adaptive card invoke activity mock microsoft_teams_api_activities_invoke_adaptive_card.AdaptiveCardInvokeActivity = MagicMock # Adaptive card response mocks microsoft_teams_api_models_adaptive_card.AdaptiveCardActionCardResponse = MagicMock microsoft_teams_api_models_adaptive_card.AdaptiveCardActionMessageResponse = MagicMock # Invoke response mocks class MockInvokeResponse: def __init__(self, status=200, body=None): self.status = status self.body = body microsoft_teams_api_models_invoke_response.InvokeResponse = MockInvokeResponse microsoft_teams_api_models_invoke_response.AdaptiveCardInvokeResponse = MagicMock # Cards mocks class MockAdaptiveCard: def with_version(self, v): return self def with_body(self, body): return self def with_actions(self, actions): return self microsoft_teams_cards.AdaptiveCard = MockAdaptiveCard microsoft_teams_cards.ExecuteAction = MagicMock microsoft_teams_cards.TextBlock = MagicMock # HttpRequest TypedDict mock def HttpRequest(body=None, headers=None): return {"body": body, "headers": headers} # HttpResponse TypedDict mock HttpResponse = dict HttpMethod = str from typing import Callable HttpRouteHandler = Callable microsoft_teams_apps_http_adapter.HttpRequest = HttpRequest microsoft_teams_apps_http_adapter.HttpResponse = HttpResponse microsoft_teams_apps_http_adapter.HttpMethod = HttpMethod microsoft_teams_apps_http_adapter.HttpRouteHandler = HttpRouteHandler # Wire the hierarchy for name, mod in { "microsoft_teams": microsoft_teams, "microsoft_teams.apps": microsoft_teams_apps, "microsoft_teams.api": microsoft_teams_api, "microsoft_teams.api.activities": microsoft_teams_api_activities, "microsoft_teams.api.activities.typing": microsoft_teams_api_activities_typing, "microsoft_teams.api.activities.invoke": microsoft_teams_api_activities_invoke, "microsoft_teams.api.activities.invoke.adaptive_card": microsoft_teams_api_activities_invoke_adaptive_card, "microsoft_teams.common": microsoft_teams_common, "microsoft_teams.common.http": microsoft_teams_common_http, "microsoft_teams.common.http.client": microsoft_teams_common_http_client, "microsoft_teams.api.models": microsoft_teams_api_models, "microsoft_teams.api.models.adaptive_card": microsoft_teams_api_models_adaptive_card, "microsoft_teams.api.models.invoke_response": microsoft_teams_api_models_invoke_response, "microsoft_teams.cards": microsoft_teams_cards, "microsoft_teams.apps.http": microsoft_teams_apps_http, "microsoft_teams.apps.http.adapter": microsoft_teams_apps_http_adapter, }.items(): sys.modules.setdefault(name, mod) _ensure_teams_mock() # Load plugins/platforms/teams/adapter.py under a unique module name # (plugin_adapter_teams) so it cannot collide with sibling plugin adapters. _teams_mod = load_plugin_adapter("teams") _teams_mod.AIOHTTP_AVAILABLE = True # SDK import is deferred (#62935); bind mocked symbols the same way connect() # does, but skip the real lazy-installer so collection does not pip-install # microsoft-teams-apps. def _bind_mock_sdk(feature, importer, target_globals, **kwargs): target_globals.update(importer()) return True with patch("tools.lazy_deps.ensure_and_bind", _bind_mock_sdk): assert _teams_mod.check_teams_requirements() is True _teams_mod.TEAMS_SDK_AVAILABLE = True # Ensure SDK symbols that were None (import failed on Python <3.12) are # replaced with the mocked versions so runtime calls don't silently no-op. import sys as _sys _mt = _sys.modules.get("microsoft_teams.api.activities.typing") if _mt and _teams_mod.TypingActivityInput is None: _teams_mod.TypingActivityInput = _mt.TypingActivityInput TeamsAdapter = _teams_mod.TeamsAdapter from plugins.platforms.teams.summary_writer import TeamsSummaryWriter # noqa: E402 check_requirements = _teams_mod.check_requirements check_teams_requirements = _teams_mod.check_teams_requirements validate_config = _teams_mod.validate_config register = _teams_mod.register # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_config(**extra): return PlatformConfig(enabled=True, extra=extra) # --------------------------------------------------------------------------- # Tests: Requirements # --------------------------------------------------------------------------- class TestTeamsRequirements: def test_returns_true_when_deps_available(self, monkeypatch): monkeypatch.setattr(_teams_mod, "TEAMS_SDK_AVAILABLE", True) monkeypatch.setattr(_teams_mod, "AIOHTTP_AVAILABLE", True) assert check_requirements() is True def test_check_teams_requirements_shortcircuits_when_present(self, monkeypatch): # When SDK symbols are already bound and aiohttp is available, the # active lazy-installer returns True immediately without re-importing. monkeypatch.setattr(_teams_mod, "App", object()) monkeypatch.setattr(_teams_mod, "AIOHTTP_AVAILABLE", True) called = {"ensure_and_bind": 0} def _fake_ensure_and_bind(*_args, **_kwargs): called["ensure_and_bind"] += 1 return True monkeypatch.setattr( "tools.lazy_deps.ensure_and_bind", _fake_ensure_and_bind ) assert check_teams_requirements() is True assert called["ensure_and_bind"] == 0 def test_check_teams_requirements_lazy_installs_when_missing(self, monkeypatch): # When deps are missing, the active installer delegates to # ensure_and_bind("platform.teams", ...) — parity with Slack/Discord. monkeypatch.setattr(_teams_mod, "App", None) monkeypatch.setattr(_teams_mod, "TEAMS_SDK_AVAILABLE", False) monkeypatch.setattr(_teams_mod, "AIOHTTP_AVAILABLE", False) seen = {} def _fake_ensure_and_bind(feature, importer, target_globals, **kwargs): seen["feature"] = feature return True monkeypatch.setattr( "tools.lazy_deps.ensure_and_bind", _fake_ensure_and_bind ) assert check_teams_requirements() is True assert seen["feature"] == "platform.teams" def test_validate_config_with_env(self, monkeypatch): monkeypatch.setenv("TEAMS_CLIENT_ID", "test-id") monkeypatch.setenv("TEAMS_CLIENT_SECRET", "test-secret") monkeypatch.setenv("TEAMS_TENANT_ID", "test-tenant") assert validate_config(_make_config()) is True def test_validate_config_from_extra(self, monkeypatch): monkeypatch.delenv("TEAMS_CLIENT_ID", raising=False) monkeypatch.delenv("TEAMS_CLIENT_SECRET", raising=False) monkeypatch.delenv("TEAMS_TENANT_ID", raising=False) cfg = _make_config(client_id="id", client_secret="secret", tenant_id="tenant") assert validate_config(cfg) is True # --------------------------------------------------------------------------- # Tests: Adapter Init # --------------------------------------------------------------------------- class TestTeamsAdapterInit: def test_reads_config_from_extra(self): config = _make_config( client_id="cfg-id", client_secret="cfg-secret", tenant_id="cfg-tenant", ) adapter = TeamsAdapter(config) assert adapter._client_id == "cfg-id" assert adapter._client_secret == "cfg-secret" assert adapter._tenant_id == "cfg-tenant" def test_custom_port_from_env(self, monkeypatch): monkeypatch.setenv("TEAMS_PORT", "5000") adapter = TeamsAdapter(_make_config(client_id="id", client_secret="secret", tenant_id="tenant")) assert adapter._port == 5000 def test_invalid_port_from_extra_falls_back_to_default(self): adapter = TeamsAdapter( _make_config(client_id="id", client_secret="secret", tenant_id="tenant", port="abc") ) assert adapter._port == 3978 # --------------------------------------------------------------------------- # Tests: Plugin registration # --------------------------------------------------------------------------- class TestTeamsPluginRegistration: def test_register_name(self): ctx = MagicMock() register(ctx) kwargs = ctx.register_platform.call_args[1] assert kwargs["name"] == "teams" def test_register_splits_passive_probe_from_active_installer(self): # check_fn is the PASSIVE probe (status displays call it freely); # the ACTIVE lazy-installer rides on ensure_deps_fn, which # create_adapter() invokes when the passive probe fails (#79812). ctx = MagicMock() register(ctx) kwargs = ctx.register_platform.call_args[1] assert kwargs["check_fn"] is check_requirements assert kwargs["ensure_deps_fn"] is check_teams_requirements def test_register_auth_env_vars(self): ctx = MagicMock() register(ctx) kwargs = ctx.register_platform.call_args[1] assert kwargs["allowed_users_env"] == "TEAMS_ALLOWED_USERS" assert kwargs["allow_all_env"] == "TEAMS_ALLOW_ALL_USERS" # --------------------------------------------------------------------------- # Tests: Interactive setup (import fix regression — #18325 / #19173) # --------------------------------------------------------------------------- class TestTeamsInteractiveSetup: def test_interactive_setup_persists_credentials(self, tmp_path, monkeypatch): """Regression for #19173: interactive_setup must import prompt helpers from hermes_cli.cli_output (not hermes_cli.config) and persist credentials to .env without crashing. """ hermes_home = tmp_path / "hermes" monkeypatch.setenv("HERMES_HOME", str(hermes_home)) import hermes_cli.cli_output as cli_output_mod answers = iter(["client-id", "client-secret", "tenant-id", "aad-1, aad-2"]) monkeypatch.setattr(cli_output_mod, "prompt", lambda *_a, **_kw: next(answers)) monkeypatch.setattr(cli_output_mod, "prompt_yes_no", lambda *_a, **_kw: True) monkeypatch.setattr(cli_output_mod, "print_info", lambda *_a, **_kw: None) monkeypatch.setattr(cli_output_mod, "print_success", lambda *_a, **_kw: None) monkeypatch.setattr(cli_output_mod, "print_warning", lambda *_a, **_kw: None) _teams_mod.interactive_setup() env_text = (hermes_home / ".env").read_text(encoding="utf-8") assert "TEAMS_CLIENT_ID=client-id" in env_text assert "TEAMS_TENANT_ID=tenant-id" in env_text class TestTeamsConnect: @pytest.mark.anyio async def test_connect_fails_without_sdk(self, monkeypatch): monkeypatch.setattr(_teams_mod, "TEAMS_SDK_AVAILABLE", False) monkeypatch.setattr(_teams_mod, "App", None) monkeypatch.setattr(_teams_mod, "ClientOptions", None) # Simulate the SDK being unavailable AND not installable (offline / # locked-down env): the lazy-installer can't rebind the globals, so # App stays None and connect() must fail without calling it. monkeypatch.setattr( "tools.lazy_deps.ensure_and_bind", lambda *_a, **_k: False, ) adapter = TeamsAdapter(_make_config( client_id="id", client_secret="secret", tenant_id="tenant", )) result = await adapter.connect() assert result is False @pytest.mark.anyio async def test_connect_fails_when_namespace_exists_but_app_unbound(self, monkeypatch): """find_spec('microsoft_teams') can be true from sibling packages without microsoft-teams-apps. connect() must not call App() while it is still None — that was ``'NoneType' object is not callable``. """ monkeypatch.setattr(_teams_mod, "TEAMS_SDK_AVAILABLE", True) monkeypatch.setattr(_teams_mod, "App", None) monkeypatch.setattr(_teams_mod, "ClientOptions", None) monkeypatch.setattr(_teams_mod, "AIOHTTP_AVAILABLE", True) monkeypatch.setattr( "tools.lazy_deps.ensure_and_bind", lambda *_a, **_k: False, ) adapter = TeamsAdapter(_make_config( client_id="id", client_secret="secret", tenant_id="tenant", )) result = await adapter.connect() assert result is False assert adapter._app is None # --------------------------------------------------------------------------- # Tests: Send # --------------------------------------------------------------------------- class TestTeamsSend: @pytest.mark.anyio async def test_send_calls_app_send(self): adapter = TeamsAdapter(_make_config( client_id="id", client_secret="secret", tenant_id="tenant", )) mock_result = MagicMock() mock_result.id = "msg-123" mock_app = MagicMock() mock_app.send = AsyncMock(return_value=mock_result) adapter._app = mock_app result = await adapter.send("conv-id", "Hello") assert result.success is True assert result.message_id == "msg-123" mock_app.send.assert_awaited_once_with("conv-id", "Hello") def _make_summary_payload(): return TeamsMeetingSummaryPayload( meeting_ref=TeamsMeetingRef(meeting_id="meeting-123"), title="Weekly Sync", summary="Discussed launch readiness.", key_decisions=["Proceed with staged rollout."], action_items=["Send launch checklist."], risks=["QA sign-off still pending."], ) class TestTeamsSummaryWriter: @pytest.mark.anyio async def test_graph_delivery_posts_to_channel(self): graph_client = SimpleNamespace( post_json=AsyncMock(return_value={"id": "msg-123", "webUrl": "https://teams.example/messages/123"}) ) writer = TeamsSummaryWriter(graph_client=graph_client) payload = _make_summary_payload() result = await writer.write_summary( payload, { "delivery_mode": "graph", "team_id": "team-1", "channel_id": "channel-1", }, ) assert result["target_type"] == "channel" assert result["message_id"] == "msg-123" graph_client.post_json.assert_awaited_once() path = graph_client.post_json.await_args.args[0] body = graph_client.post_json.await_args.kwargs["json_body"] assert path == "/teams/team-1/channels/channel-1/messages" assert body["body"]["contentType"] == "html" assert "Weekly Sync" in body["body"]["content"] # --------------------------------------------------------------------------- # Tests: Message Handling # --------------------------------------------------------------------------- class TestTeamsMessageHandling: def _make_activity( self, *, text="Hello", from_id="user-123", from_aad_id="aad-456", from_name="Test User", conversation_id="19:abc@thread.v2", conversation_type="personal", tenant_id="tenant-789", activity_id="activity-001", attachments=None, ): activity = MagicMock() activity.text = text activity.id = activity_id activity.from_ = MagicMock() activity.from_.id = from_id activity.from_.aad_object_id = from_aad_id activity.from_.name = from_name activity.conversation = MagicMock() activity.conversation.id = conversation_id activity.conversation.conversation_type = conversation_type activity.conversation.name = "Test Chat" activity.conversation.tenant_id = tenant_id activity.attachments = attachments or [] return activity def _make_ctx(self, activity): ctx = MagicMock() ctx.activity = activity return ctx @pytest.mark.anyio async def test_personal_message_creates_dm_event(self): adapter = TeamsAdapter(_make_config( client_id="bot-id", client_secret="secret", tenant_id="tenant", )) adapter._app = MagicMock() adapter._app.id = "bot-id" adapter.handle_message = AsyncMock() activity = self._make_activity(conversation_type="personal") await adapter._on_message(self._make_ctx(activity)) adapter.handle_message.assert_awaited_once() event = adapter.handle_message.call_args[0][0] assert event.source.chat_type == "dm" @pytest.mark.anyio async def test_group_message_creates_group_event(self): adapter = TeamsAdapter(_make_config( client_id="bot-id", client_secret="secret", tenant_id="tenant", )) adapter._app = MagicMock() adapter._app.id = "bot-id" adapter.handle_message = AsyncMock() activity = self._make_activity(conversation_type="groupChat") await adapter._on_message(self._make_ctx(activity)) event = adapter.handle_message.call_args[0][0] assert event.source.chat_type == "group" @pytest.mark.anyio async def test_aad_user_route_survives_conversation_changes(self, monkeypatch): from gateway.profile_routing import parse_profile_routes from gateway.run import GatewayRunner routes = parse_profile_routes([ {"name": "owner", "platform": "teams", "user_id": "aad-456", "profile": "owner"}, {"name": "other", "platform": "teams", "user_id": "aad-789", "profile": "other"}, ]) runner = GatewayRunner.__new__(GatewayRunner) runner.config = SimpleNamespace(multiplex_profiles=True, profile_routes=routes) monkeypatch.setattr( "gateway.run._multiplex_profile_homes", lambda _config: [("owner", None), ("other", None)], ) adapter = TeamsAdapter(_make_config( client_id="bot-id", client_secret="secret", tenant_id="tenant", )) adapter.gateway_runner = runner adapter._app = MagicMock() adapter._app.id = "bot-id" adapter.handle_message = AsyncMock() for activity_id, conversation_id, conversation_type, user_id in ( ("activity-group", "19:shared@thread.v2", "groupChat", "aad-456"), ("activity-channel", "19:channel@thread.v2", "channel", "aad-456"), ("activity-dm", "19:dm@thread.v2", "personal", "aad-456"), ("activity-other", "19:shared@thread.v2", "groupChat", "aad-789"), ): await adapter._on_message(self._make_ctx(self._make_activity( activity_id=activity_id, conversation_id=conversation_id, conversation_type=conversation_type, from_aad_id=user_id, ))) sources = [call.args[0].source for call in adapter.handle_message.await_args_list] assert [source.profile for source in sources] == ["owner", "owner", "owner", "other"] assert [source.chat_type for source in sources] == ["group", "channel", "dm", "group"] assert [runner._session_key_for_source(source).split(":", 2)[1] for source in sources] == [ "owner", "owner", "owner", "other", ] class TestTeamsAttachmentClassification: """Document attachments must set MessageType.DOCUMENT so run.py's document-context injection surfaces the cached file to the agent (same bug class as Signal/Email/SimpleX, PR #44695).""" def _make_adapter(self): adapter = TeamsAdapter(_make_config( client_id="bot-id", client_secret="secret", tenant_id="tenant", )) adapter._app = MagicMock() adapter._app.id = "bot-id" adapter.handle_message = AsyncMock() return adapter def _make_activity(self, attachments, text="see attached"): activity = MagicMock() activity.text = text activity.id = "activity-att-001" activity.from_ = MagicMock() activity.from_.id = "user-123" activity.from_.aad_object_id = "aad-456" activity.from_.name = "Test User" activity.conversation = MagicMock() activity.conversation.id = "19:abc@thread.v2" activity.conversation.conversation_type = "personal" activity.conversation.name = "Test Chat" activity.conversation.tenant_id = "tenant-789" activity.attachments = attachments return activity def _make_ctx(self, activity): ctx = MagicMock() ctx.activity = activity return ctx def _file_download_attachment(self, name="report.pdf", file_type="pdf"): att = MagicMock() att.content_type = "application/vnd.microsoft.teams.file.download.info" att.content_url = None att.name = name att.content = { "downloadUrl": "https://contoso.sharepoint.com/download/x", "fileType": file_type, } return att def _image_attachment(self): att = MagicMock() att.content_type = "image/png" att.content_url = "https://smba.example.com/img.png" att.name = "img.png" return att def _html_body_attachment(self): # Teams mirrors the message body as a text/html attachment att = MagicMock() att.content_type = "text/html" att.content_url = None att.name = "" return att @pytest.mark.anyio async def test_file_download_info_sets_document_type(self): from gateway.platforms.event import MessageType adapter = self._make_adapter() adapter._fetch_attachment_bytes = AsyncMock(return_value=b"%PDF-1.4 fake") activity = self._make_activity([self._file_download_attachment()]) await adapter._on_message(self._make_ctx(activity)) event = adapter.handle_message.call_args[0][0] assert event.message_type == MessageType.DOCUMENT, ( f"Expected DOCUMENT, got {event.message_type}. " "Documents must be classified as DOCUMENT so run.py injects file context." ) assert len(event.media_urls) == 1 assert event.media_types == ["application/pdf"] @pytest.mark.anyio async def test_mixed_image_and_document_prefers_document(self): from gateway.platforms.event import MessageType adapter = self._make_adapter() adapter._fetch_attachment_bytes = AsyncMock(return_value=b"%PDF-1.4 fake") async def fake_cache_image(url, *a, **kw): return "/tmp/img.png" with pytest.MonkeyPatch.context() as mp: mp.setattr(_teams_mod, "cache_image_from_url", fake_cache_image) activity = self._make_activity([ self._image_attachment(), self._file_download_attachment(), ]) await adapter._on_message(self._make_ctx(activity)) event = adapter.handle_message.call_args[0][0] assert event.message_type == MessageType.DOCUMENT assert len(event.media_urls) == 2 # ── Bot Framework connector attachments (pasted images) ────────────────── class TestTeamsBotFrameworkAttachments: """Pasted/inline images arrive on smba.trafficmanager.net hosts and need the bot's own bearer token (unlike SharePoint downloadUrls). These tests pin the auth routing, the token cache, the attacker-host block, and the failure fallbacks of that path.""" def _make_adapter(self): adapter = TeamsAdapter(_make_config( client_id="bot-id", client_secret="secret", tenant_id="tenant", )) adapter._app = MagicMock() adapter._app.id = "bot-id" adapter.handle_message = AsyncMock() return adapter def _make_activity(self, attachments): activity = MagicMock() activity.text = "see attached" activity.id = "activity-att-001" activity.from_ = MagicMock() activity.from_.id = "user-123" activity.from_.aad_object_id = "aad-456" activity.from_.name = "Test User" activity.conversation = MagicMock() activity.conversation.id = "19:abc@thread.v2" activity.conversation.conversation_type = "personal" activity.conversation.name = "Test Chat" activity.conversation.tenant_id = "tenant-789" activity.attachments = attachments return activity def _make_ctx(self, activity): ctx = MagicMock() ctx.activity = activity return ctx def _bf_image_attachment(self, url=None): att = MagicMock() att.content_type = "image/png" att.content_url = url or "https://smba.trafficmanager.net/emea/b1/v3/attachments/0-abc/views/original" att.name = "pasted.png" return att @pytest.mark.anyio async def test_bf_url_predicate_exact_match_allowlist(self): """Only exact allowlisted hosts on https default port may receive the bot's bearer token — lookalikes, other schemes, and non-443 ports must NOT (any Azure customer can register .trafficmanager.net).""" f = _teams_mod._is_botframework_attachment_url assert f("https://smba.trafficmanager.net/emea/v3/attachments/x") assert f("https://smba.infra.gov.teams.microsoft.us/amer/v3/attachments/x") assert f("https://smba.trafficmanager.net:443/emea/v3/attachments/x") # Attacker lookalikes / non-allowlisted / wrong scheme / wrong port assert not f("https://evil-trafficmanager.net/steal") assert not f("https://emea.smba.trafficmanager.net/v3/attachments/x") assert not f("https://notbotframework.com/steal") assert not f("https://trafficmanager.net.evil.com/steal") assert not f("http://smba.trafficmanager.net/v3/attachments/x") assert not f("https://smba.trafficmanager.net:444/v3/attachments/x") assert not f("") assert not f("https://sharepoint.com/x") @pytest.mark.anyio async def test_bf_image_routes_through_authenticated_fetch(self): adapter = self._make_adapter() adapter._fetch_attachment_bytes = AsyncMock(return_value=b"\x89PNG fake") adapter._get_botframework_token = AsyncMock(return_value="tok") async def fake_cache_media_bytes(data, **kwargs): return SimpleNamespace( path="/tmp/img.png", media_type="image/png", kind="image" ) with patch.object(_teams_mod, "cache_media_bytes_async", fake_cache_media_bytes): activity = self._make_activity([self._bf_image_attachment()]) await adapter._on_message(self._make_ctx(activity)) event = adapter.handle_message.call_args[0][0] assert len(event.media_urls) == 1 assert event.media_types == ["image/png"] # URL was fetched with auth (via _fetch_attachment_bytes, which the # token routing test below exercises end-to-end) adapter._fetch_attachment_bytes.assert_awaited_once_with( "https://smba.trafficmanager.net/emea/b1/v3/attachments/0-abc/views/original" ) @pytest.mark.anyio async def test_non_bf_image_uses_generic_cache_helper(self): adapter = self._make_adapter() adapter._fetch_attachment_bytes = AsyncMock(side_effect=AssertionError("must not be called")) async def fake_cache_image(url, *a, **kw): return "/tmp/img.jpg" with patch.object(_teams_mod, "cache_image_from_url", fake_cache_image): activity = self._make_activity( [self._bf_image_attachment(url="https://contoso.sharepoint.com/img.png")] ) await adapter._on_message(self._make_ctx(activity)) event = adapter.handle_message.call_args[0][0] assert len(event.media_urls) == 1 assert event.media_urls[0] == "/tmp/img.jpg" @pytest.mark.anyio async def test_fetch_attachment_bytes_sends_bearer_for_bf_host(self): """End-to-end over _fetch_attachment_bytes: BF host → token acquired and Authorization attached; non-BF host → no token call.""" adapter = self._make_adapter() adapter._get_botframework_token = AsyncMock(return_value="the-token") captured = {} class _FakeStreamResponse: def __init__(self): self.headers = {} def raise_for_status(self): pass async def aiter_bytes(self): yield b"\x89PNG fake" class _FakeStreamCtx: def __init__(self, response): self._response = response async def __aenter__(self): return self._response async def __aexit__(self, *a): return None class _FakeClient: def __init__(self, **kw): pass async def __aenter__(self): return self async def __aexit__(self, *a): return None def stream(self, method, url, headers=None): captured["headers"] = headers or {} return _FakeStreamCtx(_FakeStreamResponse()) with patch("tools.url_safety.create_ssrf_safe_async_client", lambda **kw: _FakeClient()), \ patch("tools.url_safety.is_safe_url", lambda url: True): # BF host: bearer attached data = await adapter._fetch_attachment_bytes("https://smba.trafficmanager.net/emea/v3/attachments/x") assert captured["headers"].get("Authorization") == "Bearer the-token" assert data == b"\x89PNG fake" adapter._get_botframework_token = AsyncMock(return_value="the-token") with patch("tools.url_safety.create_ssrf_safe_async_client", lambda **kw: _FakeClient()), \ patch("tools.url_safety.is_safe_url", lambda url: True): # Attacker lookalike host: NO bearer (exact-match allowlist) await adapter._fetch_attachment_bytes("https://evil-trafficmanager.net/steal") assert "Authorization" not in captured["headers"], ( "bearer token must not be sent to attacker lookalike hosts" ) adapter._get_botframework_token.assert_not_awaited() @pytest.mark.anyio async def test_token_refresh_is_serialized_under_lock(self): """Two concurrent token fetches on a cold cache share ONE POST — the lock prevents a token-endpoint stampede.""" import asyncio as _asyncio adapter = self._make_adapter() posts = [] release = _asyncio.Event() class _TokenResp: status_code = 200 def raise_for_status(self): pass def json(self): return {"access_token": "tok-1", "expires_in": 3600} class _SlowTokenClient: def __init__(self, **kw): pass async def __aenter__(self): return self async def __aexit__(self, *a): return None async def post(self, url, data=None): posts.append((url, dict(data or {}))) await release.wait() # hold both callers at the STS door return _TokenResp() async def release_later(): await _asyncio.sleep(0.05) release.set() with patch("httpx.AsyncClient", _SlowTokenClient): t1 = _asyncio.create_task(adapter._get_botframework_token()) t2 = _asyncio.create_task(adapter._get_botframework_token()) await release_later() tok1, tok2 = await t1, await t2 assert tok1 == "tok-1" and tok2 == "tok-1" assert len(posts) == 1, f"concurrent cold-cache fetches must share one POST, got {len(posts)}" @pytest.mark.anyio async def test_token_acquisition_and_cache_reuse(self): adapter = self._make_adapter() posts = [] class _TokenResp: status_code = 200 def raise_for_status(self): pass def json(self): return {"access_token": "tok-1", "expires_in": 3600} class _TokenClient: def __init__(self, **kw): pass async def __aenter__(self): return self async def __aexit__(self, *a): return None async def post(self, url, data=None): posts.append((url, dict(data or {}))) return _TokenResp() with patch("httpx.AsyncClient", _TokenClient): tok1 = await adapter._get_botframework_token() tok2 = await adapter._get_botframework_token() assert tok1 == "tok-1" and tok2 == "tok-1" assert len(posts) == 1, "second call must hit the cache" assert posts[0][0] == "https://login.microsoftonline.com/tenant/oauth2/v2.0/token" assert posts[0][1]["scope"] == "https://api.botframework.com/.default" assert posts[0][1]["client_id"] == "bot-id" assert posts[0][1]["client_secret"] == "secret" @pytest.mark.anyio async def test_token_acquisition_failure_degrades_to_unauthenticated_fetch(self): """Token failure must not break the fetch: warning + fetch without Authorization (same net behavior as the pre-fix path).""" import httpx as _httpx adapter = self._make_adapter() adapter._get_botframework_token = AsyncMock(side_effect=ValueError("no creds")) captured = {} class _FakeStreamResponse: def __init__(self): self.headers = {} def raise_for_status(self): raise _httpx.HTTPStatusError( "401", request=MagicMock(), response=MagicMock(status_code=401) ) async def aiter_bytes(self): yield b"" class _FakeStreamCtx: def __init__(self, response): self._response = response async def __aenter__(self): return self._response async def __aexit__(self, *a): return None class _FakeClient: def __init__(self, **kw): pass async def __aenter__(self): return self async def __aexit__(self, *a): return None def stream(self, method, url, headers=None): captured["headers"] = headers or {} return _FakeStreamCtx(_FakeStreamResponse()) with patch("tools.url_safety.create_ssrf_safe_async_client", lambda **kw: _FakeClient()), \ patch("tools.url_safety.is_safe_url", lambda url: True): with pytest.raises(_httpx.HTTPStatusError): await adapter._fetch_attachment_bytes("https://smba.trafficmanager.net/v3/attachments/x") assert "Authorization" not in captured["headers"] @pytest.mark.anyio async def test_bf_image_invalid_bytes_logs_warning(self): """Non-image bytes from the BF endpoint must not be silently dropped — the else branch warns (regression guard for the silent-drop).""" adapter = self._make_adapter() adapter._fetch_attachment_bytes = AsyncMock(return_value=b"error page") async def _no_media(*a, **kw): return None with patch.object(_teams_mod, "cache_media_bytes_async", _no_media): with patch.object(_teams_mod.logger, "warning") as warn: activity = self._make_activity([self._bf_image_attachment()]) await adapter._on_message(self._make_ctx(activity)) event = adapter.handle_message.call_args[0][0] assert event.media_urls == [] assert warn.called, "silent drop of invalid BF image bytes must log a warning" logged = " ".join(str(c) for c in warn.call_args_list) assert "image validation" in logged or "failed image validation" in logged # ── _standalone_send (out-of-process cron delivery) ────────────────────── class _FakeAiohttpResponse: def __init__(self, status: int, payload, text_body: str = ""): self.status = status self._payload = payload self._text = text_body or (str(payload) if payload is not None else "") async def json(self): return self._payload async def text(self): return self._text async def __aenter__(self): return self async def __aexit__(self, exc_type, exc, tb): return None class _FakeAiohttpSession: """Scripted aiohttp.ClientSession with a queue of responses so tests can assert calls in order.""" def __init__(self, scripts): self._scripts = list(scripts) self.calls: list[tuple[str, dict]] = [] async def __aenter__(self): return self async def __aexit__(self, exc_type, exc, tb): return None def post(self, url, **kwargs): self.calls.append((url, kwargs)) if not self._scripts: raise AssertionError(f"No scripted response for POST {url}") return self._scripts.pop(0) def _install_fake_aiohttp(monkeypatch, session): """Replace ``aiohttp`` in ``sys.modules`` so ``import aiohttp as _aiohttp`` inside ``_standalone_send`` picks up our fake.""" fake_aiohttp = types.SimpleNamespace( ClientSession=lambda timeout=None, **kwargs: session, ClientTimeout=lambda total=None: None, ) monkeypatch.setitem(sys.modules, "aiohttp", fake_aiohttp) class TestTeamsStandaloneSend: @pytest.mark.asyncio async def test_standalone_send_acquires_token_and_posts_activity(self, monkeypatch): monkeypatch.setenv("TEAMS_CLIENT_ID", "client-id") monkeypatch.setenv("TEAMS_CLIENT_SECRET", "secret") monkeypatch.setenv("TEAMS_TENANT_ID", "tenant") monkeypatch.delenv("TEAMS_SERVICE_URL", raising=False) token_resp = _FakeAiohttpResponse(200, {"access_token": "the-token"}) activity_resp = _FakeAiohttpResponse(200, {"id": "msg-99"}) session = _FakeAiohttpSession([token_resp, activity_resp]) _install_fake_aiohttp(monkeypatch, session) result = await _teams_mod._standalone_send( PlatformConfig(enabled=True, extra={}), "19:abc@thread.skype", "hello cron", ) assert result == {"success": True, "message_id": "msg-99"} assert len(session.calls) == 2 token_url, token_kwargs = session.calls[0] assert "login.microsoftonline.com/tenant/oauth2/v2.0/token" in token_url assert token_kwargs["data"]["client_id"] == "client-id" assert token_kwargs["data"]["client_secret"] == "secret" assert token_kwargs["data"]["scope"] == "https://api.botframework.com/.default" activity_url, activity_kwargs = session.calls[1] # Default service URL when TEAMS_SERVICE_URL is unset assert "smba.trafficmanager.net" in activity_url assert "/v3/conversations/19:abc@thread.skype/activities" in activity_url assert activity_kwargs["headers"]["Authorization"] == "Bearer the-token" assert activity_kwargs["json"]["text"] == "hello cron" assert activity_kwargs["json"]["type"] == "message" @pytest.mark.asyncio async def test_standalone_send_propagates_token_failure(self, monkeypatch): monkeypatch.setenv("TEAMS_CLIENT_ID", "client-id") monkeypatch.setenv("TEAMS_CLIENT_SECRET", "secret") monkeypatch.setenv("TEAMS_TENANT_ID", "tenant") token_resp = _FakeAiohttpResponse( 401, {"error": "unauthorized_client"}, text_body='{"error":"unauthorized_client"}', ) session = _FakeAiohttpSession([token_resp]) _install_fake_aiohttp(monkeypatch, session) result = await _teams_mod._standalone_send( PlatformConfig(enabled=True, extra={}), "19:abc@thread.skype", "hi", ) assert "error" in result assert "401" in result["error"] assert "token" in result["error"].lower() class TestTeamsMediaAttachments: """send_video / send_voice / send_document route through the same Attachment mechanism as send_image so the gateway's media dispatch (run.py) delivers native attachments instead of the base-class text fallback (file path sent as plain text).""" def _make_adapter(self): adapter = TeamsAdapter(_make_config( client_id="bot-id", client_secret="secret", tenant_id="tenant", )) adapter._app = MagicMock() adapter._app.id = "bot-id" adapter._app.send = AsyncMock(return_value=MagicMock(id="msg-001")) return adapter @pytest.mark.asyncio async def test_send_voice_local_file_base64(self, tmp_path): adapter = self._make_adapter() audio = tmp_path / "reply.mp3" audio.write_bytes(b"ID3fakeaudio") result = await adapter.send_voice("19:abc@thread.v2", str(audio), caption="here you go") assert result.success adapter._app.send.assert_awaited_once() @pytest.mark.asyncio async def test_send_document_local_file_base64(self, tmp_path): adapter = self._make_adapter() doc = tmp_path / "report.pdf" doc.write_bytes(b"%PDF-1.4 fake") result = await adapter.send_document("19:abc@thread.v2", str(doc)) assert result.success adapter._app.send.assert_awaited_once() # --------------------------------------------------------------------------- # Tests: require_mention gating (RSC-delivered history) # --------------------------------------------------------------------------- class TestTeamsRequireMention: """With resource-specific consent Teams delivers every channel/groupChat message, not just mentions. ``require_mention`` must drop unaddressed non-personal posts BEFORE the attachment loop, keep @mentions (wire id ``28:``) / replies to the bot / personal chats, and be read env-over-YAML like every other adapter.""" APP_ID = "bot-id" def _make_adapter(self, monkeypatch=None, **extra): adapter = TeamsAdapter(_make_config( client_id=self.APP_ID, client_secret="secret", tenant_id="tenant", **extra)) adapter._app = MagicMock() adapter._app.id = self.APP_ID adapter.handle_message = AsyncMock() adapter._fetch_attachment_bytes = AsyncMock(return_value=b"\x89PNG" + b"\0" * 32) return adapter def _activity(self, conversation_type, *, text="hello", mentioned_id=None, reply_to_id=None): activity = MagicMock() activity.text = text activity.id = f"act-{conversation_type}-{mentioned_id}-{reply_to_id}" activity.from_ = MagicMock(aad_object_id="aad-456", name="Test User") activity.from_.id = "29:user-123" activity.recipient = MagicMock() activity.recipient.id = f"28:{self.APP_ID}" activity.conversation = MagicMock(conversation_type=conversation_type, tenant_id="t") activity.conversation.id = "19:conv@thread.v2" activity.conversation.name = "Conv" att = MagicMock(content_type="image/png") att.name = "a.png" att.content_url = "https://smba.trafficmanager.net/emea/v3/attachments/1/views/original" activity.attachments = [att] activity.reply_to_id = reply_to_id activity.entities = [] if mentioned_id: entity = MagicMock(type="mention") entity.mentioned = MagicMock() entity.mentioned.id = mentioned_id activity.entities = [entity] return activity @pytest.mark.anyio @pytest.mark.parametrize("conversation_type, kwargs, dispatched", [ ("channel", {}, False), ("groupChat", {}, False), ("channel", {"text": "Alice hi", "mentioned_id": "29:alice"}, False), # someone else ("channel", {"text": "Hermes hi", "mentioned_id": "28:bot-id"}, True), # wire form of the bot id ("groupChat", {"text": "Hermes hi", "mentioned_id": "bot-id"}, True), ("channel", {"reply_to_id": "bot-msg-1"}, True), ("personal", {}, True), ]) async def test_gate_drops_unaddressed_non_personal_before_attachment_download( self, conversation_type, kwargs, dispatched, ): adapter = self._make_adapter(require_mention=True) adapter._sent_ids.append("bot-msg-1") ctx = MagicMock() ctx.activity = self._activity(conversation_type, **kwargs) await adapter._on_message(ctx) assert adapter.handle_message.await_count == (1 if dispatched else 0) assert adapter._fetch_attachment_bytes.await_count == (1 if dispatched else 0) @pytest.mark.parametrize("yaml_value, env_value, expected", [ (None, None, False), # opt-in: absent key leaves every conversation ungated (True, None, True), ("false", None, False), (True, "false", False), # explicit env beats YAML, like MATRIX_/MATTERMOST_REQUIRE_MENTION (False, "true", True), ]) def test_require_mention_read_env_over_yaml(self, monkeypatch, yaml_value, env_value, expected): monkeypatch.delenv("TEAMS_REQUIRE_MENTION", raising=False) if env_value is not None: monkeypatch.setenv("TEAMS_REQUIRE_MENTION", env_value) extra = {} if yaml_value is None else {"require_mention": yaml_value} adapter = self._make_adapter(**extra) assert adapter._require_mention is expected assert adapter._extra.get("require_mention") == yaml_value # extras stay readable on the instance