Adds --repo-file, can be used to load info from a json file instead o…

…f string.

docs/common/mappings.rst

@@ -4,7 +4,8 @@ Destination Mapping
 The destination mapping is given by an internal service named ``StArMap`` throught the `Starmap Client`_ library.
 
 The tool expects a valid endpoint to the StArMap service and/or a JSON string containing the list of mappings
-using the parameter ``--repo``.
+using the parameter ``--repo``. Instead of using ``--repo`` you can instead use ``--repo-file`` and provide
+a json file to be parsed.
 
 The following sections will cover all the details about the destination mappings.
 

src/pubtools/_marketplacesvm/arguments.py

@@ -166,18 +166,43 @@ class RepoQueryLoad(Action):
     This action is intended to allow the optional load of mappings right in the command call
     instead of having to request data from server.
 
-    It will evaluate the input data and set it as a StArMap's QueryResponse.
+    It will evaluate the input data and set it as a StArMap's QueryResponseContainer.
     """
 
     def __init__(self, *args, **kwargs):
         """Instantiate the RepoQueryLoad action."""
         super(RepoQueryLoad, self).__init__(*args, **kwargs)
 
     def __call__(self, parser, namespace, values, options=None):
-        """Convert the received args into QueryResponse."""
+        """Convert the received args into QueryResponseContainer."""
         items = getattr(namespace, self.dest, None) or []
         if values and isinstance(values, str):
             items = json.loads(values)
             if not isinstance(items, list):
                 raise ArgumentError(self, f"Expected value to be a list, got: {type(items)}")
         setattr(namespace, self.dest, items)
+
+
+class RepoFileQueryLoad(Action):
+    """
+    Argparse Action subclass for loading StArMap mappings from the ``repo-file`` argument.
+
+    This action is intended to allow the optional load of mappings from a json file
+    instead of having to request data from server.
+
+    It will evaluate the input file and set it as a StArMap's QueryResponseContainer.
+    """
+
+    def __init__(self, *args, **kwargs):
+        """Instantiate the RepoQueryLoad action."""
+        super(RepoFileQueryLoad, self).__init__(*args, **kwargs)
+
+    def __call__(self, parser, namespace, values, options=None):
+        """Convert the received args into QueryResponseContainer."""
+        items = getattr(namespace, self.dest, None) or []
+        if values and isinstance(values, str):
+            with open(values, "r") as v:
+                items = json.load(v)
+                if not isinstance(items, list):
+                    raise ArgumentError(self, f"Expected value to be a list, got: {type(items)}")
+        setattr(namespace, self.dest, items)

src/pubtools/_marketplacesvm/services/starmap.py

@@ -8,7 +8,7 @@
 from starmap_client.providers import InMemoryMapProviderV2
 from starmap_client.session import StarmapMockSession, StarmapSession
 
-from ..arguments import RepoQueryLoad
+from ..arguments import RepoFileQueryLoad, RepoQueryLoad
 from .base import Service
 
 log = logging.getLogger("pubtools.marketplacesvm")
@@ -57,6 +57,13 @@ def add_service_args(self, parser: ArgumentParser) -> None:
             action=RepoQueryLoad,
         )
 
+        group.add_argument(
+            "--repo-file",
+            help="Override the StArMap mappings for push items with json file.",
+            type=str,
+            action=RepoFileQueryLoad,
+        )
+
         group.add_argument(
             "--offline",
             help="Do not connect to a real StArMap server, use a mock session instead. "
@@ -65,12 +72,12 @@ def add_service_args(self, parser: ArgumentParser) -> None:
         )
 
     def _get_repo(self) -> Dict[str, Any]:
-        """Instantiate the InMemoryMapProvider when ``--repo`` is passed.
+        """Instantiate the InMemoryMapProvider when ``--repo`` or ``--repo-file`` is passed.
 
         This will make starmap_client load the list of mappings from memory
         first and only call the server whenever the local mapping is not found.
         """
-        local_mappings = self._service_args.repo
+        local_mappings = self._service_args.repo or self._service_args.repo_file
         if local_mappings:
             self._container = QueryResponseContainer.from_json(local_mappings)
             provider = InMemoryMapProviderV2(container=self._container)

tests/combined_push/test_combined_push.py

@@ -398,6 +398,7 @@ def test_do_combined_push_overriden_destination(
     starmap_query_aws_community: QueryResponseContainer,
     command_tester: CommandTester,
     monkeypatch: pytest.MonkeyPatch,
+    tmpdir: pytest.Testdir,
 ) -> None:
     """Test a successfull combined push for marketplaces and community workflows."""
     # Store the auto-assigned mocks for StArMap on both workflows
