|
3 | 3 | from unittest.mock import AsyncMock, MagicMock, patch |
4 | 4 |
|
5 | 5 | import httpx |
| 6 | +import pytest |
| 7 | +from pydantic import ValidationError |
6 | 8 |
|
7 | 9 | from a2a.server.tasks.base_push_notification_sender import ( |
8 | 10 | BasePushNotificationSender, |
@@ -130,6 +132,40 @@ async def test_send_notification_with_auth_header(self) -> None: |
130 | 132 | ) |
131 | 133 | mock_response.raise_for_status.assert_called_once() |
132 | 134 |
|
| 135 | + def test_authorization_header_no_credentials(self) -> None: |
| 136 | + auth = PushNotificationAuthenticationInfo( |
| 137 | + schemes=['Bearer'], credentials=None |
| 138 | + ) |
| 139 | + config = create_sample_push_config(authentication=auth) |
| 140 | + assert self.sender._authorization_header(config) is None |
| 141 | + |
| 142 | + def test_authorization_header_empty_schemes(self) -> None: |
| 143 | + auth = PushNotificationAuthenticationInfo( |
| 144 | + schemes=[], credentials='token' |
| 145 | + ) |
| 146 | + config = create_sample_push_config(authentication=auth) |
| 147 | + assert self.sender._authorization_header(config) is None |
| 148 | + |
| 149 | + def test_authorization_header_non_bearer_scheme(self) -> None: |
| 150 | + auth = PushNotificationAuthenticationInfo( |
| 151 | + schemes=['Basic'], credentials='token' |
| 152 | + ) |
| 153 | + config = create_sample_push_config(authentication=auth) |
| 154 | + assert self.sender._authorization_header(config) == 'Basic token' |
| 155 | + |
| 156 | + def test_authorization_header_filters_empty_schemes(self) -> None: |
| 157 | + auth = PushNotificationAuthenticationInfo( |
| 158 | + schemes=['', 'Bearer'], credentials='token' |
| 159 | + ) |
| 160 | + config = create_sample_push_config(authentication=auth) |
| 161 | + assert self.sender._authorization_header(config) == 'Bearer token' |
| 162 | + |
| 163 | + def test_authorization_header_none_scheme_rejected(self) -> None: |
| 164 | + with pytest.raises(ValidationError): |
| 165 | + PushNotificationAuthenticationInfo( |
| 166 | + schemes=['Bearer', None], credentials='token' |
| 167 | + ) |
| 168 | + |
133 | 169 | async def test_send_notification_no_config(self) -> None: |
134 | 170 | task_id = 'task_send_no_config' |
135 | 171 | task_data = create_sample_task(task_id=task_id) |
|
0 commit comments