Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion testsuite/page_objects/nav_bar.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

from testsuite.page_objects.navigator import step, Navigable
from testsuite.page_objects.policies.policies import PoliciesPage
from testsuite.page_objects.overview.overview_page import OverviewPage


class NavBar(Navigable):
Expand All @@ -20,9 +21,11 @@ def expand_kuadrant(self):
if self.kuadrant_nav.get_attribute("aria-expanded") == "false":
self.kuadrant_nav.click(timeout=60000)

@step(OverviewPage)
def overview(self):
"""Navigates to the console plugin Overview page"""
"""Navigates to the console plugin Overview page and returns an OverviewPage object"""
self.expand_kuadrant()
self.page.locator("//a[contains(@class, 'nav__link') and contains(@href, '/kuadrant/overview')]").click()

@step(PoliciesPage)
def policies(self):
Expand Down
Empty file.
45 changes: 45 additions & 0 deletions testsuite/page_objects/overview/overview_page.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
"""Page object for the console plugin Overview page"""

import backoff
from playwright.sync_api import Page
from testsuite.page_objects.navigator import Navigable


class OverviewPage(Navigable):
"""Page object for Overview page"""

def __init__(self, page: Page):
super().__init__(page)
self.page_heading = page.locator("h1").filter(has_text="Overview")

def is_displayed(self):
"""Returns the page heading locator"""
return self.page_heading

def page_displayed(self):
"""Check if the overview page is displayed"""
self.page_heading.wait_for(state="visible", timeout=60000)
return self.page_heading.is_visible()

def get_metric_count(self, metric_name: str):
"""Get the count from a gateway metric card"""
metric_card = self.page.locator(f"//div[contains(., '{metric_name}')]").first
count_element = metric_card.locator("strong").first
return int(count_element.inner_text().strip())

@backoff.on_predicate(backoff.constant, interval=2, max_tries=30, jitter=None)
def wait_for_healthy_gateways(self, expected_count: int):
"""Wait for the healthy gateways count to reach the expected value"""
return self.get_metric_count("Healthy Gateways") >= expected_count

def has_gateway_in_traffic_analysis(self, gateway_name: str):
"""Check if gateway appears in traffic analysis section"""
return self.page.wait_for_selector(f"//tr//a[@data-test-id='{gateway_name}']")

def has_httproute_in_section(self, route_name: str):
"""Check if HTTPRoute appears in HTTPRoutes section"""
return self.page.wait_for_selector(f"//tr//a[@data-test-id='{route_name}']")

def has_policy_in_section(self, policy_name: str):
"""Check if policy appears in Policies section"""
return self.page.wait_for_selector(f"//tr//a[@data-test-id='{policy_name}']")
13 changes: 13 additions & 0 deletions testsuite/tests/singlecluster/ui/console_plugin/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,19 @@ def navigate_console(page, exposer, cluster, skip_or_fail):
NavBar(page).kuadrant_nav.wait_for(state="visible", timeout=30000)


@pytest.fixture(scope="session")
def openshift_version(cluster):
"""Get OpenShift cluster version"""
result = cluster.do_action(
"get", "clusterversion", "version", "-o", "jsonpath={.status.desired.version}", auto_raise=False
)
if result.status() != 0:
return None
version_str = result.out().strip()
parts = version_str.split(".")
return tuple(int(p.split("-")[0]) for p in parts[:2]) # Convert "4.20.0" -> (4, 20)


@pytest.fixture
def navigator(page):
"""Return a Navigator bound to the current Playwright page"""
Expand Down
Empty file.
Original file line number Diff line number Diff line change
@@ -0,0 +1,159 @@
"""UI tests for console plugin Overview page"""

import pytest
from testsuite.gateway import GatewayListener
from testsuite.gateway.gateway_api.gateway import KuadrantGateway
from testsuite.gateway.gateway_api.route import HTTPRoute
from testsuite.kuadrant.policy.authorization.auth_policy import AuthPolicy
from testsuite.page_objects.overview.overview_page import OverviewPage

pytestmark = [pytest.mark.ui]


def test_overview_page_panels_and_links(navigator):
"""Verify all section panels are visible and Getting started resources has clickable links"""
# Navigate to overview page
overview_page = navigator.navigate(OverviewPage)
assert overview_page.page_displayed(), "Overview page did not load"

# Verify all expected section panels exist
getting_started = overview_page.page.get_by_text("Getting started resources")
assert getting_started.is_visible()
assert overview_page.page.get_by_role("heading", name="Gateways", exact=True).is_visible()
assert overview_page.page.get_by_role("heading", name="Gateways - Traffic Analysis", exact=True).is_visible()
assert overview_page.page.get_by_role("heading", name="Policies", exact=True).is_visible()
assert overview_page.page.get_by_role("heading", name="HTTPRoutes", exact=True).is_visible()

# Verify Getting started resources panel has clickable links
getting_started_section = getting_started.locator("xpath=ancestor::section").first
links = getting_started_section.get_by_role("link").all()
clickable_links = [link for link in links if link.is_visible() and link.is_enabled()]
assert len(clickable_links) > 0, "No clickable links found in Getting started resources panel"


def test_creation_buttons(navigator):
"""Verify creation buttons are clickable and policy dropdown shows available policy types"""
# Navigate to overview page
overview_page = navigator.navigate(OverviewPage)
assert overview_page.page_displayed(), "Overview page did not load"

# Verify Create Gateway button is visible and clickable
create_gateway = overview_page.page.get_by_text("Create Gateway")
assert create_gateway.is_visible() and create_gateway.is_enabled()

# Verify Create HTTPRoute button is visible and clickable
create_httproute = overview_page.page.get_by_text("Create HTTPRoute")
assert create_httproute.is_visible() and create_httproute.is_enabled()