@@ -444,3 +445,65 @@ def test_do_combined_push_overriden_destination(
     # Ensure the "server" was not called
     mock_starmap_mkt.query_image_by_name.assert_not_called()
     mock_starmap_cmt.query_image_by_name.assert_not_called()
+
+
+@mock.patch("pubtools._marketplacesvm.tasks.community_push.command.Source")
+@mock.patch("pubtools._marketplacesvm.tasks.push.command.Source")
+def test_do_combined_push_overriden_destination_file(
+    marketplace_source: mock.MagicMock,
+    community_source: mock.MagicMock,
+    ami_push_item: AmiPushItem,
+    starmap_query_aws_marketplace: QueryResponseContainer,
+    starmap_query_aws_community: QueryResponseContainer,
+    command_tester: CommandTester,
+    monkeypatch: pytest.MonkeyPatch,
+    tmpdir: pytest.Testdir,
+) -> None:
+    """Test a successfull combined push for marketplaces and community workflows."""
+    # Store the auto-assigned mocks for StArMap on both workflows
+    mock_starmap_mkt = MarketplacesVMPush.starmap
+    mock_starmap_cmt = CommunityVMPush.starmap
+
+    # The policy name must be the same for community and marketplace workflows
+    qre = starmap_query_aws_marketplace.responses.pop()
+    qre = evolve(qre, name="sample_product")
+    starmap_query_aws_marketplace.responses.append(qre)
+    binfo = KojiBuildInfo(name="sample_product", version="7.0", release="20230101")
+    ami_push_item = evolve(ami_push_item, build_info=binfo)
+
+    # Create a testing StArMap instance which will fail to resolve the server
+    # it is, it can only be used offline
+    responses = starmap_query_aws_community.responses + starmap_query_aws_marketplace.responses
+    provider = InMemoryMapProviderV2(QueryResponseContainer(responses))
+    starmap = StarmapClient("https://foo.com/bar", provider=provider)
+    monkeypatch.setattr(MarketplacesVMPush, 'starmap', starmap)
+    monkeypatch.setattr(CommunityVMPush, 'starmap', starmap)
+
+    # Add the push items in the queue
+    marketplace_source.get.return_value.__enter__.return_value = [ami_push_item]
+    community_source.get.return_value.__enter__.return_value = [ami_push_item]
+
+    p = tmpdir.mkdir('data').join('test.json')
+    p.write(json.dumps([asdict(x) for x in responses], default=str))
+
+    # Test file
+    command_tester.test(
+        lambda: entry_point(CombinedVMPush),
+        [
+            "test-push",
+            "--starmap-url",
+            "https://starmap-example.com",
+            "--credentials",
+            "eyJtYXJrZXRwbGFjZV9hY2NvdW50IjogInRlc3QtbmEiLCAiYXV0aCI6eyJmb28iOiJiYXIifQo=",
+            "--rhsm-url",
+            "https://rhsm.com/test/api/",
+            "--repo-file",
+            f"{p}",
+            "--debug",
+            "koji:https://fakekoji.com?vmi_build=ami_build",
+        ],
+    )
+
+    # Ensure the "server" was not called
+    mock_starmap_mkt.query_image_by_name.assert_not_called()
+    mock_starmap_cmt.query_image_by_name.assert_not_called()

tests/community_push/test_community_push.py

@@ -222,6 +222,7 @@ def test_do_community_push_overridden_destination(
     command_tester: CommandTester,
     starmap_ami_billing_config: Dict[str, Any],
     ami_push_item: AmiPushItem,
+    tmpdir: pytest.Testdir,
 ) -> None:
     """Test a community push success with the destinations overriden from command line."""
     binfo = KojiBuildInfo(name="sample-product", version="7.0", release="20230101")
@@ -278,6 +279,73 @@ def test_do_community_push_overridden_destination(
     )
 
 
+@mock.patch("pubtools._marketplacesvm.tasks.community_push.command.Source")
+def test_do_community_push_overridden_destination_file(
+    mock_source: mock.MagicMock,
+    fake_cloud_instance: mock.MagicMock,
+    command_tester: CommandTester,
+    starmap_ami_billing_config: Dict[str, Any],
+    ami_push_item: AmiPushItem,
+    tmpdir: pytest.Testdir,
+) -> None:
+    """Test a community push success with the destinations overriden from command line."""
+    binfo = KojiBuildInfo(name="sample-product", version="7.0", release="20230101")
+    ami_push_item = evolve(ami_push_item, build_info=binfo)
+    mock_source.get.return_value.__enter__.return_value = [ami_push_item]
+
+    policy = [
+        {
+            "mappings": {
+                "aws-na": {
+                    "destinations": [
+                        {
+                            "destination": "new_aws-na_destination-access",
+                            "overwrite": False,
+                            "restrict_version": False,
+                        }
+                    ],
+                    "provider": "awstest",
+                },
+                "aws-emea": {
+                    "destinations": [
+                        {
+                            "destination": "new_aws-emea_destination-hourly",
+                            "overwrite": True,
+                            "restrict_version": False,
+                        }
+                    ],
+                    "provider": "awstest",
+                },
+            },
+            "billing-code-config": starmap_ami_billing_config,
+            "cloud": "aws",
+            "meta": {"release": {"type": "ga"}},
+            "name": "sample-product",
+            "workflow": "community",
+        }
+    ]
+
+    p = tmpdir.mkdir('data').join('test.json')
+    p.write(json.dumps(policy))
+
+    command_tester.test(
+        lambda: entry_point(CommunityVMPush),
+        [
+            "test-push",
+            "--starmap-url",
+            "https://starmap-example.com",
+            "--credentials",
+            "eyJtYXJrZXRwbGFjZV9hY2NvdW50IjogInRlc3QtbmEiLCAiYXV0aCI6eyJmb28iOiJiYXIifQo=",
+            "--repo-file",
+            f"{p}",
+            "--rhsm-url",
+            "https://rhsm.com/test/api/",
+            "--debug",
+            "koji:https://fakekoji.com?vmi_build=ami_build",
+        ],
+    )
+
+
 @mock.patch("pubtools._marketplacesvm.tasks.community_push.command.Source")
 def test_do_community_push_offline_starmap(
     mock_source: mock.MagicMock,

tests/logs/combined_push/test_combined_push/test_do_combined_push_overriden_destination_file.jsonl

@@ -0,0 +1 @@
+