Integration testing framework

dev/integration/README.md

@@ -0,0 +1,2 @@
+# Microsoft 365 Agents SDK for Python Integration Testing Framework
+

dev/integration/requirements.txt

@@ -0,0 +1 @@
+aioresponses

dev/integration/src/core/__init__.py

@@ -0,0 +1,21 @@
+from .application_runner import ApplicationRunner
+from .aiohttp import AiohttpEnvironment
+from .client import (
+    AgentClient,
+    ResponseClient,
+)
+from .environment import Environment
+from .integration import integration, IntegrationFixtures
+from .sample import Sample
+
+
+__all__ = [
+    "AgentClient",
+    "ApplicationRunner",
+    "AiohttpEnvironment",
+    "ResponseClient",
+    "Environment",
+    "integration",
+    "IntegrationFixtures",
+    "Sample",
+]

dev/integration/src/core/aiohttp/__init__.py

@@ -0,0 +1,4 @@
+from .aiohttp_environment import AiohttpEnvironment
+from .aiohttp_runner import AiohttpRunner
+
+__all__ = ["AiohttpEnvironment", "AiohttpRunner"]

dev/integration/src/core/aiohttp/aiohttp_environment.py

@@ -0,0 +1,58 @@
+from tkinter import E
+from aiohttp.web import Request, Response, Application, run_app
+
+from microsoft_agents.hosting.aiohttp import (
+    CloudAdapter,
+    jwt_authorization_middleware,
+    start_agent_process,
+)
+from microsoft_agents.hosting.core import (
+    Authorization,
+    AgentApplication,
+    TurnState,
+    MemoryStorage,
+)
+from microsoft_agents.authentication.msal import MsalConnectionManager
+from microsoft_agents.activity import load_configuration_from_env
+
+from ..application_runner import ApplicationRunner
+from ..environment import Environment
+from .aiohttp_runner import AiohttpRunner
+
+
+class AiohttpEnvironment(Environment):
+    """An environment for aiohttp-hosted agents."""
+
+    async def init_env(self, environ_config: dict) -> None:
+        environ_config = environ_config or {}
+
+        self.config = load_configuration_from_env(environ_config)
+
+        self.storage = MemoryStorage()
+        self.connection_manager = MsalConnectionManager(**self.config)
+        self.adapter = CloudAdapter(connection_manager=self.connection_manager)
+        self.authorization = Authorization(
+            self.storage, self.connection_manager, **self.config
+        )
+
+        self.agent_application = AgentApplication[TurnState](
+            storage=self.storage,
+            adapter=self.adapter,
+            authorization=self.authorization,
+            **self.config
+        )
+
+    def create_runner(self, host: str, port: int) -> ApplicationRunner:
+
+        async def entry_point(req: Request) -> Response:
+            agent: AgentApplication = req.app["agent_app"]
+            adapter: CloudAdapter = req.app["adapter"]
+            return await start_agent_process(req, agent, adapter)
+
+        APP = Application(middlewares=[jwt_authorization_middleware])
+        APP.router.add_post("/api/messages", entry_point)
+        APP["agent_configuration"] = self.connection_manager.get_default_connection_configuration()
+        APP["agent_app"] = self.agent_application
+        APP["adapter"] = self.adapter
+
+        return AiohttpRunner(APP, host, port)

dev/integration/src/core/aiohttp/aiohttp_runner.py

