verify data

No-Issue

galaxy_ng/tests/integration/api/test_load_data.py

@@ -1,11 +1,11 @@
 import logging
 import json
 import pytest
-import yaml
 
 from galaxy_ng.tests.integration.utils.repo_management_utils import create_repo_and_dist, \
-    upload_new_artifact, add_content_units
-from galaxykit.collections import sign_collection, deprecate_collection
+    upload_new_artifact
+from galaxykit.collections import sign_collection, deprecate_collection, \
+    move_or_copy_collection
 from galaxykit.containers import create_container, delete_container
 from galaxykit.namespaces import add_group
 from galaxykit.registries import create_registry, delete_registry
@@ -22,16 +22,12 @@
 class TestLoadData:
 
     @pytest.mark.load_data
-    def test_load_data(self, galaxy_client):
+    def test_load_data(self, galaxy_client, data):
         """
         Test loading data that will be verified at a later stage
         after the AAP upgrade or backup/restore
         """
         gc = galaxy_client("admin")
-        path = 'galaxy_ng/tests/integration/load_data.yaml'
-
-        with open(path, 'r') as yaml_file:
-            data = yaml.safe_load(yaml_file)
 
         for group in data["groups"]:
             # creates a group, nothing happens if it already exists
@@ -94,11 +90,12 @@ def test_load_data(self, galaxy_client):
                 collection_resp_1 = gc.get(
                     f"pulp/api/v3/content/ansible/collection_versions/?name={artifact.name}"
                 )
-                repo_pulp_href = get_repository_href(gc, collection["repository"])
-                content_units = [collection_resp_1["results"][0]["pulp_href"]]
-                add_content_units(gc, content_units, repo_pulp_href)
+                move_or_copy_collection(gc, artifact.namespace, artifact.name,
+                                        artifact.version, "staging",
+                                        destination=collection["repository"])
                 if collection["signed"]:
                     logger.debug("Signing collection")
+                    repo_pulp_href = get_repository_href(gc, collection["repository"])
                     sign_collection(gc, collection_resp_1["results"][0]["pulp_href"],
                                     repo_pulp_href)
                 if collection["deprecated"]:

galaxy_ng/tests/integration/api/test_verify_data.py

