Improve returned objects of Metadata API methods

Return dataclasses instead of untyped dicts.

Author: bjoernricks

gvm/protocols/http/openvasd/_metadata.py

@@ -6,13 +6,44 @@
 API wrapper for retrieving metadata from the openvasd HTTP API using HEAD requests.
 """
 
+from dataclasses import dataclass
 from typing import Union
 
 import httpx
 
 from ._api import OpenvasdAPI
 
 
+@dataclass
+class Metadata:
+    """
+    Represents metadata returned by the openvasd API.
+
+    Attributes:
+        api_version: Comma separated list of available API versions
+        feed_version: Version of the feed.
+        authentication: Supported authentication methods
+    """
+
+    api_version: str
+    feed_version: str
+    authentication: str
+
+
+@dataclass
+class MetadataError:
+    """
+    Represents an error response from the metadata API.
+
+    Attributes:
+        error: Error message.
+        status_code: HTTP status code of the error response.
+    """
+
+    error: str
+    status_code: int
+
+
 class MetadataAPI(OpenvasdAPI):
     """
     Provides access to metadata endpoints exposed by the openvasd server
@@ -27,20 +58,12 @@ class MetadataAPI(OpenvasdAPI):
     is handled gracefully.
     """
 
-    def get(self) -> dict[str, Union[str, int]]:
+    def get(self) -> Union[Metadata, MetadataError]:
         """
         Perform a HEAD request to `/` to retrieve top-level API metadata.
 
         Returns:
-            A dictionary containing:
-
-              - "api-version"
-              - "feed-version"
-              - "authentication"
-
-            Or if exceptions are suppressed and error occurs:
-
-              - {"error": str, "status_code": int}
+            A Metadata instance or MetadataError if exceptions are suppressed and an error occurs.
 
         Raises:
             httpx.HTTPStatusError: For non-401 HTTP errors if exceptions are not suppressed.
@@ -50,30 +73,24 @@ def get(self) -> dict[str, Union[str, int]]:
         try:
             response = self._client.head("/")
             response.raise_for_status()
-            return {
-                "api-version": response.headers.get("api-version"),
-                "feed-version": response.headers.get("feed-version"),
-                "authentication": response.headers.get("authentication"),
-            }
+            return Metadata(
+                api_version=response.headers.get("api-version"),
+                feed_version=response.headers.get("feed-version"),
+                authentication=response.headers.get("authentication"),
+            )
         except httpx.HTTPStatusError as e:
             if self._suppress_exceptions:
-                return {"error": str(e), "status_code": e.response.status_code}
+                return MetadataError(
+                    error=str(e), status_code=e.response.status_code
+                )
             raise
 
-    def get_scans(self) -> dict[str, Union[str, int]]:
+    def get_scans(self) -> Union[Metadata, MetadataError]:
         """
-         Perform a HEAD request to `/scans` to retrieve scan endpoint metadata.
+        Perform a HEAD request to `/scans` to retrieve scan endpoint metadata.
 
         Returns:
-             A dictionary containing:
-
-               - "api-version"
-               - "feed-version"
-               - "authentication"
-
-             Or if safe=True and error occurs:
-
-               - {"error": str, "status_code": int}
+            A Metadata instance or MetadataError if exceptions are suppressed and an error occurs.
 
          Raises:
              httpx.HTTPStatusError: For non-401 HTTP errors if exceptions are not suppressed.
@@ -83,12 +100,14 @@ def get_scans(self) -> dict[str, Union[str, int]]:
         try:
             response = self._client.head("/scans")
             response.raise_for_status()
-            return {
-                "api-version": response.headers.get("api-version"),
-                "feed-version": response.headers.get("feed-version"),
-                "authentication": response.headers.get("authentication"),
-            }
+            return Metadata(
+                api_version=response.headers.get("api-version"),
+                feed_version=response.headers.get("feed-version"),
+                authentication=response.headers.get("authentication"),
+            )
         except httpx.HTTPStatusError as e:
             if self._suppress_exceptions:
-                return {"error": str(e), "status_code": e.response.status_code}
+                return MetadataError(
+                    error=str(e), status_code=e.response.status_code
+                )
             raise

tests/protocols/http/openvasd/test_metadata.py

@@ -7,7 +7,11 @@
 
 import httpx
 
-from gvm.protocols.http.openvasd._metadata import MetadataAPI
+from gvm.protocols.http.openvasd._metadata import (
+    Metadata,
+    MetadataAPI,
+    MetadataError,
+)
 
 
 def _mock_head_response(status_code=200, headers=None):
@@ -35,7 +39,14 @@ def test_get_successful(self):
         result = api.get()
         self.mock_client.head.assert_called_once_with("/")
         mock_response.raise_for_status.assert_called_once()
-        self.assertEqual(result, headers)
+        self.assertEqual(
+            result,
+            Metadata(
+                api_version="1.0",
+                feed_version="2025.01",
+                authentication="API-KEY",
+            ),
+        )
 
     def test_get_unauthorized_with_suppress_exceptions(self):
         mock_response = MagicMock(spec=httpx.Response)
@@ -47,7 +58,9 @@ def test_get_unauthorized_with_suppress_exceptions(self):
 
         api = MetadataAPI(self.mock_client, suppress_exceptions=True)
         result = api.get()
-        self.assertEqual(result, {"error": "Unauthorized", "status_code": 401})
+        self.assertEqual(
+            result, MetadataError(error="Unauthorized", status_code=401)
+        )
 
     def test_get_failure_raises_httpx_error(self):
         mock_response = _mock_head_response(500, None)
@@ -77,7 +90,14 @@ def test_get_scans_successful(self):
         result = api.get_scans()
         self.mock_client.head.assert_called_once_with("/scans")
         mock_response.raise_for_status.assert_called_once()
-        self.assertEqual(result, headers)
+        self.assertEqual(
+            result,
+            Metadata(
+                api_version="1.0",
+                feed_version="2025.01",
+                authentication="API-KEY",
+            ),
+        )
 
     def test_get_scans_unauthorized_with_suppress_exceptions(self):
         mock_response = MagicMock(spec=httpx.Response)
@@ -89,7 +109,9 @@ def test_get_scans_unauthorized_with_suppress_exceptions(self):
 
         api = MetadataAPI(self.mock_client, suppress_exceptions=True)
         result = api.get_scans()
-        self.assertEqual(result, {"error": "Unauthorized", "status_code": 401})
+        self.assertEqual(
+            result, MetadataError(error="Unauthorized", status_code=401)
+        )
 
     def test_get_scans_failure_raises_httpx_error(self):
         mock_response = _mock_head_response(500, None)