"""Tests for the Microsoft Teams platform adapter plugin.""" import sys import types from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch import pytest from gateway.config import PlatformConfig 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("pm.extras.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_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_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( "pm.extras.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( "pm.extras.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 # --------------------------------------------------------------------------- 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" # ── _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() # --------------------------------------------------------------------------- # 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