Merge pull request #41 from rohanpm/pulp-fake

Introduce --pulp-fake option for development/testing

CHANGELOG.md

@@ -7,7 +7,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
-- n/a
+- The `--pulp-fake` option was added for development and testing.
 
 ## [1.2.0] - 2021-08-18
 

pubtools/_pulp/services/pulp.py

@@ -51,6 +51,14 @@ def add_service_args(self, parser):
             default=None,
             type=pulp_throttle,
         )
+        group.add_argument(
+            "--pulp-fake",
+            help=(
+                "Use a fake in-memory Pulp client rather than interacting with a real server. "
+                + "For development/testing only, may have limited functionality."
+            ),
+            action="store_true",
+        )
 
     @property
     def pulp_client(self):
@@ -64,6 +72,11 @@ def __get_instance(self):
         auth = None
         args = self._service_args
 
+        if args.pulp_fake:
+            LOG.warning("Using a fake Pulp client, no changes will be made to Pulp!")
+            controller = pulplib.FakeController()
+            return controller.client
+
         # checks if pulp password is available as environment variable
         if args.pulp_user:
             pulp_password = args.pulp_password or os.environ.get("PULP_PASSWORD")

tests/shared/test_pulp_task.py

@@ -41,6 +41,23 @@ def test_pulp_client():
     assert isinstance(client, Client)
 
 
+def test_pulp_fake_client():
+    """Checks that a fake client is created if --pulp-fake is given"""
+    task = TaskWithPulpClient()
+    arg = ["", "--pulp-url", "https://pulp.example.com/", "--pulp-fake"]
+    with patch("sys.argv", arg):
+        client = task.pulp_client
+
+    # Fake client doesn't advertise itself in any obvious way.
+    # Just do some rough checks...
+    assert "Fake" in type(client).__name__
+
+    # Should be able to use the API even though it's obviously not connected
+    # to a real Pulp server
+    assert "rpm" in client.get_content_type_ids().result()
+    assert list(client.search_repository().result()) == []
+
+
 def test_main():
     """Checks main returns without exception when invoked with minimal args
     assuming run() and add_args() are implemented