tests: move helpers to fixtures

canonical · Sep 6, 2024 · b5d3710 · b5d3710
1 parent 7fbb0d2
commit b5d3710
Show file tree

Hide file tree

Showing 3 changed files with 57 additions and 77 deletions.
diff --git a/oauth_tools/fixtures.py b/oauth_tools/fixtures.py
@@ -2,13 +2,15 @@
 # Copyright 2024 Canonical Ltd.
 # See LICENSE file for licensing details.
 
+import functools
 import logging
 import os
-from typing import Any, AsyncGenerator, Callable, Coroutine, Dict, Generator
+from typing import Any, AsyncGenerator, Callable, Coroutine, Dict, Generator, Optional
 
 import pytest
 import pytest_asyncio
-from lightkube import Client, KubeConfig
+from lightkube import ApiError, Client, KubeConfig
+from lightkube.resources.core_v1 import Service
 from playwright.async_api import async_playwright
 from playwright.async_api._generated import Browser, BrowserContext, BrowserType, Page
 from playwright.async_api._generated import Playwright as AsyncPlaywright
@@ -26,6 +28,29 @@ def client() -> Client:
     return Client(config=KubeConfig.from_file(KUBECONFIG), field_manager="dex-test")
 
 
+async def get_k8s_service_address(ops_test: OpsTest, service_name: str, client: Client) -> Optional[str]:
+    """Get the address of a LoadBalancer Kubernetes service using kubectl."""
+    try:
+        result = client.get(Service, name=service_name, namespace=ops_test.model.name)
+        return result.status.loadBalancer.ingress[0].ip
+    except ApiError as e:
+        logger.error(f"Error retrieving service address: {e}")
+        return None
+
+
+async def reverse_proxy_app_url(ops_test: OpsTest, ingress_app_name: str, app_name: str, client: Client) -> str:
+    """Get the address of a proxied application."""
+    address = await get_k8s_service_address(ops_test, f"{ingress_app_name}-lb", client)
+    proxy_app_url = f"https://{address}/{ops_test.model.name}-{app_name}/"
+    logger.debug(f"Retrieved IP address: {proxy_app_url}")
+    return proxy_app_url
+
+
+@pytest_asyncio.fixture(scope="module")
+async def get_reverse_proxy_app_url(ops_test: OpsTest, client: Client) -> Callable:
+    return functools.partial(reverse_proxy_app_url, ops_test=ops_test, client=client)
+
+
 @pytest.fixture(scope="module")
 def ext_idp_service(ops_test: OpsTest, client: Client) -> Generator[DexIdpService, None, None]:
     """Deploy and manage the lifecycle of an Dex service."""

diff --git a/oauth_tools/oauth_helpers.py b/oauth_tools/oauth_helpers.py
@@ -5,10 +5,8 @@
 import logging
 import re
 from os.path import join
-from typing import Dict, List, Optional
+from typing import Callable, Dict, List, Optional
 
-from lightkube import ApiError, Client
-from lightkube.resources.core_v1 import Service
 from playwright.async_api import expect
 from playwright.async_api._generated import BrowserContext, Page
 from pytest_operator.plugin import OpsTest
@@ -19,33 +17,6 @@
 logger = logging.getLogger(__name__)
 
 
