Merge pull request #11 from release-engineering/offline_mappings

Support Local Mappings Provider

starmap_client/client.py

@@ -10,6 +10,7 @@
     QueryResponse,
     Workflow,
 )
+from starmap_client.providers import StarmapProvider
 from starmap_client.session import StarmapSession
 
 log = logging.getLogger(__name__)
@@ -22,7 +23,11 @@ class StarmapClient(object):
     """Number of policies to retrieve per call."""
 
     def __init__(
-        self, url: str, api_version: str = "v1", session_params: Optional[Dict[str, Any]] = None
+        self,
+        url: str,
+        api_version: str = "v1",
+        session_params: Optional[Dict[str, Any]] = None,
+        provider: Optional[StarmapProvider] = None,
     ):
         """
         Create a new StArMapClient.
@@ -35,13 +40,25 @@ def __init__(
                 The StArMap API version. Defaults to `v1`.
             session_params (dict, optional)
                 Additional keyword arguments for StarmapSession
+            provider (StarmapProvider, optional):
+                Object responsible to provide mappings locally. When set the client will be query it
+                first and if no mapping is found the subsequent request will be made to the server.
         """
         session_params = session_params or {}
         self.session = StarmapSession(url, api_version, **session_params)
+        self._provider = provider
         self._policies: List[Policy] = []
 
     def _query(self, params: Dict[str, Any]) -> Optional[QueryResponse]:
-        rsp = self.session.get("/query", params=params)
+        qr = None
+        if self._provider:
+            qr = self._provider.query(params)
+        rsp = qr or self.session.get("/query", params=params)
+        if isinstance(rsp, QueryResponse):
+            log.debug(
+                "Returning response from the local provider %s", self._provider.__class__.__name__
+            )
+            return rsp
         if rsp.status_code == 404:
             log.error(f"Marketplace mappings not defined for {params}")
             return None

starmap_client/providers/__init__.py

@@ -0,0 +1,3 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+from starmap_client.providers.base import StarmapProvider  # noqa: F401
+from starmap_client.providers.memory import InMemoryMapProvider  # noqa: F401

starmap_client/providers/base.py

@@ -0,0 +1,35 @@
+from abc import ABC, abstractmethod
+from typing import Any, Dict, List, Optional
+
+from starmap_client.models import QueryResponse
+
+
+class StarmapProvider(ABC):
+    """Define the interface for a local mappings provider."""
+
+    @abstractmethod
+    def query(self, params: Dict[str, Any]) -> Optional[QueryResponse]:
+        """Retrieve the mapping without using the server.
+
+        It relies in the local provider to retrieve the correct mapping
+        according to the parameters.
+
+        Args:
+            params (dict):
+                The request params to retrieve the mapping.
+        Returns:
+            The requested mapping when found.
+        """
+
+    @abstractmethod
+    def list_content(self) -> List[QueryResponse]:
+        """Return a list with all stored QueryResponse objects."""
+
+    @abstractmethod
+    def store(self, query_response: QueryResponse) -> None:
+        """Store a single query_response into the local provider.
+
+        Args:
+            query_response (query_response):
+                The object to store.
+        """

starmap_client/providers/memory.py

