Add Microsoft Teams channel on current nightly base
This commit is contained in:
T3chC0wb0y
Xubin Ren
@@ -0,0 +1,684 @@
|
||||
import json
|
||||
|
||||
import jwt
|
||||
import pytest
|
||||
from cryptography.hazmat.primitives.asymmetric import rsa
|
||||
|
||||
from nanobot.bus.events import OutboundMessage
|
||||
from nanobot.channels.msteams import ConversationRef, MSTeamsChannel, MSTeamsConfig
|
||||
|
||||
|
||||
class DummyBus:
|
||||
def __init__(self):
|
||||
self.inbound = []
|
||||
|
||||
async def publish_inbound(self, msg):
|
||||
self.inbound.append(msg)
|
||||
|
||||
|
||||
class FakeResponse:
|
||||
def __init__(self, payload):
|
||||
self._payload = payload
|
||||
|
||||
def raise_for_status(self):
|
||||
return None
|
||||
|
||||
def json(self):
|
||||
return self._payload
|
||||
|
||||
|
||||
class FakeHttpClient:
|
||||
def __init__(self, payload=None):
|
||||
self.payload = payload or {"access_token": "tok", "expires_in": 3600}
|
||||
self.calls = []
|
||||
|
||||
async def post(self, url, **kwargs):
|
||||
self.calls.append((url, kwargs))
|
||||
return FakeResponse(self.payload)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_activity_personal_message_publishes_and_stores_ref(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr("nanobot.channels.msteams.get_workspace_path", lambda: tmp_path)
|
||||
|
||||
bus = DummyBus()
|
||||
ch = MSTeamsChannel(
|
||||
{
|
||||
"enabled": True,
|
||||
"appId": "app-id",
|
||||
"appPassword": "secret",
|
||||
"tenantId": "tenant-id",
|
||||
"allowFrom": ["*"],
|
||||
},
|
||||
bus,
|
||||
)
|
||||
|
||||
activity = {
|
||||
"type": "message",
|
||||
"id": "activity-1",
|
||||
"text": "Hello from Teams",
|
||||
"serviceUrl": "https://smba.trafficmanager.net/amer/",
|
||||
"conversation": {
|
||||
"id": "conv-123",
|
||||
"conversationType": "personal",
|
||||
},
|
||||
"from": {
|
||||
"id": "29:user-id",
|
||||
"aadObjectId": "aad-user-1",
|
||||
"name": "Bob",
|
||||
},
|
||||
"recipient": {
|
||||
"id": "28:bot-id",
|
||||
"name": "nanobot",
|
||||
},
|
||||
"channelData": {
|
||||
"tenant": {"id": "tenant-id"},
|
||||
},
|
||||
}
|
||||
|
||||
await ch._handle_activity(activity)
|
||||
|
||||
assert len(bus.inbound) == 1
|
||||
msg = bus.inbound[0]
|
||||
assert msg.channel == "msteams"
|
||||
assert msg.sender_id == "aad-user-1"
|
||||
assert msg.chat_id == "conv-123"
|
||||
assert msg.content == "Hello from Teams"
|
||||
assert msg.metadata["msteams"]["conversation_id"] == "conv-123"
|
||||
assert "conv-123" in ch._conversation_refs
|
||||
|
||||
saved = json.loads((tmp_path / "state" / "msteams_conversations.json").read_text(encoding="utf-8"))
|
||||
assert saved["conv-123"]["conversation_id"] == "conv-123"
|
||||
assert saved["conv-123"]["tenant_id"] == "tenant-id"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_activity_ignores_group_messages(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr("nanobot.channels.msteams.get_workspace_path", lambda: tmp_path)
|
||||
|
||||
bus = DummyBus()
|
||||
ch = MSTeamsChannel(
|
||||
{
|
||||
"enabled": True,
|
||||
"appId": "app-id",
|
||||
"appPassword": "secret",
|
||||
"tenantId": "tenant-id",
|
||||
"allowFrom": ["*"],
|
||||
},
|
||||
bus,
|
||||
)
|
||||
|
||||
activity = {
|
||||
"type": "message",
|
||||
"id": "activity-2",
|
||||
"text": "Hello group",
|
||||
"serviceUrl": "https://smba.trafficmanager.net/amer/",
|
||||
"conversation": {
|
||||
"id": "conv-group",
|
||||
"conversationType": "channel",
|
||||
},
|
||||
"from": {
|
||||
"id": "29:user-id",
|
||||
"aadObjectId": "aad-user-1",
|
||||
"name": "Bob",
|
||||
},
|
||||
"recipient": {
|
||||
"id": "28:bot-id",
|
||||
"name": "nanobot",
|
||||
},
|
||||
}
|
||||
|
||||
await ch._handle_activity(activity)
|
||||
|
||||
assert bus.inbound == []
|
||||
assert ch._conversation_refs == {}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_activity_mention_only_uses_default_response(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr("nanobot.channels.msteams.get_workspace_path", lambda: tmp_path)
|
||||
|
||||
bus = DummyBus()
|
||||
ch = MSTeamsChannel(
|
||||
{
|
||||
"enabled": True,
|
||||
"appId": "app-id",
|
||||
"appPassword": "secret",
|
||||
"tenantId": "tenant-id",
|
||||
"allowFrom": ["*"],
|
||||
},
|
||||
bus,
|
||||
)
|
||||
|
||||
activity = {
|
||||
"type": "message",
|
||||
"id": "activity-3",
|
||||
"text": "<at>Nanobot</at>",
|
||||
"serviceUrl": "https://smba.trafficmanager.net/amer/",
|
||||
"conversation": {
|
||||
"id": "conv-empty",
|
||||
"conversationType": "personal",
|
||||
},
|
||||
"from": {
|
||||
"id": "29:user-id",
|
||||
"aadObjectId": "aad-user-1",
|
||||
"name": "Bob",
|
||||
},
|
||||
"recipient": {
|
||||
"id": "28:bot-id",
|
||||
"name": "nanobot",
|
||||
},
|
||||
}
|
||||
|
||||
await ch._handle_activity(activity)
|
||||
|
||||
assert len(bus.inbound) == 1
|
||||
assert bus.inbound[0].content == "Hi — what can I help with?"
|
||||
assert "conv-empty" in ch._conversation_refs
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_activity_mention_only_ignores_when_response_disabled(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr("nanobot.channels.msteams.get_workspace_path", lambda: tmp_path)
|
||||
|
||||
bus = DummyBus()
|
||||
ch = MSTeamsChannel(
|
||||
{
|
||||
"enabled": True,
|
||||
"appId": "app-id",
|
||||
"appPassword": "secret",
|
||||
"tenantId": "tenant-id",
|
||||
"allowFrom": ["*"],
|
||||
"mentionOnlyResponse": " ",
|
||||
},
|
||||
bus,
|
||||
)
|
||||
|
||||
activity = {
|
||||
"type": "message",
|
||||
"id": "activity-4",
|
||||
"text": "<at>Nanobot</at>",
|
||||
"serviceUrl": "https://smba.trafficmanager.net/amer/",
|
||||
"conversation": {
|
||||
"id": "conv-empty-disabled",
|
||||
"conversationType": "personal",
|
||||
},
|
||||
"from": {
|
||||
"id": "29:user-id",
|
||||
"aadObjectId": "aad-user-1",
|
||||
"name": "Bob",
|
||||
},
|
||||
"recipient": {
|
||||
"id": "28:bot-id",
|
||||
"name": "nanobot",
|
||||
},
|
||||
}
|
||||
|
||||
await ch._handle_activity(activity)
|
||||
|
||||
assert bus.inbound == []
|
||||
assert ch._conversation_refs == {}
|
||||
|
||||
|
||||
def test_strip_possible_bot_mention_removes_generic_at_tags(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr("nanobot.channels.msteams.get_workspace_path", lambda: tmp_path)
|
||||
|
||||
bus = DummyBus()
|
||||
ch = MSTeamsChannel(
|
||||
{
|
||||
"enabled": True,
|
||||
"appId": "app-id",
|
||||
"appPassword": "secret",
|
||||
"tenantId": "tenant-id",
|
||||
"allowFrom": ["*"],
|
||||
},
|
||||
bus,
|
||||
)
|
||||
|
||||
assert ch._strip_possible_bot_mention("<at>Nanobot</at> hello") == "hello"
|
||||
assert ch._strip_possible_bot_mention("hi <at>Some Bot</at> there") == "hi there"
|
||||
|
||||
|
||||
def test_sanitize_inbound_text_keeps_normal_inline_message(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr("nanobot.channels.msteams.get_workspace_path", lambda: tmp_path)
|
||||
|
||||
bus = DummyBus()
|
||||
ch = MSTeamsChannel(
|
||||
{
|
||||
"enabled": True,
|
||||
"appId": "app-id",
|
||||
"appPassword": "secret",
|
||||
"tenantId": "tenant-id",
|
||||
"allowFrom": ["*"],
|
||||
},
|
||||
bus,
|
||||
)
|
||||
|
||||
activity = {
|
||||
"text": "<at>Nanobot</at> normal inline message",
|
||||
"channelData": {},
|
||||
}
|
||||
|
||||
assert ch._sanitize_inbound_text(activity) == "normal inline message"
|
||||
|
||||
|
||||
def test_sanitize_inbound_text_normalizes_fwdioc_wrapper_without_reply_metadata(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr("nanobot.channels.msteams.get_workspace_path", lambda: tmp_path)
|
||||
|
||||
bus = DummyBus()
|
||||
ch = MSTeamsChannel(
|
||||
{
|
||||
"enabled": True,
|
||||
"appId": "app-id",
|
||||
"appPassword": "secret",
|
||||
"tenantId": "tenant-id",
|
||||
"allowFrom": ["*"],
|
||||
},
|
||||
bus,
|
||||
)
|
||||
|
||||
activity = {
|
||||
"text": "FWDIOC-BOT \r\nQuoted prior message\r\n\r\nThis is a reply with quote test",
|
||||
"channelData": {},
|
||||
}
|
||||
|
||||
assert ch._sanitize_inbound_text(activity) == (
|
||||
"User is replying to: Quoted prior message\n"
|
||||
"User reply: This is a reply with quote test"
|
||||
)
|
||||
|
||||
|
||||
def test_sanitize_inbound_text_structures_reply_quote_prefix(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr("nanobot.channels.msteams.get_workspace_path", lambda: tmp_path)
|
||||
|
||||
bus = DummyBus()
|
||||
ch = MSTeamsChannel(
|
||||
{
|
||||
"enabled": True,
|
||||
"appId": "app-id",
|
||||
"appPassword": "secret",
|
||||
"tenantId": "tenant-id",
|
||||
"allowFrom": ["*"],
|
||||
},
|
||||
bus,
|
||||
)
|
||||
|
||||
activity = {
|
||||
"text": "Replying to Bob Smith\nactual reply text",
|
||||
"replyToId": "parent-activity",
|
||||
"channelData": {"messageType": "reply"},
|
||||
}
|
||||
|
||||
assert ch._sanitize_inbound_text(activity) == "User is replying to: Bob Smith\nUser reply: actual reply text"
|
||||
|
||||
|
||||
def test_sanitize_inbound_text_structures_live_fwdioc_quote_shape(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr("nanobot.channels.msteams.get_workspace_path", lambda: tmp_path)
|
||||
|
||||
bus = DummyBus()
|
||||
ch = MSTeamsChannel(
|
||||
{
|
||||
"enabled": True,
|
||||
"appId": "app-id",
|
||||
"appPassword": "secret",
|
||||
"tenantId": "tenant-id",
|
||||
"allowFrom": ["*"],
|
||||
},
|
||||
bus,
|
||||
)
|
||||
|
||||
activity = {
|
||||
"text": "FWDIOC-BOT Got it. I’ll watch for the exact text reply with quote test and then inspect that turn specifically. Reply with quote test",
|
||||
"replyToId": "parent-activity",
|
||||
"channelData": {"messageType": "reply"},
|
||||
}
|
||||
|
||||
assert ch._sanitize_inbound_text(activity) == (
|
||||
"User is replying to: Got it. I’ll watch for the exact text reply with quote test and then inspect that turn specifically.\n"
|
||||
"User reply: Reply with quote test"
|
||||
)
|
||||
|
||||
|
||||
def test_sanitize_inbound_text_structures_multiline_fwdioc_quote_shape(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr("nanobot.channels.msteams.get_workspace_path", lambda: tmp_path)
|
||||
|
||||
bus = DummyBus()
|
||||
ch = MSTeamsChannel(
|
||||
{
|
||||
"enabled": True,
|
||||
"appId": "app-id",
|
||||
"appPassword": "secret",
|
||||
"tenantId": "tenant-id",
|
||||
"allowFrom": ["*"],
|
||||
},
|
||||
bus,
|
||||
)
|
||||
|
||||
activity = {
|
||||
"text": (
|
||||
"FWDIOC-BOT\r\n"
|
||||
"Understood — then the restart already happened, and the new Teams quote normalization should now be live. "
|
||||
"Next best step: • send one more real reply-with-quote message in Teams • I&rsquo…\r\n"
|
||||
"\r\n"
|
||||
"This is a reply with quote"
|
||||
),
|
||||
"replyToId": "parent-activity",
|
||||
"channelData": {"messageType": "reply"},
|
||||
}
|
||||
|
||||
assert ch._sanitize_inbound_text(activity) == (
|
||||
"User is replying to: Understood — then the restart already happened, and the new Teams quote normalization should now be live. "
|
||||
"Next best step: • send one more real reply-with-quote message in Teams • I’…\n"
|
||||
"User reply: This is a reply with quote"
|
||||
)
|
||||
|
||||
|
||||
def test_sanitize_inbound_text_structures_exact_live_crlf_fwdioc_shape(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr("nanobot.channels.msteams.get_workspace_path", lambda: tmp_path)
|
||||
|
||||
bus = DummyBus()
|
||||
ch = MSTeamsChannel(
|
||||
{
|
||||
"enabled": True,
|
||||
"appId": "app-id",
|
||||
"appPassword": "secret",
|
||||
"tenantId": "tenant-id",
|
||||
"allowFrom": ["*"],
|
||||
},
|
||||
bus,
|
||||
)
|
||||
|
||||
activity = {
|
||||
"text": (
|
||||
"FWDIOC-BOT \r\n"
|
||||
"Please send one real reply-with-quote message in Teams. That single test should be enough now: "
|
||||
"• I’ll check the new MSTeams sanitized inbound text ... log • and compare it to the prompt…\r\n"
|
||||
"\r\n"
|
||||
"This is a reply with quote test"
|
||||
),
|
||||
"replyToId": "parent-activity",
|
||||
"channelData": {"messageType": "reply"},
|
||||
}
|
||||
|
||||
assert ch._sanitize_inbound_text(activity) == (
|
||||
"User is replying to: Please send one real reply-with-quote message in Teams. That single test should be enough now: "
|
||||
"• I’ll check the new MSTeams sanitized inbound text ... log • and compare it to the prompt…\n"
|
||||
"User reply: This is a reply with quote test"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_access_token_uses_configured_tenant(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr("nanobot.channels.msteams.get_workspace_path", lambda: tmp_path)
|
||||
|
||||
bus = DummyBus()
|
||||
ch = MSTeamsChannel(
|
||||
{
|
||||
"enabled": True,
|
||||
"appId": "app-id",
|
||||
"appPassword": "secret",
|
||||
"tenantId": "tenant-123",
|
||||
"allowFrom": ["*"],
|
||||
},
|
||||
bus,
|
||||
)
|
||||
|
||||
fake_http = FakeHttpClient()
|
||||
ch._http = fake_http
|
||||
|
||||
token = await ch._get_access_token()
|
||||
|
||||
assert token == "tok"
|
||||
assert len(fake_http.calls) == 1
|
||||
url, kwargs = fake_http.calls[0]
|
||||
assert url == "https://login.microsoftonline.com/tenant-123/oauth2/v2.0/token"
|
||||
assert kwargs["data"]["client_id"] == "app-id"
|
||||
assert kwargs["data"]["client_secret"] == "secret"
|
||||
assert kwargs["data"]["scope"] == "https://api.botframework.com/.default"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_replies_to_activity_when_reply_in_thread_enabled(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr("nanobot.channels.msteams.get_workspace_path", lambda: tmp_path)
|
||||
|
||||
bus = DummyBus()
|
||||
ch = MSTeamsChannel(
|
||||
{
|
||||
"enabled": True,
|
||||
"appId": "app-id",
|
||||
"appPassword": "secret",
|
||||
"tenantId": "tenant-id",
|
||||
"allowFrom": ["*"],
|
||||
"replyInThread": True,
|
||||
},
|
||||
bus,
|
||||
)
|
||||
|
||||
fake_http = FakeHttpClient()
|
||||
ch._http = fake_http
|
||||
ch._token = "tok"
|
||||
ch._token_expires_at = 9999999999
|
||||
ch._conversation_refs["conv-123"] = ConversationRef(
|
||||
service_url="https://smba.trafficmanager.net/amer/",
|
||||
conversation_id="conv-123",
|
||||
activity_id="activity-1",
|
||||
)
|
||||
|
||||
await ch.send(OutboundMessage(channel="msteams", chat_id="conv-123", content="Reply text"))
|
||||
|
||||
assert len(fake_http.calls) == 1
|
||||
url, kwargs = fake_http.calls[0]
|
||||
assert url == "https://smba.trafficmanager.net/amer/v3/conversations/conv-123/activities/activity-1"
|
||||
assert kwargs["headers"]["Authorization"] == "Bearer tok"
|
||||
assert kwargs["json"]["text"] == "Reply text"
|
||||
assert kwargs["json"]["replyToId"] == "activity-1"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_posts_to_conversation_when_thread_reply_disabled(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr("nanobot.channels.msteams.get_workspace_path", lambda: tmp_path)
|
||||
|
||||
bus = DummyBus()
|
||||
ch = MSTeamsChannel(
|
||||
{
|
||||
"enabled": True,
|
||||
"appId": "app-id",
|
||||
"appPassword": "secret",
|
||||
"tenantId": "tenant-id",
|
||||
"allowFrom": ["*"],
|
||||
"replyInThread": False,
|
||||
},
|
||||
bus,
|
||||
)
|
||||
|
||||
fake_http = FakeHttpClient()
|
||||
ch._http = fake_http
|
||||
ch._token = "tok"
|
||||
ch._token_expires_at = 9999999999
|
||||
ch._conversation_refs["conv-123"] = ConversationRef(
|
||||
service_url="https://smba.trafficmanager.net/amer/",
|
||||
conversation_id="conv-123",
|
||||
activity_id="activity-1",
|
||||
)
|
||||
|
||||
await ch.send(OutboundMessage(channel="msteams", chat_id="conv-123", content="Reply text"))
|
||||
|
||||
assert len(fake_http.calls) == 1
|
||||
url, kwargs = fake_http.calls[0]
|
||||
assert url == "https://smba.trafficmanager.net/amer/v3/conversations/conv-123/activities"
|
||||
assert kwargs["headers"]["Authorization"] == "Bearer tok"
|
||||
assert kwargs["json"]["text"] == "Reply text"
|
||||
assert "replyToId" not in kwargs["json"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_posts_to_conversation_when_thread_reply_enabled_but_no_activity_id(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr("nanobot.channels.msteams.get_workspace_path", lambda: tmp_path)
|
||||
|
||||
bus = DummyBus()
|
||||
ch = MSTeamsChannel(
|
||||
{
|
||||
"enabled": True,
|
||||
"appId": "app-id",
|
||||
"appPassword": "secret",
|
||||
"tenantId": "tenant-id",
|
||||
"allowFrom": ["*"],
|
||||
"replyInThread": True,
|
||||
},
|
||||
bus,
|
||||
)
|
||||
|
||||
fake_http = FakeHttpClient()
|
||||
ch._http = fake_http
|
||||
ch._token = "tok"
|
||||
ch._token_expires_at = 9999999999
|
||||
ch._conversation_refs["conv-123"] = ConversationRef(
|
||||
service_url="https://smba.trafficmanager.net/amer/",
|
||||
conversation_id="conv-123",
|
||||
activity_id=None,
|
||||
)
|
||||
|
||||
await ch.send(OutboundMessage(channel="msteams", chat_id="conv-123", content="Reply text"))
|
||||
|
||||
assert len(fake_http.calls) == 1
|
||||
url, kwargs = fake_http.calls[0]
|
||||
assert url == "https://smba.trafficmanager.net/amer/v3/conversations/conv-123/activities"
|
||||
assert kwargs["headers"]["Authorization"] == "Bearer tok"
|
||||
assert kwargs["json"]["text"] == "Reply text"
|
||||
assert "replyToId" not in kwargs["json"]
|
||||
|
||||
|
||||
def _make_test_rsa_jwk(kid: str = "test-kid"):
|
||||
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
||||
public_key = private_key.public_key()
|
||||
jwk = json.loads(jwt.algorithms.RSAAlgorithm.to_jwk(public_key))
|
||||
jwk["kid"] = kid
|
||||
jwk["use"] = "sig"
|
||||
jwk["kty"] = "RSA"
|
||||
jwk["alg"] = "RS256"
|
||||
return private_key, jwk
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_validate_inbound_auth_accepts_observed_botframework_shape(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr("nanobot.channels.msteams.get_workspace_path", lambda: tmp_path)
|
||||
|
||||
bus = DummyBus()
|
||||
ch = MSTeamsChannel(
|
||||
{
|
||||
"enabled": True,
|
||||
"appId": "app-id",
|
||||
"appPassword": "secret",
|
||||
"tenantId": "tenant-id",
|
||||
"allowFrom": ["*"],
|
||||
"validateInboundAuth": True,
|
||||
},
|
||||
bus,
|
||||
)
|
||||
|
||||
private_key, jwk = _make_test_rsa_jwk()
|
||||
ch._botframework_jwks = {"keys": [jwk]}
|
||||
ch._botframework_jwks_expires_at = 9999999999
|
||||
|
||||
service_url = "https://smba.trafficmanager.net/amer/tenant/"
|
||||
token = jwt.encode(
|
||||
{
|
||||
"iss": "https://api.botframework.com",
|
||||
"aud": "app-id",
|
||||
"serviceurl": service_url,
|
||||
"nbf": 1700000000,
|
||||
"exp": 4100000000,
|
||||
},
|
||||
private_key,
|
||||
algorithm="RS256",
|
||||
headers={"kid": jwk["kid"]},
|
||||
)
|
||||
|
||||
await ch._validate_inbound_auth(
|
||||
f"Bearer {token}",
|
||||
{"serviceUrl": service_url},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_validate_inbound_auth_rejects_service_url_mismatch(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr("nanobot.channels.msteams.get_workspace_path", lambda: tmp_path)
|
||||
|
||||
bus = DummyBus()
|
||||
ch = MSTeamsChannel(
|
||||
{
|
||||
"enabled": True,
|
||||
"appId": "app-id",
|
||||
"appPassword": "secret",
|
||||
"tenantId": "tenant-id",
|
||||
"allowFrom": ["*"],
|
||||
"validateInboundAuth": True,
|
||||
},
|
||||
bus,
|
||||
)
|
||||
|
||||
private_key, jwk = _make_test_rsa_jwk()
|
||||
ch._botframework_jwks = {"keys": [jwk]}
|
||||
ch._botframework_jwks_expires_at = 9999999999
|
||||
|
||||
token = jwt.encode(
|
||||
{
|
||||
"iss": "https://api.botframework.com",
|
||||
"aud": "app-id",
|
||||
"serviceurl": "https://smba.trafficmanager.net/amer/tenant-a/",
|
||||
"nbf": 1700000000,
|
||||
"exp": 4100000000,
|
||||
},
|
||||
private_key,
|
||||
algorithm="RS256",
|
||||
headers={"kid": jwk["kid"]},
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="serviceUrl claim mismatch"):
|
||||
await ch._validate_inbound_auth(
|
||||
f"Bearer {token}",
|
||||
{"serviceUrl": "https://smba.trafficmanager.net/amer/tenant-b/"},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_validate_inbound_auth_rejects_missing_bearer_token(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr("nanobot.channels.msteams.get_workspace_path", lambda: tmp_path)
|
||||
|
||||
bus = DummyBus()
|
||||
ch = MSTeamsChannel(
|
||||
{
|
||||
"enabled": True,
|
||||
"appId": "app-id",
|
||||
"appPassword": "secret",
|
||||
"tenantId": "tenant-id",
|
||||
"allowFrom": ["*"],
|
||||
"validateInboundAuth": True,
|
||||
},
|
||||
bus,
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="missing bearer token"):
|
||||
await ch._validate_inbound_auth("", {"serviceUrl": "https://smba.trafficmanager.net/amer/tenant/"})
|
||||
|
||||
|
||||
def test_msteams_default_config_includes_restart_notify_fields():
|
||||
cfg = MSTeamsChannel.default_config()
|
||||
|
||||
assert cfg["restartNotifyEnabled"] is False
|
||||
assert "restartNotifyPreMessage" in cfg
|
||||
assert "restartNotifyPostMessage" in cfg
|
||||
|
||||
|
||||
def test_msteams_config_accepts_restart_notify_aliases():
|
||||
cfg = MSTeamsConfig.model_validate(
|
||||
{
|
||||
"restartNotifyEnabled": True,
|
||||
"restartNotifyPreMessage": "Restarting now.",
|
||||
"restartNotifyPostMessage": "Back online.",
|
||||
}
|
||||
)
|
||||
|
||||
assert cfg.restart_notify_enabled is True
|
||||
assert cfg.restart_notify_pre_message == "Restarting now."
|
||||
assert cfg.restart_notify_post_message == "Back online."
|
||||
Reference in New Issue
Block a user