Cherrypick-6.11.z Manifest Refactor

pytest_fixtures/component/taxonomy.py

@@ -77,9 +77,9 @@ def module_org_with_manifest(module_org):
 
 
 @pytest.fixture(scope='module')
-def module_entitlement_manifest_org(module_org, module_entitlement_manifest):
+def module_entitlement_manifest_org(module_org, module_entitlement_manifest, module_target_sat):
     """Creates an organization and uploads an entitlement mode manifest generated with manifester"""
-    upload_manifest(module_org.id, module_entitlement_manifest.content)
+    module_target_sat.upload_manifest(module_org.id, module_entitlement_manifest.content)
     return module_org
 
 
@@ -90,10 +90,23 @@ def module_sca_manifest_org(module_org, module_sca_manifest):
     return module_org
 
 
+@pytest.fixture(scope='module')
+def module_extra_rhel_entitlement_manifest_org(
+    module_target_sat,
+    module_org,
+    module_extra_rhel_entitlement_manifest,
+):
+    """Creates an organization and uploads an entitlement mode manifest generated by manifester and
+    containing more RHEL entitlements than the default entitlement manifest"""
+    module_org.sca_disable()
+    module_target_sat.upload_manifest(module_org.id, module_extra_rhel_entitlement_manifest.content)
+    return module_org
+
+
 @pytest.fixture(scope='function')
-def function_entitlement_manifest_org(function_org, function_entitlement_manifest):
+def function_entitlement_manifest_org(function_org, function_entitlement_manifest, target_sat):
     """Creates an organization and uploads an entitlement mode manifest generated with manifester"""
-    upload_manifest(function_org.id, function_entitlement_manifest.content)
+    target_sat.upload_manifest(function_org.id, function_entitlement_manifest.content)
     return function_org
 
 
@@ -106,7 +119,7 @@ def upgrade_entitlement_manifest_org(function_org, upgrade_entitlement_manifest,
     return function_org
 
 
-@pytest.fixture
+@pytest.fixture(scope='function')
 def function_sca_manifest_org(function_org, function_sca_manifest, target_sat):
     """Creates an organization and uploads an SCA mode manifest generated with manifester"""
     upload_manifest(function_org.id, function_sca_manifest.content)

robottelo/cli/factory.py

@@ -66,6 +66,7 @@
 from robottelo.cli.virt_who_config import VirtWhoConfig
 from robottelo.config import settings
 from robottelo.logging import logger
+from robottelo.utils import clone
 from robottelo.utils import ssh
 from robottelo.utils.datafactory import valid_cron_expressions
 from robottelo.utils.decorators import cacheable
@@ -1735,7 +1736,7 @@ def setup_org_for_a_custom_repo(options=None):
     }
 
 
