Merge PR #3440: fix: Automatically clean up unsupported or expired MSTeams session
fix: Automatically clean up unsupported or expired MSTeams session
This commit is contained in:
+10
-2
@@ -644,7 +644,11 @@ Create or reuse a Microsoft Teams / Azure bot app registration. Set the bot mess
|
|||||||
"allowFrom": ["*"],
|
"allowFrom": ["*"],
|
||||||
"replyInThread": true,
|
"replyInThread": true,
|
||||||
"mentionOnlyResponse": "Hi — what can I help with?",
|
"mentionOnlyResponse": "Hi — what can I help with?",
|
||||||
"validateInboundAuth": true
|
"validateInboundAuth": true,
|
||||||
|
"refTtlDays": 30,
|
||||||
|
"pruneWebChatRefs": true,
|
||||||
|
"pruneNonPersonalRefs": true,
|
||||||
|
"refTouchIntervalS": 300
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -653,6 +657,10 @@ Create or reuse a Microsoft Teams / Azure bot app registration. Set the bot mess
|
|||||||
> - `replyInThread: true` replies to the triggering Teams activity when a stored `activity_id` is available.
|
> - `replyInThread: true` replies to the triggering Teams activity when a stored `activity_id` is available.
|
||||||
> - `mentionOnlyResponse` controls what Nanobot receives when a user sends only a bot mention (`<at>Nanobot</at>`). Set to `""` to ignore mention-only messages.
|
> - `mentionOnlyResponse` controls what Nanobot receives when a user sends only a bot mention (`<at>Nanobot</at>`). Set to `""` to ignore mention-only messages.
|
||||||
> - `validateInboundAuth: true` enables inbound Bot Framework bearer-token validation (signature, issuer, audience, lifetime, `serviceUrl`). This is the safe default for public deployments. Only set it to `false` for local development or tightly controlled testing.
|
> - `validateInboundAuth: true` enables inbound Bot Framework bearer-token validation (signature, issuer, audience, lifetime, `serviceUrl`). This is the safe default for public deployments. Only set it to `false` for local development or tightly controlled testing.
|
||||||
|
> - `refTtlDays` (default `30`) controls how old stored conversation refs can be before they are pruned.
|
||||||
|
> - `pruneWebChatRefs` (default `true`) drops refs with `webchat.botframework.com` service URLs.
|
||||||
|
> - `pruneNonPersonalRefs` (default `true`) drops refs whose `conversation_type` is not `personal`.
|
||||||
|
> - `refTouchIntervalS` (default `300`) throttles how often successful sends refresh `updated_at` for active refs.
|
||||||
|
|
||||||
**4. Run**
|
**4. Run**
|
||||||
|
|
||||||
@@ -660,4 +668,4 @@ Create or reuse a Microsoft Teams / Azure bot app registration. Set the bot mess
|
|||||||
nanobot gateway
|
nanobot gateway
|
||||||
```
|
```
|
||||||
|
|
||||||
</details>
|
</details>
|
||||||
|
|||||||
+265
-57
@@ -15,12 +15,21 @@ import asyncio
|
|||||||
import html
|
import html
|
||||||
import importlib.util
|
import importlib.util
|
||||||
import json
|
import json
|
||||||
|
import os
|
||||||
import re
|
import re
|
||||||
|
import tempfile
|
||||||
import threading
|
import threading
|
||||||
import time
|
import time
|
||||||
|
from contextlib import contextmanager
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||||
from typing import TYPE_CHECKING, Any
|
from typing import TYPE_CHECKING, Any
|
||||||
|
from urllib.parse import urlparse
|
||||||
|
|
||||||
|
try: # pragma: no cover - Windows fallback path
|
||||||
|
import fcntl
|
||||||
|
except ImportError: # pragma: no cover
|
||||||
|
fcntl = None
|
||||||
|
|
||||||
import httpx
|
import httpx
|
||||||
from loguru import logger
|
from loguru import logger
|
||||||
@@ -43,6 +52,13 @@ if TYPE_CHECKING:
|
|||||||
if MSTEAMS_AVAILABLE:
|
if MSTEAMS_AVAILABLE:
|
||||||
import jwt
|
import jwt
|
||||||
|
|
||||||
|
MSTEAMS_REF_TTL_DAYS = 30
|
||||||
|
MSTEAMS_REF_TTL_S = MSTEAMS_REF_TTL_DAYS * 24 * 60 * 60
|
||||||
|
MSTEAMS_WEBCHAT_HOST = "webchat.botframework.com"
|
||||||
|
MSTEAMS_REF_META_FILENAME = "msteams_conversations_meta.json"
|
||||||
|
MSTEAMS_REF_LOCK_FILENAME = "msteams_conversations.lock"
|
||||||
|
MSTEAMS_REF_TOUCH_INTERVAL_S = 300
|
||||||
|
|
||||||
|
|
||||||
class MSTeamsConfig(Base):
|
class MSTeamsConfig(Base):
|
||||||
"""Microsoft Teams channel configuration."""
|
"""Microsoft Teams channel configuration."""
|
||||||
@@ -58,6 +74,10 @@ class MSTeamsConfig(Base):
|
|||||||
reply_in_thread: bool = True
|
reply_in_thread: bool = True
|
||||||
mention_only_response: str = "Hi — what can I help with?"
|
mention_only_response: str = "Hi — what can I help with?"
|
||||||
validate_inbound_auth: bool = True
|
validate_inbound_auth: bool = True
|
||||||
|
ref_ttl_days: int = Field(default=MSTEAMS_REF_TTL_DAYS, ge=1)
|
||||||
|
prune_web_chat_refs: bool = True
|
||||||
|
prune_non_personal_refs: bool = True
|
||||||
|
ref_touch_interval_s: int = Field(default=MSTEAMS_REF_TOUCH_INTERVAL_S, ge=0)
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
@@ -103,7 +123,13 @@ class MSTeamsChannel(BaseChannel):
|
|||||||
self._botframework_jwks_expires_at: float = 0.0
|
self._botframework_jwks_expires_at: float = 0.0
|
||||||
self._refs_path = get_workspace_path() / "state" / "msteams_conversations.json"
|
self._refs_path = get_workspace_path() / "state" / "msteams_conversations.json"
|
||||||
self._refs_path.parent.mkdir(parents=True, exist_ok=True)
|
self._refs_path.parent.mkdir(parents=True, exist_ok=True)
|
||||||
|
self._refs_meta_path = self._refs_path.parent / MSTEAMS_REF_META_FILENAME
|
||||||
|
self._refs_lock_path = self._refs_path.parent / MSTEAMS_REF_LOCK_FILENAME
|
||||||
|
self._refs_guard = threading.RLock()
|
||||||
self._conversation_refs: dict[str, ConversationRef] = self._load_refs()
|
self._conversation_refs: dict[str, ConversationRef] = self._load_refs()
|
||||||
|
with self._refs_guard:
|
||||||
|
if self._prune_conversation_refs():
|
||||||
|
self._save_refs_locked(prune=True)
|
||||||
|
|
||||||
async def start(self) -> None:
|
async def start(self) -> None:
|
||||||
"""Start the Teams webhook listener."""
|
"""Start the Teams webhook listener."""
|
||||||
@@ -236,6 +262,7 @@ class MSTeamsChannel(BaseChannel):
|
|||||||
resp = await self._http.post(base_url, headers=headers, json=payload)
|
resp = await self._http.post(base_url, headers=headers, json=payload)
|
||||||
resp.raise_for_status()
|
resp.raise_for_status()
|
||||||
logger.info("MSTeams message sent to {}", ref.conversation_id)
|
logger.info("MSTeams message sent to {}", ref.conversation_id)
|
||||||
|
self._touch_conversation_ref(str(msg.chat_id), persist=True)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error("MSTeams send failed: {}", e)
|
logger.error("MSTeams send failed: {}", e)
|
||||||
raise
|
raise
|
||||||
@@ -282,17 +309,17 @@ class MSTeamsChannel(BaseChannel):
|
|||||||
)
|
)
|
||||||
return
|
return
|
||||||
|
|
||||||
self._conversation_refs[conversation_id] = ConversationRef(
|
with self._refs_guard:
|
||||||
service_url=service_url,
|
self._conversation_refs[conversation_id] = ConversationRef(
|
||||||
conversation_id=conversation_id,
|
service_url=service_url,
|
||||||
bot_id=str(recipient.get("id") or "") or None,
|
conversation_id=conversation_id,
|
||||||
activity_id=activity_id or None,
|
bot_id=str(recipient.get("id") or "") or None,
|
||||||
conversation_type=conversation_type or None,
|
activity_id=activity_id or None,
|
||||||
tenant_id=str((channel_data.get("tenant") or {}).get("id") or "") or None,
|
conversation_type=conversation_type or None,
|
||||||
updated_at=time.time(),
|
tenant_id=str((channel_data.get("tenant") or {}).get("id") or "") or None,
|
||||||
)
|
updated_at=time.time(),
|
||||||
|
)
|
||||||
self._save_refs()
|
self._save_refs_locked()
|
||||||
|
|
||||||
await self._handle_message(
|
await self._handle_message(
|
||||||
sender_id=sender_id,
|
sender_id=sender_id,
|
||||||
@@ -487,61 +514,242 @@ class MSTeamsChannel(BaseChannel):
|
|||||||
self._botframework_jwks_expires_at = now + 3600
|
self._botframework_jwks_expires_at = now + 3600
|
||||||
return self._botframework_jwks
|
return self._botframework_jwks
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _safe_float(value: Any) -> float | None:
|
||||||
|
try:
|
||||||
|
out = float(value)
|
||||||
|
if out > 0:
|
||||||
|
return out
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
return None
|
||||||
|
return None
|
||||||
|
|
||||||
|
def _normalize_ref_record(self, value: Any) -> ConversationRef | None:
|
||||||
|
"""Normalize a stored ref record from legacy/current schema."""
|
||||||
|
if not isinstance(value, dict):
|
||||||
|
return None
|
||||||
|
service_url = str(value.get("service_url") or "").strip()
|
||||||
|
conversation_id = str(value.get("conversation_id") or "").strip()
|
||||||
|
if not service_url or not conversation_id:
|
||||||
|
return None
|
||||||
|
return ConversationRef(
|
||||||
|
service_url=service_url,
|
||||||
|
conversation_id=conversation_id,
|
||||||
|
bot_id=str(value.get("bot_id") or "") or None,
|
||||||
|
activity_id=str(value.get("activity_id") or "") or None,
|
||||||
|
conversation_type=str(value.get("conversation_type") or "") or None,
|
||||||
|
tenant_id=str(value.get("tenant_id") or "") or None,
|
||||||
|
updated_at=self._safe_float(value.get("updated_at")),
|
||||||
|
)
|
||||||
|
|
||||||
|
def _load_refs_raw(self) -> tuple[dict[str, Any], dict[str, Any], bool]:
|
||||||
|
"""Load raw refs/main+meta JSON payloads."""
|
||||||
|
main_data: dict[str, Any] = {}
|
||||||
|
meta_data: dict[str, Any] = {}
|
||||||
|
meta_exists = self._refs_meta_path.exists()
|
||||||
|
|
||||||
|
if self._refs_path.exists():
|
||||||
|
try:
|
||||||
|
loaded = json.loads(self._refs_path.read_text(encoding="utf-8"))
|
||||||
|
if isinstance(loaded, dict):
|
||||||
|
main_data = loaded
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning("Failed to load MSTeams conversation refs: {}", e)
|
||||||
|
|
||||||
|
if meta_exists:
|
||||||
|
try:
|
||||||
|
loaded_meta = json.loads(self._refs_meta_path.read_text(encoding="utf-8"))
|
||||||
|
if isinstance(loaded_meta, dict):
|
||||||
|
meta_data = loaded_meta
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning("Failed to load MSTeams conversation refs metadata: {}", e)
|
||||||
|
|
||||||
|
return main_data, meta_data, meta_exists
|
||||||
|
|
||||||
|
def _load_refs_from_disk(self) -> dict[str, ConversationRef]:
|
||||||
|
"""Load refs from disk with compatibility fallback for legacy layouts."""
|
||||||
|
main_data, meta_data, meta_exists = self._load_refs_raw()
|
||||||
|
if not main_data:
|
||||||
|
return {}
|
||||||
|
|
||||||
|
out: dict[str, ConversationRef] = {}
|
||||||
|
now = time.time()
|
||||||
|
for key, value in main_data.items():
|
||||||
|
ref = self._normalize_ref_record(value)
|
||||||
|
if not ref:
|
||||||
|
continue
|
||||||
|
|
||||||
|
meta_entry = meta_data.get(key) if isinstance(meta_data, dict) else None
|
||||||
|
meta_ts = None
|
||||||
|
if isinstance(meta_entry, dict):
|
||||||
|
meta_ts = self._safe_float(meta_entry.get("updated_at"))
|
||||||
|
elif meta_entry is not None:
|
||||||
|
meta_ts = self._safe_float(meta_entry)
|
||||||
|
|
||||||
|
if meta_ts is not None:
|
||||||
|
ref.updated_at = meta_ts
|
||||||
|
elif not meta_exists:
|
||||||
|
# First run after introducing meta sidecar: keep legacy refs alive
|
||||||
|
# by initializing timestamps to "now" instead of purging immediately.
|
||||||
|
ref.updated_at = now
|
||||||
|
elif ref.updated_at is None:
|
||||||
|
ref.updated_at = now
|
||||||
|
|
||||||
|
out[key] = ref
|
||||||
|
return out
|
||||||
|
|
||||||
def _load_refs(self) -> dict[str, ConversationRef]:
|
def _load_refs(self) -> dict[str, ConversationRef]:
|
||||||
"""Load stored conversation references."""
|
"""Load stored conversation references."""
|
||||||
if not self._refs_path.exists():
|
return self._load_refs_from_disk()
|
||||||
return {}
|
|
||||||
try:
|
|
||||||
data = json.loads(self._refs_path.read_text(encoding="utf-8"))
|
|
||||||
out: dict[str, ConversationRef] = {}
|
|
||||||
for key, value in data.items():
|
|
||||||
out[key] = ConversationRef(**value)
|
|
||||||
return out
|
|
||||||
except Exception as e:
|
|
||||||
logger.warning("Failed to load MSTeams conversation refs: {}", e)
|
|
||||||
return {}
|
|
||||||
|
|
||||||
def _save_refs(self) -> None:
|
@contextmanager
|
||||||
"""Persist conversation references."""
|
def _refs_file_lock(self):
|
||||||
|
"""Cross-process lock while merging and writing refs state."""
|
||||||
|
self._refs_path.parent.mkdir(parents=True, exist_ok=True)
|
||||||
|
lock_fp = self._refs_lock_path.open("a+", encoding="utf-8")
|
||||||
try:
|
try:
|
||||||
stale_keys = [
|
if fcntl is not None:
|
||||||
key
|
fcntl.flock(lock_fp.fileno(), fcntl.LOCK_EX)
|
||||||
for key, ref in self._conversation_refs.items()
|
yield
|
||||||
if self._is_stale_or_unsupported_ref(ref)
|
finally:
|
||||||
]
|
try:
|
||||||
for key in stale_keys:
|
if fcntl is not None:
|
||||||
self._conversation_refs.pop(key, None)
|
fcntl.flock(lock_fp.fileno(), fcntl.LOCK_UN)
|
||||||
|
finally:
|
||||||
|
lock_fp.close()
|
||||||
|
|
||||||
data = {
|
def _is_webchat_service_url(self, service_url: str) -> bool:
|
||||||
key: {
|
"""Return True when service URL points to unsupported Bot Framework Web Chat."""
|
||||||
"service_url": ref.service_url,
|
normalized = service_url.strip()
|
||||||
"conversation_id": ref.conversation_id,
|
if not normalized:
|
||||||
"bot_id": ref.bot_id,
|
return False
|
||||||
"activity_id": ref.activity_id,
|
host = (urlparse(normalized).hostname or "").strip().lower()
|
||||||
"conversation_type": ref.conversation_type,
|
if host:
|
||||||
"tenant_id": ref.tenant_id,
|
return host == MSTEAMS_WEBCHAT_HOST or host.endswith(f".{MSTEAMS_WEBCHAT_HOST}")
|
||||||
"updated_at": ref.updated_at,
|
return MSTEAMS_WEBCHAT_HOST in normalized.lower()
|
||||||
|
|
||||||
|
def _prune_conversation_refs(self, *, now: float | None = None) -> bool:
|
||||||
|
"""Remove stale and unsupported conversation refs from memory."""
|
||||||
|
if not self._conversation_refs:
|
||||||
|
return False
|
||||||
|
|
||||||
|
now_ts = time.time() if now is None else now
|
||||||
|
ttl_days = int(self.config.ref_ttl_days)
|
||||||
|
stale_before = now_ts - (ttl_days * 24 * 60 * 60)
|
||||||
|
keys_to_drop: list[str] = []
|
||||||
|
|
||||||
|
for key, ref in self._conversation_refs.items():
|
||||||
|
if self.config.prune_web_chat_refs and self._is_webchat_service_url(ref.service_url):
|
||||||
|
keys_to_drop.append(key)
|
||||||
|
continue
|
||||||
|
|
||||||
|
conv_type = str(ref.conversation_type or "").strip().lower()
|
||||||
|
if self.config.prune_non_personal_refs and conv_type and conv_type != "personal":
|
||||||
|
keys_to_drop.append(key)
|
||||||
|
continue
|
||||||
|
|
||||||
|
try:
|
||||||
|
updated_at = float(ref.updated_at) if ref.updated_at is not None else 0.0
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
updated_at = 0.0
|
||||||
|
if updated_at <= 0 or updated_at < stale_before:
|
||||||
|
keys_to_drop.append(key)
|
||||||
|
|
||||||
|
if not keys_to_drop:
|
||||||
|
return False
|
||||||
|
|
||||||
|
for key in keys_to_drop:
|
||||||
|
self._conversation_refs.pop(key, None)
|
||||||
|
logger.info(
|
||||||
|
"MSTeams pruned {} stale/unsupported conversation refs (ttl={} days)",
|
||||||
|
len(keys_to_drop),
|
||||||
|
ttl_days,
|
||||||
|
)
|
||||||
|
return True
|
||||||
|
|
||||||
|
def _merge_refs_from_disk_locked(self) -> None:
|
||||||
|
"""Merge disk refs into memory to reduce lost updates across processes."""
|
||||||
|
disk_refs = self._load_refs_from_disk()
|
||||||
|
for key, disk_ref in disk_refs.items():
|
||||||
|
mem_ref = self._conversation_refs.get(key)
|
||||||
|
if mem_ref is None:
|
||||||
|
self._conversation_refs[key] = disk_ref
|
||||||
|
continue
|
||||||
|
disk_ts = self._safe_float(disk_ref.updated_at) or 0.0
|
||||||
|
mem_ts = self._safe_float(mem_ref.updated_at) or 0.0
|
||||||
|
if disk_ts > mem_ts:
|
||||||
|
self._conversation_refs[key] = disk_ref
|
||||||
|
|
||||||
|
def _touch_conversation_ref(self, chat_id: str, *, persist: bool = False) -> None:
|
||||||
|
"""Refresh updated_at for an active ref to keep it from expiring while used."""
|
||||||
|
with self._refs_guard:
|
||||||
|
ref = self._conversation_refs.get(str(chat_id))
|
||||||
|
if not ref:
|
||||||
|
return
|
||||||
|
now = time.time()
|
||||||
|
prev = self._safe_float(ref.updated_at) or 0.0
|
||||||
|
min_interval = max(0, int(self.config.ref_touch_interval_s))
|
||||||
|
if min_interval > 0 and prev > 0 and now - prev < min_interval:
|
||||||
|
return
|
||||||
|
ref.updated_at = now
|
||||||
|
if persist:
|
||||||
|
self._save_refs_locked()
|
||||||
|
|
||||||
|
def _write_json_atomically(self, path, data: dict[str, Any]) -> None:
|
||||||
|
"""Write refs JSON atomically to reduce corruption risk during crashes."""
|
||||||
|
payload = json.dumps(data, indent=2)
|
||||||
|
tmp_path: str | None = None
|
||||||
|
try:
|
||||||
|
fd, tmp_path = tempfile.mkstemp(
|
||||||
|
dir=str(path.parent),
|
||||||
|
prefix=f"{path.name}.",
|
||||||
|
suffix=".tmp",
|
||||||
|
)
|
||||||
|
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
||||||
|
f.write(payload)
|
||||||
|
f.flush()
|
||||||
|
os.fsync(f.fileno())
|
||||||
|
os.replace(tmp_path, path)
|
||||||
|
finally:
|
||||||
|
if tmp_path and os.path.exists(tmp_path):
|
||||||
|
try:
|
||||||
|
os.unlink(tmp_path)
|
||||||
|
except OSError:
|
||||||
|
pass
|
||||||
|
|
||||||
|
def _save_refs_locked(self, *, prune: bool = True) -> None:
|
||||||
|
"""Persist conversation references (caller must hold _refs_guard)."""
|
||||||
|
try:
|
||||||
|
with self._refs_file_lock():
|
||||||
|
self._merge_refs_from_disk_locked()
|
||||||
|
if prune:
|
||||||
|
self._prune_conversation_refs()
|
||||||
|
refs_data = {
|
||||||
|
key: {
|
||||||
|
"service_url": ref.service_url,
|
||||||
|
"conversation_id": ref.conversation_id,
|
||||||
|
"bot_id": ref.bot_id,
|
||||||
|
"activity_id": ref.activity_id,
|
||||||
|
"conversation_type": ref.conversation_type,
|
||||||
|
"tenant_id": ref.tenant_id,
|
||||||
|
}
|
||||||
|
for key, ref in self._conversation_refs.items()
|
||||||
}
|
}
|
||||||
for key, ref in self._conversation_refs.items()
|
refs_meta = {
|
||||||
}
|
key: {
|
||||||
self._refs_path.write_text(json.dumps(data, indent=2), encoding="utf-8")
|
"updated_at": self._safe_float(ref.updated_at),
|
||||||
|
}
|
||||||
|
for key, ref in self._conversation_refs.items()
|
||||||
|
}
|
||||||
|
self._write_json_atomically(self._refs_path, refs_data)
|
||||||
|
self._write_json_atomically(self._refs_meta_path, refs_meta)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.warning("Failed to save MSTeams conversation refs: {}", e)
|
logger.warning("Failed to save MSTeams conversation refs: {}", e)
|
||||||
|
|
||||||
def _is_stale_or_unsupported_ref(self, ref: ConversationRef) -> bool:
|
def _save_refs(self, *, prune: bool = True) -> None:
|
||||||
"""Reject unsupported refs and prune old refs."""
|
"""Persist conversation references."""
|
||||||
service_url = (ref.service_url or "").strip().lower()
|
with self._refs_guard:
|
||||||
conversation_type = (ref.conversation_type or "").strip().lower()
|
self._save_refs_locked(prune=prune)
|
||||||
updated_at = ref.updated_at or 0.0
|
|
||||||
max_age_seconds = 30 * 24 * 60 * 60
|
|
||||||
|
|
||||||
if "webchat.botframework.com" in service_url:
|
|
||||||
return True
|
|
||||||
if conversation_type and conversation_type != "personal":
|
|
||||||
return True
|
|
||||||
if updated_at and updated_at < time.time() - max_age_seconds:
|
|
||||||
return True
|
|
||||||
return False
|
|
||||||
|
|
||||||
async def _get_access_token(self) -> str:
|
async def _get_access_token(self) -> str:
|
||||||
"""Fetch an access token for Bot Framework / Azure Bot auth."""
|
"""Fetch an access token for Bot Framework / Azure Bot auth."""
|
||||||
|
|||||||
+286
-4
@@ -18,7 +18,7 @@ from cryptography.hazmat.primitives.asymmetric import rsa
|
|||||||
|
|
||||||
import nanobot.channels.msteams as msteams_module
|
import nanobot.channels.msteams as msteams_module
|
||||||
from nanobot.bus.events import OutboundMessage
|
from nanobot.bus.events import OutboundMessage
|
||||||
from nanobot.channels.msteams import ConversationRef, MSTeamsChannel, MSTeamsConfig
|
from nanobot.channels.msteams import ConversationRef, MSTeamsChannel
|
||||||
|
|
||||||
|
|
||||||
class DummyBus:
|
class DummyBus:
|
||||||
@@ -116,6 +116,258 @@ async def test_handle_activity_personal_message_publishes_and_stores_ref(make_ch
|
|||||||
saved = json.loads((tmp_path / "state" / "msteams_conversations.json").read_text(encoding="utf-8"))
|
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"]["conversation_id"] == "conv-123"
|
||||||
assert saved["conv-123"]["tenant_id"] == "tenant-id"
|
assert saved["conv-123"]["tenant_id"] == "tenant-id"
|
||||||
|
saved_meta = json.loads(
|
||||||
|
(tmp_path / "state" / msteams_module.MSTEAMS_REF_META_FILENAME).read_text(encoding="utf-8"),
|
||||||
|
)
|
||||||
|
assert float(saved_meta["conv-123"]["updated_at"]) > 0
|
||||||
|
|
||||||
|
|
||||||
|
def test_init_prunes_stale_and_unsupported_conversation_refs(make_channel, tmp_path, monkeypatch):
|
||||||
|
now = 1_800_000_000.0
|
||||||
|
monkeypatch.setattr(msteams_module.time, "time", lambda: now)
|
||||||
|
|
||||||
|
state_dir = tmp_path / "state"
|
||||||
|
state_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
refs_path = state_dir / "msteams_conversations.json"
|
||||||
|
refs_meta_path = state_dir / msteams_module.MSTEAMS_REF_META_FILENAME
|
||||||
|
refs_path.write_text(
|
||||||
|
json.dumps(
|
||||||
|
{
|
||||||
|
"conv-valid": {
|
||||||
|
"service_url": "https://smba.trafficmanager.net/amer/",
|
||||||
|
"conversation_id": "conv-valid",
|
||||||
|
"conversation_type": "personal",
|
||||||
|
},
|
||||||
|
"conv-webchat": {
|
||||||
|
"service_url": "https://webchat.botframework.com/",
|
||||||
|
"conversation_id": "conv-webchat",
|
||||||
|
"conversation_type": "personal",
|
||||||
|
},
|
||||||
|
"conv-group": {
|
||||||
|
"service_url": "https://smba.trafficmanager.net/amer/",
|
||||||
|
"conversation_id": "conv-group",
|
||||||
|
"conversation_type": "channel",
|
||||||
|
},
|
||||||
|
"conv-stale": {
|
||||||
|
"service_url": "https://smba.trafficmanager.net/amer/",
|
||||||
|
"conversation_id": "conv-stale",
|
||||||
|
"conversation_type": "personal",
|
||||||
|
},
|
||||||
|
"conv-missing-ts": {
|
||||||
|
"service_url": "https://smba.trafficmanager.net/amer/",
|
||||||
|
"conversation_id": "conv-missing-ts",
|
||||||
|
"conversation_type": "personal",
|
||||||
|
},
|
||||||
|
},
|
||||||
|
indent=2,
|
||||||
|
),
|
||||||
|
encoding="utf-8",
|
||||||
|
)
|
||||||
|
refs_meta_path.write_text(
|
||||||
|
json.dumps(
|
||||||
|
{
|
||||||
|
"conv-valid": {"updated_at": now - 60},
|
||||||
|
"conv-webchat": {"updated_at": now - 60},
|
||||||
|
"conv-group": {"updated_at": now - 60},
|
||||||
|
"conv-stale": {"updated_at": now - msteams_module.MSTEAMS_REF_TTL_S - 1},
|
||||||
|
},
|
||||||
|
indent=2,
|
||||||
|
),
|
||||||
|
encoding="utf-8",
|
||||||
|
)
|
||||||
|
|
||||||
|
ch = make_channel()
|
||||||
|
|
||||||
|
assert set(ch._conversation_refs.keys()) == {"conv-valid", "conv-missing-ts"}
|
||||||
|
assert ch._conversation_refs["conv-valid"].conversation_id == "conv-valid"
|
||||||
|
assert ch._conversation_refs["conv-missing-ts"].conversation_id == "conv-missing-ts"
|
||||||
|
|
||||||
|
persisted = json.loads(refs_path.read_text(encoding="utf-8"))
|
||||||
|
assert set(persisted.keys()) == {"conv-valid", "conv-missing-ts"}
|
||||||
|
|
||||||
|
|
||||||
|
def test_save_prunes_unsupported_conversation_refs(make_channel, tmp_path, monkeypatch):
|
||||||
|
now = 1_800_000_000.0
|
||||||
|
monkeypatch.setattr(msteams_module.time, "time", lambda: now)
|
||||||
|
|
||||||
|
ch = make_channel()
|
||||||
|
ch._conversation_refs = {
|
||||||
|
"conv-valid": ConversationRef(
|
||||||
|
service_url="https://smba.trafficmanager.net/amer/",
|
||||||
|
conversation_id="conv-valid",
|
||||||
|
conversation_type="personal",
|
||||||
|
updated_at=now,
|
||||||
|
),
|
||||||
|
"conv-webchat": ConversationRef(
|
||||||
|
service_url="https://webchat.botframework.com/",
|
||||||
|
conversation_id="conv-webchat",
|
||||||
|
conversation_type="personal",
|
||||||
|
updated_at=now,
|
||||||
|
),
|
||||||
|
"conv-group": ConversationRef(
|
||||||
|
service_url="https://smba.trafficmanager.net/amer/",
|
||||||
|
conversation_id="conv-group",
|
||||||
|
conversation_type="groupChat",
|
||||||
|
updated_at=now,
|
||||||
|
),
|
||||||
|
}
|
||||||
|
|
||||||
|
ch._save_refs()
|
||||||
|
|
||||||
|
assert set(ch._conversation_refs.keys()) == {"conv-valid"}
|
||||||
|
|
||||||
|
saved = json.loads((tmp_path / "state" / "msteams_conversations.json").read_text(encoding="utf-8"))
|
||||||
|
assert set(saved.keys()) == {"conv-valid"}
|
||||||
|
saved_meta = json.loads(
|
||||||
|
(tmp_path / "state" / msteams_module.MSTEAMS_REF_META_FILENAME).read_text(encoding="utf-8"),
|
||||||
|
)
|
||||||
|
assert set(saved_meta.keys()) == {"conv-valid"}
|
||||||
|
|
||||||
|
|
||||||
|
def test_init_respects_prune_toggle_flags(make_channel, tmp_path, monkeypatch):
|
||||||
|
now = 1_800_000_000.0
|
||||||
|
monkeypatch.setattr(msteams_module.time, "time", lambda: now)
|
||||||
|
|
||||||
|
state_dir = tmp_path / "state"
|
||||||
|
state_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
refs_path = state_dir / "msteams_conversations.json"
|
||||||
|
refs_path.write_text(
|
||||||
|
json.dumps(
|
||||||
|
{
|
||||||
|
"conv-webchat": {
|
||||||
|
"service_url": "https://webchat.botframework.com/",
|
||||||
|
"conversation_id": "conv-webchat",
|
||||||
|
"conversation_type": "personal",
|
||||||
|
"updated_at": now - 60,
|
||||||
|
},
|
||||||
|
"conv-group": {
|
||||||
|
"service_url": "https://smba.trafficmanager.net/amer/",
|
||||||
|
"conversation_id": "conv-group",
|
||||||
|
"conversation_type": "channel",
|
||||||
|
"updated_at": now - 60,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
indent=2,
|
||||||
|
),
|
||||||
|
encoding="utf-8",
|
||||||
|
)
|
||||||
|
|
||||||
|
ch = make_channel(pruneWebChatRefs=False, pruneNonPersonalRefs=False)
|
||||||
|
|
||||||
|
assert set(ch._conversation_refs.keys()) == {"conv-webchat", "conv-group"}
|
||||||
|
persisted = json.loads(refs_path.read_text(encoding="utf-8"))
|
||||||
|
assert set(persisted.keys()) == {"conv-webchat", "conv-group"}
|
||||||
|
|
||||||
|
|
||||||
|
def test_init_respects_custom_ref_ttl_days(make_channel, tmp_path, monkeypatch):
|
||||||
|
now = 1_800_000_000.0
|
||||||
|
monkeypatch.setattr(msteams_module.time, "time", lambda: now)
|
||||||
|
|
||||||
|
state_dir = tmp_path / "state"
|
||||||
|
state_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
refs_path = state_dir / "msteams_conversations.json"
|
||||||
|
refs_meta_path = state_dir / msteams_module.MSTEAMS_REF_META_FILENAME
|
||||||
|
refs_path.write_text(
|
||||||
|
json.dumps(
|
||||||
|
{
|
||||||
|
"conv-fresh": {
|
||||||
|
"service_url": "https://smba.trafficmanager.net/amer/",
|
||||||
|
"conversation_id": "conv-fresh",
|
||||||
|
"conversation_type": "personal",
|
||||||
|
},
|
||||||
|
"conv-old": {
|
||||||
|
"service_url": "https://smba.trafficmanager.net/amer/",
|
||||||
|
"conversation_id": "conv-old",
|
||||||
|
"conversation_type": "personal",
|
||||||
|
},
|
||||||
|
},
|
||||||
|
indent=2,
|
||||||
|
),
|
||||||
|
encoding="utf-8",
|
||||||
|
)
|
||||||
|
refs_meta_path.write_text(
|
||||||
|
json.dumps(
|
||||||
|
{
|
||||||
|
"conv-fresh": {"updated_at": now - 12 * 60 * 60},
|
||||||
|
"conv-old": {"updated_at": now - 10 * 24 * 60 * 60},
|
||||||
|
},
|
||||||
|
indent=2,
|
||||||
|
),
|
||||||
|
encoding="utf-8",
|
||||||
|
)
|
||||||
|
|
||||||
|
ch = make_channel(refTtlDays=1)
|
||||||
|
|
||||||
|
assert set(ch._conversation_refs.keys()) == {"conv-fresh"}
|
||||||
|
persisted = json.loads(refs_path.read_text(encoding="utf-8"))
|
||||||
|
assert set(persisted.keys()) == {"conv-fresh"}
|
||||||
|
|
||||||
|
|
||||||
|
def test_init_without_meta_keeps_legacy_refs_alive(make_channel, tmp_path, monkeypatch):
|
||||||
|
now = 1_800_000_000.0
|
||||||
|
monkeypatch.setattr(msteams_module.time, "time", lambda: now)
|
||||||
|
|
||||||
|
state_dir = tmp_path / "state"
|
||||||
|
state_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
refs_path = state_dir / "msteams_conversations.json"
|
||||||
|
refs_path.write_text(
|
||||||
|
json.dumps(
|
||||||
|
{
|
||||||
|
"conv-legacy": {
|
||||||
|
"service_url": "https://smba.trafficmanager.net/amer/",
|
||||||
|
"conversation_id": "conv-legacy",
|
||||||
|
"conversation_type": "personal",
|
||||||
|
}
|
||||||
|
},
|
||||||
|
indent=2,
|
||||||
|
),
|
||||||
|
encoding="utf-8",
|
||||||
|
)
|
||||||
|
|
||||||
|
ch = make_channel(refTtlDays=1)
|
||||||
|
|
||||||
|
assert set(ch._conversation_refs.keys()) == {"conv-legacy"}
|
||||||
|
assert ch._conversation_refs["conv-legacy"].updated_at == now
|
||||||
|
assert not (state_dir / msteams_module.MSTEAMS_REF_META_FILENAME).exists()
|
||||||
|
|
||||||
|
|
||||||
|
def test_save_uses_atomic_replace_and_keeps_existing_file_on_replace_error(make_channel, tmp_path, monkeypatch):
|
||||||
|
ch = make_channel()
|
||||||
|
refs_path = tmp_path / "state" / "msteams_conversations.json"
|
||||||
|
refs_path.write_text(
|
||||||
|
json.dumps(
|
||||||
|
{
|
||||||
|
"conv-old": {
|
||||||
|
"service_url": "https://smba.trafficmanager.net/amer/",
|
||||||
|
"conversation_id": "conv-old",
|
||||||
|
"conversation_type": "personal",
|
||||||
|
"updated_at": 1_700_000_000.0,
|
||||||
|
}
|
||||||
|
},
|
||||||
|
indent=2,
|
||||||
|
),
|
||||||
|
encoding="utf-8",
|
||||||
|
)
|
||||||
|
|
||||||
|
ch._conversation_refs = {
|
||||||
|
"conv-new": ConversationRef(
|
||||||
|
service_url="https://smba.trafficmanager.net/amer/",
|
||||||
|
conversation_id="conv-new",
|
||||||
|
conversation_type="personal",
|
||||||
|
updated_at=1_800_000_000.0,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
def _raise_replace(_src, _dst):
|
||||||
|
raise OSError("replace failed")
|
||||||
|
|
||||||
|
monkeypatch.setattr(msteams_module.os, "replace", _raise_replace)
|
||||||
|
ch._save_refs()
|
||||||
|
|
||||||
|
persisted = json.loads(refs_path.read_text(encoding="utf-8"))
|
||||||
|
assert set(persisted.keys()) == {"conv-old"}
|
||||||
|
tmp_files = list((tmp_path / "state").glob("msteams_conversations.json.*.tmp"))
|
||||||
|
assert tmp_files == []
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
@@ -405,6 +657,33 @@ async def test_send_posts_to_conversation_with_reply_to_id_when_reply_in_thread_
|
|||||||
assert kwargs["json"]["replyToId"] == "activity-1"
|
assert kwargs["json"]["replyToId"] == "activity-1"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_send_success_refreshes_updated_at_and_persists_meta(make_channel, tmp_path, monkeypatch):
|
||||||
|
now = {"value": 1_800_000_000.0}
|
||||||
|
monkeypatch.setattr(msteams_module.time, "time", lambda: now["value"])
|
||||||
|
|
||||||
|
ch = make_channel(refTouchIntervalS=0)
|
||||||
|
fake_http = FakeHttpClient()
|
||||||
|
ch._http = fake_http
|
||||||
|
ch._token = "tok"
|
||||||
|
ch._token_expires_at = 9_999_999_999
|
||||||
|
ch._conversation_refs["conv-123"] = ConversationRef(
|
||||||
|
service_url="https://smba.trafficmanager.net/amer/",
|
||||||
|
conversation_id="conv-123",
|
||||||
|
activity_id="activity-1",
|
||||||
|
updated_at=now["value"] - 100,
|
||||||
|
)
|
||||||
|
|
||||||
|
now["value"] += 5
|
||||||
|
await ch.send(OutboundMessage(channel="msteams", chat_id="conv-123", content="Reply text"))
|
||||||
|
|
||||||
|
assert ch._conversation_refs["conv-123"].updated_at == now["value"]
|
||||||
|
saved_meta = json.loads(
|
||||||
|
(tmp_path / "state" / msteams_module.MSTEAMS_REF_META_FILENAME).read_text(encoding="utf-8"),
|
||||||
|
)
|
||||||
|
assert saved_meta["conv-123"]["updated_at"] == now["value"]
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_send_posts_to_conversation_when_thread_reply_disabled(make_channel):
|
async def test_send_posts_to_conversation_when_thread_reply_disabled(make_channel):
|
||||||
ch = make_channel(replyInThread=False)
|
ch = make_channel(replyInThread=False)
|
||||||
@@ -592,15 +871,18 @@ def test_save_refs_prunes_webchat_and_stale_refs(make_channel):
|
|||||||
assert set(ch._conversation_refs) == {"teams-good"}
|
assert set(ch._conversation_refs) == {"teams-good"}
|
||||||
saved = json.loads(ch._refs_path.read_text(encoding="utf-8"))
|
saved = json.loads(ch._refs_path.read_text(encoding="utf-8"))
|
||||||
assert set(saved) == {"teams-good"}
|
assert set(saved) == {"teams-good"}
|
||||||
assert saved["teams-good"]["updated_at"] == pytest.approx(now)
|
saved_meta = json.loads(ch._refs_meta_path.read_text(encoding="utf-8"))
|
||||||
|
assert saved_meta["teams-good"]["updated_at"] == pytest.approx(now)
|
||||||
|
|
||||||
|
|
||||||
def test_msteams_default_config_includes_restart_notify_fields():
|
def test_msteams_default_config_includes_restart_notify_fields():
|
||||||
cfg = MSTeamsChannel.default_config()
|
cfg = MSTeamsChannel.default_config()
|
||||||
|
|
||||||
assert cfg["validateInboundAuth"] is True
|
assert cfg["validateInboundAuth"] is True
|
||||||
|
assert cfg["refTtlDays"] == msteams_module.MSTEAMS_REF_TTL_DAYS
|
||||||
|
assert cfg["pruneWebChatRefs"] is True
|
||||||
|
assert cfg["pruneNonPersonalRefs"] is True
|
||||||
|
assert cfg["refTouchIntervalS"] == msteams_module.MSTEAMS_REF_TOUCH_INTERVAL_S
|
||||||
assert "restartNotifyEnabled" not in cfg
|
assert "restartNotifyEnabled" not in cfg
|
||||||
assert "restartNotifyPreMessage" not in cfg
|
assert "restartNotifyPreMessage" not in cfg
|
||||||
assert "restartNotifyPostMessage" not in cfg
|
assert "restartNotifyPostMessage" not in cfg
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user