@@ -0,0 +1,68 @@
+from typing import Any, Dict, List, Optional
+
+from starmap_client.models import QueryResponse
+from starmap_client.providers.base import StarmapProvider
+from starmap_client.providers.utils import get_image_name
+
+
+class InMemoryMapProvider(StarmapProvider):
+    """Provide in memory (RAM) QueryResponse mapping objects."""
+
+    def __init__(
+        self, map_responses: Optional[List[QueryResponse]] = None, *args, **kwargs
+    ) -> None:
+        """Crete a new InMemoryMapProvider object.
+
+        Args:
+            map_responses (list, optional)
+                List of QueryResponse objects to load into memory. They will be
+                used by query to fetch the correct response based on name
+                and workflow.
+        """
+        self._separator = str(kwargs.pop("separator", "+"))
+        self._content: Dict[str, QueryResponse] = {}
+        super(StarmapProvider, self).__init__()
+        self._boostrap(map_responses)
+
+    def _boostrap(self, map_responses: Optional[List[QueryResponse]]) -> None:
+        """Initialize the internal content dictionary.
+
+        Args:
+            map_responses (list, optional)
+                List of QueryResponse objects to load into memory.
+        """
+        if not map_responses:
+            return None
+
+        # The in memory content is made of a combination of name and workflow
+        for map in map_responses:
+            key = f"{map.name}{self._separator}{map.workflow.value}"
+            self._content[key] = map
+
+    def list_content(self) -> List[QueryResponse]:
+        """Return a list of stored content."""
+        return list(self._content.values())
+
+    def store(self, query_response: QueryResponse) -> None:
+        """Store/replace a single QueryResponse object.
+
+        Args:
+            query_response (query_response):
+                The object to store.
+        """
+        key = f"{query_response.name}{self._separator}{query_response.workflow.value}"
+        self._content[key] = query_response
+
+    def query(self, params: Dict[str, Any]) -> Optional[QueryResponse]:
+        """Return the mapping from memory according to the received params.
+
+        Args:
+            params (dict):
+                The request params to retrieve the mapping.
+        Returns:
+            The requested mapping when found.
+        """
+        name = params.get("name") or get_image_name(params.get("image"))
+        workflow = str(params.get("workflow", ""))
+        search_key = f"{name}{self._separator}{workflow}"
+        return self._content.get(search_key)

starmap_client/providers/utils.py

@@ -0,0 +1,77 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+# The functions below were adapted from Kobo's RPMLib:
+# https://github.com/release-engineering/kobo/blob/master/kobo/rpmlib.py
+import logging
+from typing import Dict, Optional, Tuple
+
+log = logging.getLogger(__name__)
+
+
+def split_nvr_epoch(nvre: str) -> Tuple[str, str]:
+    """
+    Split nvre to N-V-R and E.
+
+    :param nvre: E:N-V-R or N-V-R:E string
+    :return: (N-V-R, E)
+    """
+    if ":" in nvre:
+        log.debug("Splitting NVR and Epoch")
+        if nvre.count(":") != 1:
+            raise RuntimeError(f"Invalid NVRE: {nvre}")
+
+        nvr, epoch = nvre.rsplit(":", 1)
+        if "-" in epoch:
+            if "-" not in nvr:
+                # switch nvr with epoch
+                nvr, epoch = epoch, nvr
+            else:
+                # it's probably N-E:V-R format, handle it after the split
+                nvr, epoch = nvre, ""
+    else:
+        log.debug("No epoch to split")
+        nvr, epoch = nvre, ""
+
+    return nvr, epoch
+
+
+def parse_nvr(nvre: str) -> Dict[str, str]:
+    """
+    Split N-V-R into a dictionary.
+
+    :param nvre: N-V-R:E, E:N-V-R or N-E:V-R string
+    :return: {name, version, release, epoch}
+    """
+    log.debug("Parsing NVR")
+    if "/" in nvre:
+        nvre = nvre.split("/")[-1]
+
+    nvr, epoch = split_nvr_epoch(nvre)
+
+    log.debug("Splitting NVR parts")
+    nvr_parts = nvr.rsplit("-", 2)
+    if len(nvr_parts) != 3:
+        raise RuntimeError(f"Invalid NVR: {nvr}")
+
+    # parse E:V
+    if epoch == "" and ":" in nvr_parts[1]:
+        log.debug("Parsing E:V")
+        epoch, nvr_parts[1] = nvr_parts[1].split(":", 1)
+
+    # check if epoch is empty or numeric
+    if epoch != "":
+        try:
+            int(epoch)
+        except ValueError:
+            raise RuntimeError(f"Invalid epoch '{epoch}' in '{nvr}'")
+
+    result = dict(zip(["name", "version", "release"], nvr_parts))
+    result["epoch"] = epoch
+    return result
+
+
+def get_image_name(image: Optional[str]) -> str:
+    """Retrieve the name from NVR."""
+    if not image:
+        return ""
+    nvr = parse_nvr(image)
+    return nvr.get("name", "")

tests/test_client.py

@@ -10,6 +10,7 @@
 
 from starmap_client import StarmapClient
 from starmap_client.models import Destination, Mapping, Policy, QueryResponse