@@ -0,0 +1,139 @@
+import logging
+import pytest
+
+from galaxy_ng.tests.integration.utils.repo_management_utils import search_collection_endpoint
+from galaxykit.collections import collection_info
+from galaxykit.groups import get_group_id
+from galaxykit.namespaces import get_namespace
+from galaxykit.registries import get_registry_pk
+from galaxykit.remotes import view_remotes
+from galaxykit.repositories import get_repository_href
+from galaxykit.roles import get_role
+from galaxykit.users import get_user
+
+logger = logging.getLogger(__name__)
+
+
+@pytest.mark.min_hub_version("4.7dev")
+class TestVerifyData:
+
+    @pytest.mark.verify_data
+    def test_verify_data_users(self, galaxy_client, data):
+        """
+        Test that verifies the data previously loaded by test_load_data
+        """
+        gc = galaxy_client("admin")
+        for expected_user in data["users"]:
+            actual_user = get_user(gc, expected_user["username"])
+            assert expected_user["username"] == actual_user["username"]
+            assert expected_user["email"] == actual_user["email"]
+            assert expected_user["is_superuser"] == actual_user["is_superuser"]
+            assert expected_user["group"] in str(actual_user["groups"])
+
+    @pytest.mark.verify_data
+    def test_verify_data_ns(self, galaxy_client, data):
+        """
+        Test that verifies the data previously loaded by test_load_data
+        """
+        gc = galaxy_client("admin")
+        for expected_ns in data["namespaces"]:
+            actual_ns = get_namespace(gc, expected_ns["name"])
+            assert expected_ns["name"] == actual_ns["name"]
+            assert expected_ns["group"] in str(actual_ns["groups"])
+
+    @pytest.mark.verify_data
+    def test_verify_data_collections(self, galaxy_client, data):
+        """
+        Test that verifies the data previously loaded by test_load_data
+        """
+        gc = galaxy_client("admin")
+        for expected_col in data["collections"]:
+            expected_name = f"collection_dep_a_{expected_col['name']}"
+            actual_col = collection_info(gc, expected_col["repository"],
+                                         expected_col["namespace"], expected_name,
+                                         expected_col["version"])
+            assert actual_col["version"] == expected_col["version"]
+            assert actual_col["name"] == expected_name
+            assert actual_col["namespace"]["name"] == expected_col["namespace"]
+            if expected_col["signed"]:
+                assert len(actual_col["signatures"]) > 0
+            else:
+                assert len(actual_col["signatures"]) == 0
+            _, actual_col = search_collection_endpoint(gc, name=expected_name)
+            assert actual_col[0]["is_deprecated"] == expected_col["deprecated"]
+            assert actual_col[0]["is_signed"] == expected_col["signed"]
+            assert actual_col[0]["cv_name"] == expected_name
+            assert actual_col[0]["cv_version"] == expected_col["version"]
+            assert actual_col[0]["repo_name"] == expected_col["repository"]
+
+    @pytest.mark.verify_data
+    def test_verify_data_groups(self, galaxy_client, data):
+        """
+        Test that verifies the data previously loaded by test_load_data
+        """
+        gc = galaxy_client("admin")
+        for expected_group in data["groups"]:
+            get_group_id(gc, expected_group["name"])
+
+    @pytest.mark.verify_data
+    def test_verify_data_repos(self, galaxy_client, data):
+        """
+        Test that verifies the data previously loaded by test_load_data
+        """
+        gc = galaxy_client("admin")
+        for expected_repo in data["repositories"]:
+            get_repository_href(gc, expected_repo["name"])
+
+    @pytest.mark.verify_data
+    def test_verify_data_rbac_roles(self, galaxy_client, data):
+        """
+        Test that verifies the data previously loaded by test_load_data
+        """
+        gc = galaxy_client("admin")
+        for expected_rbac_role in data["roles"]:
+            role_info_1 = get_role(gc, expected_rbac_role["name"])
+            assert role_info_1["name"] == expected_rbac_role["name"]
+            assert role_info_1["description"] == expected_rbac_role["description"]
+            assert sorted(role_info_1["permissions"]) == sorted(expected_rbac_role["permissions"])
+
+    @pytest.mark.verify_data
+    def test_verify_data_ee(self, galaxy_client, data):
+        """
+        Test that verifies the data previously loaded by test_load_data
+        """
+        gc = galaxy_client("admin")
+        for ee in data["execution_environments"]:
+            # this needs to be moved to galaxykit
+            actual_ee = gc.get(f"v3/plugin/execution-environments/repositories/{ee['name']}/")
+            assert actual_ee["name"] == ee["name"]
+            assert (actual_ee["pulp"]["repository"]["remote"]["upstream_name"]
+                    == ee["upstream_name"])
+            actual_registry = actual_ee["pulp"]["repository"]["remote"]["registry"]
+            expected_registry = get_registry_pk(gc, ee["remote_registry"])
+            assert expected_registry == actual_registry
+
+    @pytest.mark.verify_data
+    def test_verify_data_remote_registries(self, galaxy_client, data):
+        """
+        Test that verifies the data previously loaded by test_load_data
+        """
+        gc = galaxy_client("admin")
+        for remote_registry in data["remote_registries"]:
+            # this needs to be moved to galaxykit
+            actual_rr = gc.get(f"_ui/v1/execution-environments/registries/"
+                               f"?name={remote_registry['name']}")
+            assert actual_rr["data"][0]["name"] == remote_registry["name"]
+            assert actual_rr["data"][0]["url"] == remote_registry["url"]
+
+    @pytest.mark.verify_data
+    def test_verify_data_remotes(self, galaxy_client, data):
+        """
+        Test that verifies the data previously loaded by test_load_data
+        """
+        gc = galaxy_client("admin")
+        for remote in data["remotes"]:
+            actual_remote = view_remotes(gc, remote["name"])
+            assert actual_remote["results"][0]["url"] == remote["url"]
+            assert actual_remote["results"][0]["name"] == remote["name"]
+            assert actual_remote["results"][0]["signed_only"] == remote["signed_only"]
+            assert actual_remote["results"][0]["tls_validation"] == remote["tls_validation"]

galaxy_ng/tests/integration/conftest.py

@@ -1,6 +1,7 @@
 import logging
 import os
 import shutil
+import yaml
 
 import pytest
 from orionutils.utils import increment_version
@@ -82,6 +83,7 @@
 galaxy_stage_ansible: tests that run against galaxy-stage.ansible.com
 installer_smoke_test: smoke tests to validate AAP installation (VM)
 load_data: tests that load data that will be verified after upgrade or backup/restore
+verify_data: tests that verify the data previously loaded by load_data test
 """
 
 logger = logging.getLogger(__name__)
@@ -547,6 +549,14 @@ def keep_generated_test_artifact(ansible_config):
     )
 
 
+@pytest.fixture(scope="session")
+def data():
+    path = 'galaxy_ng/tests/integration/load_data.yaml'
+    with open(path, 'r') as yaml_file:
+        data = yaml.safe_load(yaml_file)
+    return data
+
+
 def min_hub_version(ansible_config, spec):
     version = get_hub_version(ansible_config)
     return Requirement.parse(f"galaxy_ng<{spec}").specifier.contains(version)