-async def get_k8s_service_address(ops_test: OpsTest, service_name: str, client: Client) -> Optional[str]:
-    """Get the address of a LoadBalancer Kubernetes service using kubectl."""
-    try:
-        result = client.get(Service, name=service_name, namespace=ops_test.model.name)
-        return result.status.loadBalancer.ingress[0].ip
-    except ApiError as e:
-        logger.error(f"Error retrieving service address: {e}")
-        return None
-
-
-async def get_reverse_proxy_app_url(
-    ops_test: OpsTest, ingress_app_name: str, app_name: str, client: Client
-) -> str:
-    """Get the address of a proxied application.
-
-    Args:
-        ops_test (OpsTest): The ops_test fixture.
-        ingress_app_name (str): The ingress app's name.
-        app_name (str): The app's name.
-        client: The lightkube client.
-    """
-    address = await get_k8s_service_address(ops_test, f"{ingress_app_name}-lb", client)
-    proxy_app_url = f"https://{address}/{ops_test.model.name}-{app_name}/"
-    logger.debug(f"Retrieved IP address: {proxy_app_url}")
-    return proxy_app_url
-
-
 async def deploy_identity_bundle(
     ops_test: OpsTest,
     bundle_url: str = "identity-platform",
@@ -170,15 +141,15 @@ async def click_on_sign_in_button_by_alt_text(page: Page, alt_text: str):
 
 
 async def complete_auth_code_login(
-    page: Page, ops_test: OpsTest, ext_idp_service: ExternalIdpService, client: Client
+    page: Page, ops_test: OpsTest, ext_idp_service: ExternalIdpService, get_reverse_proxy_app_url: Callable
 ) -> None:
     """Take a page that is in the identity-platform's login page and login the user.
 
     Args:
         page (page): The page fixture.
         ops_test (OpsTest): The ops_test fixture.
         ext_idp_service (ExternalIdpService): The ExternalIdpService.
-        client: The lightkube client.
+        get_reverse_proxy_app_url: The get_reverse_proxy_app_url fixture.
     """
     if not isinstance(ext_idp_service, ExternalIdpService):
         raise ValueError(
@@ -187,7 +158,7 @@ async def complete_auth_code_login(
 
     expected_url = join(
         await get_reverse_proxy_app_url(
-            ops_test, APPS.TRAEFIK_PUBLIC, APPS.IDENTITY_PLATFORM_LOGIN_UI_OPERATOR, client
+            ingress_app_name=APPS.TRAEFIK_PUBLIC, app_name=APPS.IDENTITY_PLATFORM_LOGIN_UI_OPERATOR
         ),
         "ui/login",
     )
@@ -205,7 +176,7 @@ async def complete_device_login(
     ops_test: OpsTest,
     verification_uri_complete: str,
     ext_idp_service: ExternalIdpService,
-    client: Client,
+    get_reverse_proxy_app_url: Callable,
 ) -> None:
     """Perform the device code login flow.
 
@@ -214,12 +185,12 @@ async def complete_device_login(
         ops_test (OpsTest): The ops_test fixture.
         verification_uri_complete (str): The `verification_uri_complete` the user is shown.
         ext_idp_service (ExternalIdpService): The ExternalIdpService.
-        client: The lightkube client.
+        get_reverse_proxy_app_url: The get_reverse_proxy_app_url fixture.
     """
     await page.goto(verification_uri_complete)
     expected_url = join(
         await get_reverse_proxy_app_url(
-            ops_test, APPS.TRAEFIK_PUBLIC, "identity-platform-login-ui-operator", client
+            ingress_app_name=APPS.TRAEFIK_PUBLIC, app_name="identity-platform-login-ui-operator"
         ),
         "ui/device_code",
     )
@@ -229,12 +200,14 @@ async def complete_device_login(
     async with page.expect_navigation():
         await page.get_by_role("button", name="Next").click()
 
-    await complete_auth_code_login(page, ops_test, ext_idp_service=ext_idp_service, client=client)
+    await complete_auth_code_login(
+        page, ops_test, ext_idp_service=ext_idp_service, get_reverse_proxy_app_url=get_reverse_proxy_app_url
+    )
 
     logger.info("Device login flow is complete")
     expected_url = join(
         await get_reverse_proxy_app_url(
-            ops_test, APPS.TRAEFIK_PUBLIC, "identity-platform-login-ui-operator", client
+            ingress_app_name=APPS.TRAEFIK_PUBLIC, app_name="identity-platform-login-ui-operator"
         ),
         "ui/device_complete",
     )
@@ -281,8 +254,6 @@ async def get_cookies_from_browser_by_url(browser_context: BrowserContext, url:
 
 
 __all__ = [
-    "get_k8s_service_address",
-    "get_reverse_proxy_app_url",
     "deploy_identity_bundle",
     "clean_up_identity_bundle",
     "access_application_login_page",

diff --git a/tests/integration/test_bundle.py b/tests/integration/test_bundle.py
@@ -7,7 +7,7 @@
 import os
 from os.path import join
 from pathlib import Path
-from typing import Optional
+from typing import Callable
 from urllib.parse import parse_qs, urlparse
 
 import pytest
@@ -20,8 +20,6 @@
     get_authorization_url,
     userinfo_request,
 )
-from lightkube import ApiError, Client
-from lightkube.resources.core_v1 import Service
 from playwright.async_api._generated import Page
 from pytest_operator.plugin import OpsTest
 
@@ -45,25 +43,6 @@ def get_bundle_template() -> Path:
     return get_this_script_dir() / ".." / ".." / "bundle.yaml.j2"
 
 
-async def get_k8s_service_address(ops_test: OpsTest, service_name: str, client: Client) -> Optional[str]:
-    """Get the address of a LoadBalancer Kubernetes service using kubectl."""
-    try:
-        result = client.get(Service, name=service_name, namespace=ops_test.model.name)
-        return result.status.loadBalancer.ingress[0].ip
-    except ApiError as e:
-        logger.error(f"Error retrieving service address: {e}")
-        return None
-
-
-async def get_reverse_proxy_app_url(
-    ops_test: OpsTest, ingress_app_name: str, app_name: str, client: Client
-) -> str:
-    address = await get_k8s_service_address(ops_test, f"{ingress_app_name}-lb", client)
-    proxy_app_url = f"https://{address}/{ops_test.model.name}-{app_name}/"
-    logger.debug(f"Retrieved IP address: {proxy_app_url}")
-    return proxy_app_url
-
-
 @pytest.mark.skip_if_deployed
 @pytest.mark.abort_on_fail
 async def test_render_and_deploy_bundle(
@@ -90,10 +69,10 @@ async def test_render_and_deploy_bundle(
 
 @pytest.mark.abort_on_fail
 async def test_hydra_is_up(
-    ops_test: OpsTest, admin_traefik_app_name: str, hydra_app_name: str, client: Client
+    ops_test: OpsTest, admin_traefik_app_name: str, hydra_app_name: str, get_reverse_proxy_app_url: Callable
 ) -> None:
     """Check that hydra and its environment dependencies (e.g. the database) are responsive."""
-    hydra_url = await get_reverse_proxy_app_url(ops_test, admin_traefik_app_name, hydra_app_name, client)
+    hydra_url = await get_reverse_proxy_app_url(ingress_app_name=admin_traefik_app_name, app_name=hydra_app_name)
 
     health_check_url = join(hydra_url, "health/ready")
 
@@ -103,10 +82,10 @@ async def test_hydra_is_up(
 
 @pytest.mark.abort_on_fail
 async def test_kratos_is_up(
-    ops_test: OpsTest, admin_traefik_app_name: str, kratos_app_name: str, client: Client
+    ops_test: OpsTest, admin_traefik_app_name: str, kratos_app_name: str, get_reverse_proxy_app_url: Callable
 ) -> None:
     """Check that kratos and its environment dependencies (e.g. the database) are responsive."""
-    kratos_url = await get_reverse_proxy_app_url(ops_test, admin_traefik_app_name, kratos_app_name, client)
+    kratos_url = await get_reverse_proxy_app_url(ingress_app_name=admin_traefik_app_name, app_name=kratos_app_name)
 
     health_check_url = join(kratos_url, "admin/health/ready")
 
@@ -249,12 +228,12 @@ async def test_authorization_code_flow(
     user_email: str,
     hydra_app_name: str,
     public_traefik_app_name: str,
-    client: Client,
+    get_reverse_proxy_app_url: Callable,
 ) -> None:
     # This is a hack, we just need a server to be running on the redirect_uri
     # so that when we get redirected there we don't get a connection_refused
     # error.
-    redirect_uri = await get_reverse_proxy_app_url(ops_test, public_traefik_app_name, "dummy", client)
+    redirect_uri = await get_reverse_proxy_app_url(ingress_app_name=public_traefik_app_name, app_name="dummy")
     app = ops_test.model.applications[hydra_app_name]
     action = await app.units[0].run_action(
         "create-oauth-client",
@@ -267,12 +246,17 @@ async def test_authorization_code_flow(
     client_id = res["client-id"]
     client_secret = res["client-secret"]
 
-    hydra_url = await get_reverse_proxy_app_url(ops_test, public_traefik_app_name, hydra_app_name, client)
+    hydra_url = await get_reverse_proxy_app_url(ingress_app_name=public_traefik_app_name, app_name=hydra_app_name)
 
     # Go to hydra authorization endpoint
     await page.goto(get_authorization_url(hydra_url, client_id, redirect_uri))
 
-    await complete_auth_code_login(page, ops_test, ext_idp_service=ext_idp_service, client=client)
+    await complete_auth_code_login(
+        page,
+        ops_test,
+        ext_idp_service=ext_idp_service,
+        get_reverse_proxy_app_url=get_reverse_proxy_app_url,
+    )
 
     await page.wait_for_url(redirect_uri + "?*")
 
@@ -302,7 +286,7 @@ async def test_client_credentials_flow(
     ops_test: OpsTest,
     hydra_app_name: str,
     public_traefik_app_name: str,
-    client: Client,
+    get_reverse_proxy_app_url: Callable,
 ) -> None:
     app = ops_test.model.applications[hydra_app_name]
     action = await app.units[0].run_action(
@@ -315,7 +299,7 @@ async def test_client_credentials_flow(
     client_id = res["client-id"]
     client_secret = res["client-secret"]
 
-    hydra_url = await get_reverse_proxy_app_url(ops_test, public_traefik_app_name, hydra_app_name, client)
+    hydra_url = await get_reverse_proxy_app_url(ingress_app_name=public_traefik_app_name, app_name=hydra_app_name)
 
     resp = client_credentials_grant_request(hydra_url, client_id, client_secret)
 
@@ -329,7 +313,7 @@ async def test_device_flow(
     user_email: str,
     hydra_app_name: str,
     public_traefik_app_name: str,
-    client: Client,
+    get_reverse_proxy_app_url: Callable,
 ) -> None:
     scopes = ["openid", "profile", "email", "offline_access"]
     app = ops_test.model.applications[hydra_app_name]
@@ -344,7 +328,7 @@ async def test_device_flow(
     client_id = res["client-id"]
     client_secret = res["client-secret"]
 
-    hydra_url = await get_reverse_proxy_app_url(ops_test, public_traefik_app_name, hydra_app_name, client)
+    hydra_url = await get_reverse_proxy_app_url(ingress_app_name=public_traefik_app_name, app_name=hydra_app_name)
 
     # Make the device auth request
     auth_resp = device_auth_request(hydra_url, client_id, client_secret, scope=" ".join(scopes))
@@ -369,7 +353,7 @@ async def test_device_flow(
         ops_test,
         device_auth_resp["verification_uri_complete"],
         ext_idp_service=ext_idp_service,
-        client=client,
+        get_reverse_proxy_app_url=get_reverse_proxy_app_url,
     )
 
     # Exchange device code for tokens