Merge pull request #4 from synkd/add_testing_capabilities

Modify Manifester class to work with MockStub

.github/workflows/codeql-analysis.yml

@@ -5,6 +5,10 @@ on:
     branches: [master]
   pull_request:
     types: [opened, synchronize, reopened]
+    paths-ignore:
+        - "*.md"
+        - "*.example"
+        - ".gitignore"
 
 jobs:
   analyze:
@@ -14,22 +18,24 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.8", "3.9", "3.10"]
+        python-version: ["3.10", "3.11", "3.12"]
 
     steps:
     - name: Checkout repository
-      uses: actions/checkout@v3
+      uses: actions/checkout@v4
+      with:
+        ref: ${{ github.ref }}
 
     - name: Initialize CodeQL
-      uses: github/codeql-action/init@v1
+      uses: github/codeql-action/init@v2
       with:
         languages: ${{ matrix.language }}
 
     - name: Perform CodeQL Analysis
-      uses: github/codeql-action/analyze@v1
+      uses: github/codeql-action/analyze@v2
 
     - name: Setup Python
-      uses: actions/setup-python@v2
+      uses: actions/setup-python@v4
       with:
         python-version: ${{ matrix.python-version }}
 
@@ -38,5 +44,4 @@ jobs:
         pip install -U pip
         pip install -U .[test]
         cp manifester_settings.yaml.example manifester_settings.yaml
-        manifester --help
-        # pytest -v tests/ --ignore tests/functional
+        pytest -v tests/

.github/workflows/python-publish.yml

@@ -12,12 +12,12 @@ jobs:
     strategy:
       matrix:
         # build/push in lowest support python version
-        python-version: [ 3.8 ]
+        python-version: [ 3.10 ]
 
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v4
 
-    - uses: actions/setup-python@v2
+    - uses: actions/setup-python@v4
       with:
         python-version: ${{ matrix.python-version }}
 
@@ -29,7 +29,7 @@ jobs:
         python -m twine check dist/*
 
     - name: Build and publish
-      uses: pypa/gh-action-pypi-publish@release/v1
+      uses: pypa/gh-action-pypi-publish@v1.8.11
       with:
         user: __token__
         password: ${{ secrets.PYPI_TOKEN }}

.github/workflows/update_manifester_image.yml

@@ -11,16 +11,16 @@ jobs:
 
     steps:
       - name: Checkout
-        uses: actions/checkout@v2
+        uses: actions/checkout@v4
 
       - name: Set up QEMU
-        uses: docker/setup-qemu-action@v1
+        uses: docker/setup-qemu-action@v3
 
       - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v1
+        uses: docker/setup-buildx-action@v3
 
       - name: Login to Quay Container Registry
-        uses: docker/login-action@v1
+        uses: docker/login-action@v3
         with:
           registry: ${{ secrets.QUAY_SERVER }}
           username: ${{ secrets.QUAY_USERNAME }}

.pre-commit-config.yaml

@@ -1,27 +1,19 @@
 # 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/astral-sh/ruff-pre-commit
+    rev: v0.1.6
+    hooks:
+      - id: ruff
+        args: [--fix, --exit-non-zero-on-fix]
+      - id: ruff-format
+  - repo: https://github.com/gitleaks/gitleaks
+    rev: v8.18.0
+    hooks:
+      - id: gitleaks

manifester/commands.py

@@ -1,3 +1,4 @@
+"""Defines the CLI commands for Manifester."""
 import click
 
 from manifester import Manifester
@@ -6,6 +7,7 @@
 # To do: add a command for returning subscription pools
 @click.group
 def cli():
+    """Command-line interface for manifester."""
     pass
 
 
@@ -15,10 +17,9 @@ 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):
+    """Return a subscription manifester based on the settings for the provided manifest_category."""
     manifester = Manifester(manifest_category, allocation_name)
     manifester.create_subscription_allocation()
     for sub in manifester.subscription_data:

manifester/helpers.py

@@ -1,10 +1,13 @@
+"""Defines helper functions used by Manifester."""
+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"""
+    """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,26 +19,81 @@ 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"""
+    """Ensure that the sat_version parameter is properly formatted for the RHSM API."""
+    expected_length = 8
     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
         # version (e.g. 'sat-6.11.0') when retrieving this value from settings. The conditional
         # below assumes that, if the length of sat_version is greated than 8 characters, it includes
         # a Z-stream version that should be removed.
-        if len(sat_version) > 8:
-            sat_version = sat_version.split('.')
+        if len(sat_version) > expected_length:
+            sat_version = sat_version.split(".")
             sat_version = sat_version[0:2]
             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=0):
+    """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):
+        """Fallback to returning self if attribute doesn't exist."""
+        return self
+
+    def __getitem__(self, key):
+        """Get an item from the dictionary-like object.
+
+        If the key is a string, this method will attempt to get an attribute with that name.
+        If the key is not found, this method will return the object itself.
+        """
+        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):
+        """Allow MockStub to be used like a function."""
+        return self

manifester/logger.py

@@ -1,3 +1,4 @@
+"""Defines manifester's internal logging."""
 import logging
 from pathlib import Path
 
@@ -7,26 +8,22 @@
 
 
 def setup_logzero(level="info", path="logs/manifester.log", silent=True):
+    """Call logzero setup with the given settings."""
     Path(path).parent.mkdir(parents=True, exist_ok=True)
     log_fmt = "%(color)s[%(levelname)s %(asctime)s]%(end_color)s %(message)s"
     debug_fmt = (
-        "%(color)s[%(levelname)1.1s %(asctime)s %(module)s:%(lineno)d]"
-        "%(end_color)s %(message)s"
+        "%(color)s[%(levelname)1.1s %(asctime)s %(module)s:%(lineno)d]%(end_color)s %(message)s"
     )
     log_level = getattr(logging, level.upper(), logging.INFO)
     # formatter for terminal
-    formatter = logzero.LogFormatter(
-        fmt=debug_fmt if log_level is logging.DEBUG else log_fmt
-    )
+    formatter = logzero.LogFormatter(fmt=debug_fmt if log_level is logging.DEBUG else log_fmt)
     logzero.setup_default_logger(formatter=formatter, disableStderrLogger=silent)
     logzero.loglevel(log_level)
     # formatter for file
     formatter = logzero.LogFormatter(
         fmt=debug_fmt if log_level is logging.DEBUG else log_fmt, color=False
     )
-    logzero.logfile(
-        path, loglevel=log_level, maxBytes=1e9, backupCount=3, formatter=formatter
-    )
+    logzero.logfile(path, loglevel=log_level, maxBytes=1e9, backupCount=3, formatter=formatter)
 
 
 setup_logzero(level=settings.get("log_level", "info"))