gateway

dev/ephemeral/run_tests.sh

@@ -12,7 +12,7 @@ ${VENV_PATH}/bin/pip install -r integration_requirements.txt
 
 echo "Running pytest ..."
 ${VENV_PATH}/bin/pytest \
-    --capture=no -m "deployment_cloud or all" \
+    --capture=no --log-cli-level=ERROR -m "deployment_cloud or all" \
     -v \
     galaxy_ng/tests/integration $@
 RC=$?

galaxy_ng/tests/integration/api/rbac_actions/exec_env.py

@@ -1,5 +1,6 @@
 import requests
 
+from galaxykit.utils import GalaxyClientError
 from .utils import (
     API_ROOT,
     PULP_API_ROOT,
@@ -177,7 +178,11 @@ def ee_namespace_remove_role(user, password, expect_pass, extra):
 
 def create_ee_local(user, password, expect_pass, extra):
     name = gen_string()
-    return_code = podman_push(user['username'], password, name)
+    try:
+        podman_push(user['username'], password, name)
+        return_code = 0
+    except GalaxyClientError:
+        return_code = 1
 
     if return_code == 0:
         del_container(name)
@@ -192,7 +197,11 @@ def create_ee_in_existing_namespace(user, password, expect_pass, extra):
     namespace = extra["local_ee"].get_namespace()["name"]
     name = f"{namespace}/{gen_string()}"
 
-    return_code = podman_push(user['username'], password, name)
+    try:
+        podman_push(user['username'], password, name)
+        return_code = 0
+    except GalaxyClientError:
+        return_code = 1
 
     if return_code == 0:
         del_container(name)
@@ -207,7 +216,11 @@ def push_updates_to_existing_ee(user, password, expect_pass, extra):
     container = extra["local_ee"].get_container()["name"]
     tag = gen_string()
 
-    return_code = podman_push(user['username'], password, container, tag=tag)
+    try:
+        podman_push(user['username'], password, container, tag=tag)
+        return_code = 0
+    except GalaxyClientError:
+        return_code = 1
 
     if expect_pass:
         assert return_code == 0

galaxy_ng/tests/integration/api/rbac_actions/utils.py

@@ -15,19 +15,22 @@
 
 from ansible.galaxy.api import GalaxyError
 
-from galaxy_ng.tests.integration.utils.iqe_utils import is_ephemeral_env, get_ansible_config, \
+from galaxy_ng.tests.integration.utils.iqe_utils import get_ansible_config, \
     get_galaxy_client, AnsibleConfigFixture
+from galaxy_ng.tests.integration.utils.rbac_utils import create_local_image_container
 from galaxykit.container_images import get_container, get_container_images_latest
 ansible_config = get_ansible_config()
 CLIENT_CONFIG = ansible_config("admin")
 ADMIN_CLIENT = get_client(CLIENT_CONFIG)
 
+'''
 ansible_config = get_ansible_config()
 galaxy_client = get_galaxy_client(ansible_config)
 if is_ephemeral_env():
     gc_admin = galaxy_client("admin", basic_token=True)
 else:
-    gc_admin = galaxy_client("admin", basic_token=False)
+    gc_admin = galaxy_client("admin", basic_token=False, ignore_cache=True)
+'''
 
 API_ROOT = CLIENT_CONFIG["url"]
 PULP_API_ROOT = f"{API_ROOT}pulp/api/v3/"
@@ -146,37 +149,14 @@ def ensure_test_container_is_pulled():
 
 
 def podman_push(username, password, container, tag="latest"):
-    ensure_test_container_is_pulled()
-    container_engine = CLIENT_CONFIG["container_engine"]
-    container_registry = CLIENT_CONFIG["container_registry"]
-
-    new_container = f"{container_registry}/{container}:{tag}"
-    tag_cmd = [container_engine, "image", "tag", TEST_CONTAINER, new_container]
-
-    subprocess.run(tag_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-
-    if container_engine == "docker":
-        login_cmd = ["docker", "login", "-u", username, "-p", password, container_registry]
-        subprocess.run(login_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-
-    if container_engine == "podman":
-        push_cmd = [
-            container_engine,
-            "push",
-            "--creds",
-            f"{username}:{password}",
-            new_container,
-            "--remove-signatures",
-            "--tls-verify=false"]
-
-    if container_engine == "docker":
-        push_cmd = [
-            container_engine,
-            "push",
-            new_container]
-
-    rc = subprocess.run(push_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-    return rc.returncode
+    _ansible_config = get_ansible_config()
+    _galaxy_client = get_galaxy_client(_ansible_config)
+    gc = _galaxy_client({
+        "username": username,
+        "password": password,
+    })
+    create_local_image_container(_ansible_config("admin"), gc, container, tag)
+    return 0
 
 
 def del_user(pk):
@@ -505,13 +485,13 @@ def _reset(self):
 
         # get roles first
         roles = session.get(
-            f"{API_ROOT}{pulp_namespace_path}/list_roles",
+            f"{API_ROOT}{pulp_namespace_path}/list_roles/",
             auth=ADMIN_CREDENTIALS
         ).json()
 
         for role in roles['roles']:
             self.pulp_namespace = session.post(
-                f"{API_ROOT}{pulp_namespace_path}/remove_role",
+                f"{API_ROOT}{pulp_namespace_path}/remove_role/",
                 json={
                     'role': role['role']
                 },
@@ -546,11 +526,11 @@ def __del__(self):
 
 
 class ReusableLocalContainer:
-    def __init__(self, name):
+    def __init__(self, name, gc):
         self._ns_name = f"ee_ns_{name}"
         self._repo_name = f"ee_local_{name}"
         self._name = f"{self._ns_name}/{self._repo_name}"
-
+        self.gc = gc
         self._reset()
 
     def _reset(self):
@@ -562,19 +542,20 @@ def _reset(self):
         # 1. get namespace pulp_id from repositories
         # 2. get roles in namespace
         # 3. remove roles and groups (clean container namespace)
-        self._container = get_container(gc_admin, self._name)
-        ns_r = gc_admin.get(f"pulp/api/v3/pulp_container/namespaces/?name={self._ns_name}")
+
+        self._container = get_container(self.gc, self._name)
+        ns_r = self.gc.get(f"pulp/api/v3/pulp_container/namespaces/?name={self._ns_name}")
         pulp_namespace_path = ns_r["results"][0]["pulp_href"]
         # get roles first
-        roles = gc_admin.get(f"{pulp_namespace_path}list_roles")
+        roles = self.gc.get(f"{pulp_namespace_path}list_roles")
         for role in roles["roles"]:
             body = {
                 'role': role["role"]
             }
-            gc_admin.post(path=f"{pulp_namespace_path}remove_role/", body=body)
+            self.gc.post(path=f"{pulp_namespace_path}remove_role/", body=body)
 
-        self._namespace = gc_admin.get(pulp_namespace_path)
-        self._manifest = get_container_images_latest(gc_admin, self._name)
+        self._namespace = self.gc.get(pulp_namespace_path)
+        self._manifest = get_container_images_latest(self.gc, self._name)
 
     def get_container(self):
         return self._container

galaxy_ng/tests/integration/api/test_aiindex.py

@@ -27,11 +27,11 @@ def namespace(ansible_config, galaxy_client) -> str:
 
 
 @pytest.fixture(scope="function")
-def pe_namespace(ansible_config) -> str:
+def pe_namespace(ansible_config, galaxy_client) -> str:
     """create a new namespace owned by PE user."""
     config = ansible_config("partner_engineer")
-    api_client = get_client(config, request_token=True, require_auth=True)
-    new_namespace = generate_unused_namespace(api_client=api_client, api_version="_ui/v1")
+    gc = galaxy_client("partner_engineer")
+    new_namespace = generate_unused_namespace(gc=gc, api_version="_ui/v1")
     with UIClient(config=config) as uclient:
         # get user
         resp = uclient.get("_ui/v1/me/")

galaxy_ng/tests/integration/api/test_api_base.py

@@ -1,10 +1,11 @@
 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,55 +15,40 @@ 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")
 @pytest.mark.all
-def test_galaxy_api_root(ansible_config, artifact):
+def test_galaxy_api_root(galaxy_client, artifact):
     """Test galaxy API root."""
 
-    config = ansible_config("basic_user")
-
-    api_prefix = config.get("api_prefix")
-    api_prefix = api_prefix.rstrip("/")
-
-    api_client = get_client(
-        config=config,
-        request_token=True,
-        require_auth=True
-    )
-
+    # TODO: change to `basic_user` profile when can access pulp-v3 api root
+    gc = galaxy_client("admin")
     # verify api root works
-    response = api_client(api_prefix + '/')
+    response = gc.get(gc.galaxy_root)
     assert "v3" in response["available_versions"]
     assert "pulp-v3" in response["available_versions"]
 
-    v3_root = api_client(api_prefix + '/' + response['available_versions']['v3'])
+    # v3_root = gc.get(response['available_versions']['v3'])
+    v3_root = gc.get("v3/plugin/ansible/content/published/collections/")
     assert "published" in v3_root
 
-    pulp_root = api_client(api_prefix + '/' + response['available_versions']['pulp-v3'])
+    pulp_root = gc.get(response['available_versions']['pulp-v3'])
     assert "tasks" in pulp_root
 
 
 @pytest.mark.max_hub_version("4.5.5")
 @pytest.mark.all
-def test_galaxy_api_root_v_4_5(ansible_config, artifact):
+def test_galaxy_api_root_v_4_5(galaxy_client, artifact):
     """Test galaxy API root."""
 
     # TODO: change to `basic_user` profile when can access pulp-v3 api root
-    config = ansible_config("admin")
-    api_prefix = config.get("api_prefix")
-    api_prefix = api_prefix.rstrip("/")
-
-    api_client = get_client(
-        config=config,
-        request_token=True,
-        require_auth=True
-    )
+    gc = galaxy_client("admin")
 
     # verify api root works
-    response = api_client(api_prefix + '/')
+    response = gc.get(gc.galaxy_root)
     assert "v3" in response["available_versions"]
 
-    v3_root = api_client(api_prefix + '/' + response['available_versions']['v3'])
+    v3_root = gc.get(response['available_versions']['v3'])
     assert "published" in v3_root