Integration testing framework

dev/integration/env.TEMPLATE

@@ -0,0 +1,3 @@
+aioresponses
+microsoft-agents-activity
+microsoft-agents-hosting-core

dev/integration/requirements.txt

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

dev/integration/src/core/__init__.py

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

dev/integration/src/core/application_runner.py

@@ -0,0 +1,35 @@
+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
+    def _start_server(self) -> None:
+        raise NotImplementedError("Start server method must be implemented by subclasses")
+
+    def _stop_server(self) -> None:
+        pass
+
+    async def __aenter__(self) -> None:
+
+        if self._thread:
+            raise RuntimeError("Server is already running")
+
+        self._thread = Thread(target=self._start_server, daemon=True)
+        self._thread.start()
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
+
+        if self._thread:
+            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,107 @@
+import os
+import json
+import asyncio
+from typing import Any, Optional, cast
+
+from aiohttp import ClientSession
+
+from microsoft_agents.activity import (
+    Activity,
+    ActivityTypes,
+    DeliveryModes
+)
+
+from msal import ConfidentialClientApplication
+
+class AgentClient:
+
+    def __init__(
+            self,
+            messaging_endpoint: str,
+            service_endpoint: str,
+            cid: str,
+            client_id: str,
+            tenant_id: str,
+            client_secret: str,
+            default_timeout: float = 5.0
+        ):
+        self._messaging_endpoint = messaging_endpoint
+        self.service_endpoint = service_endpoint
+        self.cid = cid
+        self.client_id = client_id
+        self.tenant_id = tenant_id
+        self.client_secret = client_secret
+        self._headers = None
+        self._default_timeout = default_timeout
+
+        self._client = ClientSession(
+            base_url=self._messaging_endpoint,
+            headers={"Content-Type": "application/json"}
+        )
+
+        self._msal_app = ConfidentialClientApplication(
+            client_id=client_id,
+            client_credential=client_secret,
+            authority=f"https://login.microsoftonline.com/{tenant_id}"
+        )
+
+    async def get_access_token(self) -> str:
+        res = self._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 _set_headers(self) -> None:
+        if not self._headers:
+            token = await self.get_access_token()
+            self._headers = {
+                "Authorization": f"Bearer {token}",
+                "Content-Type": "application/json"
+            }
+
+    async def send_request(self, activity: Activity) -> str:
+
+        await self._set_headers()
+
+        if activity.conversation:
+            activity.conversation.id = self.cid
+
+        async with self._client.post(
+            self._messaging_endpoint,
+            headers=self._headers,
+            json=activity.model_dump(by_alias=True, exclude_none=True)
+        ) as response:
+            if not response.ok:
+                raise Exception(f"Failed to send activity: {response.status}")
+            content = await response.text()
+            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, timeout: Optional[float] = None) -> str:
+        timeout = timeout or self._default_timeout
+        activity = self._to_activity(activity_or_text)
+        content = await self.send_request(activity)
+        return content
+
+    async def send_expect_replies(self, activity_or_text: Activity | str, 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
+
+        content = await self.send_request(activity)
+
+        activities_data = json.loads(content).get("activities", [])
+        activities = [Activity.model_validate(act) for act in activities_data]
+        return activities

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

@@ -0,0 +1,18 @@
+# from microsoft_agents.activity import Activity
+
+# from ..agent_client import AgentClient
+
+# class AutoClient:
+
+#     def __init__(self, agent_client: AgentClient):
+#         self._agent_client = agent_client
+
+#     async def generate_message(self) -> str:
+#         pass
+
+#     async def run(self, max_turns: int = 10, time_between_turns: float = 2.0) -> None:
+
+#         for i in range(max_turns):
+#             await self._agent_client.send_activity(
+#                 Activity(type="message", text=self.generate_message())
+#             )

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

@@ -0,0 +1,182 @@
+import asyncio
+import json
+import sys
+import threading
+import uuid
+from io import StringIO
+from typing import Dict, List, Optional
+from threading import Lock
+from collections import defaultdict
+
+from aiohttp import web, ClientSession
+from aiohttp.web import Request, Response
+import aiohttp_security
+from microsoft_agents.core.models import Activity, EntityTypes, ActivityTypes, StreamInfo, StreamTypes
+from microsoft_agents.core.serialization import ProtocolJsonSerializer
+
+class BotResponse:
+    """Python equivalent of the C# BotResponse class using aiohttp web framework."""
+
+    def __init__(self):
+        self._app: Optional[web.Application] = None
+        self._runner: Optional[web.AppRunner] = None
+        self._site: Optional[web.TCPSite] = None
+        self._multiple_activities: Dict[str, List[Activity]] = defaultdict(list)
+        self._activity_locks: Dict[str, Lock] = defaultdict(Lock)
+        self.test_id: str = str(uuid.uuid4())
+        self.service_endpoint: str = "http://localhost:9873"
+
+        # Suppress console output (equivalent to Console.SetOut(TextWriter.Null))
+        sys.stdout = StringIO()
+
+        # Initialize the web application
+        self._setup_app()
+
+    def _setup_app(self):
+        """Setup the aiohttp web application with routes and middleware."""
+        self._app = web.Application()
+
+        # Add JWT authentication middleware (placeholder - would need proper implementation)
+        # self._app.middlewares.append(self._auth_middleware)
+
+        # Add routes
+        self._app.router.add_post('/v3/conversations/{path:.*}', self._handle_conversation)
+
+    async def _auth_middleware(self, request: Request, handler):
+        """JWT authentication middleware (placeholder implementation)."""
+        # TODO: Implement proper JWT authentication
+        return await handler(request)
+
+    async def _handle_conversation(self, request: Request) -> Response:
+        """Handle POST requests to /v3/conversations/{*text}."""
+        try:
+            # Read request body
+            body = await request.text()
+            act = ProtocolJsonSerializer.to_object(body, Activity)
+            cid = act.conversation.id if act.conversation else None
+
+            if not cid:
+                return web.Response(status=400, text="Missing conversation ID")
+
+            # Add activity to the list (thread-safe)
+            with self._activity_locks[cid]:
+                self._multiple_activities[cid].append(act)
+
+            # Create response
+            response_data = {
+                "Id": str(uuid.uuid4())
+            }
+
+            # Check if the activity is a streamed activity
+            if (act.entities and 
+                any(e.type == EntityTypes.STREAM_INFO for e in act.entities)):
+
+                entities_json = ProtocolJsonSerializer.to_json(act.entities[0])
+                sact = ProtocolJsonSerializer.to_object(entities_json, StreamInfo)
+
+                handled = self._handle_streamed_activity(act, sact, cid)
+
+                response = web.Response(
+                    status=200,
+                    content_type="application/json",
+                    text=json.dumps(response_data)
+                )
+
+                # Handle task completion (would need BotClient implementation)
+                if handled:
+                    await self._complete_streaming_task(cid)
+
+                return response
+            else:
+                # Handle non-streamed activities
+                if act.type != ActivityTypes.TYPING:
+                    # Start background task with 5-second delay
+                    asyncio.create_task(self._delayed_task_completion(cid))
+
+                return web.Response(
+                    status=200,
+                    content_type="application/json",
+                    text=json.dumps(response_data)
+                )
+
+        except Exception as e:
+            return web.Response(status=500, text=str(e))
+
+    def _handle_streamed_activity(self, act: Activity, sact: StreamInfo, cid: str) -> bool:
+        """Handle streamed activity logic."""
+
+        # Check if activity is the final message
+        if sact.stream_type == StreamTypes.FINAL:
+            if act.type == ActivityTypes.MESSAGE:
+                return True
+            else:
+                raise Exception("final streamed activity should be type message")
+
+        # Handler for streaming types which allows us to verify later if the text length has increased
+        elif sact.stream_type == StreamTypes.STREAMING:
+            if sact.stream_sequence <= 0 and act.type == ActivityTypes.TYPING:
+                raise Exception("streamed activity's stream sequence should be a positive number")
+
+        # Activity is being streamed but isn't the final message
+        return False
+
+    async def _complete_streaming_task(self, cid: str):
+        """Complete streaming task (placeholder for BotClient.TaskList logic)."""
+        # TODO: Implement BotClient.TaskList equivalent
+        # This would require the BotClient class to be implemented
+        if cid in self._multiple_activities:
+            activities = self._multiple_activities[cid].copy()
+            # Complete the task with activities
+            # BotClient.complete_task(cid, activities)
+            # Clean up
+            del self._multiple_activities[cid]
+            if cid in self._activity_locks:
+                del self._activity_locks[cid]
+
+    async def _delayed_task_completion(self, cid: str):
+        """Handle delayed task completion after 5 seconds."""
+        await asyncio.sleep(5.0)
+        # TODO: Implement BotClient.TaskList equivalent
+        # if BotClient.has_task(cid):
+        #     activities = self._multiple_activities.get(cid, [])
+        #     BotClient.complete_task(cid, activities)
+
+    async def start(self):
+        """Start the web server."""
+        self._runner = web.AppRunner(self._app)
+        await self._runner.setup()
+
+        # Extract port from service_endpoint
+        port = int(self.service_endpoint.split(':')[-1])
+        self._site = web.TCPSite(self._runner, 'localhost', port)
+        await self._site.start()
+
+        print(f"Bot server started at {self.service_endpoint}")
+
+    async def dispose(self):
+        """Cleanup resources (equivalent to DisposeAsync)."""
+        if self._site:
+            await self._site.stop()
+        if self._runner:
+            await self._runner.cleanup()
+
+        # Restore stdout
+        sys.stdout = sys.__stdout__
+
+
+# Example usage
+async def main():
+    bot_response = BotResponse()
+    try:
+        await bot_response.start()
+        # Keep the server running
+        while True:
+            await asyncio.sleep(1)
+    except KeyboardInterrupt:
+        print("Shutting down...")
+    finally:
+        await bot_response.dispose()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())