# -*- coding: utf-8 -*-
"""Location: ./tests/playwright/test_gateways.py
Copyright 2025
SPDX-License-Identifier: Apache-2.0
Authors: Mihai Criveti
Test cases for MCP Servers & Federated Gateways (MCP Registry) management.
"""
# Standard
import logging
# Third-Party
from playwright.sync_api import expect
import pytest
# Local
from .pages.gateways_page import GatewaysPage
logger = logging.getLogger(__name__)
@pytest.mark.ui
@pytest.mark.gateways
class TestGatewaysPage:
"""Test cases for Gateways page functionality."""
def test_gateways_panel_loads(self, gateways_page: GatewaysPage):
"""Test that gateways panel loads successfully."""
gateways_page.navigate_to_gateways_tab()
expect(gateways_page.gateways_panel).to_be_visible()
expect(gateways_page.panel_title).to_be_visible()
expect(gateways_page.gateways_table).to_be_visible()
def test_gateways_table_structure(self, gateways_page: GatewaysPage):
"""Test that gateways table has correct structure."""
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
# Verify table headers exist
table = gateways_page.gateways_table
expect(table.locator('th:has-text("Actions")')).to_be_visible()
expect(table.locator('th:has-text("S. No.")')).to_be_visible()
expect(table.locator('th:has-text("Name")')).to_be_visible()
expect(table.locator('th:has-text("URL")')).to_be_visible()
expect(table.locator('th:has-text("Tags")')).to_be_visible()
expect(table.locator('th:has-text("Status")')).to_be_visible()
expect(table.locator('th:has-text("Visibility")')).to_be_visible()
def test_add_gateway_form_visible(self, gateways_page: GatewaysPage):
"""Test that add gateway form is visible and has required fields."""
gateways_page.navigate_to_gateways_tab()
expect(gateways_page.add_gateway_form).to_be_visible()
expect(gateways_page.gateway_name_input).to_be_visible()
expect(gateways_page.gateway_url_input).to_be_visible()
expect(gateways_page.gateway_description_input).to_be_visible()
expect(gateways_page.gateway_tags_input).to_be_visible()
expect(gateways_page.transport_select).to_be_visible()
expect(gateways_page.auth_type_select).to_be_visible()
expect(gateways_page.add_gateway_btn).to_be_visible()
def test_search_functionality(self, gateways_page: GatewaysPage):
"""Test gateway search functionality."""
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
# Verify search input is visible
expect(gateways_page.search_input).to_be_visible()
expect(gateways_page.clear_search_btn).to_be_visible()
# Test search (if gateways exist)
initial_count = gateways_page.get_gateway_count()
if initial_count > 0:
gateways_page.search_gateways("nonexistent-gateway-xyz")
# search_gateways already dispatches events and waits 500ms internally
search_count = gateways_page.get_gateway_count()
assert search_count <= initial_count
def test_show_inactive_toggle(self, gateways_page: GatewaysPage):
"""Test show inactive gateways toggle."""
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
# Verify checkbox exists
expect(gateways_page.show_inactive_checkbox).to_be_visible()
# Test toggle
initial_state = gateways_page.show_inactive_checkbox.is_checked()
gateways_page.toggle_show_inactive(not initial_state)
gateways_page.page.wait_for_timeout(500)
assert gateways_page.show_inactive_checkbox.is_checked() == (not initial_state)
def test_transport_type_options(self, gateways_page: GatewaysPage):
"""Test that transport type select has correct options."""
gateways_page.navigate_to_gateways_tab()
transport_select = gateways_page.transport_select
expect(transport_select).to_be_visible()
# Verify SSE and STREAMABLEHTTP options exist
expect(transport_select.locator('option[value="SSE"]')).to_be_attached()
expect(transport_select.locator('option[value="STREAMABLEHTTP"]')).to_be_attached()
def test_auth_type_options(self, gateways_page: GatewaysPage):
"""Test that authentication type select has correct options."""
gateways_page.navigate_to_gateways_tab()
auth_select = gateways_page.auth_type_select
expect(auth_select).to_be_visible()
# Verify auth type options
expect(auth_select.locator('option[value=""]')).to_be_attached() # None
expect(auth_select.locator('option[value="basic"]')).to_be_attached()
expect(auth_select.locator('option[value="bearer"]')).to_be_attached()
expect(auth_select.locator('option[value="authheaders"]')).to_be_attached()
expect(auth_select.locator('option[value="oauth"]')).to_be_attached()
expect(auth_select.locator('option[value="query_param"]')).to_be_attached()
def test_visibility_radio_buttons(self, gateways_page: GatewaysPage):
"""Test that visibility radio buttons are present."""
gateways_page.navigate_to_gateways_tab()
expect(gateways_page.visibility_public_radio).to_be_attached()
expect(gateways_page.visibility_team_radio).to_be_attached()
expect(gateways_page.visibility_private_radio).to_be_attached()
# Public should be checked by default
expect(gateways_page.visibility_public_radio).to_be_checked()
def test_basic_auth_fields_visibility(self, gateways_page: GatewaysPage):
"""Test that basic auth fields appear when basic auth is selected."""
gateways_page.navigate_to_gateways_tab()
# Initially hidden
expect(gateways_page.auth_basic_fields).to_be_hidden()
# Select basic auth
gateways_page.auth_type_select.select_option("basic")
gateways_page.page.wait_for_timeout(300)
# Should now be visible
expect(gateways_page.auth_basic_fields).to_be_visible()
expect(gateways_page.auth_username_input).to_be_visible()
expect(gateways_page.auth_password_input).to_be_visible()
def test_bearer_auth_fields_visibility(self, gateways_page: GatewaysPage):
"""Test that bearer token fields appear when bearer auth is selected."""
gateways_page.navigate_to_gateways_tab()
# Initially hidden
expect(gateways_page.auth_bearer_fields).to_be_hidden()
# Select bearer auth
gateways_page.auth_type_select.select_option("bearer")
gateways_page.page.wait_for_timeout(300)
# Should now be visible
expect(gateways_page.auth_bearer_fields).to_be_visible()
expect(gateways_page.auth_token_input).to_be_visible()
def test_oauth_fields_visibility(self, gateways_page: GatewaysPage):
"""Test that OAuth fields appear when OAuth is selected."""
gateways_page.navigate_to_gateways_tab()
# Initially hidden
expect(gateways_page.oauth_fields).to_be_hidden()
# Select OAuth
gateways_page.auth_type_select.select_option("oauth")
gateways_page.page.wait_for_timeout(300)
# Should now be visible
expect(gateways_page.oauth_fields).to_be_visible()
expect(gateways_page.oauth_grant_type_select).to_be_visible()
expect(gateways_page.oauth_issuer_input).to_be_visible()
expect(gateways_page.oauth_client_id_input).to_be_visible()
expect(gateways_page.oauth_client_secret_input).to_be_visible()
def test_query_param_auth_fields_visibility(self, gateways_page: GatewaysPage):
"""Test that query parameter auth fields appear when selected."""
gateways_page.navigate_to_gateways_tab()
# Initially hidden
expect(gateways_page.auth_query_param_fields).to_be_hidden()
# Select query param auth
gateways_page.auth_type_select.select_option("query_param")
gateways_page.page.wait_for_timeout(300)
# Should now be visible with security warning
expect(gateways_page.auth_query_param_fields).to_be_visible()
expect(gateways_page.auth_query_param_key_input).to_be_visible()
expect(gateways_page.auth_query_param_value_input).to_be_visible()
def test_custom_headers_auth_fields_visibility(self, gateways_page: GatewaysPage):
"""Test that custom headers auth fields appear when selected."""
gateways_page.navigate_to_gateways_tab()
# Initially hidden
expect(gateways_page.auth_headers_fields).to_be_hidden()
# Select custom headers auth
gateways_page.auth_type_select.select_option("authheaders")
gateways_page.page.wait_for_timeout(300)
# Should now be visible
expect(gateways_page.auth_headers_fields).to_be_visible()
def test_oauth_grant_type_options(self, gateways_page: GatewaysPage):
"""Test OAuth grant type options."""
gateways_page.navigate_to_gateways_tab()
# Select OAuth to show fields
gateways_page.auth_type_select.select_option("oauth")
gateways_page.page.wait_for_timeout(300)
grant_type_select = gateways_page.oauth_grant_type_select
expect(grant_type_select).to_be_visible()
# Verify grant type options
expect(grant_type_select.locator('option[value="authorization_code"]')).to_be_attached()
expect(grant_type_select.locator('option[value="client_credentials"]')).to_be_attached()
expect(grant_type_select.locator('option[value="password"]')).to_be_attached()
def test_one_time_auth_checkbox(self, gateways_page: GatewaysPage):
"""Test one-time authentication checkbox."""
gateways_page.navigate_to_gateways_tab()
expect(gateways_page.one_time_auth_checkbox).to_be_visible()
# Should not be checked by default
expect(gateways_page.one_time_auth_checkbox).not_to_be_checked()
# Test toggle
gateways_page.toggle_one_time_auth(True)
expect(gateways_page.one_time_auth_checkbox).to_be_checked()
def test_passthrough_headers_input(self, gateways_page: GatewaysPage):
"""Test passthrough headers input field."""
gateways_page.navigate_to_gateways_tab()
expect(gateways_page.passthrough_headers_input).to_be_visible()
# Test filling
test_headers = "Authorization, X-Tenant-Id, X-Trace-Id"
gateways_page.passthrough_headers_input.fill(test_headers)
expect(gateways_page.passthrough_headers_input).to_have_value(test_headers)
def test_ca_certificate_upload_elements(self, gateways_page: GatewaysPage):
"""Test CA certificate upload elements are present."""
gateways_page.navigate_to_gateways_tab()
expect(gateways_page.ca_certificate_upload_input).to_be_attached()
expect(gateways_page.ca_certificate_drop_zone).to_be_visible()
def test_pagination_controls_present(self, gateways_page: GatewaysPage):
"""Test that pagination controls are present."""
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
# Pagination controls should be visible
expect(gateways_page.pagination_controls).to_be_visible()
@pytest.mark.ui
@pytest.mark.gateways
@pytest.mark.smoke
class TestGatewayCreation:
"""Test cases for creating gateways."""
@staticmethod
def _submit_and_wait(gateways_page: GatewaysPage, gateway_name: str):
"""Submit gateway form, wait for POST response, and skip on init failure.
The server connects to the external MCP URL during POST. If initialization
fails, the gateway is NOT saved to the database (returns 502). Since this
depends on external service availability, we skip rather than fail.
After a successful POST, the JS handler redirects to /admin#gateways via
window.location.href. We wait for that navigation to settle.
Returns:
The response object from the POST.
"""
with gateways_page.page.expect_response(
lambda r: "/admin/gateways" in r.url and r.request.method == "POST",
timeout=120000,
) as response_info:
gateways_page.click_locator(gateways_page.add_gateway_btn)
response = response_info.value
if response.status >= 400:
pytest.skip(f"Gateway creation failed for '{gateway_name}' (HTTP {response.status} — external service or server error)")
# The JS success handler does window.location.href = redirectUrl which
# triggers a full page navigation. Wait for it to complete so the
# subsequent reload doesn't race with the redirect.
gateways_page.page.wait_for_load_state("domcontentloaded")
return response
@staticmethod
def _verify_gateway_in_table(gateways_page: GatewaysPage, gateway_name: str, max_retries: int = 3):
"""Reload, search, and assert gateway exists in the table.
After gateway creation, the JS handler redirects to /admin#gateways.
We reload to ensure the DB commit (which runs after the POST response
is sent) has completed, then navigate to the gateways tab and search.
"""
for attempt in range(max_retries):
# Wait progressively longer on retries for DB commit to complete
gateways_page.page.wait_for_timeout(1000 * (attempt + 1))
gateways_page.page.reload(wait_until="domcontentloaded")
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
gateways_page.page.wait_for_selector('#gateways-table-body tr[id*="gateway-row"]', state="attached", timeout=20000)
gateways_page.search_gateways(gateway_name)
if gateways_page.gateway_exists(gateway_name):
return
if attempt < max_retries - 1:
logger.warning("Gateway '%s' not found on attempt %d, retrying", gateway_name, attempt + 1)
assert gateways_page.gateway_exists(gateway_name), f"Gateway '{gateway_name}' was not found in the table after creation"
def test_create_simple_gateway(self, gateways_page: GatewaysPage, test_gateway_data: dict):
"""Test creating a simple gateway without authentication."""
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
# Delete any existing gateway with the same URL first (system prevents duplicate URLs)
if gateways_page.delete_gateway_by_url(test_gateway_data["url"]):
logger.info("Deleted existing gateway with URL '%s' before test", test_gateway_data["url"])
# Fill form fields (create_gateway fills AND clicks submit, so fill manually)
gateways_page.fill_gateway_form(
name=test_gateway_data["name"],
url=test_gateway_data["url"],
description=test_gateway_data.get("description", ""),
tags=test_gateway_data.get("tags", ""),
transport=test_gateway_data.get("transport", "SSE"),
)
# Submit and wait for POST response (skips on 502 / external service failure)
self._submit_and_wait(gateways_page, test_gateway_data["name"])
# Verify gateway was created
self._verify_gateway_in_table(gateways_page, test_gateway_data["name"])
logger.info("Gateway '%s' created successfully", test_gateway_data["name"])
# Cleanup: Delete the created gateway by URL
if gateways_page.delete_gateway_by_url(test_gateway_data["url"]):
logger.info("Cleanup: Deleted gateway with URL '%s'", test_gateway_data["url"])
def test_create_gateway_with_basic_auth(self, gateways_page: GatewaysPage, test_gateway_with_basic_auth_data: dict):
"""Test creating a gateway with basic authentication."""
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
# Delete any existing gateway with the same URL first (system prevents duplicate URLs)
if gateways_page.delete_gateway_by_url(test_gateway_with_basic_auth_data["url"]):
logger.info("Deleted existing gateway with URL '%s' before test", test_gateway_with_basic_auth_data["url"])
# Fill basic gateway info
gateways_page.fill_gateway_form(
name=test_gateway_with_basic_auth_data["name"],
url=test_gateway_with_basic_auth_data["url"],
description=test_gateway_with_basic_auth_data["description"],
tags=test_gateway_with_basic_auth_data["tags"],
)
# Configure basic auth
gateways_page.configure_basic_auth(username=test_gateway_with_basic_auth_data["auth_username"], password=test_gateway_with_basic_auth_data["auth_password"])
# Submit and wait for POST response (skips on 502 / external service failure)
self._submit_and_wait(gateways_page, test_gateway_with_basic_auth_data["name"])
# Verify gateway was created
self._verify_gateway_in_table(gateways_page, test_gateway_with_basic_auth_data["name"])
logger.info("Gateway '%s' created successfully", test_gateway_with_basic_auth_data["name"])
# Cleanup: Delete the created gateway by URL
if gateways_page.delete_gateway_by_url(test_gateway_with_basic_auth_data["url"]):
logger.info("Cleanup: Deleted gateway with URL '%s'", test_gateway_with_basic_auth_data["url"])
def test_create_gateway_with_bearer_auth(self, gateways_page: GatewaysPage, test_gateway_with_bearer_auth_data: dict):
"""Test creating a gateway with bearer token authentication."""
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
# Delete any existing gateway with the same URL first (system prevents duplicate URLs)
if gateways_page.delete_gateway_by_url(test_gateway_with_bearer_auth_data["url"]):
logger.info("Deleted existing gateway with URL '%s' before test", test_gateway_with_bearer_auth_data["url"])
# Fill basic gateway info
gateways_page.fill_gateway_form(
name=test_gateway_with_bearer_auth_data["name"],
url=test_gateway_with_bearer_auth_data["url"],
description=test_gateway_with_bearer_auth_data["description"],
tags=test_gateway_with_bearer_auth_data["tags"],
)
# Set visibility to private
gateways_page.visibility_private_radio.click()
# Configure bearer auth
gateways_page.configure_bearer_auth(token=test_gateway_with_bearer_auth_data["auth_token"])
# Submit and wait for POST response (skips on 502 / external service failure)
self._submit_and_wait(gateways_page, test_gateway_with_bearer_auth_data["name"])
# Verify gateway was created
self._verify_gateway_in_table(gateways_page, test_gateway_with_bearer_auth_data["name"])
logger.info("Gateway '%s' created successfully", test_gateway_with_bearer_auth_data["name"])
# Cleanup: Delete the created gateway by URL
if gateways_page.delete_gateway_by_url(test_gateway_with_bearer_auth_data["url"]):
logger.info("Cleanup: Deleted gateway with URL '%s'", test_gateway_with_bearer_auth_data["url"])
def test_create_gateway_with_streamable_http(self, gateways_page: GatewaysPage, test_gateway_data: dict):
"""Test creating a gateway with STREAMABLEHTTP transport."""
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
# Modify test data for STREAMABLEHTTP
gateway_data = test_gateway_data.copy()
gateway_data["name"] = f"{test_gateway_data['name']}-http"
gateway_data["description"] = "Test gateway with STREAMABLEHTTP transport"
gateway_data["tags"] = "test,http,streamable"
gateway_data["transport"] = "STREAMABLEHTTP"
gateway_name = gateway_data["name"]
gateway_url = gateway_data["url"]
# Delete any existing gateway with the same URL first (system prevents duplicate URLs)
if gateways_page.delete_gateway_by_url(gateway_url):
logger.info("Deleted existing gateway with URL '%s' before test", gateway_url)
# Fill form fields
gateways_page.fill_gateway_form(
name=gateway_name,
url=gateway_url,
description=gateway_data["description"],
tags=gateway_data["tags"],
transport="STREAMABLEHTTP",
)
# Submit and wait for POST response (skips on 502 / external service failure)
self._submit_and_wait(gateways_page, gateway_name)
# Verify gateway was created
self._verify_gateway_in_table(gateways_page, gateway_name)
logger.info("Gateway '%s' created successfully", gateway_name)
# Cleanup: Delete the created gateway by URL
if gateways_page.delete_gateway_by_url(gateway_url):
logger.info("Cleanup: Deleted gateway with URL '%s'", gateway_url)
def test_form_validation_empty_name(self, gateways_page: GatewaysPage):
"""Test form validation for empty gateway name."""
gateways_page.navigate_to_gateways_tab()
# Try to submit with empty name
gateways_page.gateway_url_input.fill("https://example.com/sse")
gateways_page.add_gateway_btn.click()
# Form should not submit (name is required)
# The browser's built-in validation should prevent submission
expect(gateways_page.gateway_name_input).to_have_attribute("required", "")
def test_form_validation_empty_url(self, gateways_page: GatewaysPage):
"""Test form validation for empty gateway URL."""
gateways_page.navigate_to_gateways_tab()
# Try to submit with empty URL
gateways_page.gateway_name_input.fill("Test Gateway")
gateways_page.add_gateway_btn.click()
# Form should not submit (URL is required)
expect(gateways_page.gateway_url_input).to_have_attribute("required", "")
@pytest.mark.ui
@pytest.mark.gateways
class TestGatewayActions:
"""Test cases for gateway row actions."""
def test_gateway_row_actions_visible(self, gateways_page: GatewaysPage):
"""Test that all gateway row action buttons are visible."""
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
# Skip if no gateways exist
if gateways_page.get_gateway_count() == 0:
pytest.skip("No gateways available for testing")
# Get first gateway row
first_row = gateways_page.get_gateway_row(0)
# Verify all action buttons exist
expect(first_row.locator('button:has-text("Test")')).to_be_visible()
expect(first_row.locator('button:has-text("View")')).to_be_visible()
expect(first_row.locator('button:has-text("Edit")')).to_be_visible()
# Either Activate or Deactivate should be visible
activate_btn = first_row.locator('button:has-text("Activate")')
deactivate_btn = first_row.locator('button:has-text("Deactivate")')
assert activate_btn.is_visible() or deactivate_btn.is_visible()
# Delete button should be visible
expect(first_row.locator('button:has-text("Delete")')).to_be_visible()
def test_test_button_click(self, gateways_page: GatewaysPage):
"""Test clicking the Test button for a gateway."""
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
# Skip if no gateways exist
if gateways_page.get_gateway_count() == 0:
pytest.skip("No gateways available for testing")
# Click Test button on first gateway
gateways_page.click_test_button(0)
gateways_page.page.wait_for_timeout(1000)
# Test button should trigger a modal or some UI feedback
# The exact behavior depends on the implementation
# For now, verify the button was clickable
# Button click verified (no exception raised)
def test_view_button_click(self, gateways_page: GatewaysPage):
"""Test clicking the View button for a gateway."""
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
# Skip if no gateways exist
if gateways_page.get_gateway_count() == 0:
pytest.skip("No gateways available for testing")
# Click View button on first gateway
gateways_page.click_view_button(0)
gateways_page.page.wait_for_timeout(1000)
# View button should open a modal or navigate to details page
# Verify some UI change occurred
# Button click verified (no exception raised)
def test_edit_button_click(self, gateways_page: GatewaysPage):
"""Test clicking the Edit button for a gateway."""
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
# Skip if no gateways exist
if gateways_page.get_gateway_count() == 0:
pytest.skip("No gateways available for testing")
# Click Edit button on first gateway
gateways_page.click_edit_button(0)
gateways_page.page.wait_for_timeout(1000)
# Edit button should open an edit modal or form
# Button click verified (no exception raised)
def test_deactivate_button_functionality(self, gateways_page: GatewaysPage):
"""Test deactivating a gateway."""
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
# Skip if no gateways exist
if gateways_page.get_gateway_count() == 0:
pytest.skip("No gateways available for testing")
# Find a gateway with Deactivate button
first_row = gateways_page.get_gateway_row(0)
deactivate_btn = first_row.locator('button:has-text("Deactivate")')
if not deactivate_btn.is_visible():
pytest.skip("No active gateways available to deactivate")
# Get gateway name before deactivation
gateway_name = first_row.locator("td").nth(2).text_content().strip()
# Click Deactivate button
gateways_page.click_deactivate_button(0)
gateways_page.page.wait_for_timeout(2000)
# Reload to see updated status
gateways_page.page.reload()
gateways_page.wait_for_gateways_table_loaded()
gateways_page.page.wait_for_timeout(1000)
# Search for the gateway
gateways_page.search_gateways(gateway_name)
gateways_page.page.wait_for_timeout(500)
# Verify Activate button is now visible (gateway was deactivated)
if gateways_page.gateway_exists(gateway_name):
gateway_row = gateways_page.get_gateway_row_by_name(gateway_name)
activate_btn = gateway_row.locator('button:has-text("Activate")')
expect(activate_btn).to_be_visible()
logger.info("Gateway '%s' deactivated successfully", gateway_name)
# Reactivate for cleanup
gateways_page.click_activate_button(0)
gateways_page.page.wait_for_timeout(1000)
def test_activate_button_functionality(self, gateways_page: GatewaysPage):
"""Test activating an inactive gateway."""
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
# Enable showing inactive gateways
gateways_page.toggle_show_inactive(True)
gateways_page.page.wait_for_timeout(1000)
# Skip if no gateways exist
if gateways_page.get_gateway_count() == 0:
pytest.skip("No gateways available for testing")
# Find a gateway with Activate button
first_row = gateways_page.get_gateway_row(0)
activate_btn = first_row.locator('button:has-text("Activate")')
if not activate_btn.is_visible():
pytest.skip("No inactive gateways available to activate")
# Get gateway name before activation
gateway_name = first_row.locator("td").nth(2).text_content().strip()
# Click Activate button
gateways_page.click_activate_button(0)
gateways_page.page.wait_for_timeout(2000)
# Reload to see updated status
gateways_page.page.reload()
gateways_page.wait_for_gateways_table_loaded()
gateways_page.page.wait_for_timeout(1000)
# Search for the gateway
gateways_page.search_gateways(gateway_name)
gateways_page.page.wait_for_timeout(500)
# Verify Deactivate button is now visible (gateway was activated)
if gateways_page.gateway_exists(gateway_name):
gateway_row = gateways_page.get_gateway_row_by_name(gateway_name)
deactivate_btn = gateway_row.locator('button:has-text("Deactivate")')
expect(deactivate_btn).to_be_visible()
logger.info("Gateway '%s' activated successfully", gateway_name)
def test_delete_button_with_confirmation(self, gateways_page: GatewaysPage, test_gateway_data: dict):
"""Test deleting a gateway with confirmation."""
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
# Create a test gateway to delete
gateway_data = test_gateway_data.copy()
gateway_data["name"] = f"{test_gateway_data['name']}-delete-test"
# Delete any existing gateway with the same name first
if gateways_page.delete_gateway_by_name(gateway_data["name"]):
logger.info("Deleted existing gateway '%s' before test", gateway_data["name"])
# Fill and submit form, wait for POST (skips on 502)
gateways_page.fill_gateway_form(
name=gateway_data["name"],
url=gateway_data["url"],
description=gateway_data.get("description", ""),
tags=gateway_data.get("tags", ""),
transport=gateway_data.get("transport", "SSE"),
)
TestGatewayCreation._submit_and_wait(gateways_page, gateway_data["name"])
# Verify gateway exists before deletion
TestGatewayCreation._verify_gateway_in_table(gateways_page, gateway_data["name"])
gateways_page.search_gateways(gateway_data["name"])
gateways_page.page.wait_for_timeout(500)
assert gateways_page.gateway_exists(gateway_data["name"]), f"Gateway '{gateway_data['name']}' should exist before deletion"
# Delete the gateway by name (search + delete first match, waits for form POST navigation)
deleted = gateways_page.delete_gateway_by_name(gateway_data["name"], confirm=True)
assert deleted, f"Failed to find and delete gateway '{gateway_data['name']}'"
# Check if the server returned an error (403 from RBAC, or error param in redirect URL).
# The delete endpoint uses require_permission("gateways.delete", allow_admin_bypass=False)
# which requires explicit RBAC role assignments beyond the is_admin JWT flag.
current_url = gateways_page.page.url
page_content = gateways_page.page.content()
if "403" in page_content or "Forbidden" in page_content or "Insufficient permissions" in page_content:
pytest.skip(f"Gateway delete blocked by RBAC permissions (allow_admin_bypass=False). URL: {current_url}")
# Verify gateway was deleted — reload to ensure DB commit is visible
gateways_page.page.reload(wait_until="domcontentloaded")
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
gateways_page.page.wait_for_selector('#gateways-table-body tr[id*="gateway-row"]', state="attached", timeout=20000)
gateways_page.search_gateways(gateway_data["name"])
gateways_page.page.wait_for_timeout(500)
assert not gateways_page.gateway_exists(gateway_data["name"]), f"Gateway '{gateway_data['name']}' should not exist after deletion"
logger.info("Gateway '%s' deleted successfully", gateway_data["name"])
def test_delete_button_without_confirmation(self, gateways_page: GatewaysPage, test_gateway_data: dict):
"""Test canceling gateway deletion."""
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
# Skip if no gateways exist
if gateways_page.get_gateway_count() == 0:
pytest.skip("No gateways available for testing")
initial_count = gateways_page.get_gateway_count()
# Try to delete but cancel the confirmation
gateways_page.delete_gateway(0, confirm=False)
gateways_page.page.wait_for_timeout(1000)
# Verify gateway count hasn't changed
final_count = gateways_page.get_gateway_count()
assert final_count == initial_count, "Gateway count should not change when deletion is canceled"
# Cancellation verified via count assertion above
def test_gateway_status_badge(self, gateways_page: GatewaysPage):
"""Test that gateway status badge is displayed."""
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
# Skip if no gateways exist
if gateways_page.get_gateway_count() == 0:
pytest.skip("No gateways available for testing")
# Get first gateway row
first_row = gateways_page.get_gateway_row(0)
# Status badge should be visible (either Active or Inactive)
status_cell = first_row.locator("td").nth(5) # Status column
expect(status_cell).to_be_visible()
def test_gateway_visibility_badge(self, gateways_page: GatewaysPage):
"""Test that gateway visibility badge is displayed."""
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
# Skip if no gateways exist
if gateways_page.get_gateway_count() == 0:
pytest.skip("No gateways available for testing")
# Get first gateway row
first_row = gateways_page.get_gateway_row(0)
# Visibility badge should be visible
visibility_cell = first_row.locator("td").nth(9) # Visibility column
expect(visibility_cell).to_be_visible()
def test_gateway_tags_display(self, gateways_page: GatewaysPage):
"""Test that gateway tags are displayed."""
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
# Skip if no gateways exist
if gateways_page.get_gateway_count() == 0:
pytest.skip("No gateways available for testing")
# Get first gateway row
first_row = gateways_page.get_gateway_row(0)
# Tags column should be visible
tags_cell = first_row.locator("td").nth(4) # Tags column
expect(tags_cell).to_be_visible()
@pytest.mark.ui
@pytest.mark.gateways
class TestGatewaySearch:
"""Test cases for gateway search and filtering."""
def test_search_by_name(self, gateways_page: GatewaysPage):
"""Test searching gateways by name."""
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
# Skip if no gateways exist
if gateways_page.get_gateway_count() == 0:
pytest.skip("No gateways available for testing")
# Get first gateway name
first_row = gateways_page.get_gateway_row(0)
gateway_name = first_row.locator("td").nth(2).text_content().strip()
# Search for it (search_gateways dispatches events and waits internally)
gateways_page.search_gateways(gateway_name)
# Should still be visible
assert gateways_page.gateway_exists(gateway_name)
def test_clear_search(self, gateways_page: GatewaysPage):
"""Test clearing gateway search."""
gateways_page.navigate_to_gateways_tab()
gateways_page.wait_for_gateways_table_loaded()
initial_count = gateways_page.get_gateway_count()
# Perform a search (search_gateways waits internally)
gateways_page.search_gateways("test-search-query")
# Clear search and wait for table to restore
gateways_page.clear_search()
gateways_page.page.wait_for_timeout(500)
# Count should return to initial
final_count = gateways_page.get_gateway_count()
assert final_count == initial_count
@pytest.mark.ui
@pytest.mark.gateways
class TestGatewayVisibility:
"""Test cases for gateway visibility settings."""
def test_public_visibility_default(self, gateways_page: GatewaysPage):
"""Test that public visibility is selected by default."""
gateways_page.navigate_to_gateways_tab()
expect(gateways_page.visibility_public_radio).to_be_checked()
expect(gateways_page.visibility_team_radio).not_to_be_checked()
expect(gateways_page.visibility_private_radio).not_to_be_checked()
def test_change_visibility_to_team(self, gateways_page: GatewaysPage):
"""Test changing visibility to team."""
gateways_page.navigate_to_gateways_tab()
gateways_page.visibility_team_radio.click()
expect(gateways_page.visibility_team_radio).to_be_checked()
expect(gateways_page.visibility_public_radio).not_to_be_checked()
expect(gateways_page.visibility_private_radio).not_to_be_checked()
def test_change_visibility_to_private(self, gateways_page: GatewaysPage):
"""Test changing visibility to private."""
gateways_page.navigate_to_gateways_tab()
gateways_page.visibility_private_radio.click()
expect(gateways_page.visibility_private_radio).to_be_checked()
expect(gateways_page.visibility_public_radio).not_to_be_checked()
expect(gateways_page.visibility_team_radio).not_to_be_checked()