@@ -0,0 +1,115 @@
+from typing import Optional
+from typing import Optional
+from threading import Thread, Event
+import asyncio
+
+from aiohttp import ClientSession
+from aiohttp.web import Application, Request, Response
+from aiohttp.web_runner import AppRunner, TCPSite
+
+from ..application_runner import ApplicationRunner
+
+
+class AiohttpRunner(ApplicationRunner):
+    """A runner for aiohttp applications."""
+
+    def __init__(self, app: Application, host: str = "localhost", port: int = 8000):
+        assert isinstance(app, Application)
+        super().__init__(app)
+
+        url = f"{host}:{port}"
+        self._host = host
+        self._port = port
+        if "http" not in url:
+            url = f"http://{url}"
+        self._url = url
+
+        self._app.router.add_get("/shutdown", self._shutdown_route)
+
+        self._server_thread: Optional[Thread] = None
+        self._shutdown_event = Event()
+        self._runner: Optional[AppRunner] = None
+        self._site: Optional[TCPSite] = None
+
+    @property
+    def url(self) -> str:
+        return self._url
+
+    async def _start_server(self) -> None:
+        try:
+            assert isinstance(self._app, Application)
+
+            self._runner = AppRunner(self._app)
+            await self._runner.setup()
+            self._site = TCPSite(self._runner, self._host, self._port)
+            await self._site.start()
+
+            # Wait for shutdown signal
+            while not self._shutdown_event.is_set():
+                await asyncio.sleep(0.1)
+
+            # Cleanup
+            await self._site.stop()
+            await self._runner.cleanup()
+
+        except Exception as error:
+            raise error
+
+    async def __aenter__(self):
+        if self._server_thread:
+            raise RuntimeError("ResponseClient is already running.")
+
+        self._shutdown_event.clear()
+        self._server_thread = Thread(
+            target=lambda: asyncio.run(self._start_server()), daemon=True
+        )
+        self._server_thread.start()
+
+        # Wait a moment to ensure the server starts
+        await asyncio.sleep(0.5)
+
+        return self
+
+    async def _stop_server(self):
+        if not self._server_thread:
+            raise RuntimeError("ResponseClient is not running.")
+
+        try:
+            async with ClientSession() as session:
+                async with session.get(
+                    f"http://{self._host}:{self._port}/shutdown"
+                ) as response:
+                    pass  # Just trigger the shutdown
+        except Exception:
+            pass  # Ignore errors during shutdown request
+
+        # Set shutdown event as fallback
+        self._shutdown_event.set()
+
+        # Wait for the server thread to finish
+        self._server_thread.join(timeout=5.0)
+        self._server_thread = None
+
+    async def _shutdown_route(self, request: Request) -> Response:
+        """Handle shutdown request by setting the shutdown event"""
+        self._shutdown_event.set()
+        return Response(status=200, text="Shutdown initiated")
+
+    async def __aexit__(self, exc_type, exc, tb):
+        if not self._server_thread:
+            raise RuntimeError("ResponseClient is not running.")
+        try:
+            async with ClientSession() as session:
+                async with session.get(
+                    f"http://{self._host}:{self._port}/shutdown"
+                ) as response:
+                    pass  # Just trigger the shutdown
+        except Exception:
+            pass  # Ignore errors during shutdown request
+
+        # Set shutdown event as fallback
+        self._shutdown_event.set()
+
+        # Wait for the server thread to finish
+        self._server_thread.join(timeout=5.0)
+        self._server_thread = None

dev/integration/src/core/application_runner.py

@@ -0,0 +1,42 @@
+import asyncio
+from abc import ABC, abstractmethod
+from typing import Any, Optional
+from threading import Thread
+
+
+class ApplicationRunner(ABC):
+    """Base class for application runners."""
+
+    def __init__(self, app: Any):
+        self._app = app
+        self._thread: Optional[Thread] = None
+
+    @abstractmethod
+    async def _start_server(self) -> None:
+        raise NotImplementedError(
+            "Start server method must be implemented by subclasses"
+        )
+
+    async def _stop_server(self) -> None:
+        pass
+
+    async def __aenter__(self) -> None:
+
+        if self._thread:
+            raise RuntimeError("Server is already running")
+
+        def target():
+            asyncio.run(self._start_server())
+
+        self._thread = Thread(target=target, daemon=True)
+        self._thread.start()
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
+
+        if self._thread:
+            await self._stop_server()
+
+            self._thread.join()
+            self._thread = None
+        else:
+            raise RuntimeError("Server is not running")

dev/integration/src/core/client/__init__.py

@@ -0,0 +1,7 @@
+from .agent_client import AgentClient
+from .response_client import ResponseClient
+
+__all__ = [
+    "AgentClient",
+    "ResponseClient",
+]

dev/integration/src/core/client/agent_client.py

