wip

No-Issue

galaxy_ng/tests/integration/api/test_iqe_rbac.py

@@ -59,7 +59,7 @@ def test_missing_role_create_user(self, galaxy_client):
         gc = galaxy_client(user)
         with pytest.raises(GalaxyClientError) as ctx:
             create_test_user(gc)
-        assert ctx.value.args[0]["status"] == "403"
+        assert ctx.value.args[0] == 403
 
     @pytest.mark.iqe_rbac_test
     def test_role_update_user(self, galaxy_client):
@@ -96,7 +96,7 @@ def test_missing_role_update_user(self, galaxy_client):
         resp["password"] = "changechangechange"
         with pytest.raises(GalaxyClientError) as ctx:
             update_user(gc_user, resp)
-        assert ctx.value.args[0]["status"] == "403"
+        assert ctx.value.args[0] == 403
 
     @pytest.mark.iqe_rbac_test
     def test_role_delete_user(self, galaxy_client):
@@ -281,7 +281,7 @@ def test_missing_permission_in_role_to_add_group(self, galaxy_client):
         gc_user = galaxy_client(user)
         with pytest.raises(GalaxyClientError) as ctx:
             gc_user.create_group(new_group_name)
-        assert ctx.value.args[0]["status"] == "403"
+        assert ctx.value.args[0] == 403
 
     @pytest.mark.iqe_rbac_test
     def test_missing_role_permission_add_namespace(self, galaxy_client):
@@ -301,7 +301,7 @@ def test_missing_role_permission_add_namespace(self, galaxy_client):
         gc_user = galaxy_client(user)
         with pytest.raises(GalaxyClientError) as ctx:
             create_namespace(gc_user, group)
-        assert ctx.value.args[0]["status"] == "403"
+        assert ctx.value.args[0] == 403
 
     @pytest.mark.iqe_rbac_test
     def test_role_add_namespace(self, galaxy_client):
@@ -385,7 +385,7 @@ def test_missing_role_upload_to_namespace(self, galaxy_client):
         gc_user = galaxy_client(user)
         with pytest.raises(GalaxyClientError) as ctx:
             upload_test_artifact(gc_user, ns2_name)
-        assert ctx.value.args[0]["status"] == "403"
+        assert ctx.value.args[0] == 403
 
     @pytest.mark.iqe_rbac_test
     def test_global_role_upload_to_namespace(self, galaxy_client):
@@ -446,7 +446,7 @@ def test_missing_role_delete_collection(self, galaxy_client):
             gc_user.delete_collection(
                 namespace_name, artifact.name, artifact.version, repository="published"
             )
-        assert ctx.value.args[0]["status"] == "403"
+        assert ctx.value.args[0] == 403
         assert collection_exists(gc, namespace_name, artifact.name, artifact.version)
 
     @pytest.mark.iqe_rbac_test
@@ -475,7 +475,7 @@ def test_missing_role_reject_collection(self, galaxy_client):
                 source="published",
                 destination="rejected",
             )
-        assert ctx.value.args[0]["status"] == "403"
+        assert ctx.value.args[0] == 403
 
     @pytest.mark.iqe_rbac_test
     def test_role_reject_collection(self, galaxy_client):
@@ -551,7 +551,7 @@ def test_missing_role_approve_collection(self, galaxy_client):
             move_or_copy_collection(
                 gc_user, namespace_name, artifact.name, artifact.version
             )  # approve collection
-        assert ctx.value.args[0]["status"] == "403"
+        assert ctx.value.args[0] == 403
 
     @pytest.mark.iqe_rbac_test
     def test_missing_role_add_remote_registry(self, galaxy_client):
@@ -568,7 +568,7 @@ def test_missing_role_add_remote_registry(self, galaxy_client):
         gc_user = galaxy_client(user)
         with pytest.raises(GalaxyClientError) as ctx:
             create_registry(gc_user, f"remote_registry_{uuid4()}", "url")
-        assert ctx.value.args[0]["status"] == "403"
+        assert ctx.value.args[0] == 403
 
     @pytest.mark.iqe_rbac_test
     def test_role_add_remote_registry(self, galaxy_client):
@@ -664,7 +664,7 @@ def test_missing_role_add_ee(self, galaxy_client):
         ee_name = f"ee_{uuid4()}"
         with pytest.raises(GalaxyClientError) as ctx:
             create_container(gc_user, ee_name, "upstream_name", remote_registry)
-        assert ctx.value.args[0]["status"] == "403"
+        assert ctx.value.args[0] == 403
 
     @pytest.mark.iqe_rbac_test
     def test_role_delete_ee(self, galaxy_client):
@@ -733,7 +733,7 @@ def test_user_missing_role_remotes(self, galaxy_client):
         gc_user = galaxy_client(user)
         with pytest.raises(GalaxyClientError) as ctx:
             community_remote_config(gc_user, "http://google.com/", "username", "password")
-        assert ctx.value.args[0]["status"] == "403"
+        assert ctx.value.args[0] == 403
 
     @pytest.mark.iqe_rbac_test
     def test_user_role_get_remotes(self, galaxy_client):

galaxy_ng/tests/integration/conftest.py

@@ -127,21 +127,19 @@ def certifiedv2(ansible_config, artifact, galaxy_client):
     """ Create and publish+certify collection version N and N+1 """
 
     # make sure the expected namespace exists ...
-    config = ansible_config("partner_engineer")
-    api_client = get_client(config)
     gc = galaxy_client("partner_engineer")
     create_namespace(gc, artifact.namespace, "")
 
     # publish v1
     config = ansible_config("partner_engineer")
     ansible_galaxy(
         f"collection publish {artifact.filename}",
-        ansible_config=config
+        galaxy_client=gc
     )
 
     # certify v1
     hub_4_5 = is_hub_4_5(ansible_config)
-    set_certification(api_client, gc, artifact, hub_4_5=hub_4_5)
+    set_certification(ansible_config(), gc, artifact, hub_4_5=hub_4_5)
 
     # Increase collection version
     new_version = increment_version(artifact.version)
@@ -153,14 +151,13 @@ def certifiedv2(ansible_config, artifact, galaxy_client):
     )
 
     # publish newer version
-    config = ansible_config("partner_engineer")
     ansible_galaxy(
         f"collection publish {artifact2.filename}",
-        ansible_config=config
+        galaxy_client=gc
     )
 
     # certify newer version
-    set_certification(api_client, gc, artifact2, hub_4_5=hub_4_5)
+    set_certification(ansible_config(), gc, artifact2, hub_4_5=hub_4_5)
 
     return (artifact, artifact2)
 

galaxy_ng/tests/integration/utils/tasks.py

@@ -75,12 +75,13 @@ def wait_for_task(api_client, resp, task_id=None, timeout=6000, raise_on_error=F
 def wait_for_task_ui_client(gc, task):
     counter = 0
     state = None
+    task_id = task["task"].split("v3/tasks/")[1][:-1]
     while state in [None, 'waiting', 'running']:
         counter += 1
         if counter >= 60:
             raise Exception('Task is taking too long')
-        ds = gc.get(task['task'])
-        state = ds['state']
+        ds = gc.get(f"pulp/api/v3/tasks/{task_id}/")
+        state = ds.json()['state']
         if state == 'completed':
             break
         time.sleep(SLEEP_SECONDS_POLLING)