|
1 |
| -# Copyright 2024 Canonical Ltd. |
| 1 | +# Copyright 2025 Canonical Ltd. |
2 | 2 | # See LICENSE file for licensing details.
|
3 | 3 |
|
4 | 4 | """Fixtures for Jenkins-k8s-operator charm integration tests."""
|
|
19 | 19 | import requests
|
20 | 20 | from juju.action import Action
|
21 | 21 | from juju.application import Application
|
22 |
| -from juju.client._definitions import FullStatus, UnitStatus |
23 | 22 | from juju.model import Controller, Model
|
24 | 23 | from juju.unit import Unit
|
25 | 24 | from keycloak import KeycloakAdmin, KeycloakOpenIDConnection
|
|
38 | 37 | get_dex_service_url,
|
39 | 38 | update_redirect_uri,
|
40 | 39 | )
|
41 |
| -from .helpers import generate_jenkins_client_from_application, get_pod_ip |
| 40 | +from .helpers import generate_jenkins_client_from_application, get_model_unit_addresses, get_pod_ip |
42 | 41 | from .types_ import KeycloakOIDCMetadata, LDAPSettings, ModelAppUnit, UnitWebClient
|
43 | 42 |
|
44 | 43 | KUBECONFIG = os.environ.get("TESTING_KUBECONFIG", "./kube-config")
|
@@ -129,13 +128,9 @@ def model_app_unit_fixture(model: Model, application: Application, unit: Unit):
|
129 | 128 | @pytest_asyncio.fixture(scope="module", name="unit_ip")
|
130 | 129 | async def unit_ip_fixture(model: Model, application: Application):
|
131 | 130 | """Get Jenkins charm unit IP."""
|
132 |
| - status: FullStatus = await model.get_status([application.name]) |
133 |
| - try: |
134 |
| - unit_status: UnitStatus = next(iter(status.applications[application.name].units.values())) |
135 |
| - assert unit_status.address, "Invalid unit address" |
136 |
| - return unit_status.address |
137 |
| - except StopIteration as exc: |
138 |
| - raise StopIteration("Invalid unit status") from exc |
| 131 | + unit_ips = await get_model_unit_addresses(model=model, app_name=application.name) |
| 132 | + assert unit_ips, f"Unit IP address not found for {application.name}" |
| 133 | + return unit_ips[0] |
139 | 134 |
|
140 | 135 |
|
141 | 136 | @pytest.fixture(scope="module", name="web_address")
|
@@ -480,15 +475,9 @@ async def jenkins_with_proxy_fixture(
|
480 | 475 | @pytest_asyncio.fixture(scope="module", name="proxy_jenkins_unit_ip")
|
481 | 476 | async def proxy_jenkins_unit_ip_fixture(model: Model, jenkins_with_proxy: Application):
|
482 | 477 | """Get Jenkins charm w/ proxy enabled unit IP."""
|
483 |
| - status: FullStatus = await model.get_status([jenkins_with_proxy.name]) |
484 |
| - try: |
485 |
| - unit_status: UnitStatus = next( |
486 |
| - iter(status.applications[jenkins_with_proxy.name].units.values()) |
487 |
| - ) |
488 |
| - assert unit_status.address, "Invalid unit address" |
489 |
| - return unit_status.address |
490 |
| - except StopIteration as exc: |
491 |
| - raise StopIteration("Invalid unit status") from exc |
| 478 | + unit_ips = await get_model_unit_addresses(model=model, app_name=jenkins_with_proxy.name) |
| 479 | + assert unit_ips, f"Unit IP address not found for {jenkins_with_proxy.name}" |
| 480 | + return unit_ips[0] |
492 | 481 |
|
493 | 482 |
|
494 | 483 | @pytest_asyncio.fixture(scope="module", name="proxy_jenkins_web_address")
|
@@ -781,10 +770,9 @@ async def traefik_application_fixture(model: Model):
|
781 | 770 | await model.wait_for_idle(
|
782 | 771 | status="active", apps=[traefik.name], timeout=20 * 60, idle_period=30, raise_on_error=False
|
783 | 772 | )
|
784 |
| - status = await model.get_status(filters=[traefik.name]) |
785 |
| - unit = next(iter(status.applications[traefik.name].units)) |
786 |
| - traefik_address = status["applications"][traefik.name]["units"][unit]["address"] |
787 |
| - return (traefik, traefik_address) |
| 773 | + unit_ips = await get_model_unit_addresses(model=model, app_name=traefik.name) |
| 774 | + assert unit_ips, f"Unit IP address not found for {traefik.name}" |
| 775 | + return (traefik, unit_ips[0]) |
788 | 776 |
|
789 | 777 |
|
790 | 778 | @pytest_asyncio.fixture(scope="module", name="oathkeeper_related")
|
|
0 commit comments