@@ -0,0 +1,136 @@
+import os
+import json
+import asyncio
+from typing import Optional, cast
+
+from aiohttp import ClientSession
+from msal import ConfidentialClientApplication
+
+from microsoft_agents.activity import Activity, ActivityTypes, DeliveryModes, ConversationAccount
+
+from ..utils import _populate_incoming_activity
+
+
+class AgentClient:
+
+    def __init__(
+        self,
+        agent_url: str,
+        cid: str,
+        client_id: str,
+        tenant_id: str,
+        client_secret: str,
+        service_url: Optional[str] = None,
+        default_timeout: float = 5.0,
+    ):
+        self._agent_url = agent_url
+        self._cid = cid
+        self._client_id = client_id
+        self._tenant_id = tenant_id
+        self._client_secret = client_secret
+        self._service_url = service_url
+        self._headers = None
+        self._default_timeout = default_timeout
+
+        self._client: Optional[ClientSession] = None
+
+    @property
+    def agent_url(self) -> str:
+        return self._agent_url
+
+    @property
+    def service_url(self) -> Optional[str]:
+        return self._service_url
+
+    async def get_access_token(self) -> str:
+
+        msal_app = ConfidentialClientApplication(
+            client_id=self._client_id,
+            client_credential=self._client_secret,
+            authority=f"https://login.microsoftonline.com/{self._tenant_id}",
+        )
+
+        res = msal_app.acquire_token_for_client(scopes=[f"{self._client_id}/.default"])
+        token = res.get("access_token") if res else None
+        if not token:
+            raise Exception("Could not obtain access token")
+        return token
+
+    async def _init_client(self) -> None:
+        if not self._client:
+            if self._client_secret:
+                token = await self.get_access_token()
+                self._headers = {
+                    "Authorization": f"Bearer {token}",
+                    "Content-Type": "application/json",
+                }
+            else:
+                self._headers = {"Content-Type": "application/json"}
+
+            self._client = ClientSession(
+                base_url=self._agent_url, headers=self._headers
+            )
+
+    async def send_request(self, activity: Activity, sleep: float = 0) -> str:
+
+        await self._init_client()
+        assert self._client
+
+        if activity.conversation:
+            activity.conversation.id = self._cid
+        else:
+            activity.conversation = ConversationAccount(id=self._cid or "<conversation_id>")
+
+        if self.service_url:
+            activity.service_url = self.service_url
+
+        activity = _populate_incoming_activity(activity)
+
+        async with self._client.post(
+            "api/messages",
+            headers=self._headers,
+            json=activity.model_dump(by_alias=True, exclude_unset=True, exclude_none=True, mode="json"),
+        ) as response:
+            content = await response.text()
+            if not response.ok:
+                raise Exception(f"Failed to send activity: {response.status}")
+            await asyncio.sleep(sleep)
+            return content
+
+    def _to_activity(self, activity_or_text: Activity | str) -> Activity:
+        if isinstance(activity_or_text, str):
+            activity = Activity(
+                type=ActivityTypes.message,
+                text=activity_or_text,
+            )
+            return activity
+        else:
+            return cast(Activity, activity_or_text)
+
+    async def send_activity(
+        self, activity_or_text: Activity | str, sleep: float = 0, timeout: Optional[float] = None
+    ) -> str:
+        timeout = timeout or self._default_timeout
+        activity = self._to_activity(activity_or_text)
+        content = await self.send_request(activity, sleep=sleep)
+        return content
+
+    async def send_expect_replies(
+        self, activity_or_text: Activity | str, sleep: float = 0, timeout: Optional[float] = None
+    ) -> list[Activity]:
+        timeout = timeout or self._default_timeout
+        activity = self._to_activity(activity_or_text)
+        activity.delivery_mode = DeliveryModes.expect_replies
+        activity.service_url = activity.service_url or "http://localhost" # temporary fix
+
+        content = await self.send_request(activity, sleep=sleep)
+
+        activities_data = json.loads(content).get("activities", [])
+        activities = [Activity.model_validate(act) for act in activities_data]
+
+        return activities
+
+    async def close(self) -> None:
+        if self._client:
+            await self._client.close()
+            self._client = None