diff --git a/libraries/microsoft-agents-a365-notifications/microsoft_agents_a365/notifications/__init__.py b/libraries/microsoft-agents-a365-notifications/microsoft_agents_a365/notifications/__init__.py index 6a0749b5..cb1fa3b6 100644 --- a/libraries/microsoft-agents-a365-notifications/microsoft_agents_a365/notifications/__init__.py +++ b/libraries/microsoft-agents-a365-notifications/microsoft_agents_a365/notifications/__init__.py @@ -12,6 +12,7 @@ from .agent_notification import ( AgentHandler, AgentNotification, + RouteHandler, ) # Import all models from the models subpackage @@ -29,6 +30,7 @@ # Main notification handler "AgentNotification", "AgentHandler", + "RouteHandler", # Models and data classes "AgentNotificationActivity", "EmailReference", diff --git a/libraries/microsoft-agents-a365-notifications/microsoft_agents_a365/notifications/agent_notification.py b/libraries/microsoft-agents-a365-notifications/microsoft_agents_a365/notifications/agent_notification.py index 034f3fb5..bcdcd3c9 100644 --- a/libraries/microsoft-agents-a365-notifications/microsoft_agents_a365/notifications/agent_notification.py +++ b/libraries/microsoft-agents-a365-notifications/microsoft_agents_a365/notifications/agent_notification.py @@ -4,7 +4,7 @@ from __future__ import annotations from collections.abc import Awaitable, Callable, Iterable -from typing import Any, TypeVar +from typing import Any, TypeAlias from microsoft_agents.activity import ChannelId from microsoft_agents.hosting.core import TurnContext @@ -14,8 +14,12 @@ from .models.agent_notification_activity import AgentNotificationActivity, NotificationTypes from .models.agent_subchannel import AgentSubChannel -TContext = TypeVar("TContext", bound=TurnContext) -TState = TypeVar("TState", bound=TurnState) +#: Type alias for the route handler function registered with the application. +#: +#: Route handlers are the inner async functions that the application framework calls +#: when a matching notification is received. They accept a :class:`~microsoft_agents.hosting.core.TurnContext` +#: and a :class:`~microsoft_agents.hosting.core.app.state.TurnState`. +RouteHandler: TypeAlias = Callable[[TurnContext, TurnState], Awaitable[None]] #: Type alias for agent notification handler functions. #: @@ -26,7 +30,8 @@ #: Args: #: context: The turn context for the current conversation turn. #: state: The application state for the current turn. -#: notification: The typed notification activity with parsed entities. +#: notification: The typed :class:`~microsoft_agents_a365.notifications.AgentNotificationActivity` +#: with parsed entities. #: #: Example: #: @@ -40,7 +45,9 @@ #: email = notification.email #: if email: #: print(f"Processing email: {email.id}") -AgentHandler = Callable[[TContext, TState, AgentNotificationActivity], Awaitable[None]] +AgentHandler: TypeAlias = Callable[ + [TurnContext, TurnState, AgentNotificationActivity], Awaitable[None] +] class AgentNotification: @@ -220,9 +227,7 @@ def decorator(handler: AgentHandler): return decorator - def on_email( - self, **kwargs: Any - ) -> Callable[[AgentHandler], Callable[[TurnContext, TurnState], Awaitable[None]]]: + def on_email(self, **kwargs: Any) -> Callable[[AgentHandler], RouteHandler]: """Register a handler for Outlook email notifications. This is a convenience decorator that registers a handler for notifications @@ -251,9 +256,7 @@ async def handle_email(context, state, notification): ChannelId(channel="agents", sub_channel=AgentSubChannel.EMAIL), **kwargs ) - def on_word( - self, **kwargs: Any - ) -> Callable[[AgentHandler], Callable[[TurnContext, TurnState], Awaitable[None]]]: + def on_word(self, **kwargs: Any) -> Callable[[AgentHandler], RouteHandler]: """Register a handler for Microsoft Word comment notifications. This is a convenience decorator that registers a handler for notifications @@ -278,9 +281,7 @@ async def handle_word_comment(context, state, notification): ChannelId(channel="agents", sub_channel=AgentSubChannel.WORD), **kwargs ) - def on_excel( - self, **kwargs: Any - ) -> Callable[[AgentHandler], Callable[[TurnContext, TurnState], Awaitable[None]]]: + def on_excel(self, **kwargs: Any) -> Callable[[AgentHandler], RouteHandler]: """Register a handler for Microsoft Excel comment notifications. This is a convenience decorator that registers a handler for notifications @@ -305,9 +306,7 @@ async def handle_excel_comment(context, state, notification): ChannelId(channel="agents", sub_channel=AgentSubChannel.EXCEL), **kwargs ) - def on_powerpoint( - self, **kwargs: Any - ) -> Callable[[AgentHandler], Callable[[TurnContext, TurnState], Awaitable[None]]]: + def on_powerpoint(self, **kwargs: Any) -> Callable[[AgentHandler], RouteHandler]: """Register a handler for Microsoft PowerPoint comment notifications. This is a convenience decorator that registers a handler for notifications @@ -332,9 +331,7 @@ async def handle_powerpoint_comment(context, state, notification): ChannelId(channel="agents", sub_channel=AgentSubChannel.POWERPOINT), **kwargs ) - def on_lifecycle( - self, **kwargs: Any - ) -> Callable[[AgentHandler], Callable[[TurnContext, TurnState], Awaitable[None]]]: + def on_lifecycle(self, **kwargs: Any) -> Callable[[AgentHandler], RouteHandler]: """Register a handler for all agent lifecycle event notifications. This is a convenience decorator that registers a handler for all lifecycle @@ -353,11 +350,9 @@ def on_lifecycle( async def handle_any_lifecycle_event(context, state, notification): print(f"Lifecycle event type: {notification.notification_type}") """ - return self.on_lifecycle_notification("*", **kwargs) + return self.on_agent_lifecycle_notification("*", **kwargs) - def on_user_created( - self, **kwargs: Any - ) -> Callable[[AgentHandler], Callable[[TurnContext, TurnState], Awaitable[None]]]: + def on_user_created(self, **kwargs: Any) -> Callable[[AgentHandler], RouteHandler]: """Register a handler for user creation lifecycle events. This is a convenience decorator that registers a handler specifically for @@ -376,11 +371,9 @@ def on_user_created( async def handle_user_created(context, state, notification): print("New agentic user identity created") """ - return self.on_lifecycle_notification(AgentLifecycleEvent.USERCREATED, **kwargs) + return self.on_agent_lifecycle_notification(AgentLifecycleEvent.USERCREATED, **kwargs) - def on_user_workload_onboarding( - self, **kwargs: Any - ) -> Callable[[AgentHandler], Callable[[TurnContext, TurnState], Awaitable[None]]]: + def on_user_workload_onboarding(self, **kwargs: Any) -> Callable[[AgentHandler], RouteHandler]: """Register a handler for user workload onboarding update events. This is a convenience decorator that registers a handler for events that occur @@ -399,13 +392,11 @@ def on_user_workload_onboarding( async def handle_onboarding_update(context, state, notification): print("User workload onboarding status updated") """ - return self.on_lifecycle_notification( + return self.on_agent_lifecycle_notification( AgentLifecycleEvent.USERWORKLOADONBOARDINGUPDATED, **kwargs ) - def on_user_deleted( - self, **kwargs: Any - ) -> Callable[[AgentHandler], Callable[[TurnContext, TurnState], Awaitable[None]]]: + def on_user_deleted(self, **kwargs: Any) -> Callable[[AgentHandler], RouteHandler]: """Register a handler for user deletion lifecycle events. This is a convenience decorator that registers a handler specifically for @@ -424,7 +415,7 @@ def on_user_deleted( async def handle_user_deleted(context, state, notification): print("Agentic user identity deleted") """ - return self.on_lifecycle_notification(AgentLifecycleEvent.USERDELETED, **kwargs) + return self.on_agent_lifecycle_notification(AgentLifecycleEvent.USERDELETED, **kwargs) @staticmethod def _normalize_subchannel(value: str | AgentSubChannel | None) -> str: diff --git a/tests/notifications/__init__.py b/tests/notifications/__init__.py new file mode 100644 index 00000000..59e481eb --- /dev/null +++ b/tests/notifications/__init__.py @@ -0,0 +1,2 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. diff --git a/tests/notifications/test_agent_notification.py b/tests/notifications/test_agent_notification.py new file mode 100644 index 00000000..900bbbc7 --- /dev/null +++ b/tests/notifications/test_agent_notification.py @@ -0,0 +1,257 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Unit tests for AgentNotification class routing decorators.""" + +from unittest.mock import MagicMock + +from microsoft_agents_a365.notifications import AgentHandler, AgentNotification, RouteHandler +from microsoft_agents_a365.notifications.models.agent_lifecycle_event import AgentLifecycleEvent +from microsoft_agents_a365.notifications.models.agent_subchannel import AgentSubChannel + + +class TestAgentNotificationTypeAliases: + """Tests verifying the public type aliases are importable and properly defined.""" + + def test_agent_handler_is_importable(self): + """AgentHandler type alias can be imported from the package.""" + assert AgentHandler is not None + + def test_route_handler_is_importable(self): + """RouteHandler type alias can be imported from the package.""" + assert RouteHandler is not None + + def test_agent_handler_is_type_alias(self): + """AgentHandler is a TypeAlias (not a TypeVar).""" + import typing + + # TypeAlias values are just the underlying type, not TypeVar instances + assert not isinstance(AgentHandler, typing.TypeVar) + + def test_route_handler_is_type_alias(self): + """RouteHandler is a TypeAlias (not a TypeVar).""" + import typing + + assert not isinstance(RouteHandler, typing.TypeVar) + + +class TestAgentNotificationRouting: + """Tests verifying that convenience decorator methods register routes correctly.""" + + def _make_app(self): + """Return a mock app with an add_route spy.""" + app = MagicMock() + app.add_route = MagicMock() + return app + + def test_on_email_calls_add_route(self): + """on_email() registers a route via app.add_route.""" + app = self._make_app() + notifications = AgentNotification(app) + + @notifications.on_email() + async def handler(context, state, notification): + pass + + app.add_route.assert_called_once() + + def test_on_word_calls_add_route(self): + """on_word() registers a route via app.add_route.""" + app = self._make_app() + notifications = AgentNotification(app) + + @notifications.on_word() + async def handler(context, state, notification): + pass + + app.add_route.assert_called_once() + + def test_on_excel_calls_add_route(self): + """on_excel() registers a route via app.add_route.""" + app = self._make_app() + notifications = AgentNotification(app) + + @notifications.on_excel() + async def handler(context, state, notification): + pass + + app.add_route.assert_called_once() + + def test_on_powerpoint_calls_add_route(self): + """on_powerpoint() registers a route via app.add_route.""" + app = self._make_app() + notifications = AgentNotification(app) + + @notifications.on_powerpoint() + async def handler(context, state, notification): + pass + + app.add_route.assert_called_once() + + def test_on_lifecycle_calls_add_route(self): + """on_lifecycle() registers a route via app.add_route.""" + app = self._make_app() + notifications = AgentNotification(app) + + @notifications.on_lifecycle() + async def handler(context, state, notification): + pass + + app.add_route.assert_called_once() + + def test_on_user_created_calls_add_route(self): + """on_user_created() registers a route via app.add_route.""" + app = self._make_app() + notifications = AgentNotification(app) + + @notifications.on_user_created() + async def handler(context, state, notification): + pass + + app.add_route.assert_called_once() + + def test_on_user_workload_onboarding_calls_add_route(self): + """on_user_workload_onboarding() registers a route via app.add_route.""" + app = self._make_app() + notifications = AgentNotification(app) + + @notifications.on_user_workload_onboarding() + async def handler(context, state, notification): + pass + + app.add_route.assert_called_once() + + def test_on_user_deleted_calls_add_route(self): + """on_user_deleted() registers a route via app.add_route.""" + app = self._make_app() + notifications = AgentNotification(app) + + @notifications.on_user_deleted() + async def handler(context, state, notification): + pass + + app.add_route.assert_called_once() + + +class TestAgentNotificationRouteSelector: + """Tests verifying that the route selector logic correctly matches activities.""" + + def _make_turn_context(self, channel: str, sub_channel: str | None = None): + """Create a mock TurnContext with the given channel_id values. + + Args: + channel: The channel identifier (e.g., "agents"). + sub_channel: The optional sub-channel identifier (e.g., "email"). + """ + channel_id = MagicMock() + channel_id.channel = channel + channel_id.sub_channel = sub_channel + + activity = MagicMock() + activity.channel_id = channel_id + + context = MagicMock() + context.activity = activity + return context + + def _make_lifecycle_context(self, value_type: str): + """Create a mock TurnContext for a lifecycle notification. + + Args: + value_type: The lifecycle event type identifier (e.g., + ``AgentLifecycleEvent.USERCREATED.value``). + """ + from microsoft_agents_a365.notifications.models.notification_types import NotificationTypes + + channel_id = MagicMock() + channel_id.channel = "agents" + channel_id.sub_channel = None + + activity = MagicMock() + activity.channel_id = channel_id + activity.name = NotificationTypes.AGENT_LIFECYCLE + activity.value_type = value_type + + context = MagicMock() + context.activity = activity + return context + + def test_on_email_selector_matches_email_subchannel(self): + """Route selector registered by on_email() matches the email subchannel.""" + app = MagicMock() + captured_selector = None + + def capture_add_route(selector, handler, **kwargs): + nonlocal captured_selector + captured_selector = selector + + app.add_route = capture_add_route + notifications = AgentNotification(app) + + @notifications.on_email() + async def handler(context, state, notification): + pass + + assert captured_selector is not None + ctx = self._make_turn_context("agents", AgentSubChannel.EMAIL.value) + assert captured_selector(ctx) is True + + def test_on_email_selector_rejects_word_subchannel(self): + """Route selector registered by on_email() rejects the word subchannel.""" + app = MagicMock() + captured_selector = None + + def capture_add_route(selector, handler, **kwargs): + nonlocal captured_selector + captured_selector = selector + + app.add_route = capture_add_route + notifications = AgentNotification(app) + + @notifications.on_email() + async def handler(context, state, notification): + pass + + assert captured_selector is not None + ctx = self._make_turn_context("agents", AgentSubChannel.WORD.value) + assert captured_selector(ctx) is False + + def test_on_lifecycle_selector_matches_any_lifecycle_event(self): + """Route selector from on_lifecycle() matches any lifecycle event (wildcard).""" + app = MagicMock() + captured_selector = None + + def capture_add_route(selector, handler, **kwargs): + nonlocal captured_selector + captured_selector = selector + + app.add_route = capture_add_route + notifications = AgentNotification(app) + + @notifications.on_lifecycle() + async def handler(context, state, notification): + pass + + assert captured_selector is not None + ctx = self._make_lifecycle_context(AgentLifecycleEvent.USERCREATED.value) + assert captured_selector(ctx) is True + + def test_on_user_created_selector_matches_user_created_event(self): + """Route selector from on_user_created() matches the user created event.""" + app = MagicMock() + captured_selector = None + + def capture_add_route(selector, handler, **kwargs): + nonlocal captured_selector + captured_selector = selector + + app.add_route = capture_add_route + notifications = AgentNotification(app) + + @notifications.on_user_created() + async def handler(context, state, notification): + pass + + assert captured_selector is not None + ctx = self._make_lifecycle_context(AgentLifecycleEvent.USERCREATED.value) + assert captured_selector(ctx) is True