|
| 1 | +# Copyright (c) Microsoft. All rights reserved. |
| 2 | + |
| 3 | +import unittest |
| 4 | +import uuid |
| 5 | +import jwt |
| 6 | + |
| 7 | +from microsoft_agents_a365.runtime.utility import Utility |
| 8 | + |
| 9 | + |
| 10 | +class TestUtility(unittest.TestCase): |
| 11 | + """Test cases for the Utility class.""" |
| 12 | + |
| 13 | + def setUp(self): |
| 14 | + """Set up test fixtures.""" |
| 15 | + self.test_app_id = "12345678-1234-1234-1234-123456789abc" |
| 16 | + self.test_azp_id = "87654321-4321-4321-4321-cba987654321" |
| 17 | + |
| 18 | + def create_test_jwt(self, claims: dict) -> str: |
| 19 | + """Create a test JWT token with the given claims.""" |
| 20 | + # Use PyJWT to create a proper JWT token (unsigned for testing) |
| 21 | + return jwt.encode(claims, key="", algorithm="none") |
| 22 | + |
| 23 | + def test_get_app_id_from_token_with_none_token(self): |
| 24 | + """Test get_app_id_from_token with None token.""" |
| 25 | + result = Utility.get_app_id_from_token(None) |
| 26 | + self.assertEqual(result, str(uuid.UUID(int=0))) |
| 27 | + |
| 28 | + def test_get_app_id_from_token_with_empty_token(self): |
| 29 | + """Test get_app_id_from_token with empty token.""" |
| 30 | + result = Utility.get_app_id_from_token("") |
| 31 | + self.assertEqual(result, str(uuid.UUID(int=0))) |
| 32 | + |
| 33 | + result = Utility.get_app_id_from_token(" ") |
| 34 | + self.assertEqual(result, str(uuid.UUID(int=0))) |
| 35 | + |
| 36 | + def test_get_app_id_from_token_with_appid_claim(self): |
| 37 | + """Test get_app_id_from_token with appid claim.""" |
| 38 | + token = self.create_test_jwt({"appid": self.test_app_id, "other": "value"}) |
| 39 | + result = Utility.get_app_id_from_token(token) |
| 40 | + self.assertEqual(result, self.test_app_id) |
| 41 | + |
| 42 | + def test_get_app_id_from_token_with_azp_claim(self): |
| 43 | + """Test get_app_id_from_token with azp claim.""" |
| 44 | + token = self.create_test_jwt({"azp": self.test_azp_id, "other": "value"}) |
| 45 | + result = Utility.get_app_id_from_token(token) |
| 46 | + self.assertEqual(result, self.test_azp_id) |
| 47 | + |
| 48 | + def test_get_app_id_from_token_with_both_claims(self): |
| 49 | + """Test get_app_id_from_token with both appid and azp claims (appid takes precedence).""" |
| 50 | + token = self.create_test_jwt({"appid": self.test_app_id, "azp": self.test_azp_id}) |
| 51 | + result = Utility.get_app_id_from_token(token) |
| 52 | + self.assertEqual(result, self.test_app_id) |
| 53 | + |
| 54 | + def test_get_app_id_from_token_without_app_claims(self): |
| 55 | + """Test get_app_id_from_token with token containing no app claims.""" |
| 56 | + token = self.create_test_jwt({"sub": "user123", "iss": "issuer"}) |
| 57 | + result = Utility.get_app_id_from_token(token) |
| 58 | + self.assertEqual(result, "") |
| 59 | + |
| 60 | + def test_get_app_id_from_token_with_invalid_token(self): |
| 61 | + """Test get_app_id_from_token with invalid token formats.""" |
| 62 | + # Invalid token format |
| 63 | + result = Utility.get_app_id_from_token("invalid.token") |
| 64 | + self.assertEqual(result, "") |
| 65 | + |
| 66 | + # Token with only two parts |
| 67 | + result = Utility.get_app_id_from_token("header.payload") |
| 68 | + self.assertEqual(result, "") |
| 69 | + |
| 70 | + # Token with invalid base64 |
| 71 | + result = Utility.get_app_id_from_token("invalid.!!!invalid!!!.signature") |
| 72 | + self.assertEqual(result, "") |
| 73 | + |
| 74 | + |
| 75 | +class MockActivity: |
| 76 | + """Mock activity class for testing.""" |
| 77 | + |
| 78 | + def __init__(self, is_agentic: bool = False, agentic_id: str = ""): |
| 79 | + self._is_agentic = is_agentic |
| 80 | + self._agentic_id = agentic_id |
| 81 | + |
| 82 | + def is_agentic_request(self) -> bool: |
| 83 | + return self._is_agentic |
| 84 | + |
| 85 | + def get_agentic_instance_id(self) -> str: |
| 86 | + return self._agentic_id |
| 87 | + |
| 88 | + |
| 89 | +class MockContext: |
| 90 | + """Mock context class for testing.""" |
| 91 | + |
| 92 | + def __init__(self, activity=None): |
| 93 | + self.activity = activity |
| 94 | + |
| 95 | + |
| 96 | +class TestUtilityResolveAgentIdentity(unittest.TestCase): |
| 97 | + """Test cases for the resolve_agent_identity method.""" |
| 98 | + |
| 99 | + def setUp(self): |
| 100 | + """Set up test fixtures.""" |
| 101 | + self.test_app_id = "token-app-id-123" |
| 102 | + self.agentic_id = "agentic-id-456" |
| 103 | + |
| 104 | + # Create a test token with PyJWT |
| 105 | + claims = {"appid": self.test_app_id} |
| 106 | + self.test_token = jwt.encode(claims, key="", algorithm="none") |
| 107 | + |
| 108 | + def test_resolve_agent_identity_with_agentic_request(self): |
| 109 | + """Test resolve_agent_identity with agentic request.""" |
| 110 | + activity = MockActivity(is_agentic=True, agentic_id=self.agentic_id) |
| 111 | + context = MockContext(activity) |
| 112 | + |
| 113 | + result = Utility.resolve_agent_identity(context, self.test_token) |
| 114 | + self.assertEqual(result, self.agentic_id) |
| 115 | + |
| 116 | + def test_resolve_agent_identity_with_non_agentic_request(self): |
| 117 | + """Test resolve_agent_identity with non-agentic request.""" |
| 118 | + activity = MockActivity(is_agentic=False) |
| 119 | + context = MockContext(activity) |
| 120 | + |
| 121 | + result = Utility.resolve_agent_identity(context, self.test_token) |
| 122 | + self.assertEqual(result, self.test_app_id) |
| 123 | + |
| 124 | + def test_resolve_agent_identity_with_context_without_activity(self): |
| 125 | + """Test resolve_agent_identity with context that has no activity.""" |
| 126 | + context = MockContext() |
| 127 | + |
| 128 | + result = Utility.resolve_agent_identity(context, self.test_token) |
| 129 | + self.assertEqual(result, self.test_app_id) |
| 130 | + |
| 131 | + def test_resolve_agent_identity_with_none_context(self): |
| 132 | + """Test resolve_agent_identity with None context.""" |
| 133 | + result = Utility.resolve_agent_identity(None, self.test_token) |
| 134 | + self.assertEqual(result, self.test_app_id) |
| 135 | + |
| 136 | + def test_resolve_agent_identity_with_agentic_but_empty_id(self): |
| 137 | + """Test resolve_agent_identity with agentic request but empty agentic ID.""" |
| 138 | + activity = MockActivity(is_agentic=True, agentic_id="") |
| 139 | + context = MockContext(activity) |
| 140 | + |
| 141 | + result = Utility.resolve_agent_identity(context, self.test_token) |
| 142 | + self.assertEqual(result, "") |
| 143 | + |
| 144 | + def test_resolve_agent_identity_fallback_on_exception(self): |
| 145 | + """Test resolve_agent_identity falls back to token when context access fails.""" |
| 146 | + |
| 147 | + # Create a context that will raise an exception when accessed |
| 148 | + class FaultyContext: |
| 149 | + @property |
| 150 | + def activity(self): |
| 151 | + raise RuntimeError("Context access failed") |
| 152 | + |
| 153 | + context = FaultyContext() |
| 154 | + result = Utility.resolve_agent_identity(context, self.test_token) |
| 155 | + self.assertEqual(result, self.test_app_id) |
| 156 | + |
| 157 | + |
| 158 | +if __name__ == "__main__": |
| 159 | + unittest.main() |
0 commit comments