Skip collection signing and validation in related test.

When there is no signing service and no requirement to sign,
don't do it.

Also auto-create autohubtest2 as necessary.

No-Issue

Signed-off-by: James Tanner <[email protected]>

galaxy_ng/tests/integration/api/test_move.py

@@ -177,21 +177,28 @@ def get_all_collections():
 @pytest.mark.deployment_standalone
 @pytest.mark.min_hub_version("4.7dev")
 @pytest.mark.skipif(is_ocp_env(), reason="Content signing not enabled in AAP Operator")
-def test_copy_associated_content(ansible_config, galaxy_client):
+def test_copy_associated_content(
+    use_collection_signatures,
+    autohubtest2,
+    ansible_config,
+    galaxy_client
+):
     """Tests whether a collection and associated content is copied from repo to repo"""
 
     # TODO: add check for ansible namespace metadata
 
     artifact = build_collection(
         "skeleton",
         config={
-            "namespace": USERNAME_PUBLISHER,
+            "namespace": autohubtest2['name'],
             "tags": ["tools", "copytest"],
         }
     )
 
-    # import and wait ...
+    # make admin client
     gc_admin = galaxy_client("admin")
+
+    # import and wait ...
     resp = upload_artifact(None, gc_admin, artifact)
     wait_for_task(gc_admin, resp)
 
@@ -223,7 +230,8 @@ def test_copy_associated_content(ansible_config, galaxy_client):
     col_marked = gc_admin.get(collection_mark)["results"]
     assert len(col_marked) == 0
 
-    sign_collection(gc_admin, cv_href, pulp_href)
+    if use_collection_signatures:
+        sign_collection(gc_admin, cv_href, pulp_href)
 
     # mark collection
     marked_collection = gc_admin.post(
@@ -243,8 +251,9 @@ def test_copy_associated_content(ansible_config, galaxy_client):
     col_deprecation = gc_admin.get(collection)["deprecated"]
     assert col_deprecation is True
 
-    col_signature = gc_admin.get(collection_version)["signatures"]
-    assert len(col_signature) == 1
+    if use_collection_signatures:
+        col_signature = gc_admin.get(collection_version)["signatures"]
+        assert len(col_signature) == 1
 
     col_marked = gc_admin.get(collection_mark)["results"]
     assert len(col_marked) == 1
@@ -268,8 +277,9 @@ def test_copy_associated_content(ansible_config, galaxy_client):
     col_deprecation = gc_admin.get(collection)["deprecated"]
     assert col_deprecation is True
 
-    col_signature = gc_admin.get(collection_version)["signatures"]
-    assert len(col_signature) == 1
+    if use_collection_signatures:
+        col_signature = gc_admin.get(collection_version)["signatures"]
+        assert len(col_signature) == 1
 
     col_marked = gc_admin.get(collection_mark)["results"]
     assert len(col_marked) == 1

galaxy_ng/tests/integration/conftest.py

@@ -386,6 +386,23 @@ def settings(galaxy_client):
     return gc.get("_ui/v1/settings/")
 
 
+@pytest.fixture(scope="function")
+def use_collection_signatures(settings):
+    """A shortcut to know if a test should attempt to work with signatures."""
+    service = settings["GALAXY_COLLECTION_SIGNING_SERVICE"]
+    required = settings["GALAXY_REQUIRE_SIGNATURE_FOR_APPROVAL"]
+    if service is not None and required:
+        return True
+    return False
+
+
+@pytest.fixture(scope="function")
+def autohubtest2(galaxy_client):
+    """A carry over pre-created namespace from the original IQE tests."""
+    gc = galaxy_client("admin")
+    return create_namespace("autohubtest2", gc=gc)
+
+
 def set_test_data(ansible_config, hub_version):
     role = "admin"
     gc = GalaxyKitClient(ansible_config).gen_authorized_client(role)

galaxy_ng/tests/integration/utils/iqe_utils.py

@@ -534,9 +534,13 @@ def __getitem__(self, key):
                 return None
 
         elif key == "username":
+            if self.profile == 'admin' and os.environ.get('HUB_ADMIN_USER'):
+                return os.environ.get('HUB_ADMIN_USER')
             return self.PROFILES[self.profile]["username"]
 
         elif key == "password":
+            if self.profile == 'admin' and os.environ.get('HUB_ADMIN_PASS'):
+                return os.environ.get('HUB_ADMIN_PASS')
             return self.PROFILES[self.profile]["password"]
 
         elif key == 'use_move_endpoint':
@@ -660,8 +664,14 @@ def has_old_credentials():
 @lru_cache()
 def get_hub_version(ansible_config):
     if aap_gateway():
-        username = ansible_config("admin").PROFILES.get("admin").get("username")
-        password = ansible_config("admin").PROFILES.get("admin").get("password")
+        # Why would we do this? ...
+        # username = ansible_config("admin").PROFILES.get("admin").get("username")
+        # password = ansible_config("admin").PROFILES.get("admin").get("password")
+
+        cfg = ansible_config("admin")
+        username = cfg.get('username')
+        password = cfg.get('password')
+
         gw_root_url = ansible_config("admin").get("gw_root_url")
         gc = GalaxyClient(galaxy_root="foo", auth={"username": username, "password": password},
                           gw_auth=True, https_verify=False, gw_root_url=gw_root_url)

galaxy_ng/tests/integration/utils/namespaces.py

@@ -4,6 +4,8 @@
 import random
 import string
 
+from galaxykit.utils import GalaxyClientError
+
 from .collections import delete_all_collections_in_namespace, \
     delete_all_collections_in_namespace_gk
 
@@ -116,3 +118,18 @@ def cleanup_namespace_gk(name, gc_admin):
 
         resp = gc_admin.get(f'v3/namespaces/?name={name}')
         assert resp['meta']['count'] == 0
+
+
+def create_namespace(namespace_name, gc=None):
+    """ Make a namespace for testing if it does not exist."""
+    assert gc is not None, "api_client is a required param"
+    # check if it already exists ...
+    try:
+        resp = gc.get(f"_ui/v1/namespaces/{namespace_name}/")
+        return resp
+    except GalaxyClientError:
+        pass
+
+    # create it
+    payload = {"name": namespace_name, "groups": []}
+    return gc.post("v3/namespaces/", body=payload)