wip

No-Issue

galaxy_ng/tests/integration/api/test_api_base.py

@@ -1,10 +1,12 @@
 import pytest
 
 from ..utils import get_client
+from ..utils.iqe_utils import remove_from_cache
 
 
 @pytest.mark.min_hub_version("4.6dev")
 @pytest.mark.deployment_standalone
+@pytest.mark.skip_in_gw
 def test_galaxy_api_root_standalone_no_auth_access(galaxy_client):
     """Test galaxy API root."""
 
@@ -14,6 +16,7 @@ def test_galaxy_api_root_standalone_no_auth_access(galaxy_client):
     response = gc.get("")
     assert "v3" in response["available_versions"]
     assert "pulp-v3" in response["available_versions"]
+    remove_from_cache("basic_user")
 
 
 @pytest.mark.min_hub_version("4.6dev")

galaxy_ng/tests/integration/api/test_auth.py

@@ -7,7 +7,7 @@
 
 from galaxykit.utils import GalaxyClientError
 from ..utils import uuid4
-from ..utils.iqe_utils import remove_from_cache
+from ..utils.iqe_utils import remove_from_cache, aap_gateway
 
 pytestmark = pytest.mark.qa  # noqa: F821
 
@@ -64,27 +64,51 @@ def test_auth_exception(galaxy_client):
 
 @pytest.mark.deployment_standalone
 @pytest.mark.galaxyapi_smoke
-@pytest.mark.skip
-def test_gateway_auth_admin(galaxy_client):
-    """Test whether admin can not access collections page using invalid token."""
-    # TODO: MODIFY FOR GW
+@pytest.mark.skipif(not aap_gateway(), reason="This test only runs if AAP Gateway is deployed")
+def test_gateway_auth_admin_gateway_sessionid(galaxy_client):
+    """Test whether admin can not access collections page using invalid gateway_sessionid."""
     gc = galaxy_client("admin")
-    gc.headers["Authorization"] = f"Bearer {uuid4()}"
+    alt_cookies = gc.gw_client.cookies
+    alt_cookies["gateway_sessionid"] = uuid4()
+    gc.headers["Cookie"] = f"csrftoken={alt_cookies['csrftoken']}; gateway_sessionid={alt_cookies['gateway_sessionid']}"
     remove_from_cache("admin")
     with pytest.raises(GalaxyClientError) as ctx:
-        gc.get("v3/collections/")
+        gc.get("v3/plugin/ansible/content/published/collections/index/", relogin=False)
     assert ctx.value.response.status_code == 403
+    remove_from_cache("admin")
 
 
 @pytest.mark.deployment_standalone
 @pytest.mark.galaxyapi_smoke
 @pytest.mark.skip
-def test_gateway_auth_exception(galaxy_client):
-    """Test whether an HTTP exception when using an invalid token."""
-    # TODO: MODIFY FOR GW
-    gc = galaxy_client("basic_user")
-    gc.headers["Authorization"] = f"Bearer {uuid4()}"
-    remove_from_cache("basic_user")
+@pytest.mark.skipif(not aap_gateway(), reason="This test only runs if AAP Gateway is deployed")
+def test_gateway_auth_admin_gateway_csrftoken(galaxy_client):
+    """Test whether admin can not access collections page using invalid csrftoken."""
+    # TODO: This test fails, invalid csrftoken does not return 403. Is it correct?
+    gc = galaxy_client("admin")
+    alt_cookies = gc.gw_client.cookies
+    alt_cookies["csrftoken"] = uuid4()
+    gc.headers["Cookie"] = f"csrftoken={alt_cookies['csrftoken']}; gateway_sessionid={alt_cookies['gateway_sessionid']}"
+    remove_from_cache("admin")
     with pytest.raises(GalaxyClientError) as ctx:
-        gc.get("v3/collections/")
-    assert ctx.value.response.status_code == 403
+        gc.get("v3/plugin/ansible/content/published/collections/index/", relogin=False)
+    assert ctx.value.response.status_code == 403
+    remove_from_cache("admin")
+
+
+
+@pytest.mark.deployment_standalone
+@pytest.mark.galaxyapi_smoke
+@pytest.mark.skipif(not aap_gateway(), reason="This test only runs if AAP Gateway is deployed")
+def test_gateway_token_auth(galaxy_client):
+    """Test whether normal auth is required and works to access APIs.
+
+    Also tests the settings for user profiles used for testing.
+    """
+    gc = galaxy_client("admin")
+    del gc.headers["Cookie"]
+    remove_from_cache("admin")
+
+    with pytest.raises(GalaxyClientError) as ctx:
+        gc.get("v3/plugin/ansible/content/published/collections/index/", relogin=False)
+    assert ctx.value.response.status_code == 403

