Modify Manifester class to work with MockStub

.pre-commit-config.yaml

@@ -1,27 +1,22 @@
 # configuration for pre-commit git hooks
 
 repos:
-- repo: https://github.com/asottile/reorder_python_imports
-  rev: v3.0.1
-  hooks:
-  - id: reorder-python-imports
-- repo: https://github.com/asottile/pyupgrade
-  rev: v2.32.0
-  hooks:
-  - id: pyupgrade
-    args: [--py36-plus]
-- repo: https://github.com/psf/black
-  rev: 22.3.0
-  hooks:
-  - id: black
-- repo: https://gitlab.com/pycqa/flake8
-  rev: 3.9.2
-  hooks:
-  - id: flake8
-- repo: https://github.com/pre-commit/pre-commit-hooks
-  rev: v4.2.0
-  hooks:
-  - id: trailing-whitespace
-  - id: end-of-file-fixer
-  - id: check-yaml
-  - id: debug-statements
+  - repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v4.4.0
+    hooks:
+    - id: trailing-whitespace
+    - id: check-yaml
+    - id: debug-statements
+  - repo: https://github.com/psf/black
+    rev: 22.10.0
+    hooks:
+    - id: black
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    rev: v0.0.277
+    hooks:
+      - id: ruff
+        args: [--fix, --exit-non-zero-on-fix]
+  - repo: https://github.com/gitleaks/gitleaks
+    rev: v8.18.0
+    hooks:
+      - id: gitleaks

manifester/commands.py

@@ -15,9 +15,7 @@ def cli():
     type=str,
     help="Category of manifest (golden_ticket or robottelo_automation by default)",
 )
-@click.option(
-    "--allocation_name", type=str, help="Name of upstream subscription allocation"
-)
+@click.option("--allocation_name", type=str, help="Name of upstream subscription allocation")
 def get_manifest(manifest_category, allocation_name):
     manifester = Manifester(manifest_category, allocation_name)
     manifester.create_subscription_allocation()

manifester/helpers.py

@@ -1,10 +1,13 @@
+from collections import UserDict
+import random
 import time
 
 from logzero import logger
 
 
 def simple_retry(cmd, cmd_args=None, cmd_kwargs=None, max_timeout=240, _cur_timeout=1):
     """Re(Try) a function given its args and kwargs up until a max timeout"""
+
     cmd_args = cmd_args if cmd_args else []
     cmd_kwargs = cmd_kwargs if cmd_kwargs else {}
     # If additional debug information is needed, the following log entry can be modified to
@@ -16,15 +19,16 @@ def simple_retry(cmd, cmd_args=None, cmd_kwargs=None, max_timeout=240, _cur_time
     if response.status_code in [429, 500, 504]:
         new_wait = _cur_timeout * 2
         if new_wait > max_timeout:
-            raise Exception("Timeout exceeded")
+            raise Exception("Retry timeout exceeded")
         logger.debug(f"Trying again in {_cur_timeout} seconds")
         time.sleep(_cur_timeout)
         response = simple_retry(cmd, cmd_args, cmd_kwargs, max_timeout, new_wait)
     return response
 
+
 def process_sat_version(sat_version, valid_sat_versions):
     """Ensure that the sat_version parameter is properly formatted for the RHSM API when creating
-       a subscription allocation with the 'POST allocations' endpoint"""
+    a subscription allocation with the 'POST allocations' endpoint"""
     if sat_version not in valid_sat_versions:
         # The valid values for the sat_version parameter when creating a subscription allocation
         # are all 8 characters or less (e.g. 'sat-6.11'). Some data sources may include a Z-stream
@@ -37,5 +41,53 @@ def process_sat_version(sat_version, valid_sat_versions):
             sat_version = ".".join(sat_version)
         # If sat_version is still not valid, default to the latest valid version.
         if sat_version not in valid_sat_versions:
-            valid_sat_versions.sort(key = lambda i: int(i.split('-')[-1].split('.')[-1]), reverse = True)
-    return valid_sat_versions[0]
+            valid_sat_versions.sort(
+                key=lambda i: int(i.split('-')[-1].split('.')[-1]), reverse=True
+            )
+            return valid_sat_versions[0]
+    return sat_version
+
+
+def fake_http_response_code(good_codes=None, bad_codes=None, fail_rate=20):
+    """Return an HTTP response code randomly selected from sets of good and bad codes"""
+
+    if random.random() > (fail_rate / 100):
+        return random.choice(good_codes)
+    else:
+        return random.choice(bad_codes)
+
+
+class MockStub(UserDict):
+    """Test helper class. Allows for both arbitrary mocking and stubbing"""
+
+    def __init__(self, in_dict=None):
+        """Initialize the class and all nested dictionaries"""
+        if in_dict is None:
+            in_dict = {}
+        for key, value in in_dict.items():
+            if isinstance(value, dict):
+                setattr(self, key, MockStub(value))
+            elif type(value) in (list, tuple):
+                setattr(
+                    self,
+                    key,
+                    [MockStub(x) if isinstance(x, dict) else x for x in value],
+                )
+            else:
+                setattr(self, key, value)
+        super().__init__(in_dict)
+
+    def __getattr__(self, name):
+        return self
+
+    def __getitem__(self, key):
+        if isinstance(key, str):
+            item = getattr(self, key, self)
+        try:
+            item = super().__getitem__(key)
+        except KeyError:
+            item = self
+        return item
+
+    def __call__(self, *args, **kwargs):
+        return self