# Verify Create Policy button is visible and clickable, then open dropdown
create_policy = overview_page.page.get_by_text("Create Policy")
assert create_policy.is_visible() and create_policy.is_enabled()
create_policy.click()

# Verify core policies are visible in the dropdown
core_policies = ["AuthPolicy", "RateLimitPolicy", "DNSPolicy", "TLSPolicy"]
for policy_type in core_policies:
menu_item = overview_page.page.get_by_role("menuitem", name=policy_type, exact=True)
assert menu_item.is_visible(), f"{policy_type} should be visible"
assert menu_item.is_enabled(), f"{policy_type} should be enabled"


def test_additional_policy_types_in_dropdown(navigator, openshift_version):
"""Verify additional policy types appear in Create Policy dropdown (OCP 4.20+)"""
if openshift_version is None:
pytest.skip("Could not detect OpenShift version")
if openshift_version < (4, 20):
pytest.skip("Requires OCP 4.20+ for OIDCPolicy, PlanPolicy, TokenRateLimitPolicy")

overview_page = navigator.navigate(OverviewPage)
assert overview_page.page_displayed(), "Overview page did not load"

# Open Create Policy dropdown
create_policy = overview_page.page.get_by_text("Create Policy")
assert create_policy.is_visible() and create_policy.is_enabled()
create_policy.click()

# Verify additional policies available in OCP 4.20+
additional_policies = ["OIDCPolicy", "PlanPolicy", "TokenRateLimitPolicy"]
for policy_type in additional_policies:
menu_item = overview_page.page.get_by_role("menuitem", name=policy_type, exact=True)
assert menu_item.is_visible(), f"{policy_type} should be visible in OCP 4.20+"
assert menu_item.is_enabled(), f"{policy_type} should be enabled"


def test_resources_appear_in_sections(request, navigator, cluster, blame, module_label, wildcard_domain, backend):
"""Verify gateway, HTTPRoute, and policy resources appear in their respective section panels"""
# Create resources programmatically
gateway_name = blame("gw")
gateway = KuadrantGateway.create_instance(cluster, gateway_name, {"app": module_label})
gateway.add_listener(GatewayListener(hostname=wildcard_domain))
request.addfinalizer(gateway.delete)
gateway.commit()

route_name = blame("route")
route = HTTPRoute.create_instance(cluster, route_name, gateway)
route.add_backend(backend)
request.addfinalizer(route.delete)
route.commit()

policy_name = blame("policy")
policy = AuthPolicy.create_instance(cluster, policy_name, gateway)
policy.authorization.add_opa_policy("denyAll", "allow = false")
request.addfinalizer(policy.delete)
policy.commit()

# Navigate to overview page
overview_page = navigator.navigate(OverviewPage)
assert overview_page.page_displayed(), "Overview page did not load"

# Verify gateway appears in Gateways - Traffic Analysis section panel
assert overview_page.has_gateway_in_traffic_analysis(
gateway_name
), f"Gateway '{gateway_name}' not visible in traffic analysis section panel"

# Verify HTTPRoute appears in HTTPRoutes section panel
assert overview_page.has_httproute_in_section(
route_name
), f"HTTPRoute '{route_name}' not visible in HTTPRoutes section panel"

# Verify policy appears in Policies section panel
assert overview_page.has_policy_in_section(
policy_name
), f"Policy '{policy_name}' not visible in Policies section panel"


def test_gateway_section_status(request, navigator, cluster, blame, module_label, wildcard_domain):
"""Verify gateway status metrics update correctly"""
# Navigate to overview page and capture initial counts
overview_page = navigator.navigate(OverviewPage)
assert overview_page.page_displayed(), "Overview page did not load"

initial_total = overview_page.get_metric_count("Total Gateways")
initial_healthy = overview_page.get_metric_count("Healthy Gateways")
initial_unhealthy = overview_page.get_metric_count("Unhealthy Gateways")

# Create gateway programmatically
gateway_name = blame("gw")
gateway = KuadrantGateway.create_instance(cluster, gateway_name, {"app": module_label})
gateway.add_listener(GatewayListener(hostname=wildcard_domain))
request.addfinalizer(gateway.delete)
gateway.commit()

# Refresh page to see updated metrics
overview_page = navigator.navigate(OverviewPage)
assert overview_page.page_displayed(), "Overview page did not load"

# Verify total gateway count increased
new_total = overview_page.get_metric_count("Total Gateways")
assert new_total > initial_total, f"Total Gateways count did not increase (was {initial_total}, now {new_total})"

# Verify unhealthy count increased (gateway starts as unhealthy while provisioning)
new_unhealthy = overview_page.get_metric_count("Unhealthy Gateways")
assert (
new_unhealthy > initial_unhealthy
), f"Unhealthy count did not increase (was {initial_unhealthy}, now {new_unhealthy})"
Comment on lines +150 to +154
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We expect the Gateway is first is state "unhealthy" but what if it reconciles so quick the ui plugin will only show "healthy"? This could happen on very fast infrastructures.
This can be left as is for now, but maybe in the future or on certain cloud providers this could cause a false negative fail.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That’s a fair point! So far I’ve always seen it report Unhealthy, even if just briefly, before becoming Healthy, but I can see how on faster infra that might not always be the case. I will for sure dig into it a bit more in the future, along with some other UI refactoring I want to do 😄


# Wait for gateway to become healthy and verify healthy count increased
overview_page.wait_for_healthy_gateways(initial_healthy + 1)
new_healthy = overview_page.get_metric_count("Healthy Gateways")
assert new_healthy > initial_healthy, f"Healthy count did not increase (was {initial_healthy}, now {new_healthy})"