+from starmap_client.providers import InMemoryMapProvider
 
 
 def load_json(json_file: str) -> Any:
@@ -56,6 +57,17 @@ def test_query_image_success(self):
         # Note: JSON need to be loaded twice as `from_json` pops its original data
         self.assertEqual(res, QueryResponse.from_json(load_json(fpath)))
 
+    def test_in_memory_query_image(self):
+        fpath = "tests/data/query/valid_quer1.json"
+        data = [QueryResponse.from_json(load_json(fpath))]
+        provider = InMemoryMapProvider(data)
+
+        self.svc = StarmapClient("https://test.starmap.com", api_version="v1", provider=provider)
+
+        res = self.svc.query_image("sample-policy-1.0-1.raw.xz")
+        self.mock_session.get.assert_not_called()
+        self.assertEqual(res, data[0])
+
     def test_query_image_not_found(self):
         self.mock_session.get.return_value = self.mock_resp_not_found
 
@@ -83,6 +95,17 @@ def test_query_image_by_name(self):
         # Note: JSON need to be loaded twice as `from_json` pops its original data
         self.assertEqual(res, QueryResponse.from_json(load_json(fpath)))
 
+    def test_in_memory_query_image_by_name(self):
+        fpath = "tests/data/query/valid_quer1.json"
+        data = [QueryResponse.from_json(load_json(fpath))]
+        provider = InMemoryMapProvider(data)
+
+        self.svc = StarmapClient("https://test.starmap.com", api_version="v1", provider=provider)
+
+        res = self.svc.query_image_by_name(name="sample-policy")
+        self.mock_session.get.assert_not_called()
+        self.assertEqual(res, data[0])
+
     def test_query_image_by_name_version(self):
         fpath = "tests/data/query/valid_quer1.json"
         self.mock_resp_success.json.return_value = load_json(fpath)

tests/test_providers/conftest.py

@@ -0,0 +1,75 @@
+from typing import Any, Dict
+
+import pytest
+
+from starmap_client.models import QueryResponse
+
+
+@pytest.fixture
+def qr1() -> Dict[str, Any]:
+    return {
+        "mappings": {
+            "aws-na": [
+                {
+                    "architecture": "x86_64",
+                    "destination": "ffffffff-ffff-ffff-ffff-ffffffffffff",
+                    "overwrite": True,
+                    "restrict_version": False,
+                    "meta": {"tag1": "aws-na-value1", "tag2": "aws-na-value2"},
+                    "tags": {"key1": "value1", "key2": "value2"},
+                }
+            ],
+            "aws-emea": [
+                {
+                    "architecture": "x86_64",
+                    "destination": "00000000-0000-0000-0000-000000000000",
+                    "overwrite": True,
+                    "restrict_version": False,
+                    "meta": {"tag1": "aws-emea-value1", "tag2": "aws-emea-value2"},
+                    "tags": {"key3": "value3", "key4": "value4"},
+                }
+            ],
+        },
+        "name": "sample-product",
+        "workflow": "stratosphere",
+    }
+
+
+@pytest.fixture
+def qr2() -> Dict[str, Any]:
+    return {
+        "mappings": {
+            "aws-na": [
+                {
+                    "architecture": "x86_64",
+                    "destination": "test-dest-1",
+                    "overwrite": True,
+                    "restrict_version": False,
+                    "meta": {"tag1": "aws-na-value1", "tag2": "aws-na-value2"},
+                    "tags": {"key1": "value1", "key2": "value2"},
+                }
+            ],
+            "aws-emea": [
+                {
+                    "architecture": "x86_64",
+                    "destination": "test-dest-2",
+                    "overwrite": True,
+                    "restrict_version": False,
+                    "meta": {"tag1": "aws-emea-value1", "tag2": "aws-emea-value2"},
+                    "tags": {"key3": "value3", "key4": "value4"},
+                }
+            ],
+        },
+        "name": "sample-product",
+        "workflow": "community",
+    }
+
+
+@pytest.fixture
+def qr1_object(qr1) -> QueryResponse:
+    return QueryResponse.from_json(qr1)
+
+
+@pytest.fixture
+def qr2_object(qr2) -> QueryResponse:
+    return QueryResponse.from_json(qr2)