galaxy_ng/tests/integration/api/test_container_push_update.py

@@ -47,16 +47,19 @@ def test_can_update_container_push(ansible_config, require_auth, galaxy_client):
     # Get an API client running with admin user credentials
     gc = galaxy_client("admin")
     if not require_auth:
-        del gc.headers["Authorization"]
+        try:
+            del gc.headers["Authorization"]
+        except KeyError:
+            gc.gw_client.logout()
         remove_from_cache("admin")
 
     # Get the pulp_href for the pushed repo
-    image = gc.get("pulp/api/v3/repositories/container/container-push/?name=alpine")
+    image = gc.get("pulp/api/v3/repositories/container/container-push/?name=alpine", relogin=False)
     container_href = image["results"][0]["pulp_href"]
 
     for value in (42, 1):
         # Make a Patch request changing the retain_repo_versions attribute to value
-        response = gc.patch(container_href, body={"retain_repo_versions": value})
+        response = gc.patch(container_href, body={"retain_repo_versions": value}, relogin=False)
 
         resp = wait_for_task(gc, response)
         assert resp["state"] == "completed"

galaxy_ng/tests/integration/api/test_container_signing.py

@@ -58,20 +58,23 @@ def test_push_and_sign_a_container(ansible_config, flags, require_auth, galaxy_c
     # Get an API client running with admin user credentials
     gc = galaxy_client("admin")
     if not require_auth:
-        del gc.headers["Authorization"]
+        try:
+            del gc.headers["Authorization"]
+        except KeyError:
+            gc.gw_client.logout()
         remove_from_cache("admin")
 
     # Get the pulp_href for the pushed image
-    image = gc.get("pulp/api/v3/repositories/container/container-push/?name=alpine")
+    image = gc.get("pulp/api/v3/repositories/container/container-push/?name=alpine", relogin=False)
     container_href = image["results"][0]["pulp_href"]
 
     # Get the pulp_href for signing service
 
-    signing_service = gc.get("pulp/api/v3/signing-services/?name=container-default")
+    signing_service = gc.get("pulp/api/v3/signing-services/?name=container-default", relogin=False)
     ss_href = signing_service["results"][0]["pulp_href"]
 
     # Sign the image
-    r = gc.post(f"{container_href}sign/", body={"manifest_signing_service": ss_href})
+    r = gc.post(f"{container_href}sign/", body={"manifest_signing_service": ss_href}, relogin=False)
     resp = wait_for_task(gc, r)
     assert resp["state"] == "completed"
 

galaxy_ng/tests/integration/api/test_pulp_api.py

@@ -6,6 +6,7 @@
 
 from galaxykit.utils import wait_for_task, GalaxyClientError
 from ..schemas import schema_pulp_objectlist, schema_pulp_roledetail, schema_task_detail
+from ..utils.iqe_utils import remove_from_cache
 from ..utils.rbac_utils import create_emtpy_local_image_container
 
 
@@ -139,14 +140,18 @@ def test_pulp_task_endpoint(galaxy_client, require_auth, ansible_config):
     name = create_emtpy_local_image_container(ansible_config("admin"), gc)
 
     if not require_auth:
-        del gc.headers["Authorization"]
+        try:
+            del gc.headers["Authorization"]
+        except KeyError:
+            gc.gw_client.logout()
+        remove_from_cache("ee_admin")
 
     delete_resp = gc.delete(
-        f"v3/plugin/execution-environments/repositories/{name}/"
+        f"v3/plugin/execution-environments/repositories/{name}/", relogin=False
     )
     task_url = delete_resp["task"]
 
-    task_detail = gc.get(task_url)
+    task_detail = gc.get(task_url, relogin=False)
     validate_json(instance=task_detail, schema=schema_task_detail)
 
     wait_for_task(gc, delete_resp)

galaxy_ng/tests/integration/api/test_tasks.py

@@ -1,6 +1,7 @@
 import pytest
 
 from ..utils import get_client, UIClient, PulpObjectBase
+from ..utils.iqe_utils import aap_gateway
 from ..utils.tools import generate_random_string
 
 REQUIREMENTS_YAML = """
@@ -39,6 +40,7 @@ def test_logging_cid_value_in_task(galaxy_client, ansible_config):
 @pytest.mark.pulp_api
 @pytest.mark.deployment_standalone
 @pytest.mark.min_hub_version("4.7dev")
+@pytest.mark.skipif(not aap_gateway(), reason="This test only runs if AAP Gateway is deployed")
 def test_gateway_logging_cid_value_in_task(galaxy_client):
     gc = galaxy_client("admin")
     ans_repo = gc.get(