[uss_qualifier] NetRIDServiceProvider function for querying user noti…

…fications

monitoring/monitorlib/fetch/__init__.py

@@ -366,6 +366,10 @@ class QueryType(str, Enum):
         "interuss.automated_testing.rid.v1.injection.deleteTest"
     )
 
+    InterussRIDAutomatedTestingV1UserNotifications = (
+        "interuss.automated_testing.rid.v1.injection.UserNotifications"
+    )
+
     # InterUSS mock_uss
     InterUSSMockUSSGetLogs = "interuss.mock_uss.logging.interaction_logs"
     InterUSSMockUSSGetLocality = "interuss.mock_uss.locality.locality_get"

monitoring/uss_qualifier/resources/netrid/service_providers.py

@@ -1,8 +1,9 @@
 import datetime
-from typing import List, Optional
+from typing import List, Optional, Tuple
 from urllib.parse import urlparse
 
 from implicitdict import ImplicitDict
+from uas_standards.interuss.automated_testing.rid.v1.injection import QueryUserNotificationsResponse
 
 from monitoring.monitorlib import fetch, infrastructure
 from monitoring.monitorlib.rid import RIDVersion
@@ -73,6 +74,29 @@ def delete_test(self, test_id: str, version: str) -> fetch.Query:
             query_type=fetch.QueryType.InterussRIDAutomatedTestingV1DeleteTest,
         )
 
+    def get_user_notifications(
+        self,
+        after: datetime.datetime,
+        before: datetime.datetime,
+    ) -> Tuple[QueryUserNotificationsResponse, fetch.Query]:
+        q = fetch.query_and_describe(
+            self.injection_client,
+            "GET",
+            url=f"/user_notifications",
+            scope=SCOPE_RID_QUALIFIER_INJECT,
+            participant_id=self.participant_id,
+            query_type=fetch.QueryType.InterussRIDAutomatedTestingV1UserNotifications,
+            params={
+                "after": after,
+                "before": before,
+            },
+        )
+
+        if q.error_message:
+            return QueryUserNotificationsResponse(user_notifications=[]), q
+
+        return ImplicitDict.parse(q.response.json, QueryUserNotificationsResponse), q
+
 
 class NetRIDServiceProviders(Resource[NetRIDServiceProvidersSpecification]):
     service_providers: List[NetRIDServiceProvider]