[CI] Skip collection signing and validation in related test. (#2200)

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

* Skip collection signing and validation in related test.
* Also auto-create autohubtest2 as necessary.
* Cleanup.
* Fixup another test when no signing service available.
* Fix another test.
* More skipping.
* Use galaxykit's create_namespace

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/api/test_repositories.py

@@ -168,7 +168,7 @@ def test_move_cv_endpoint(self, galaxy_client):
     @pytest.mark.repositories
     @pytest.mark.deployment_standalone
     @pytest.mark.skipif(is_ocp_env(), reason="Content signing not enabled in AAP Operator")
-    def test_copy_signed_cv_endpoint(self, galaxy_client):
+    def test_copy_signed_cv_endpoint(self, use_collection_signatures, galaxy_client):
         """
         Verifies a signed cv can be copied to a different repo
         """
@@ -190,13 +190,22 @@ def test_copy_signed_cv_endpoint(self, galaxy_client):
         test_repo_name_2 = f"repo-test-{generate_random_string()}"
         repo_pulp_href_2 = create_repo_and_dist(gc_admin, test_repo_name_2)
 
-        sign_collection(gc_admin, content_units[0], repo_pulp_href_1)
+        if use_collection_signatures:
+            sign_collection(gc_admin, content_units[0], repo_pulp_href_1)
 
         copy_content_between_repos(gc_admin, content_units, repo_pulp_href_1, [repo_pulp_href_2])
         matches, results = search_collection_endpoint(gc_admin, name=artifact.name)
         expected = [
-            {"cv_name": artifact.name, "repo_name": test_repo_name_1, "is_signed": True},
-            {"cv_name": artifact.name, "repo_name": test_repo_name_2, "is_signed": True},
+            {
+                "cv_name": artifact.name,
+                "repo_name": test_repo_name_1,
+                "is_signed": use_collection_signatures
+            },
+            {
+                "cv_name": artifact.name,
+                "repo_name": test_repo_name_2,
+                "is_signed": use_collection_signatures
+            },
         ]
         assert verify_repo_data(expected, results)
 
@@ -209,7 +218,7 @@ def test_copy_signed_cv_endpoint(self, galaxy_client):
     @pytest.mark.repositories
     @pytest.mark.deployment_standalone
     @pytest.mark.skipif(is_ocp_env(), reason="Content signing not enabled in AAP Operator")
-    def test_move_signed_cv_endpoint(self, galaxy_client):
+    def test_move_signed_cv_endpoint(self, use_collection_signatures, galaxy_client):
         """
         Verifies a signed cv can be moved to a different repo
         """
@@ -231,11 +240,16 @@ def test_move_signed_cv_endpoint(self, galaxy_client):
         test_repo_name_2 = f"repo-test-{generate_random_string()}"
         repo_pulp_href_2 = create_repo_and_dist(gc_admin, test_repo_name_2)
 
-        sign_collection(gc_admin, content_units[0], repo_pulp_href_1)
+        if use_collection_signatures:
+            sign_collection(gc_admin, content_units[0], repo_pulp_href_1)
 
         move_content_between_repos(gc_admin, content_units, repo_pulp_href_1, [repo_pulp_href_2])
         _, results = search_collection_endpoint(gc_admin, name=artifact.name)
-        expected = [{"cv_name": artifact.name, "repo_name": test_repo_name_2, "is_signed": True}]
+        expected = [{
+            "cv_name": artifact.name,
+            "repo_name": test_repo_name_2,
+            "is_signed": use_collection_signatures
+        }]
         assert verify_repo_data(expected, results)
         matches, _ = search_collection_endpoint(
             gc_admin, name=artifact.name, repository_name=test_repo_name_1

galaxy_ng/tests/integration/api/test_x_repo_search.py

@@ -560,10 +560,22 @@ def test_search_by_q(self, galaxy_client):
     @pytest.mark.parametrize("is_signed", [True, False])
     @pytest.mark.x_repo_search
     @pytest.mark.skipif(is_ocp_env(), reason="Content signing not enabled in AAP Operator")
-    def test_search_by_is_signed_true_false(self, galaxy_client, is_signed):
+    def test_search_by_is_signed_true_false(
+        self,
+        use_collection_signatures,
+        galaxy_client,
+        is_signed
+    ):
         """
         Verifies that search endpoint can search by is_signed
         """
+
+        # we shouldn't be here if the system doesn't have signatures
+        # enabled and a valid signing service, but we have pipelines
+        # that seem to have issues using the correct marks.
+        if not use_collection_signatures:
+            pytest.skip("signatures are required for this test")
+
         test_repo_name = f"repo-test-1-{generate_random_string()}"
         gc = galaxy_client("iqe_admin")
         repo_pulp_href = create_repo_and_dist(gc, test_repo_name)

galaxy_ng/tests/integration/conftest.py

@@ -386,6 +386,24 @@ 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")
+    create_namespace(gc, "autohubtest2", "")
+    return {"name": "autohubtest2"}
+
+
 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,9 +664,11 @@ 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")
-        gw_root_url = ansible_config("admin").get("gw_root_url")
+        cfg = ansible_config("admin")
+        username = cfg.get('username')
+        password = cfg.get('password')
+        gw_root_url = cfg.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_version = gc.get_server_version()

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)