| #!/usr/bin/env python3 |
| |
| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| |
| import argparse |
| import dataclasses |
| import glob |
| import json |
| import logging |
| import os |
| import re |
| import socket |
| import subprocess |
| import sys |
| import time |
| from collections.abc import Callable |
| from typing import Any, Final |
| |
| import netifaces |
| import rich.logging |
| from playwright.sync_api import BrowserContext, Dialog, Page, Request, Response, expect, sync_playwright |
| |
| ATR_BASE_URL: Final[str] = os.environ.get("ATR_BASE_URL", "https://localhost.apache.org:8080") |
| OPENPGP_TEST_UID: Final[str] = "<apache-tooling@example.invalid>" |
| SSH_KEY_COMMENT: Final[str] = "atr-playwright-test@127.0.0.1" |
| SSH_KEY_PATH: Final[str] = "/root/.ssh/id_ed25519" |
| TEST_PROJECT: Final[str] = "test" |
| |
| |
| @dataclasses.dataclass |
| class Credentials: |
| username: str |
| password: str |
| |
| |
| # If we did this then we'd have to call e.g. test.page, which is verbose |
| # @dataclasses.dataclass |
| # class TestArguments: |
| # page: Page |
| # credentials: Credentials |
| |
| |
| def esc_id(text: str) -> str: |
| return re.escape(text) |
| |
| |
| def get_credentials() -> Credentials | None: |
| return Credentials(username="test", password="test") |
| |
| |
| # def get_credentials_custom() -> Credentials | None: |
| # try: |
| # username = input("Enter ASF Username: ") |
| # password = getpass.getpass("Enter ASF Password: ") |
| # except (EOFError, KeyboardInterrupt): |
| # print() |
| # logging.error("EOFError: No credentials provided") |
| # return None |
| # |
| # if (not username) or (not password): |
| # logging.error("Username and password cannot be empty") |
| # return None |
| # |
| # return Credentials(username=username, password=password) |
| |
| |
| def get_default_gateway_ip() -> str | None: |
| gateways = netifaces.gateways() |
| default_gateway = gateways.get("default", {}) |
| if not isinstance(default_gateway, dict): |
| logging.error("Could not determine gateway IP: default gateway is not a dictionary") |
| return None |
| |
| match default_gateway.get(socket.AF_INET): |
| case (str(ip_address), _): |
| return ip_address |
| case _: |
| return None |
| |
| |
| def go_to_path(page: Page, path: str, wait: bool = True) -> None: |
| page.goto(f"{ATR_BASE_URL}{path}") |
| if wait: |
| wait_for_path(page, path) |
| |
| |
| def lifecycle_01_add_draft(page: Page, credentials: Credentials, version_key: str) -> None: |
| logging.info("Following link to start a new release") |
| go_to_path(page, f"/start/{TEST_PROJECT}") |
| |
| logging.info("Waiting for the start new release page") |
| version_key_locator = page.locator("input#version_key") |
| if not version_key_locator.is_visible(timeout=1000): |
| logging.error(f"Version name input not found. Page content:\n{page.content()}") |
| expect(version_key_locator).to_be_visible() |
| logging.info("Start new release page loaded") |
| |
| logging.info(f"Filling version '{version_key}'") |
| version_key_locator.fill(version_key) |
| |
| logging.info("Submitting the start new release form") |
| submit_button_locator = page.get_by_role("button", name="Start new release") |
| expect(submit_button_locator).to_be_enabled() |
| submit_button_locator.click() |
| |
| logging.info(f"Waiting for navigation to /compose/{TEST_PROJECT}/{version_key} after adding draft") |
| wait_for_path(page, f"/compose/{TEST_PROJECT}/{version_key}") |
| logging.info("Add draft actions completed successfully") |
| |
| |
| def lifecycle_02_check_draft_added(page: Page, credentials: Credentials, version_key: str) -> None: |
| logging.info(f"Checking for draft '{TEST_PROJECT} {version_key}'") |
| go_to_path(page, f"/compose/{TEST_PROJECT}/{version_key}") |
| h1_strong_locator = page.locator("h1 strong:has-text('Test')") |
| expect(h1_strong_locator).to_be_visible() |
| h1_em_locator = page.locator(f"h1 em:has-text('{esc_id(version_key)}')") |
| expect(h1_em_locator).to_be_visible() |
| logging.info(f"Draft '{TEST_PROJECT} {version_key}' found successfully") |
| |
| |
| def lifecycle_03_add_file(page: Page, credentials: Credentials, version_key: str) -> None: |
| logging.info(f"Navigating to the upload file page for {TEST_PROJECT} {version_key}") |
| go_to_path(page, f"/upload/{TEST_PROJECT}/{version_key}") |
| logging.info("Upload file page loaded") |
| |
| logging.info("Locating the file input") |
| file_input_locator = page.locator('input[name="file_data"]') |
| expect(file_input_locator).to_be_visible() |
| |
| logging.info("Setting the input file to /run/tests/example.txt") |
| file_input_locator.set_input_files("/run/tests/example.txt") |
| |
| logging.info("Locating and activating the add files button") |
| submit_button_locator = page.get_by_role("button", name="Add files") |
| expect(submit_button_locator).to_be_enabled() |
| submit_button_locator.click() |
| |
| # We used to check here that the #upload-progress-container was visible |
| # But this caused occasional timing issues, so we removed that check |
| logging.info("Waiting for upload to complete and redirect to compose page") |
| page.wait_for_url(f"**/compose/{TEST_PROJECT}/{version_key}*", timeout=30000) |
| wait_for_path(page, f"/compose/{TEST_PROJECT}/{version_key}") |
| logging.info("Add file actions completed successfully") |
| |
| logging.info(f"Navigating back to /compose/{TEST_PROJECT}/{version_key}") |
| go_to_path(page, f"/compose/{TEST_PROJECT}/{version_key}") |
| |
| logging.info("Extracting latest revision from compose page") |
| revision_link_locator = page.locator(f'a[href^="/revisions/{TEST_PROJECT}/{version_key}#"]') |
| expect(revision_link_locator).to_be_visible() |
| revision_href = revision_link_locator.get_attribute("href") |
| if not revision_href: |
| raise RuntimeError("Could not find revision link href") |
| revision = revision_href.split("#", 1)[-1] |
| logging.info(f"Found revision: {revision}") |
| |
| logging.info("Polling for task completion after file upload") |
| poll_for_tasks_completion(page, TEST_PROJECT, version_key, revision) |
| |
| logging.info(f"Navigation back to /compose/{TEST_PROJECT}/{version_key} completed successfully") |
| |
| |
| def lifecycle_04_start_vote(page: Page, credentials: Credentials, version_key: str) -> None: |
| logging.info(f"Navigating to the compose/{TEST_PROJECT} page for {TEST_PROJECT} {version_key}") |
| go_to_path(page, f"/compose/{TEST_PROJECT}/{version_key}") |
| logging.info(f"Compose/{TEST_PROJECT} page loaded successfully") |
| |
| logging.info(f"Locating start vote link for {TEST_PROJECT} {version_key}") |
| start_vote_link_locator = page.locator('a[title="Start a vote on this draft"]') |
| expect(start_vote_link_locator).to_be_visible() |
| |
| logging.info("Follow the start vote link") |
| start_vote_link_locator.click() |
| |
| logging.info("Waiting for page load after following the start vote link") |
| page.wait_for_load_state() |
| logging.info("Page loaded after following the start vote link") |
| logging.info(f"Current URL: {page.url}") |
| |
| logging.info("Locating and activating the button to prepare the vote email") |
| submit_button_locator = page.get_by_role("button", name="Send vote email") |
| expect(submit_button_locator).to_be_enabled() |
| submit_button_locator.click() |
| |
| logging.info(f"Waiting for navigation to /vote/{TEST_PROJECT}/{version_key} after submitting vote email") |
| wait_for_path(page, f"/vote/{TEST_PROJECT}/{version_key}") |
| |
| logging.info("Vote initiation actions completed successfully") |
| |
| |
| def lifecycle_05_resolve_vote(page: Page, credentials: Credentials, version_key: str) -> None: |
| logging.info(f"Navigating to the vote page for {TEST_PROJECT} {version_key}") |
| go_to_path(page, f"/vote/{TEST_PROJECT}/{version_key}") |
| logging.info("Vote page loaded successfully") |
| |
| # Wait until the vote initiation background task has completed |
| # When it finishes the page shows a "view thread" link |
| # We poll for that link before moving on |
| # Otherwise the subsequent Resolve step cannot find the completed VOTE_INITIATE task |
| # Note that this actually doesn't matter in dev testing |
| thread_link_locator = page.locator('a:has-text("view thread")') |
| link_found = False |
| for _ in range(30): |
| if thread_link_locator.is_visible(timeout=500): |
| link_found = True |
| logging.info("Vote thread link detected, task completed") |
| break |
| time.sleep(0.5) |
| page.reload() |
| if not link_found: |
| logging.warning("Vote thread link not detected after 15s, proceeding anyway") |
| |
| logging.info("Locating the 'Resolve vote' button") |
| tabulate_form_locator = page.locator(f'form[action="/resolve/{TEST_PROJECT}/{version_key}"]') |
| expect(tabulate_form_locator).to_be_visible() |
| |
| tabulate_button_locator = tabulate_form_locator.get_by_role("button", name="Resolve vote") |
| expect(tabulate_button_locator).to_be_enabled() |
| logging.info("Clicking 'Tabulate votes' button") |
| tabulate_button_locator.click() |
| |
| logging.info("Waiting for navigation to tabulated votes page") |
| wait_for_path(page, f"/resolve/{TEST_PROJECT}/{version_key}") |
| |
| logging.info("Locating the resolve vote form on the tabulated votes page") |
| resolve_form_locator = page.locator(f'form[action="/resolve/{TEST_PROJECT}/{version_key}"]') |
| expect(resolve_form_locator).to_be_visible() |
| |
| logging.info("Selecting 'Passed' radio button in resolve form") |
| passed_radio_locator = resolve_form_locator.locator('input[name="vote_result"][value="Passed"]') |
| expect(passed_radio_locator).to_be_enabled() |
| passed_radio_locator.check() |
| |
| logging.info("Submitting resolve vote form") |
| resolve_submit_locator = page.get_by_role("button", name="Resolve vote") |
| expect(resolve_submit_locator).to_be_enabled() |
| resolve_submit_locator.click() |
| |
| logging.info(f"Waiting for navigation to /finish/{TEST_PROJECT}/{version_key} after resolving the vote") |
| wait_for_path(page, f"/finish/{TEST_PROJECT}/{version_key}") |
| logging.info("Vote resolution actions completed successfully") |
| |
| |
| def lifecycle_06_announce_preview(page: Page, credentials: Credentials, version_key: str) -> None: |
| go_to_path(page, f"/finish/{TEST_PROJECT}/{version_key}") |
| logging.info("Finish page loaded successfully") |
| |
| logging.info(f"Locating the announce link for {TEST_PROJECT} {version_key}") |
| announce_link_locator = page.locator(f'a[href="/announce/{TEST_PROJECT}/{esc_id(version_key)}"]') |
| expect(announce_link_locator).to_be_visible() |
| announce_link_locator.click() |
| |
| logging.info(f"Waiting for navigation to /announce/{TEST_PROJECT}/{version_key} after announcing preview") |
| wait_for_path(page, f"/announce/{TEST_PROJECT}/{version_key}") |
| |
| logging.info(f"Locating the announcement form for {TEST_PROJECT} {version_key}") |
| form_locator = page.locator(f'form[action="/announce/{TEST_PROJECT}/{esc_id(version_key)}"]') |
| expect(form_locator).to_be_visible() |
| |
| logging.info("Locating the confirmation input within the form") |
| confirm_input_locator = form_locator.locator('input[name="confirm_announce"]') |
| expect(confirm_input_locator).to_be_visible() |
| |
| logging.info("Verifying submit button is initially disabled") |
| submit_button_locator = form_locator.get_by_role("button", name="Send announcement email") |
| expect(submit_button_locator).to_be_disabled() |
| |
| logging.info("Typing CONFIRM in the confirmation input") |
| confirm_input_locator.fill("CONFIRM") |
| |
| logging.info("Verifying submit button is now enabled") |
| expect(submit_button_locator).to_be_enabled() |
| |
| logging.info("Clicking the announce button") |
| submit_button_locator.click() |
| |
| logging.info("Waiting for navigation to /releases after submitting announcement") |
| wait_for_path(page, f"/releases/finished/{TEST_PROJECT}") |
| logging.info("Preview announcement actions completed successfully") |
| |
| |
| def lifecycle_07_release_exists(page: Page, credentials: Credentials, version_key: str) -> None: |
| logging.info(f"Checking for release {TEST_PROJECT} {version_key} on /releases/finished/{TEST_PROJECT}") |
| go_to_path(page, f"/releases/finished/{TEST_PROJECT}") |
| logging.info("Releases finished page loaded successfully") |
| |
| release_card_locator = page.locator(f'div.card:has(strong.card-title:has-text("{version_key}"))') |
| expect(release_card_locator).to_be_visible() |
| logging.info(f"Found card for {TEST_PROJECT} {version_key} release") |
| logging.info(f"Release {TEST_PROJECT} {version_key} confirmed exists on /releases/finished/{TEST_PROJECT}") |
| |
| |
| def main() -> None: |
| # TODO: Only members of ASF Tooling can run these tests |
| parser = argparse.ArgumentParser(description="Run Playwright debugging test") |
| parser.add_argument( |
| "--log", |
| default="INFO", |
| choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], |
| help="Set the logging level, default is INFO", |
| ) |
| parser.add_argument( |
| "--skip-slow", |
| action="store_true", |
| help="Skip slow tests", |
| ) |
| parser.add_argument( |
| "--tidy", |
| action="store_true", |
| help="Run cleanup tasks after tests complete", |
| ) |
| args = parser.parse_args() |
| log_level = getattr(logging, args.log.upper(), logging.INFO) |
| |
| logging.basicConfig( |
| level=log_level, |
| format="%(message)s", |
| datefmt="[%X]", |
| handlers=[rich.logging.RichHandler(rich_tracebacks=True, show_path=False)], |
| force=True, |
| ) |
| |
| logging.debug(f"Log level set to {args.log.upper()}") |
| if "ATR_BASE_URL" not in os.environ: |
| # Add localhost.apache.org to /etc/hosts |
| default_gateway_ip = get_default_gateway_ip() |
| if default_gateway_ip is not None: |
| with open("/etc/hosts", "a") as f: |
| f.write(f"{default_gateway_ip} localhost.apache.org\n") |
| logging.info(f"Added localhost.apache.org to /etc/hosts with IP {default_gateway_ip}") |
| else: |
| logging.warning("Could not determine default gateway IP, skipping /etc/hosts modification") |
| |
| run_tests(args.skip_slow, args.tidy) |
| |
| |
| def poll_for_tasks_completion(page: Page, project_key: str, version_key: str, revision: str) -> None: |
| rev_path = f"{project_key}/{version_key}/{revision}" |
| polling_url = f"{ATR_BASE_URL}/admin/ongoing-tasks/{rev_path}" |
| logging.info(f"Polling URL: {polling_url}") |
| |
| max_wait_seconds = 18 |
| poll_interval_seconds = 0.01 |
| start_time = time.monotonic() |
| |
| for attempt in range(int(max_wait_seconds / poll_interval_seconds)): |
| if attempt > 0: |
| time.sleep(poll_interval_seconds) |
| |
| elapsed_time = time.monotonic() - start_time |
| if elapsed_time > max_wait_seconds: |
| raise TimeoutError(f"Tasks did not complete within {max_wait_seconds} seconds") |
| |
| response = page.request.get(polling_url) |
| if not response.ok: |
| raise RuntimeError(f"Polling request failed with status {response.status}") |
| try: |
| ongoing_count_str = response.text() |
| if not ongoing_count_str: |
| raise RuntimeError("Polling request returned empty body") |
| ongoing_count = int(ongoing_count_str) |
| if ongoing_count == 0: |
| elapsed_time = time.monotonic() - start_time |
| logging.info(f"All tasks completed in {elapsed_time} seconds") |
| return |
| except ValueError: |
| raise RuntimeError(f"Polling request returned non-integer body: {response.text()}") |
| except Exception: |
| logging.exception("Unexpected error during polling response processing") |
| raise |
| |
| raise TimeoutError(f"Tasks did not complete within {max_wait_seconds} seconds") |
| |
| |
| def ensure_success_results_are_visible(page: Page, result_type: str) -> None: |
| button_id = f"#btn-toggle-{result_type}-success" |
| show_success_btn = page.locator(button_id) |
| |
| if show_success_btn.is_visible(timeout=500): |
| raw_button_text = show_success_btn.text_content() or "" |
| if "Show Success" in " ".join(raw_button_text.split()): |
| show_success_btn.click() |
| expect(show_success_btn).to_contain_text("Hide Success", timeout=2000) |
| first_success_row = page.locator(f".atr-result-{result_type}.atr-result-status-success").first |
| if first_success_row.is_visible(timeout=500): |
| expect(first_success_row).not_to_have_class("atr-hide", timeout=1000) |
| |
| |
| def release_remove(page: Page, release_key: str) -> None: |
| logging.info(f"Checking whether the {release_key} release exists") |
| release_checkbox_locator = page.locator(f'input[name="releases_to_delete"][value="{release_key}"]') |
| |
| if release_checkbox_locator.is_visible(): |
| logging.info(f"Found the {release_key} release, proceeding with deletion") |
| logging.info(f"Selecting {release_key} for deletion") |
| release_checkbox_locator.check() |
| |
| logging.info(f"Filling deletion confirmation for {release_key}") |
| page.locator("#confirm_delete").fill("DELETE") |
| |
| logging.info(f"Submitting deletion form for {release_key}") |
| submit_button_locator = page.get_by_role("button", name="Delete selected releases permanently") |
| expect(submit_button_locator).to_be_enabled() |
| submit_button_locator.click() |
| |
| logging.info(f"Waiting for page load after deletion submission for {release_key}") |
| page.wait_for_load_state() |
| logging.info(f"Page loaded after deletion for {release_key}") |
| |
| logging.info(f"Checking for success flash message for {release_key}") |
| flash_message_locator = page.locator("div.flash-success") |
| expect(flash_message_locator).to_be_visible() |
| expect(flash_message_locator).to_contain_text("Successfully deleted 1 release") |
| logging.info(f"Deletion successful for {release_key}") |
| else: |
| logging.info(f"Could not find the {release_key} release, no deletion needed") |
| |
| |
| def run_tests(skip_slow: bool, tidy_after: bool) -> None: |
| if (credentials := get_credentials()) is None: |
| logging.error("Cannot run tests: no credentials provided") |
| sys.exit(1) |
| |
| with sync_playwright() as p: |
| browser = None |
| context = None |
| try: |
| browser = p.chromium.launch() |
| context = browser.new_context(ignore_https_errors=True) |
| run_tests_in_context(context, credentials, skip_slow, tidy_after) |
| |
| except Exception as e: |
| logging.error(f"Error during page interaction: {e}", exc_info=True) |
| sys.exit(1) |
| finally: |
| if context: |
| context.close() |
| if browser: |
| browser.close() |
| |
| |
| def run_tests_in_context(context: BrowserContext, credentials: Credentials, skip_slow: bool, tidy_after: bool) -> None: |
| ssh_keys_generate() |
| page = context.new_page() |
| test_all(page, credentials, skip_slow) |
| logging.info("Tests finished successfully") |
| |
| if tidy_after: |
| logging.info("Tidying up after the tests") |
| test_tidy_up(page) |
| logging.info("Tidying up after the tests finished") |
| |
| |
| def run_tests_skipping_slow( |
| tests: list[Callable[..., Any]], page: Page, credentials: Credentials, skip_slow: bool |
| ) -> None: |
| for test in tests: |
| if skip_slow and ("slow" in test.__annotations__): |
| logging.info(f"Skipping slow test: {test.__name__}") |
| continue |
| # if "credentials" in test.__code__.co_varnames: |
| test(page, credentials) |
| |
| |
| def show_default_gateway_ip() -> None: |
| match get_default_gateway_ip(): |
| case str(ip_address): |
| logging.info(f"Default gateway IP: {ip_address}") |
| case None: |
| logging.warning("Could not determine gateway IP") |
| |
| |
| def slow(func: Callable[..., Any]) -> Callable[..., Any]: |
| func.__annotations__["slow"] = True |
| return func |
| |
| |
| def ssh_keys_generate() -> None: |
| ssh_key_path = SSH_KEY_PATH |
| ssh_dir = os.path.dirname(ssh_key_path) |
| |
| try: |
| if os.path.exists(ssh_key_path): |
| os.remove(ssh_key_path) |
| logging.info(f"Removed existing SSH key at {ssh_key_path}") |
| if os.path.exists(f"{ssh_key_path}.pub"): |
| os.remove(f"{ssh_key_path}.pub") |
| logging.info(f"Removed existing SSH public key at {ssh_key_path}.pub") |
| |
| os.makedirs(ssh_dir, mode=0o700, exist_ok=True) |
| |
| logging.info(f"Generating new SSH key at {ssh_key_path} with comment {SSH_KEY_COMMENT}") |
| subprocess.run( |
| ["ssh-keygen", "-t", "ed25519", "-f", ssh_key_path, "-N", "", "-C", SSH_KEY_COMMENT], |
| check=True, |
| capture_output=True, |
| text=True, |
| ) |
| logging.info("SSH key generated successfully") |
| |
| except (OSError, subprocess.CalledProcessError) as e: |
| logging.error(f"Failed to generate SSH key: {e}", exc_info=True) |
| if isinstance(e, subprocess.CalledProcessError): |
| logging.error(f"ssh-keygen stderr: {e.stderr}") |
| raise RuntimeError("SSH key generation failed") from e |
| |
| |
| def test_all(page: Page, credentials: Credentials, skip_slow: bool) -> None: |
| start = time.perf_counter() |
| test_login(page, credentials) |
| test_tidy_up(page) |
| |
| # Declare all tests |
| # The order here is important |
| tests: dict[str, list[Callable[..., Any]]] = {} |
| tests["projects"] = [ |
| test_projects_01_update, |
| test_projects_02_check_directory, |
| ] |
| tests["lifecycle"] = [ |
| test_lifecycle_01_add_draft, |
| test_lifecycle_02_check_draft_added, |
| test_lifecycle_03_add_file, |
| test_lifecycle_04_start_vote, |
| test_lifecycle_05_resolve_vote, |
| test_lifecycle_06_announce_preview, |
| test_lifecycle_07_release_exists, |
| ] |
| tests["openpgp"] = [ |
| test_openpgp_01_upload, |
| ] |
| tests["ssh"] = [ |
| test_ssh_01_add_key, |
| test_ssh_02_rsync_upload, |
| ] |
| tests["checks"] = [ |
| test_checks_01_hashing_sha512, |
| test_checks_02_license_files, |
| test_checks_03_license_headers, |
| test_checks_04_paths, |
| test_checks_05_signature, |
| test_checks_06_targz, |
| test_checks_07_cache, |
| ] |
| |
| # Order between our tests must be preserved |
| # Insertion order is reliable since Python 3.6 |
| # Therefore iteration over tests matches the insertion order above |
| for key in tests: |
| run_tests_skipping_slow(tests[key], page, credentials, skip_slow) |
| |
| finish = time.perf_counter() |
| logging.info(f"Tests took {round(finish - start, 3)} seconds") |
| |
| |
| def test_checks_01_hashing_sha512(page: Page, credentials: Credentials) -> None: |
| project_key = TEST_PROJECT |
| version_key = "0.2" |
| filename_sha512 = f"apache-{project_key}-{version_key}.tar.gz.sha512" |
| compose_path = f"/compose/{project_key}/{version_key}" |
| report_file_path = f"/report/{project_key}/{version_key}/{filename_sha512}" |
| |
| logging.info(f"Starting hashing check test for {filename_sha512}") |
| |
| logging.info(f"Navigating to compose page {compose_path}") |
| go_to_path(page, compose_path) |
| |
| logging.info(f"Locating 'Show report' link for {filename_sha512}") |
| row_locator = page.locator(f"tr:has(:text('{filename_sha512}'))") |
| evaluate_link_title = f"Show report for {filename_sha512}" |
| evaluate_link_locator = row_locator.locator(f'a[title="{evaluate_link_title}"]') |
| expect(evaluate_link_locator).to_be_visible() |
| |
| logging.info(f"Clicking 'Show report' link for {filename_sha512}") |
| evaluate_link_locator.click() |
| |
| logging.info(f"Waiting for navigation to {report_file_path}") |
| wait_for_path(page, report_file_path) |
| logging.info(f"Successfully navigated to {report_file_path}") |
| |
| ensure_success_results_are_visible(page, "primary") |
| |
| logging.info("Verifying Hashing Check status") |
| check_row_locator = page.locator("tr.atr-result-primary:has(th:has-text('Hashing Check'))") |
| expect(check_row_locator).to_be_visible() |
| logging.info("Located Hashing Check row") |
| |
| success_badge_locator = check_row_locator.locator("td span.badge.bg-success:text-is('Success')") |
| expect(success_badge_locator).to_be_visible() |
| logging.info("Hashing Check status verified as Success") |
| |
| |
| def test_checks_02_license_files(page: Page, credentials: Credentials) -> None: |
| project_key = TEST_PROJECT |
| version_key = "0.2" |
| filename_targz = f"apache-{project_key}-{version_key}.tar.gz" |
| compose_path = f"/compose/{project_key}/{version_key}" |
| report_file_path = f"/report/{project_key}/{version_key}/{filename_targz}" |
| |
| logging.info(f"Starting License Files check test for {filename_targz}") |
| |
| logging.info(f"Navigating to compose page {compose_path}") |
| go_to_path(page, compose_path) |
| |
| logging.info(f"Locating 'Show report' link for {filename_targz}") |
| row_locator = page.locator(f"tr:has(:text('{filename_targz}'))") |
| evaluate_link_title = f"Show report for {filename_targz}" |
| evaluate_link_locator = row_locator.locator(f'a[title="{evaluate_link_title}"]') |
| expect(evaluate_link_locator).to_be_visible() |
| |
| logging.info(f"Clicking 'Show report' link for {filename_targz}") |
| evaluate_link_locator.click() |
| |
| logging.info(f"Waiting for navigation to {report_file_path}") |
| wait_for_path(page, report_file_path) |
| logging.info(f"Successfully navigated to {report_file_path}") |
| |
| ensure_success_results_are_visible(page, "primary") |
| |
| logging.info("Verifying License Files check status") |
| check_row_locator = page.locator("tr.atr-result-primary:has(th:has-text('License Files'))") |
| expect(check_row_locator).to_have_count(2) |
| logging.info("Located 2 License Files check rows") |
| |
| success_badge_locator = check_row_locator.locator("td span.badge.bg-success:text-is('Success')") |
| expect(success_badge_locator).to_have_count(2) |
| logging.info("License Files check status verified as Success for 2 rows") |
| |
| |
| def test_checks_03_license_headers(page: Page, credentials: Credentials) -> None: |
| project_key = TEST_PROJECT |
| version_key = "0.2" |
| filename_targz = f"apache-{project_key}-{version_key}.tar.gz" |
| report_file_path = f"/report/{project_key}/{version_key}/{filename_targz}" |
| |
| logging.info(f"Starting License Headers check test for {filename_targz}") |
| |
| # Don't repeat the link test, just go straight there |
| logging.info(f"Navigating to report page {report_file_path}") |
| go_to_path(page, report_file_path) |
| logging.info(f"Successfully navigated to {report_file_path}") |
| |
| ensure_success_results_are_visible(page, "primary") |
| |
| logging.info("Verifying License Headers check status") |
| check_row_locator = page.locator("tr.atr-result-primary:has(th:has-text('License Headers'))") |
| expect(check_row_locator).to_be_visible() |
| logging.info("Located License Headers check row") |
| |
| success_badge_locator = check_row_locator.locator("td span.badge.bg-success:text-is('Success')") |
| expect(success_badge_locator).to_be_visible() |
| logging.info("License Headers check status verified as Success") |
| |
| |
| def test_checks_04_paths(page: Page, credentials: Credentials) -> None: |
| project_key = TEST_PROJECT |
| version_key = "0.2" |
| filename_sha512 = f"apache-{project_key}-{version_key}.tar.gz.sha512" |
| report_file_path = f"/report/{project_key}/{version_key}/{filename_sha512}" |
| |
| logging.info(f"Starting Paths check test for {filename_sha512}") |
| |
| logging.info(f"Navigating to report page {report_file_path}") |
| go_to_path(page, report_file_path) |
| logging.info(f"Successfully navigated to {report_file_path}") |
| |
| ensure_success_results_are_visible(page, "primary") |
| |
| # TODO: It's a bit strange to have the status in the check name |
| # But we have to do this because we need separate Recorder objects |
| logging.info("Verifying Paths Check Success status") |
| check_row_locator = page.locator( |
| f"tr.atr-result-primary:has(th:has-text('Paths Check Success')):has(td:has-text('{filename_sha512}'))" |
| ) |
| expect(check_row_locator).to_be_visible() |
| logging.info("Located Paths Check Success row") |
| |
| success_badge_locator = check_row_locator.locator("td span.badge.bg-success:text-is('Success')") |
| expect(success_badge_locator).to_be_visible() |
| logging.info("Paths Check Success status verified as Success") |
| |
| |
| def test_checks_05_signature(page: Page, credentials: Credentials) -> None: |
| project_key = TEST_PROJECT |
| version_key = "0.2" |
| filename_asc = f"apache-{project_key}-{version_key}.tar.gz.asc" |
| report_file_path = f"/report/{project_key}/{version_key}/{filename_asc}" |
| |
| logging.info(f"Starting Signature check test for {filename_asc}") |
| |
| logging.info(f"Navigating to report page {report_file_path}") |
| go_to_path(page, report_file_path) |
| logging.info(f"Successfully navigated to {report_file_path}") |
| |
| ensure_success_results_are_visible(page, "primary") |
| |
| logging.info("Verifying Signature Check status") |
| check_row_locator = page.locator("tr.atr-result-primary:has(th:has-text('Signature Check'))") |
| expect(check_row_locator).to_be_visible() |
| logging.info("Located Signature Check row") |
| |
| success_badge_locator = check_row_locator.locator("td span.badge.bg-success:text-is('Success')") |
| expect(success_badge_locator).to_be_visible() |
| logging.info("Signature Check status verified as Success") |
| |
| |
| def test_checks_06_targz(page: Page, credentials: Credentials) -> None: |
| project_key = TEST_PROJECT |
| version_key = "0.2" |
| filename_targz = f"apache-{project_key}-{version_key}.tar.gz" |
| report_file_path = f"/report/{project_key}/{version_key}/{filename_targz}" |
| |
| logging.info(f"Starting Targz checks for {filename_targz}") |
| |
| logging.info(f"Navigating to report page {report_file_path}") |
| go_to_path(page, report_file_path) |
| logging.info(f"Successfully navigated to {report_file_path}") |
| |
| ensure_success_results_are_visible(page, "primary") |
| |
| logging.info("Verifying Targz Structure status") |
| structure_row_locator = page.locator("tr.atr-result-primary:has(th:has-text('Targz Structure'))") |
| expect(structure_row_locator).to_be_visible() |
| logging.info("Located Targz Structure row") |
| structure_success_badge = structure_row_locator.locator("td span.badge.bg-success:text-is('Success')") |
| expect(structure_success_badge).to_be_visible() |
| logging.info("Targz Structure status verified as Success") |
| |
| |
| def test_checks_07_cache(page: Page, credentials: Credentials) -> None: |
| project_key = TEST_PROJECT |
| version_key = "0.2" |
| filename_targz = f"apache-{project_key}-{version_key}.tar.gz" |
| report_file_path = f"/report/{project_key}/{version_key}/{filename_targz}" |
| |
| logging.info(f"Starting check cache checks for {filename_targz}") |
| logging.info("Uploading new file to create new revision") |
| logging.info(f"Navigating to the upload file page for {TEST_PROJECT} {version_key}") |
| go_to_path(page, f"/upload/{TEST_PROJECT}/{version_key}") |
| logging.info("Upload file page loaded") |
| |
| logging.info("Locating the file input") |
| file_input_locator = page.locator('input[name="file_data"]') |
| expect(file_input_locator).to_be_visible() |
| |
| logging.info("Setting the input file to /run/tests/example.txt") |
| file_input_locator.set_input_files("/run/tests/example.txt") |
| |
| logging.info("Locating and activating the add files button") |
| submit_button_locator = page.get_by_role("button", name="Add files") |
| expect(submit_button_locator).to_be_enabled() |
| submit_button_locator.click() |
| |
| logging.info("Waiting for upload to complete and redirect to compose page") |
| page.wait_for_url(f"**/compose/{TEST_PROJECT}/{version_key}*", timeout=30000) |
| wait_for_path(page, f"/compose/{TEST_PROJECT}/{version_key}") |
| logging.info("Add file actions completed successfully") |
| |
| logging.info(f"Navigating to report page {report_file_path}") |
| go_to_path(page, report_file_path) |
| logging.info(f"Successfully navigated to {report_file_path}") |
| |
| ensure_success_results_are_visible(page, "primary") |
| |
| logging.info("Verifying Targz Structure status exists") |
| structure_row_locator = page.locator("tr.atr-result-primary:has(th:has-text('Targz Structure'))") |
| expect(structure_row_locator).to_be_visible() |
| |
| logging.info("Verifying Targz Structure result is from previous revision") |
| check_link_locator = structure_row_locator.locator("th:has-text('Targz Structure') a") |
| check_link_url = (check_link_locator.get_attribute("href") or "").replace(ATR_BASE_URL, "") |
| check_link_locator.click() |
| logging.info(f"Waiting for raw check result to load: {check_link_url}") |
| wait_for_path(page, check_link_url) |
| check_result = json.loads(page.locator("pre").inner_text()) |
| logging.info("Verifying revision number") |
| assert check_result["revision_number"] == "00001" |
| |
| |
| def test_openpgp_01_upload(page: Page, credentials: Credentials) -> None: |
| for key_path in glob.glob("/run/tests/*.asc"): |
| key_fingerprint_lower = os.path.basename(key_path).split(".")[0].lower() |
| key_fingerprint_upper = key_fingerprint_lower.upper() |
| break |
| else: |
| raise RuntimeError("No test key found") |
| |
| logging.info("Starting OpenPGP key upload test") |
| go_to_path(page, "/keys") |
| |
| logging.info("Following link to add OpenPGP key") |
| add_key_link_locator = page.locator('a:has-text("Add your OpenPGP key")') |
| expect(add_key_link_locator).to_be_visible() |
| add_key_link_locator.click() |
| |
| logging.info("Waiting for Add OpenPGP key page") |
| wait_for_path(page, "/keys/add") |
| |
| try: |
| logging.info(f"Reading public key from {key_path}") |
| with open(key_path, encoding="utf-8") as f: |
| public_key_content = f.read().strip() |
| logging.info("Public key read successfully") |
| except OSError as e: |
| logging.error(f"Failed to read public key file {key_path}: {e}") |
| raise RuntimeError("Failed to read public key file") from e |
| |
| logging.info("Filling public key textarea") |
| public_key_textarea_locator = page.locator('textarea[name="public_key"]') |
| expect(public_key_textarea_locator).to_be_visible() |
| public_key_textarea_locator.fill(public_key_content) |
| |
| logging.info("Clicking Select all committees button") |
| select_all_button_locator = page.locator("#toggleCommitteesBtn") |
| expect(select_all_button_locator).to_be_visible() |
| select_all_button_locator.click() |
| |
| logging.info("Submitting the Add OpenPGP key form") |
| submit_button_locator = page.get_by_role("button", name="Add OpenPGP key") |
| expect(submit_button_locator).to_be_enabled() |
| submit_button_locator.click() |
| |
| logging.info("Waiting for navigation back to /keys page") |
| wait_for_path(page, "/keys") |
| |
| logging.info("Checking for success flash message on /keys page") |
| try: |
| flash_message_locator = page.locator("div.flash-success") |
| expect(flash_message_locator).to_be_visible() |
| expect(flash_message_locator).to_contain_text(f"OpenPGP key {key_fingerprint_upper} added successfully.") |
| logging.info("OpenPGP key upload successful message shown") |
| except AssertionError: |
| flash_message_locator = page.locator("div.flash-warning") |
| expect(flash_message_locator).to_be_visible() |
| expect(flash_message_locator).to_contain_text( |
| f"OpenPGP key {key_fingerprint_upper} was already in the database." |
| ) |
| logging.info("OpenPGP key already in database message shown") |
| |
| logging.info(f"Verifying OpenPGP key with fingerprint {key_fingerprint_upper} is visible") |
| key_row_locator = page.locator(f'tr.page-user-openpgp-key:has(a[href="/keys/details/{key_fingerprint_lower}"])') |
| expect(key_row_locator).to_be_visible() |
| logging.info("OpenPGP key fingerprint verified successfully on /keys page") |
| |
| |
| def test_lifecycle_01_add_draft(page: Page, credentials: Credentials) -> None: |
| lifecycle_01_add_draft(page, credentials, version_key="0.1+draft") |
| lifecycle_01_add_draft(page, credentials, version_key="0.1+candidate") |
| lifecycle_01_add_draft(page, credentials, version_key="0.1+preview") |
| lifecycle_01_add_draft(page, credentials, version_key="0.1+release") |
| |
| |
| def test_lifecycle_02_check_draft_added(page: Page, credentials: Credentials) -> None: |
| lifecycle_02_check_draft_added(page, credentials, version_key="0.1+draft") |
| lifecycle_02_check_draft_added(page, credentials, version_key="0.1+candidate") |
| lifecycle_02_check_draft_added(page, credentials, version_key="0.1+preview") |
| lifecycle_02_check_draft_added(page, credentials, version_key="0.1+release") |
| |
| |
| def test_lifecycle_03_add_file(page: Page, credentials: Credentials) -> None: |
| lifecycle_03_add_file(page, credentials, version_key="0.1+draft") |
| lifecycle_03_add_file(page, credentials, version_key="0.1+candidate") |
| lifecycle_03_add_file(page, credentials, version_key="0.1+preview") |
| lifecycle_03_add_file(page, credentials, version_key="0.1+release") |
| |
| |
| def test_lifecycle_04_start_vote(page: Page, credentials: Credentials) -> None: |
| lifecycle_04_start_vote(page, credentials, version_key="0.1+candidate") |
| lifecycle_04_start_vote(page, credentials, version_key="0.1+preview") |
| lifecycle_04_start_vote(page, credentials, version_key="0.1+release") |
| |
| |
| def test_lifecycle_05_resolve_vote(page: Page, credentials: Credentials) -> None: |
| lifecycle_05_resolve_vote(page, credentials, version_key="0.1+preview") |
| lifecycle_05_resolve_vote(page, credentials, version_key="0.1+release") |
| |
| |
| def test_lifecycle_06_announce_preview(page: Page, credentials: Credentials) -> None: |
| lifecycle_06_announce_preview(page, credentials, version_key="0.1+release") |
| |
| |
| def test_lifecycle_07_release_exists(page: Page, credentials: Credentials) -> None: |
| lifecycle_07_release_exists(page, credentials, version_key="0.1+release") |
| |
| |
| def test_login(page: Page, credentials: Credentials) -> None: |
| debugging = False |
| |
| def remove_debugging() -> None: |
| pass |
| |
| if debugging: |
| remove_debugging = test_logging_debug(page, credentials) |
| |
| if credentials.username == "test": |
| go_to_path(page, "/test/login", wait=False) |
| wait_for_path(page, "/") |
| logging.info("Test login completed successfully") |
| return |
| |
| go_to_path(page, "/") |
| logging.info(f"Initial page title: {page.title()}") |
| |
| logging.info("Following link to log in") |
| login_link_locator = page.get_by_role("link", name="Login") |
| expect(login_link_locator).to_be_visible() |
| login_link_locator.click() |
| |
| logging.info("Waiting for the login page") |
| username_field_locator = page.locator('input[name="username"]') |
| expect(username_field_locator).to_be_visible() |
| logging.info("Login page loaded") |
| |
| logging.info("Filling credentials") |
| username_field_locator.fill(credentials.username) |
| page.locator('input[name="password"]').fill(credentials.password) |
| |
| logging.info("Submitting the login form") |
| submit_button_locator = page.get_by_role("button", name="Authenticate") |
| expect(submit_button_locator).to_be_enabled() |
| submit_button_locator.click() |
| |
| logging.info("Waiting for the page to load") |
| page.wait_for_load_state() |
| logging.info("Page loaded after login") |
| logging.info(f"Initial URL after login: {page.url}") |
| |
| logging.info("Waiting for the redirect to /") |
| # We can't use wait_for_path here because it goes through /auth |
| page.wait_for_url("https://*/") |
| logging.info("Redirected to /") |
| logging.info(f"Page URL: {page.url}") |
| logging.info("Login actions completed successfully") |
| |
| if debugging: |
| remove_debugging() |
| |
| |
| def test_logging_debug(page: Page, credentials: Credentials) -> Callable[[], None]: |
| def log_request(request: Request) -> None: |
| logging.info(f">> REQUEST: {request.method} {request.url}") |
| for key, value in request.headers.items(): |
| logging.info(f" REQ HEADER: {key}: {value}") |
| |
| def log_response(response: Response) -> None: |
| logging.info(f"<< RESPONSE: {response.status} {response.url}") |
| headers = response.headers |
| for key, value in headers.items(): |
| logging.info(f" RESP HEADER: {key}: {value}") |
| if "location" in headers: |
| logging.info(f" >> REDIRECTING TO: {headers['location']}") |
| if response.status == 500: |
| logging.info(f" BODY: {response.text()}") |
| |
| page.on("request", log_request) |
| page.on("response", log_response) |
| |
| def remove_debugging() -> None: |
| page.remove_listener("request", log_request) |
| page.remove_listener("response", log_response) |
| |
| return remove_debugging |
| |
| |
| @slow |
| def test_projects_01_update(page: Page, credentials: Credentials) -> None: |
| logging.info("Navigating to the admin update projects page") |
| go_to_path(page, "/admin/projects/update") |
| logging.info("Admin update projects page loaded") |
| |
| logging.info("Locating and activating the button to update projects") |
| update_button_locator = page.get_by_role("button", name="Update projects") |
| expect(update_button_locator).to_be_enabled() |
| update_button_locator.click() |
| |
| logging.info("Waiting for project update completion message") |
| success_message_locator = page.locator("div.status-message.success") |
| expect(success_message_locator).to_contain_text( |
| re.compile( |
| r"Successfully added \d+ and updated \d+ committees and projects \(PMCs and PPMCs\) with membership data" |
| ) |
| ) |
| logging.info("Project update completed successfully") |
| |
| |
| def test_projects_02_check_directory(page: Page, credentials: Credentials) -> None: |
| logging.info("Navigating to the project directory page") |
| go_to_path(page, "/projects") |
| logging.info("Project directory page loaded") |
| |
| logging.info("Checking for the Apache Test project card") |
| h3_locator = page.get_by_text("Apache Test", exact=True) |
| test_card_locator = h3_locator.locator("xpath=ancestor::div[contains(@class, 'project-card')]") |
| expect(test_card_locator).to_be_visible() |
| logging.info("Apache Test project card found successfully") |
| |
| |
| def test_projects_03_add_project(page: Page, credentials: Credentials) -> None: |
| base_project_label = "test" |
| project_key = "Apache Test Example" |
| project_label = "test-example" |
| |
| logging.info("Navigating to the add derived project page") |
| go_to_path(page, f"/project/add/{base_project_label}") |
| logging.info("Add a new project page loaded") |
| |
| logging.info(f"Filling display name '{project_key}'") |
| page.locator('input[name="display_name"]').fill(project_key) |
| |
| logging.info(f"Filling label '{project_label}'") |
| page.locator('input[name="label"]').fill(project_label) |
| |
| logging.info("Submitting the add derived project form") |
| submit_button_locator = page.get_by_role("button", name="Add project") |
| expect(submit_button_locator).to_be_enabled() |
| submit_button_locator.click() |
| |
| logging.info(f"Waiting for navigation to project view page for {project_label}") |
| wait_for_path(page, f"/projects/{project_label}") |
| logging.info("Navigated to project view page successfully") |
| |
| logging.info(f"Checking for project title '{project_key}' on view page") |
| title_locator = page.locator(f'h1:has-text("{project_key}")') |
| expect(title_locator).to_be_visible() |
| logging.info("Project title confirmed on view page") |
| |
| |
| def test_ssh_01_add_key(page: Page, credentials: Credentials) -> None: |
| logging.info("Starting SSH key addition test") |
| go_to_path(page, "/keys") |
| logging.info("Public Keys page loaded") |
| |
| logging.info("Clicking Add your SSH key button") |
| # There can be two buttons with the same text if the user did not upload an SSH key yet |
| page.locator('a[href="/keys/ssh/add"]:has-text("Add your SSH key")').first.click() |
| wait_for_path(page, "/keys/ssh/add") |
| logging.info("Navigated to Add your SSH key page") |
| |
| public_key_path = f"{SSH_KEY_PATH}.pub" |
| try: |
| logging.info(f"Reading public key from {public_key_path}") |
| with open(public_key_path, encoding="utf-8") as f: |
| public_key_content = f.read().strip() |
| logging.info("Public key read successfully") |
| except OSError as e: |
| logging.error(f"Failed to read public key file {public_key_path}: {e}") |
| raise RuntimeError("Failed to read public key file") from e |
| |
| logging.info("Pasting public key into textarea") |
| page.locator('textarea[name="key"]').fill(public_key_content) |
| |
| logging.info("Submitting the Add SSH key form") |
| page.get_by_role("button", name="Add SSH key").click() |
| |
| logging.info("Waiting for navigation back to /keys page") |
| wait_for_path(page, "/keys") |
| logging.info("Navigated back to /keys page") |
| |
| try: |
| logging.info("Calculating expected SSH key fingerprint using ssh-keygen -lf") |
| result = subprocess.run( |
| ["ssh-keygen", "-lf", public_key_path], |
| check=True, |
| capture_output=True, |
| text=True, |
| ) |
| fingerprint_output = result.stdout.strip() |
| match = re.search(r"SHA256:([\w\+/=]+)", fingerprint_output) |
| if not match: |
| logging.error(f"Could not parse fingerprint from ssh-keygen output: {fingerprint_output}") |
| raise RuntimeError("Failed to parse SSH key fingerprint") |
| expected_fingerprint = f"SHA256:{match.group(1)}" |
| logging.info(f"Expected fingerprint: {expected_fingerprint}") |
| |
| except (subprocess.CalledProcessError, FileNotFoundError, RuntimeError) as e: |
| logging.error(f"Failed to get SSH key fingerprint: {e}") |
| if isinstance(e, subprocess.CalledProcessError): |
| logging.error(f"ssh-keygen stderr: {e.stderr}") |
| raise RuntimeError("Failed to get SSH key fingerprint") from e |
| |
| logging.info("Verifying that the added SSH key fingerprint is visible") |
| key_card_locator = page.locator(f'div.card:has(td:has-text("{expected_fingerprint}"))') |
| expect(key_card_locator).to_be_visible() |
| logging.info("SSH key fingerprint verified successfully on /keys page") |
| |
| |
| def test_ssh_02_rsync_upload(page: Page, credentials: Credentials) -> None: |
| project_key = TEST_PROJECT |
| version_key = "0.2" |
| source_dir_rel = f"apache-{project_key}-{version_key}" |
| source_dir_abs = f"/run/tests/{source_dir_rel}" |
| file1 = f"apache-{project_key}-{version_key}.tar.gz" |
| file2 = f"{file1}.sha512" |
| |
| logging.info(f"Starting rsync upload test for {project_key}-{version_key}") |
| |
| if "ATR_BASE_URL" in os.environ: |
| ssh_host = os.environ.get("ATR_BASE_URL", "").replace("https://", "").replace(":8080", "") |
| else: |
| gateway_ip = get_default_gateway_ip() |
| if not gateway_ip: |
| raise RuntimeError("Cannot proceed without gateway IP") |
| ssh_host = gateway_ip |
| |
| username = credentials.username |
| ssh_command = "ssh -p 2222 -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null" |
| source_path = f"{source_dir_abs}/" |
| destination = f"{username}@{ssh_host}:/{project_key}/{version_key}/" |
| |
| rsync_cmd = [ |
| "rsync", |
| "-av", |
| "-e", |
| ssh_command, |
| source_path, |
| destination, |
| ] |
| |
| logging.info(f"Executing rsync command: {' '.join(rsync_cmd)}") |
| try: |
| result = subprocess.run(rsync_cmd, check=True, capture_output=True, text=True) |
| logging.info(f"rsync completed successfully. stdout:\n{result.stdout}") |
| if result.stderr: |
| logging.warning(f"rsync stderr:\n{result.stderr}") |
| except subprocess.CalledProcessError as e: |
| logging.error(f"rsync command failed with exit code {e.returncode}") |
| logging.error(f"rsync stdout:\n{e.stdout}") |
| logging.error(f"rsync stderr:\n{e.stderr}") |
| raise RuntimeError("rsync upload failed") from e |
| except FileNotFoundError: |
| logging.error("rsync command not found. Is rsync installed in the container?") |
| raise RuntimeError("rsync command not found") |
| |
| compose_path = f"/compose/{project_key}/{version_key}" |
| logging.info(f"Waiting for quarantine validation and files to appear on {compose_path}") |
| max_wait_seconds = 30 |
| start_time = time.monotonic() |
| while True: |
| go_to_path(page, compose_path) |
| file1_locator = page.get_by_role("cell", name=file1, exact=True) |
| if file1_locator.count() > 0: |
| break |
| elapsed = time.monotonic() - start_time |
| if elapsed > max_wait_seconds: |
| raise TimeoutError(f"Files did not appear on {compose_path} within {max_wait_seconds} seconds") |
| time.sleep(1) |
| |
| logging.info(f"Found file: {file1}") |
| file2_locator = page.get_by_role("cell", name=file2, exact=True) |
| expect(file2_locator).to_be_visible() |
| logging.info(f"Found file: {file2}") |
| logging.info("rsync upload test completed successfully") |
| |
| logging.info(f"Extracting latest revision from {compose_path}") |
| revision_link_locator = page.locator(f'a[href^="/revisions/{project_key}/{version_key}#"]') |
| expect(revision_link_locator).to_be_visible() |
| revision_href = revision_link_locator.get_attribute("href") |
| if not revision_href: |
| raise RuntimeError("Could not find revision link href") |
| revision = revision_href.split("#", 1)[-1] |
| logging.info(f"Found revision: {revision}") |
| |
| logging.info(f"Polling for task completion for revision {revision}") |
| poll_for_tasks_completion(page, project_key, version_key, revision) |
| |
| |
| def test_tidy_up(page: Page) -> None: |
| test_tidy_up_releases(page) |
| test_tidy_up_ssh_keys(page) |
| test_tidy_up_openpgp_keys(page) |
| |
| |
| def test_tidy_up_openpgp_keys(page: Page) -> None: |
| logging.info("Starting OpenPGP key tidy up") |
| |
| # First, delete the test key if it exists with wrong apache_uid |
| # (it may exist from real usage due to on_conflict_do_nothing in the INSERT) |
| # TODO: Don't hardcode this |
| logging.info("Deleting test key from database via admin route") |
| |
| # Navigate to the delete route and submit the form |
| go_to_path(page, "/admin/delete-test-openpgp-keys") |
| delete_button = page.get_by_role("button", name="Delete all OpenPGP keys for test user") |
| if delete_button.is_visible(): |
| delete_button.click() |
| page.wait_for_load_state() |
| logging.info("Test key deletion form submitted") |
| else: |
| logging.info("Test key deletion button not found, key may not exist") |
| |
| go_to_path(page, "/keys") |
| logging.info("Navigated to /keys page for OpenPGP key cleanup") |
| |
| openpgp_key_section_locator = page.locator("h3:has-text('OpenPGP keys')") |
| table_locator = openpgp_key_section_locator.locator("xpath=following-sibling::div//table") |
| key_rows_locator = table_locator.locator("tbody tr.page-user-openpgp-key") |
| |
| key_rows = key_rows_locator.all() |
| logging.info(f"Found {len(key_rows)} OpenPGP key rows to check") |
| |
| fingerprints_to_delete = [] |
| |
| for row in key_rows: |
| link_locator = row.locator('a[href^="/keys/details/"]') |
| href = link_locator.get_attribute("href") |
| if not href: |
| logging.warning("Could not find href for key details link in a row, skipping") |
| continue |
| fingerprint = href.split("/")[-1] |
| |
| go_to_path(page, href, wait=False) |
| |
| pre_locator = page.locator("pre") |
| expect(pre_locator).to_be_visible() |
| key_content = pre_locator.inner_text() |
| |
| if OPENPGP_TEST_UID in key_content: |
| logging.info(f"Found test OpenPGP key with fingerprint {fingerprint} for deletion") |
| fingerprints_to_delete.append(fingerprint) |
| |
| go_to_path(page, "/keys") |
| |
| # For the complexity linter only |
| test_tidy_up_openpgp_keys_continued(page, fingerprints_to_delete) |
| |
| |
| def test_tidy_up_openpgp_keys_continued(page: Page, fingerprints_to_delete: list[str]) -> None: |
| if not fingerprints_to_delete: |
| logging.info("No test OpenPGP keys found to delete") |
| return |
| |
| # Delete identified keys |
| logging.info(f"Attempting to delete {len(fingerprints_to_delete)} test OpenPGP keys") |
| for fingerprint in fingerprints_to_delete: |
| logging.info(f"Locating delete form for fingerprint: {fingerprint}") |
| # Locate again by fingerprint for robustness |
| row_to_delete_locator = page.locator(f'tr:has(a[href="/keys/details/{fingerprint}"])') |
| delete_button_locator = row_to_delete_locator.get_by_role("button", name="Delete key") |
| |
| if delete_button_locator.is_visible(): |
| logging.info(f"Delete button found for {fingerprint}, proceeding with deletion") |
| |
| def handle_dialog(dialog: Dialog) -> None: |
| logging.info(f"Accepting dialog for OpenPGP key deletion: {dialog.message}") |
| dialog.accept() |
| |
| page.once("dialog", handle_dialog) |
| delete_button_locator.click() |
| |
| logging.info(f"Waiting for page reload after deleting OpenPGP key {fingerprint}") |
| page.wait_for_load_state() |
| wait_for_path(page, "/keys") |
| |
| flash_message_locator = page.locator("div.flash-success") |
| expect(flash_message_locator).to_contain_text("OpenPGP key deleted successfully") |
| logging.info(f"Deletion successful for OpenPGP key {fingerprint}") |
| |
| else: |
| logging.warning(f"Could not find delete button for OpenPGP fingerprint {fingerprint} after re-locating") |
| |
| logging.info("OpenPGP key tidy up finished") |
| |
| |
| def test_tidy_up_project(page: Page) -> None: |
| project_key = "Apache Test" |
| logging.info(f"Checking for project '{project_key}' at /projects") |
| go_to_path(page, "/projects") |
| logging.info("Project directory page loaded") |
| |
| h3_locator = page.get_by_text(project_key, exact=True) |
| example_card_locator = h3_locator.locator("xpath=ancestor::div[contains(@class, 'project-card')]") |
| |
| if example_card_locator.is_visible(): |
| logging.info(f"Found project card for '{project_key}'") |
| delete_button_locator = example_card_locator.get_by_role("button", name="Delete Project") |
| |
| if delete_button_locator.is_visible(): |
| logging.info(f"Delete button found for '{project_key}', proceeding with deletion") |
| |
| def handle_dialog(dialog: Dialog) -> None: |
| logging.info(f"Accepting dialog: {dialog.message}") |
| dialog.accept() |
| |
| page.once("dialog", handle_dialog) |
| delete_button_locator.click() |
| |
| logging.info("Waiting for navigation back to /projects after deletion") |
| wait_for_path(page, "/projects") |
| |
| logging.info(f"Verifying project card for '{project_key}' is no longer visible") |
| h3_locator_check = page.get_by_text(project_key, exact=True) |
| card_locator_check = h3_locator_check.locator("xpath=ancestor::div[contains(@class, 'project-card')]") |
| expect(card_locator_check).not_to_be_visible() |
| logging.info(f"Project '{project_key}' deleted successfully") |
| else: |
| logging.info(f"Delete button not visible for '{project_key}', no deletion performed") |
| else: |
| logging.info(f"Project card for '{project_key}' not found, no deletion needed") |
| |
| |
| def test_tidy_up_ssh_keys(page: Page) -> None: |
| logging.info("Starting SSH key tidy up") |
| go_to_path(page, "/keys") |
| logging.info("Navigated to /keys page for SSH key cleanup") |
| |
| ssh_key_section_locator = page.locator("h3:has-text('SSH keys')") |
| key_cards_container_locator = ssh_key_section_locator.locator( |
| "xpath=following-sibling::div[contains(@class, 'mb-5')]//div[contains(@class, 'd-grid')]" |
| ) |
| key_cards_locator = key_cards_container_locator.locator("> div.card") |
| |
| key_cards = key_cards_locator.all() |
| logging.info(f"Found {len(key_cards)} potential SSH key cards to check") |
| |
| fingerprints_to_delete = [] |
| |
| for card in key_cards: |
| details_element = card.locator("details").first |
| summary_element = details_element.locator("summary").first |
| |
| if not details_element.is_visible(timeout=500): |
| logging.warning("SSH key card: <details> element not found or not visible, skipping") |
| continue |
| if not summary_element.is_visible(timeout=500): |
| logging.warning("SSH key card: <summary> element not found or not visible, skipping") |
| continue |
| |
| is_already_open = details_element.evaluate("el => el.hasAttribute('open')") |
| |
| if not is_already_open: |
| logging.info("SSH key card: details is not open, clicking summary to open") |
| summary_element.click() |
| try: |
| expect(details_element).to_have_attribute("open", "", timeout=2000) |
| logging.info("SSH key card: details successfully opened") |
| except Exception as e: |
| logging.warning( |
| f"SSH key card: failed to confirm details opened after clicking summary: {e}, skipping card" |
| ) |
| continue |
| else: |
| logging.info("SSH key card: details already open") |
| |
| details_pre_locator = details_element.locator("pre").first |
| try: |
| expect(details_pre_locator).to_be_visible(timeout=1000) |
| except Exception as e: |
| logging.warning( |
| f"SSH key card: <pre> tag not visible even after attempting to open details: {e}, skipping card" |
| ) |
| continue |
| |
| key_content = details_pre_locator.inner_text() |
| if SSH_KEY_COMMENT in key_content: |
| fingerprint_td_locator = card.locator('td:has-text("SHA256:")') |
| if fingerprint_td_locator.is_visible(timeout=500): |
| fingerprint = fingerprint_td_locator.inner_text().strip() |
| if fingerprint: |
| logging.info(f"Found test SSH key with fingerprint {fingerprint} for deletion") |
| fingerprints_to_delete.append(fingerprint) |
| else: |
| logging.warning("Found test SSH key card but could not extract fingerprint text from td") |
| else: |
| logging.warning("Could not locate fingerprint td for a test key card") |
| else: |
| logging.debug(f"SSH key card: test comment '{SSH_KEY_COMMENT}' not found in key content") |
| |
| # For the complexity linter only |
| test_tidy_up_ssh_keys_continued(page, fingerprints_to_delete) |
| |
| |
| def test_tidy_up_ssh_keys_continued(page: Page, fingerprints_to_delete: list[str]) -> None: |
| if not fingerprints_to_delete: |
| logging.info("No test SSH keys found to delete") |
| return |
| |
| # Delete identified keys |
| logging.info(f"Attempting to delete {len(fingerprints_to_delete)} test SSH keys") |
| for fingerprint in fingerprints_to_delete: |
| logging.info(f"Locating delete form for fingerprint: {fingerprint}") |
| # Locate again by fingerprint for robustness in case of changes |
| card_to_delete_locator = page.locator(f"div.card:has(td:has-text('{fingerprint}'))") |
| delete_button_locator = card_to_delete_locator.get_by_role("button", name="Delete key") |
| |
| if delete_button_locator.is_visible(): |
| logging.info(f"Delete button found for {fingerprint}, proceeding with deletion") |
| |
| def handle_dialog(dialog: Dialog) -> None: |
| logging.info(f"Accepting dialog for key deletion: {dialog.message}") |
| dialog.accept() |
| |
| page.once("dialog", handle_dialog) |
| delete_button_locator.click() |
| |
| logging.info(f"Waiting for page reload after deleting key {fingerprint}") |
| page.wait_for_load_state() |
| wait_for_path(page, "/keys") |
| |
| flash_message_locator = page.locator("div.flash-success") |
| expect(flash_message_locator).to_contain_text("SSH key deleted successfully") |
| logging.info(f"Deletion successful for key {fingerprint}") |
| |
| else: |
| logging.warning(f"Could not find delete button for fingerprint {fingerprint} after re-locating") |
| |
| logging.info("SSH key tidy up finished") |
| |
| |
| def test_tidy_up_releases(page: Page) -> None: |
| logging.info("Navigating to the admin delete release page") |
| go_to_path(page, "/admin/delete-release") |
| logging.info("Admin delete release page loaded") |
| |
| # TODO: Get these names automatically |
| release_remove(page, f"{TEST_PROJECT}-0.1+draft") |
| release_remove(page, f"{TEST_PROJECT}-0.1+candidate") |
| release_remove(page, f"{TEST_PROJECT}-0.1+preview") |
| release_remove(page, f"{TEST_PROJECT}-0.1+release") |
| release_remove(page, f"{TEST_PROJECT}-0.2") |
| |
| |
| def wait_for_path(page: Page, path: str, timeout: int = 30000) -> None: |
| page.wait_for_url(f"**{path}", timeout=timeout) |
| page.wait_for_load_state() |
| logging.info(f"Current URL: {page.url}") |
| |
| |
| if __name__ == "__main__": |
| main() |