-def _setup_org_for_a_rh_repo(options=None):
+def _setup_org_for_a_rh_repo(target_sat, options=None):
     """Sets up Org for the given Red Hat repository by:
 
     1. Checks if organization and lifecycle environment were given, otherwise
@@ -1774,10 +1775,10 @@ def _setup_org_for_a_rh_repo(options=None):
     else:
         env_id = options['lifecycle-environment-id']
     # Clone manifest and upload it
-    with manifests.clone() as manifest:
-        ssh.get_client().put(manifest, manifest.filename)
+    with clone() as manifest:
+        target_sat.put(manifest.path, manifest.name)
     try:
-        Subscription.upload({'file': manifest.filename, 'organization-id': org_id})
+        Subscription.upload({'file': manifest.name, 'organization-id': org_id})
     except CLIReturnCodeError as err:
         raise CLIFactoryError(f'Failed to upload manifest\n{err.msg}')
     # Enable repo from Repository Set

robottelo/host_helpers/repository_mixins.py

@@ -6,7 +6,6 @@
 import sys
 
 from robottelo import constants
-from robottelo import manifests
 from robottelo.config import settings
 from robottelo.exceptions import DistroNotSupportedError
 from robottelo.exceptions import OnlyOneOSRepositoryAllowed
@@ -660,7 +659,7 @@ def setup_content(
         if self.need_subscription:
             # upload manifest only when needed
             if upload_manifest and not self.organization_has_manifest(org_id):
-                manifests.upload_manifest_locked(org_id, interface=manifests.INTERFACE_CLI)
+                self.satellite.upload_manifest(org_id, interface='CLI')
             if not rh_subscriptions:
                 # add the default subscription if no subscription provided
                 rh_subscriptions = [constants.DEFAULT_SUBSCRIPTION_NAME]

robottelo/host_helpers/satellite_mixins.py

@@ -15,6 +15,7 @@
 from robottelo.host_helpers.api_factory import APIFactory
 from robottelo.host_helpers.cli_factory import CLIFactory
 from robottelo.logging import logger
+from robottelo.manifests import clone
 from robottelo.utils.installer import InstallerCommand
 
 
@@ -127,6 +128,41 @@ def md5_by_url(self, url):
             f'wget -qO - {url} | tee {filename} | md5sum | awk \'{{print $1}}\''
         ).stdout
 
+    def upload_manifest(self, org_id, manifest=None, interface='API', timeout=None):
+        """Upload a manifest using the requested interface.
+
+        :type org_id: int
+        :type manifest: Manifester object or None
+        :type interface: str
+        :type timeout: int
+
+        :returns: the manifest upload result
+
+        """
+        if manifest is None:
+            manifest = clone()
+        if timeout is None:
+            # Set the timeout to 1500 seconds to align with the API timeout.
+            timeout = 1500000
+        if interface == 'CLI':
+            if isinstance(manifest.content, bytes):
+                self.put(f'{manifest.path}', f'{manifest.name}')
+                result = self.cli.Subscription.upload(
+                    {'file': manifest.name, 'organization-id': org_id}, timeout=timeout
+                )
+            else:
+                self.put(manifest, manifest.filename)
+                result = self.cli.Subscription.upload(
+                    {'file': manifest.filename, 'organization-id': org_id}, timeout=timeout
+                )
+        else:
+            if not isinstance(manifest, bytes):
+                manifest = manifest.content
+            result = self.api.Subscription().upload(
+                data={'organization_id': org_id}, files={'content': manifest}
+            )
+        return result
+
     def is_sca_mode_enabled(self, org_id):
         """This method checks whether Simple Content Access (SCA) mode is enabled for a
         given organization.

robottelo/utils/__init__.py

@@ -1,14 +1,23 @@
 # General utility functions which does not fit into other util modules OR
 # Independent utility functions that doesnt need separate module
 import base64
+import io
+import json
 import os
 import re
+import time
+import uuid
+import zipfile
 from pathlib import Path
 
+import requests
 from cryptography.hazmat.backends import default_backend as crypto_default_backend
+from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives import serialization as crypto_serialization
+from cryptography.hazmat.primitives.asymmetric import padding
 from cryptography.hazmat.primitives.asymmetric import rsa
 
+from robottelo.config import settings
 from robottelo.constants import Colored
 from robottelo.exceptions import InvalidVaultURLForOIDC
 
@@ -80,3 +89,171 @@ def slugify_component(string, keep_hyphens=True):
     if not keep_hyphens:
         string = string.replace('-', '_')
     return re.sub("[^-_a-zA-Z0-9]", "", string.lower())
+
+
+# Manifest Cloning
+class ManifestCloner:
+    """Manifest cloning utility class."""
+
+    def __init__(self, template=None, private_key=None, signing_key=None):
+        self.template = template
+        self.signing_key = signing_key
+        self.private_key = private_key
+
+    def _download_manifest_info(self, name='default'):
+        """Download and cache the manifest information."""
+        if self.template is None:
+            self.template = {}
+        self.template[name] = requests.get(settings.fake_manifest.url[name], verify=False).content
+        if self.signing_key is None:
+            self.signing_key = requests.get(settings.fake_manifest.key_url, verify=False).content
+        if self.private_key is None:
+            self.private_key = crypto_serialization.load_pem_private_key(
+                self.signing_key, password=None, backend=crypto_default_backend()
+            )
+
+    def manifest_clone(self, org_environment_access=False, name='default'):
+        """Clones a RedHat-manifest file.
+
+        Change the consumer ``uuid`` and sign the new manifest with
+        signing key. The certificate for the key must be installed on the
+        candlepin server in order to accept uploading the cloned
+        manifest.
+
+        :param org_environment_access: Whether to modify consumer content
+            access mode to org_environment (Golden ticket enabled manifest).
+
+        :param name: which manifest url to clone (named key-value pairs
+            are defined as fake_manifest.url value in robottelo.properties
+            (default: 'default')
+
+        :return: A file-like object (``BytesIO`` on Python 3 and
+            ``StringIO`` on Python 2) with the contents of the cloned
+            manifest.
+        """
+        if self.signing_key is None or self.template is None or self.template.get(name) is None:
+            self._download_manifest_info(name)
+
+        template_zip = zipfile.ZipFile(io.BytesIO(self.template[name]))
+        # Extract the consumer_export.zip from the template manifest.
+        consumer_export_zip = zipfile.ZipFile(io.BytesIO(template_zip.read('consumer_export.zip')))
+
+        # Generate a new consumer_export.zip file changing the consumer
+        # uuid.
+        consumer_export = io.BytesIO()
+        with zipfile.ZipFile(consumer_export, 'w') as new_consumer_export_zip:
+            for name in consumer_export_zip.namelist():
+                if name == 'export/consumer.json':
+                    consumer_data = json.loads(consumer_export_zip.read(name).decode('utf-8'))
+                    consumer_data['uuid'] = str(uuid.uuid1())
+                    if org_environment_access:
+                        consumer_data['contentAccessMode'] = 'org_environment'
+                        consumer_data['owner'][
+                            'contentAccessModeList'
+                        ] = 'entitlement,org_environment'
+                    new_consumer_export_zip.writestr(name, json.dumps(consumer_data))
+                else:
+                    new_consumer_export_zip.writestr(name, consumer_export_zip.read(name))
+
+        # Generate a new manifest.zip file with the generated
+        # consumer_export.zip and new signature.
+        manifest = io.BytesIO()
+        with zipfile.ZipFile(manifest, 'w', zipfile.ZIP_DEFLATED) as manifest_zip:
+            consumer_export.seek(0)
+            manifest_zip.writestr('consumer_export.zip', consumer_export.read())
+            consumer_export.seek(0)
+            signature = self.private_key.sign(
+                consumer_export.read(), padding.PKCS1v15(), hashes.SHA256()
+            )
+            manifest_zip.writestr('signature', signature)
+        # Make sure that the file-like object is at the beginning and
+        # ready to be read.
+        manifest.seek(0)
+        return manifest
+
+    def original(self, name='default'):
+        """Returns the original manifest as a file-like object.
+
+        :param name: A name of the manifest as defined in robottelo.properties
+
+        Be aware that using the original manifest and not removing it
+        afterwards will make it impossible to import it to any other
+        Organization.
+
+        Make sure to close the returned file-like object in order to clean up
+        the memory used to store it.
+        """
+        if self.signing_key is None or self.template is None or self.template.get(name) is None:
+            self._download_manifest_info(name)
+        return io.BytesIO(self.template[name])
+
+
+# Cache the ManifestCloner in order to avoid downloading the manifest template
+# every single time.
+_manifest_cloner = ManifestCloner()
+
+
+class Manifest:
+    """Class that holds the contents of a manifest with a generated filename
+    based on ``time.time``.
+
+    To ensure that the manifest content is closed use this class as a context
+    manager with the ``with`` statement::
+
+        with Manifest() as manifest:
+            # my fancy stuff
+    """
+
+    def __init__(self, content=None, filename=None, org_environment_access=False, name='default'):
+        self._content = content
+        self.filename = filename
+
+        if self._content is None:
+            self._content = _manifest_cloner.manifest_clone(
+                org_environment_access=org_environment_access, name=name
+            )
+        if self.filename is None:
+            self.filename = f'/var/tmp/manifest-{int(time.time())}.zip'
+
+    @property
+    def content(self):
+        if not self._content.closed:
+            # Make sure that the content is always ready to read
+            self._content.seek(0)
+        return self._content
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type, value, traceback):
+        if not self.content.closed:
+            self.content.close()
+
+
+def clone(org_environment_access=False, name='default'):
+    """Clone the cached manifest and return a ``Manifest`` object.
+
+    :param org_environment_access: Whether to modify consumer content
+        access mode to org_environment (Golden ticket enabled manifest).
+
+    :param name: key name of the fake_manifests.url dict defined in
+        robottelo.properties
+
+    Is hightly recommended to use this with the ``with`` statement to make that
+    the content of the manifest (file-like object) is closed properly::
+
+        with clone() as manifest:
+            # my fancy stuff
+    """
+
+    return Manifest(org_environment_access=org_environment_access, name=name)
+
+
+def original_manifest(name='default'):
+    """Returns a ``Manifest`` object filed with the template manifest.
+
+    :param name: key name of the fake_manifests.url dict defined in
+        robottelo.properties
+    """
+
+    return Manifest(_manifest_cloner.original(name=name))