File _service:obs_scm:hawk_test-1713859822.01332c78.... of Package hawk_test

Overview Repositories Revisions Requests Users Attributes Meta

File _service:obs_scm:hawk_test-1713859822.01332c78.obscpio of Package hawk_test (Revision 473e005e0bbb8224c03f382699ce97f1)

Currently displaying revision 473e005e0bbb8224c03f382699ce97f1 , Show latest

07070100000000000081A400000000000000000000000166276CEE0000001D000000000000000000000000000000000000002C00000000hawk_test-1713859822.01332c78/.dockerignore.git
**/__pycache__
**/*.pyc
07070100000001000081A400000000000000000000000166276CEE000000C1000000000000000000000000000000000000002800000000hawk_test-1713859822.01332c78/.pylintrc[MESSAGES CONTROL]
disable=missing-docstring,line-too-long,too-many-public-methods,too-many-instance-attributes,too-many-arguments,too-many-statements,too-few-public-methods,raise-missing-from
07070100000002000081A400000000000000000000000166276CEE000000AE000000000000000000000000000000000000002A00000000hawk_test-1713859822.01332c78/.travis.ymllanguage: python

dist: focal

python:
  - "3.6"
  - "3.7"
  - "3.8"
  - "3.9"

install:
  - pip install -r requirements-test.txt

script:
  - make test
  - make test-docker
07070100000003000081A400000000000000000000000166276CEE0000007D000000000000000000000000000000000000002600000000hawk_test-1713859822.01332c78/AUTHORSAlexei.Tighineanu@suse.com	@atighineanu
Alvaro.Carvajal@suse.com	@alvarocarvajald
Ricardo.Branco@suse.com		@ricardobranco777
07070100000004000081A400000000000000000000000166276CEE000003B9000000000000000000000000000000000000002900000000hawk_test-1713859822.01332c78/Dockerfile# Defines the tag for OBS and build script builds:
#!BuildTag: hawk_test
# Use the repositories defined in OBS for installing packages
#!UseOBSRepositories
FROM	registry.opensuse.org/opensuse/leap:15.4

RUN	zypper -n install -y --no-recommends \
		MozillaFirefox \
		MozillaFirefox-branding-upstream \
		chromium \
		file \
		python3-paramiko \
		python3-PyVirtualDisplay \
		python3-selenium \
		xauth \
		xdpyinfo \
		xorg-x11-fonts \
		xorg-x11-Xvnc && \
	zypper -n clean -a

RUN	zypper -n install -y --no-recommends \
		gzip \
		tar \
		wget && \
	zypper -n clean -a && \
	wget -O- https://github.com/mozilla/geckodriver/releases/download/v0.32.0/geckodriver-v0.32.0-linux64.tar.gz | tar zxf - -C /usr/local/bin/

RUN	chmod +x /usr/local/bin/*

RUN	useradd -l -m -d /test test

COPY	*.py /

ENV     PYTHONPATH /
ENV	PYTHONUNBUFFERED 1
ENV	DBUS_SESSION_BUS_ADDRESS /dev/null

WORKDIR	/test

USER	test
ENTRYPOINT ["/usr/bin/python3", "/hawk_test.py"]
07070100000005000081A400000000000000000000000166276CEE00000047000000000000000000000000000000000000002700000000hawk_test-1713859822.01332c78/Makefiletest:
	@flake8 --ignore=E501
	@pylint --ignored-modules=distutils *.py
07070100000006000081A400000000000000000000000166276CEE0000088C000000000000000000000000000000000000002800000000hawk_test-1713859822.01332c78/README.md# Hawk end to end tests.

This Docker image runs a set of Selenium tests for testing [Hawk](https://github.com/ClusterLabs/hawk/)

The following tests are executed by openQA during ci regularly.

As developer you can execute them manually when you do an update on hawk.

# Pre-requisites:

* docker
* 2 Hawk vms running 
 (normally it is a cluster)
See https://github.com/SUSE/pacemaker-deploy  for deploying hawk

# Quickstart:

1) Create the docker image
`docker build . -t hawk_test `

2) Run the tests with:
``` docker run --ipc=host -v /tmp/.X11-unix:/tmp/.X11-unix -e DISPLAY=$DISPLAY hawk_test -H 10.162.32.175 -S 10.162.29.122 -s linux --xvfb ```

Notes:
  - You may want to add `--net=host` if you have problems with DNS resolution.

## Dependencies

- OS packages:
  - Xvfb (optional)
  - Docker (optional)
  - Firefox
  - [Geckodriver](https://github.com/mozilla/geckodriver/releases)
  - Chromium (optional)
  - [Chromedriver](https://chromedriver.chromium.org/downloads) (optional)
  - Python 3.6+
- Python packages:
  - paramiko
  - selenium
  - PyVirtualDisplay

## Options

```
  -h, --help            show this help message and exit
  -b {firefox,chrome,chromium}, --browser {firefox,chrome,chromium}
                        Browser to use in the test
  -H HOST, --host HOST  Host or IP address where HAWK is running
  -S SLAVE, --slave SLAVE
                        Host or IP address of the slave
  -I VIRTUAL_IP, --virtual-ip VIRTUAL_IP
                        Virtual IP address in CIDR notation
  -P PORT, --port PORT  TCP port where HAWK is running
  -s SECRET, --secret SECRET
                        root SSH Password of the HAWK node
  -r RESULTS, --results RESULTS
                        Generate hawk_test.results file for use with openQA.
  --xvfb                Use Xvfb. Headless mode
```

## FAQ

- Why Xvfb?
  - The `-headless` in both browsers still have bugs, specially with modal dialogs.
  - Having Xvfb prevents it from connecting to our X system.
- Why docker?
  - The Docker image packs the necessary dependencies in such a way that fits the compatibility matrix between Python, Selenium, Firefox (and Geckodriver) & Chromium (and Chromedriver).
07070100000007000081A400000000000000000000000166276CEE000014AC000000000000000000000000000000000000002B00000000hawk_test-1713859822.01332c78/hawk_test.py#!/usr/bin/python3
# Copyright (C) 2019 SUSE LLC
"""HAWK GUI interface Selenium test: tests hawk GUI with Selenium using firefox or chrome"""

import argparse
import ipaddress
import socket
import sys

from pyvirtualdisplay import Display

from hawk_test_driver import HawkTestDriver
from hawk_test_results import ResultSet
from hawk_test_ssh import HawkTestSSH

def hostname(string):
    try:
        socket.getaddrinfo(string, 1)
        return string
    except socket.gaierror:
        raise argparse.ArgumentTypeError(f"Unknown host: {string}")  # pylint: disable=raise-missing-from

def cidr_address(string):
    try:
        ipaddress.ip_network(string, False)
        return string
    except ValueError as exc:
        raise argparse.ArgumentTypeError(f"Invalid CIDR address: {string}") from exc

def port(string):
    if string.isdigit() and 1 <= int(string) <= 65535:
        return string
    raise argparse.ArgumentTypeError(f"Invalid port number: {string}") from ValueError

def parse_args():
    parser = argparse.ArgumentParser(description='HAWK GUI interface Selenium test')
    parser.add_argument('-b', '--browser', default='firefox', choices=['firefox', 'chrome', 'chromium'],
                        help='Browser to use in the test')
    parser.add_argument('-H', '--host', default='localhost', type=hostname,
                        help='Host or IP address where HAWK is running')
    parser.add_argument('-S', '--slave', type=hostname,
                        help='Host or IP address of the slave')
    parser.add_argument('-I', '--virtual-ip', type=cidr_address,
                        help='Virtual IP address in CIDR notation')
    parser.add_argument('-P', '--port', default='7630', type=port,
                        help='TCP port where HAWK is running')
    parser.add_argument('-s', '--secret',
                        help='root SSH Password of the HAWK node')
    parser.add_argument('-r', '--results',
                        help='Generate hawk_test.results file')
    parser.add_argument('--xvfb', action='store_true',
                        help='Use Xvfb. Headless mode')
    args = parser.parse_args()
    return args

def main():
    args = parse_args()

if args.xvfb:
        global DISPLAY  # pylint: disable=global-statement
        DISPLAY = Display()
        DISPLAY.start()

# Initialize results set
    results = ResultSet()
    results.add_ssh_tests()

# Establish SSH connection to verify status
    ssh = HawkTestSSH(args.host, args.secret)

# Get version from /etc/os-release
    test_version = ssh.ssh.exec_command("grep VERSION= /etc/os-release")[1].read().decode().strip().split("=")[1].strip('"')

# Create driver instance
    browser = HawkTestDriver(addr=args.host, port=args.port,
                             browser=args.browser, headless=args.xvfb,
                             version=test_version)

# Resources to create
    cluster = 'Anderes'
    primitive = 'cool_primitive'
    clone = 'cool_clone'
    group = 'cool_group'

# Tests to perform
    if args.virtual_ip:
        browser.test('test_add_virtual_ip', results, args.virtual_ip)
        browser.test('test_remove_virtual_ip', results)
    else:
        results.set_test_status('test_add_virtual_ip', 'skipped')
        results.set_test_status('test_remove_virtual_ip', 'skipped')
    browser.test('test_set_stonith_maintenance', results)
    ssh.verify_stonith_in_maintenance(results)
    browser.test('test_disable_stonith_maintenance', results)
    browser.test('test_view_details_first_node', results)
    browser.test('test_clear_state_first_node', results)
    browser.test('test_set_first_node_maintenance', results)
    ssh.verify_node_maintenance(results)
    browser.test('test_disable_maintenance_first_node', results)
    browser.test('test_add_new_cluster', results, cluster)
    browser.test('test_remove_cluster', results, cluster)
    browser.test('test_click_on_history', results)
    browser.test('test_generate_report', results)
    browser.test('test_click_on_command_log', results)
    browser.test('test_click_on_status', results)
    browser.test('test_add_primitive', results, primitive)
    ssh.verify_primitive(primitive, test_version, results)
    browser.test('test_remove_primitive', results, primitive)
    ssh.verify_primitive_removed(primitive, results)
    browser.test('test_add_clone', results, clone)
    browser.test('test_remove_clone', results, clone)
    browser.test('test_add_group', results, group)
    browser.test('test_remove_group', results, group)
    browser.test('test_click_around_edit_conf', results)
    if args.slave:
        browser.addr = args.slave
        browser.test('test_fencing', results)
    else:
        results.set_test_status('test_fencing', 'skipped')

# Save results if run with -r or --results
    if args.results:
        results.logresults(args.results)

return results.get_failed_tests_total()

if __name__ == "__main__":
    import warnings
    warnings.filterwarnings(action='ignore', module='.*paramiko.*')

DISPLAY = None
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        if DISPLAY is not None:
            DISPLAY.stop()
        sys.exit(1)
    finally:
        if DISPLAY is not None:
            DISPLAY.stop()
07070100000008000081A400000000000000000000000166276CEE00006E55000000000000000000000000000000000000003200000000hawk_test-1713859822.01332c78/hawk_test_driver.py#!/usr/bin/python3
# Copyright (C) 2019 SUSE LLC
"""Define Selenium driver related functions and classes to test the HAWK GUI"""

import ipaddress
import time
from distutils.version import LooseVersion as Version

from selenium import webdriver
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.common.by import By
from selenium.common.exceptions import TimeoutException, WebDriverException
from selenium.common.exceptions import ElementNotInteractableException
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

BIG_TIMEOUT = 6

# Error messages
class Error:
    MAINT_TOGGLE_ERR = "Could not find Switch to Maintenance toggle button for node"
    PRIMITIVE_TARGET_ROLE_ERR = "Couldn't find value [Started] for primitive target-role"
    STONITH_ERR = "Couldn't find stonith-sbd menu to place it in maintenance mode"
    STONITH_ERR_OFF = "Could not find Disable Maintenance Mode button for stonith-sbd"

# XPATH constants
class Xpath:
    ANYTHING_OPT_LIST = '//option[contains(@value, "anything")]'
    CLICK_OK_SUBMIT = '//*[@id="modal"]/div/div/form/div[3]/input'
    CLONE_CHILD = '//select[contains(@data-help-filter, ".row.resource") and contains(@name, "clone[child]")]'
    CLONE_DATA_HELP_FILTER = '//a[contains(@data-help-filter, ".clone")]'
    COMMIT_BTN_DANGER = '//button[contains(@class, "btn-danger") and contains(@class, "commit")]'
    DISMISS_MODAL = '//*[@id="modal"]/div/div/div[3]/button'
    DROP_DOWN_FORMAT = '//*[@id="resources"]/div[1]/div[2]/div[2]/table/tbody/tr[{}]/td[6]/div/div'
    EDIT_MONITOR_TIMEOUT = '//*[@id="oplist"]/fieldset/div/div[1]/div[3]/div[2]/div/div/a[1]'
    EDIT_START_TIMEOUT = '//*[@id="oplist"]/fieldset/div/div[1]/div[1]/div[2]/div/div/a[1]'
    EDIT_STOP_TIMEOUT = '//*[@id="oplist"]/fieldset/div/div[1]/div[2]/div[2]/div/div/a[1]'
    GENERATE_REPORT = '//*[@id="generate"]/form/div/div[2]/button'
    GROUP_DATA_FILTER = '//a[contains(@data-help-filter, ".group")]'
    HREF_ALERTS = '//a[contains(@href, "#alerts")]'
    HREF_CONFIGURATION = '//a[contains(@href, "#configurationMenu")]'
    HREF_CONFIG_EDIT = '//a[contains(@href, "config/edit")]'
    HREF_CONSTRAINTS = '//a[contains(@href, "#constraints")]'
    HREF_DASHBOARD = '//a[contains(@href, "/dashboard")]'
    HREF_DELETE_FORMAT = '//a[contains(@href, "{}") and contains(@title, "Delete")]'
    HREF_FENCING = '//a[contains(@href, "#fencing")]'
    HREF_NODES = '//a[contains(@href, "#nodes")]'
    HREF_REPORTS = '//a[contains(@href, "/reports")]'
    HREF_TAGS = '//a[contains(@href, "#tags")]'
    MODAL_MONITOR_TIMEOUT = '//*[@id="modal"]/div/div/form/div[2]/fieldset/div/div[1]/div'
    MODAL_STOP = '//*[@id="modal"]/div/div/form/div[2]/fieldset/div/div[2]/div/div/select/option[6]'
    MODAL_TIMEOUT = '//*[@id="modal"]/div/div/form/div[2]/fieldset/div/div[1]/div/div'
    NODE_DETAILS = '//*[@id="nodes"]/div[1]/div[2]/div[2]/table/tbody/tr[1]/td[5]/div/a[2]'
    NODE_MAINT = '//a[contains(@href, "maintenance") and contains(@title, "Switch to maintenance")]'
    NODE_READY = '//a[contains(@href, "ready") and contains(@title, "Switch to ready")]'
    OCF_OPT_LIST = '//option[contains(@value, "ocf")]'
    OPERATIONS = '//*[@id="nodes"]/div[1]/div[2]/div[2]/table/tbody/tr[1]/td[5]/div/div/button'
    OPT_STONITH = '//option[contains(@value, "stonith-sbd")]'
    RESOURCES_TYPES = '//a[contains(@href, "resources/types")]'
    RSC_OK_SUBMIT = '//input[contains(@class, "submit")]'
    RSC_ROWS = '//*[@id="resources"]/div[1]/div[2]/div[2]/table/tbody/tr'
    STONITH_CHKBOX = '//input[contains(@type, "checkbox") and contains(@value, "stonith-sbd")]'
    STONITH_MAINT_OFF = '//a[contains(@href, "stonith-sbd") and contains(@title, "Disable Maintenance Mode")]'
    STONITH_MAINT_ON = '//a[contains(@href, "stonith-sbd/maintenance_on")]'
    TARGET_ROLE_FORMAT = '//select[contains(@class, "form-control") and contains(@name, "{}[meta][target-role]")]'
    TARGET_ROLE_STARTED = '//option[contains(@value, "tarted")]'
    WIZARDS_BASIC = '//span[contains(@href, "basic")]'

class HawkTestDriver:
    def __init__(self, addr='localhost', port='7630', browser='firefox', headless=False, version='15-SP5'):
        self.addr = addr
        self.port = port
        self.driver = None
        self.test_version = version
        self.test_status = True
        self.headless = headless
        self.browser = browser
        if browser == 'firefox':
            self.timeout_scale = 2.5
        else:
            self.timeout_scale = 1

def _connect(self):
        if self.browser in ['chrome', 'chromium']:
            options = webdriver.ChromeOptions()
            options.add_argument('--no-sandbox')
            options.add_argument('--disable-gpu')
            if self.headless:
                options.add_argument('--headless')
            options.add_argument('--disable-dev-shm-usage')
            self.driver = webdriver.Chrome(chrome_options=options)
        else:
            profile = webdriver.FirefoxProfile()
            profile.accept_untrusted_certs = True
            profile.assume_untrusted_cert_issuer = True
            self.driver = webdriver.Firefox(firefox_profile=profile)
        self.driver.maximize_window()
        return self.driver

def _close(self):
        self.click_on('Logout')
        self.driver.quit()

@staticmethod
    def set_test_status(results, testname, status):
        results.set_test_status(testname, status)

def _do_login(self):
        mainlink = f'https://{self.addr}:{self.port}'
        self.driver.get(mainlink)
        elem = self.find_element(By.NAME, "session[username]")
        if not elem:
            print("ERROR: couldn't find element [session[username]]. Cannot login")
            self.driver.quit()
            return False
        elem.send_keys("hacluster")
        elem = self.find_element(By.NAME, "session[password]")
        if not elem:
            print("ERROR: Couldn't find element [session[password]]. Cannot login")
            self.driver.quit()
            return False
        elem.send_keys("linux")
        elem.send_keys(Keys.RETURN)
        return True

# Clicks on element identified by clicker if major version from the test is greater or
    # equal than the version to check
    def click_if_major_version(self, version_to_check, text):
        if Version(self.test_version) >= Version(version_to_check):
            self.find_element(By.XPATH, f"//*[text()='{text.capitalize()}']").click()

# Internal support function click_on partial link test. Sets test_status to False on failure
    def click_on(self, text):
        print(f"INFO: Main page. Click on {text}")
        elem = self.find_element(By.PARTIAL_LINK_TEXT, text)
        if not elem:
            print(f"ERROR: Couldn't find element '{text}'")
            self.test_status = False
            return False
        try:
            elem.click()
        except ElementNotInteractableException:
            # Element is obscured. Wait and click again
            time.sleep(2 * self.timeout_scale)
            elem.click()
        time.sleep(self.timeout_scale)
        return True

def find_element(self, bywhat, texto, tout=60):
        try:
            elem = WebDriverWait(self.driver,
                                 tout).until(EC.presence_of_element_located((bywhat, texto)))
        except TimeoutException:
            print(f"INFO: {tout} seconds timeout while looking for element [{texto}] by [{bywhat}]")
            return False
        return elem

def verify_success(self):
        elem = self.find_element(By.CLASS_NAME, 'alert-success', 60 * self.timeout_scale)
        if not elem:
            elem = self.find_element(By.PARTIAL_LINK_TEXT, 'Rename', 5)
            if not elem:
                return False
        return True

def fill_value(self, field, tout):
        elem = self.find_element(By.NAME, field)
        if not elem:
            print(f"ERROR: couldn't find element [{field}].")
            return
        elem.clear()
        elem.send_keys(f"{tout}")

def submit_operation_params(self, errmsg):
        self.check_and_click_by_xpath(errmsg, [Xpath.CLICK_OK_SUBMIT])

def check_edit_conf(self):
        print("INFO: Check edit configuration")
        time.sleep(1)
        if Version(self.test_version) >= Version("15"):
            self.check_and_click_by_xpath("Couldn't find Configuration element", [Xpath.HREF_CONFIGURATION])
        time.sleep(1)
        self.check_and_click_by_xpath("Couldn't find Edit Configuration element", [Xpath.HREF_CONFIG_EDIT])

# Internal support function to find element(s) by xpath and click them
    # Sets test_status to False on failure.
    def check_and_click_by_xpath(self, errmsg, xpath_exps):
        for xpath in xpath_exps:
            elem = self.find_element(By.XPATH, xpath)
            if not elem:
                print(f"ERROR: Couldn't find element by xpath [{xpath}] {errmsg}")
                self.test_status = False
                return
            time.sleep(5)
            try:
                elem.click()
            except ElementNotInteractableException:
                # Element is obscured. Wait and click again
                time.sleep(10 * self.timeout_scale)
                elem.click()

# Generic function to perform the tests
    def test(self, testname, results, *extra):
        self.test_status = True    # Clear internal test status before testing
        self._connect()
        if self._do_login():
            time.sleep(5)
            if getattr(self, testname)(*extra):
                self.set_test_status(results, testname, 'passed')
            else:
                self.set_test_status(results, testname, 'failed')
                self.driver.save_screenshot(f'{testname}.png')
        self._close()

# Set STONITH/sbd in maintenance. Assumes stonith-sbd resource is the last one listed on the
    # resources table
    def test_set_stonith_maintenance(self):
        # wait for page to fully load
        if self.find_element(By.XPATH, Xpath.RSC_ROWS):
            totalrows = len(self.driver.find_elements_by_xpath(Xpath.RSC_ROWS))
            if not totalrows:
                totalrows = 1
            print("TEST: test_set_stonith_maintenance: Placing stonith-sbd in maintenance")
            self.check_and_click_by_xpath(Error.STONITH_ERR, [Xpath.DROP_DOWN_FORMAT.format(totalrows),
                                                              Xpath.STONITH_MAINT_ON, Xpath.COMMIT_BTN_DANGER])
        if self.verify_success():
            print("INFO: stonith-sbd successfully placed in maintenance mode")
            return True
        print("ERROR: failed to place stonith-sbd in maintenance mode")
        return False

def test_disable_stonith_maintenance(self):
        print("TEST: test_disable_stonith_maintenance: Re-activating stonith-sbd")
        self.check_and_click_by_xpath(Error.STONITH_ERR_OFF, [Xpath.STONITH_MAINT_OFF, Xpath.COMMIT_BTN_DANGER])
        if self.verify_success():
            print("INFO: stonith-sbd successfully reactivated")
            return True
        print("ERROR: failed to reactive stonith-sbd from maintenance mode")
        return False

def test_view_details_first_node(self):
        print("TEST: test_view_details_first_node: Checking details of first cluster node")
        self.click_on('Nodes')
        self.check_and_click_by_xpath("Click on Nodes", [Xpath.HREF_NODES])
        self.check_and_click_by_xpath("Could not find first node pull down menu", [Xpath.NODE_DETAILS])
        self.check_and_click_by_xpath("Could not find button to dismiss node details popup",
                                      [Xpath.DISMISS_MODAL])
        time.sleep(self.timeout_scale)
        return self.test_status

def test_clear_state_first_node(self):
        print("TEST: test_clear_state_first_node")
        self.click_on('Nodes')
        self.check_and_click_by_xpath("Click on Nodes", [Xpath.HREF_NODES])
        self.check_and_click_by_xpath("Could not find pull down menu for first cluster node",
                                      [Xpath.OPERATIONS])
        self.click_on('Clear state')
        self.check_and_click_by_xpath("Could not clear the state of the first node",
                                      [Xpath.COMMIT_BTN_DANGER])
        if self.verify_success():
            print("INFO: cleared state of first node successfully")
            time.sleep(2 * self.timeout_scale)
            return True
        print("ERROR: failed to clear state of the first node")
        return False

def test_set_first_node_maintenance(self):
        print("TEST: test_set_first_node_maintenance: switching node to maintenance")
        self.click_on('Nodes')
        self.check_and_click_by_xpath("Click on Nodes", [Xpath.HREF_NODES])
        self.check_and_click_by_xpath(Error.MAINT_TOGGLE_ERR, [Xpath.NODE_MAINT, Xpath.COMMIT_BTN_DANGER])
        if self.verify_success():
            print("INFO: node successfully switched to maintenance mode")
            return True
        print("ERROR: failed to switch node to maintenance mode")
        return False

def test_disable_maintenance_first_node(self):
        print("TEST: test_disable_maintenance_first_node: switching node to ready")
        self.click_on('Nodes')
        self.check_and_click_by_xpath("Click on Nodes", [Xpath.HREF_NODES])
        self.check_and_click_by_xpath(Error.MAINT_TOGGLE_ERR, [Xpath.NODE_READY, Xpath.COMMIT_BTN_DANGER])
        if self.verify_success():
            print("INFO: node successfully switched to ready mode")
            return True
        print("ERROR: failed to switch node to ready mode")
        return False

def test_add_new_cluster(self, cluster):
        print("TEST: test_add_new_cluster")
        self.click_on('Dashboard')
        self.check_and_click_by_xpath("Click on Dashboard", [Xpath.HREF_DASHBOARD])
        elem = self.find_element(By.CLASS_NAME, "btn-default")
        if not elem:
            print("ERROR: Couldn't find class 'btn-default'")
            return False
        elem.click()
        elem = self.find_element(By.NAME, "cluster[name]")
        if not elem:
            print("ERROR: Couldn't find element [cluster[name]]. Cannot add cluster")
            return False
        elem.send_keys(cluster)
        elem = self.find_element(By.NAME, "cluster[host]")
        if not elem:
            print("ERROR: Couldn't find element [cluster[host]]. Cannot add cluster")
            return False
        elem.send_keys(self.addr)
        elem = self.find_element(By.NAME, "submit")
        if not elem:
            print("ERROR: Couldn't find submit button")
            return False
        elem.click()
        while True:
            elem = self.find_element(By.PARTIAL_LINK_TEXT, 'Dashboard')
            try:
                elem.click()
                return True
            except WebDriverException:
                time.sleep(1 + self.timeout_scale)
        return False

def test_remove_cluster(self, cluster):
        print("TEST: test_remove_cluster")
        self.click_on('Dashboard')
        self.check_and_click_by_xpath("Click on Dashboard", [Xpath.HREF_DASHBOARD])
        elem = self.find_element(By.PARTIAL_LINK_TEXT, cluster)
        if not elem:
            print(f"ERROR: Couldn't find cluster [{cluster}]. Cannot remove")
            return False
        elem.click()
        time.sleep(BIG_TIMEOUT)
        elem = self.find_element(By.CLASS_NAME, 'close')
        if not elem:
            print("ERROR: Cannot find cluster remove button")
            return False
        elem.click()
        time.sleep(2 * self.timeout_scale)
        elem = self.find_element(By.CLASS_NAME, 'cancel')
        if not elem:
            print(f"ERROR: No cancel button while removing cluster [{cluster}]")
        else:
            elem.click()
        time.sleep(self.timeout_scale)
        elem = self.find_element(By.CLASS_NAME, 'close')
        elem.click()
        time.sleep(2 * self.timeout_scale)
        elem = self.find_element(By.CLASS_NAME, 'btn-danger')
        if not elem:
            print(f"ERROR: No OK button found while removing cluster [{cluster}]")
        else:
            elem.click()
        if self.verify_success():
            print(f"INFO: Successfully removed cluster: [{cluster}]")
            return True
        print(f"ERROR: Could not remove cluster [{cluster}]")
        return False

def test_click_on_history(self):
        print("TEST: test_click_on_history")
        self.click_if_major_version("15", 'troubleshooting')
        if not self.test_status:
            return False
        return self.click_on('History')

def test_generate_report(self):
        print("TEST: test_generate_report: click on Generate report")
        self.click_if_major_version("15", 'troubleshooting')
        self.click_on('History')
        self.check_and_click_by_xpath("Click on History", [Xpath.HREF_REPORTS])
        if self.find_element(By.XPATH, Xpath.GENERATE_REPORT):
            self.check_and_click_by_xpath("Could not find button for Generate report",
                                          [Xpath.GENERATE_REPORT])
            # Look for actual report
            self.find_element(By.LINK_TEXT, "hawk")
            print("INFO: successfully generated report")
            return True
        print("ERROR: failed to generate report")
        return False

def test_click_on_command_log(self):
        print("TEST: test_click_on_command_log")
        self.click_if_major_version("15", 'troubleshooting')
        if not self.test_status:
            return False
        return self.click_on('Command Log')

def test_click_on_status(self):
        print("TEST: test_click_on_status")
        return self.click_on('Status')

def test_add_primitive(self, primitive):
        print(f"TEST: test_add_primitive: Add Resources: Primitive {primitive}")
        self.click_if_major_version("15", 'configuration')
        self.click_on('Resource')
        self.check_and_click_by_xpath("Click on Add Resource", [Xpath.RESOURCES_TYPES])
        self.click_on('rimitive')
        # Fill the primitive
        elem = self.find_element(By.NAME, 'primitive[id]')
        if not elem:
            print(f"ERROR: Couldn't find element [primitive[id]]. Cannot add primitive [{primitive}].")
            return False
        elem.send_keys(primitive)
        elem = self.find_element(By.NAME, 'primitive[clazz]')
        if not elem:
            print(f"ERROR: Couldn't find element [primitive[clazz]]. Cannot add primitive [{primitive}]")
            return False
        elem.click()
        self.check_and_click_by_xpath("Couldn't find value [ocf] for primitive class",
                                      [Xpath.OCF_OPT_LIST])
        elem = self.find_element(By.NAME, 'primitive[type]')
        if not elem:
            print(f"ERROR: Couldn't find element [primitive[type]]. Cannot add primitive [{primitive}].")
            return False
        elem.click()
        self.check_and_click_by_xpath("Couldn't find value [anything] for primitive type",
                                      [Xpath.ANYTHING_OPT_LIST])
        elem = self.find_element(By.NAME, 'primitive[params][binfile]')
        if not elem:
            print("ERROR: Couldn't find element [primitive[params][binfile]]")
            return False
        elem.send_keys("file")
        # Set start timeout value in 35s
        self.check_and_click_by_xpath("Couldn't find edit button for start operation",
                                      [Xpath.EDIT_START_TIMEOUT, Xpath.MODAL_TIMEOUT])
        self.fill_value('op[timeout]', "35s")
        self.submit_operation_params(". Couldn't Apply changes for start operation")
        # Set stop timeout value in 15s and on-fail
        self.check_and_click_by_xpath("Couldn't find edit button for stop operation",
                                      [Xpath.EDIT_STOP_TIMEOUT, Xpath.MODAL_TIMEOUT])
        self.fill_value('op[timeout]', "15s")
        self.check_and_click_by_xpath("Couldn't add on-fail option for stop operation",
                                      [Xpath.MODAL_STOP])
        self.submit_operation_params(". Couldn't Apply changes for stop operation")
        # Set monitor timeout value in 9s and interval in 13s
        self.check_and_click_by_xpath("Couldn't find edit button for monitor operation",
                                      [Xpath.EDIT_MONITOR_TIMEOUT, Xpath.MODAL_MONITOR_TIMEOUT])
        self.fill_value('op[timeout]', "9s")
        self.fill_value('op[interval]', "13s")
        self.submit_operation_params(". Couldn't Apply changes for monitor operation")
        elem = self.find_element(By.NAME, 'primitive[meta][target-role]')
        if not elem:
            print("ERROR: Couldn't find element [primitive[meta][target-role]]. "
                  f"Cannot add primitive [{primitive}].")
            return False
        time.sleep(1)
        elem.click()
        self.check_and_click_by_xpath(Error.PRIMITIVE_TARGET_ROLE_ERR, [Xpath.TARGET_ROLE_STARTED])
        elem = self.find_element(By.NAME, 'submit')
        if not elem:
            print(f"ERROR: Couldn't find submit button for primitive [{primitive}] creation.")
        else:
            elem.click()
        status = self.verify_success()
        if status:
            print(f"INFO: Successfully added primitive [{primitive}] of class [ocf:heartbeat:anything]")
        else:
            print(f"ERROR: Could not create primitive [{primitive}]")
        return status

def remove_rsc(self, name):
        print(f"INFO: Remove Resource: {name}")
        self.check_edit_conf()
        # resources list does load again after edit configuration page is loaded
        time.sleep(10)
        self.check_and_click_by_xpath(f"Cannot delete resource [{name}]", [Xpath.HREF_DELETE_FORMAT.format(name)])
        time.sleep(2)
        self.check_and_click_by_xpath(f"Cannot confirm delete of resource [{name}]", [Xpath.COMMIT_BTN_DANGER])
        time.sleep(2)
        self.check_and_click_by_xpath("Couldn't find Edit Configuration element", [Xpath.HREF_CONFIG_EDIT])
        time.sleep(3)
        if not self.test_status:
            print(f"ERROR: One of the elements required to remove resource [{name}] wasn't found")
            return False
        elem = self.find_element(By.XPATH, Xpath.HREF_DELETE_FORMAT.format(name), 5)
        if not elem:
            print(f"INFO: Successfully removed resource [{name}]")
            return True
        print(f"ERROR: Failed to remove resource [{name}]")
        return False

def test_remove_primitive(self, primitive):
        print(f"TEST: test_remove_primitive: Remove Primitive: {primitive}")
        return self.remove_rsc(primitive)

def test_remove_clone(self, clone):
        print(f"TEST: test_remove_clone: Remove Clone: {clone}")
        return self.remove_rsc(clone)

def test_remove_group(self, group):
        print(f"TEST: test_remove_group: Remove Group: {group}")
        return self.remove_rsc(group)

def test_add_clone(self, clone):
        print(f"TEST: test_add_clone: Adding clone [{clone}]")
        self.click_if_major_version("15", 'configuration')
        self.click_on('Resource')
        self.check_and_click_by_xpath("Click on Add Resource", [Xpath.RESOURCES_TYPES])
        self.check_and_click_by_xpath(f"on Create Clone [{clone}]", [Xpath.CLONE_DATA_HELP_FILTER])
        elem = self.find_element(By.NAME, 'clone[id]')
        if not elem:
            print("ERROR: Couldn't find element [clone[id]]. No text-field where to type clone id")
            return False
        elem.send_keys(clone)
        self.check_and_click_by_xpath(f"while adding clone [{clone}]",
                                      [Xpath.CLONE_CHILD, Xpath.OPT_STONITH, Xpath.TARGET_ROLE_FORMAT.format('clone'),
                                       Xpath.TARGET_ROLE_STARTED, Xpath.RSC_OK_SUBMIT])
        if self.verify_success():
            print(f"INFO: Successfully added clone [{clone}] of [stonith-sbd]")
            return True
        print(f"ERROR: Could not create clone [{clone}]")
        return False

def test_add_group(self, group):
        print(f"TEST: test_add_group: Adding group [{group}]")
        self.click_if_major_version("15", 'configuration')
        self.click_on('Resource')
        self.check_and_click_by_xpath("Click on Add Resource", [Xpath.RESOURCES_TYPES])
        self.check_and_click_by_xpath(f"while adding group [{group}]", [Xpath.GROUP_DATA_FILTER])
        elem = self.find_element(By.NAME, 'group[id]')
        if not elem:
            print("ERROR: Couldn't find text-field [group[id]] to input group id")
            return False
        elem.send_keys(group)
        self.check_and_click_by_xpath(f"while adding group [{group}]",
                                      [Xpath.STONITH_CHKBOX, Xpath.TARGET_ROLE_FORMAT.format('group'),
                                       Xpath.TARGET_ROLE_STARTED, Xpath.RSC_OK_SUBMIT])
        if self.verify_success():
            print(f"INFO: Successfully added group [{group}] of [stonith-sbd]")
            return True
        print(f"ERROR: Could not create group [{group}]")
        return False

def test_click_around_edit_conf(self):
        print("TEST: test_click_around_edit_conf")
        print("TEST: Will click on Constraints, Nodes, Tags, Alerts and Fencing")
        self.check_edit_conf()

click_list = [
            Xpath.HREF_CONSTRAINTS, Xpath.HREF_NODES, Xpath.HREF_TAGS,
            Xpath.HREF_ALERTS, Xpath.HREF_FENCING
        ]

self.check_and_click_by_xpath("while checking around edit configuration", click_list)
        return self.test_status

def test_add_virtual_ip(self, virtual_ip):
        print("TEST: test_add_virtual_ip: Add virtual IP from the Wizard")
        self.click_if_major_version("15", 'configuration')
        broadcast = str(ipaddress.IPv4Network(virtual_ip, False).broadcast_address)
        virtual_ip, netmask = virtual_ip.split('/')
        self.find_element(By.LINK_TEXT, 'Wizards').click()
        self.check_and_click_by_xpath('while clicking Basic', [Xpath.WIZARDS_BASIC])
        self.click_on('Virtual IP')
        self.fill_value('virtual-ip.id', 'vip')
        self.fill_value('virtual-ip.ip', virtual_ip)
        self.fill_value('virtual-ip.cidr_netmask', netmask)
        self.fill_value('virtual-ip.broadcast', broadcast)
        elem = self.find_element(By.NAME, 'submit')
        if not elem:
            print("ERROR: Couldn't find [Verify] button on Virtual IP Wizard")
            return False
        elem.click()
        time.sleep(3)
        elem = self.find_element(By.NAME, 'submit')
        if not elem:
            print("ERROR: Couldn't find [Apply] button on Virtual IP Wizard")
            return False
        elem.click()
        time.sleep(3)
        # Check that we can connect to the Wizard on the virtual IP
        old_addr = self.addr
        self.addr = virtual_ip
        self._close()
        time.sleep(10)
        self._connect()
        try:
            self._do_login()
        except WebDriverException:
            print("ERROR: Could not connect to virtual IP")
            # Reset address to old_addr so remaining tests can connect to HAWK
            self.addr = old_addr
            return False
        print("INFO: Successfully added virtual IP")
        # Reset address to old_addr
        self.addr = old_addr
        return True

def test_remove_virtual_ip(self):
        print("TEST: test_remove_virtual_ip: Remove virtual IP")
        return self.remove_rsc("vip")

def test_fencing(self):
        print("TEST: test_fencing")
        self.click_on('Nodes')
        self.check_and_click_by_xpath("Click on Nodes", [Xpath.OPERATIONS])
        self.click_on('Fence')
        self.check_and_click_by_xpath("Could not fence first node",
                                      [Xpath.COMMIT_BTN_DANGER])
        if self.verify_success():
            print("INFO: Master node successfully fenced")
            return True
        print("ERROR: Could not fence master node")
        return False
07070100000009000081A400000000000000000000000166276CEE00000BA0000000000000000000000000000000000000003300000000hawk_test-1713859822.01332c78/hawk_test_results.py#!/usr/bin/python3
# Copyright (C) 2019 SUSE LLC
"""Define classes and functions to handle results in HAWK GUI test"""

import time
import json

from hawk_test_driver import HawkTestDriver
from hawk_test_ssh import HawkTestSSH

class ResultSet:
    def __init__(self):
        self.my_tests = []
        self.start_time = time.time()
        for func in dir(HawkTestDriver):
            if func.startswith('test_') and callable(getattr(HawkTestDriver, func)):
                self.my_tests.append(func)
        self.results_set = {'tests': [], 'info': {}, 'summary': {}}
        for test in self.my_tests:
            auxd = {'name': test, 'test_index': 0, 'outcome': 'failed'}
            self.results_set['tests'].append(auxd)
        self.results_set['info']['timestamp'] = time.time()
        with open('/etc/os-release', encoding="utf-8") as file:
            lines = file.read().splitlines()
        osrel = {k: v[1:-1] for (k, v) in [line.split('=') for line in lines if '=' in line]}
        self.results_set['info']['distro'] = osrel['PRETTY_NAME']
        self.results_set['info']['results_file'] = 'hawk_test.results'
        self.results_set['summary']['duration'] = 0
        self.results_set['summary']['passed'] = 0
        self.results_set['summary']['num_tests'] = len(self.my_tests)

def add_ssh_tests(self):
        for func in dir(HawkTestSSH):
            if func.startswith('verify_') and callable(getattr(HawkTestSSH, func)):
                self.my_tests.append(func)
                auxd = {'name': str(func), 'test_index': 0, 'outcome': 'failed'}
                self.results_set['tests'].append(auxd)
        self.results_set['summary']['num_tests'] = len(self.my_tests)

def logresults(self, filename):
        with open(filename, "w", encoding="utf-8") as resfh:
            resfh.write(json.dumps(self.results_set))

def set_test_status(self, testname, status):
        if status not in ['passed', 'failed', 'skipped']:
            raise ValueError('test status must be either [passed] or [failed]')
        if status == 'passed' and \
                self.results_set['tests'][self.my_tests.index(testname)]['outcome'] != 'passed':
            self.results_set['summary']['passed'] += 1
        elif status == 'failed' and \
                self.results_set['tests'][self.my_tests.index(testname)]['outcome'] != 'failed':
            self.results_set['summary']['passed'] -= 1
        elif status == 'skipped' and \
                self.results_set['tests'][self.my_tests.index(testname)]['outcome'] != 'skipped':
            self.results_set['summary']['num_tests'] -= 1
        self.results_set['tests'][self.my_tests.index(testname)]['outcome'] = status
        self.results_set['summary']['duration'] = time.time() - self.start_time
        self.results_set['info']['timestamp'] = time.time()

def get_failed_tests_total(self):
        return self.results_set['summary']['num_tests'] - self.results_set['summary']['passed']
0707010000000A000081A400000000000000000000000166276CEE00001040000000000000000000000000000000000000002F00000000hawk_test-1713859822.01332c78/hawk_test_ssh.py#!/usr/bin/python3
# Copyright (C) 2019 SUSE LLC
"""Define SSH related functions to test the HAWK GUI"""

from distutils.version import LooseVersion as Version
import paramiko

class HawkTestSSH:
    def __init__(self, hostname, secret=None):
        self.ssh = paramiko.SSHClient()
        self.ssh.load_system_host_keys()
        self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy)
        self.ssh.connect(hostname=hostname.lower(), username="root", password=secret)

def check_cluster_conf_ssh(self, command, mustmatch, anycheck=False):
        _, out, err = self.ssh.exec_command(command)
        out, err = map(lambda f: f.read().decode().rstrip('\n'), (out, err))
        print(f"INFO: ssh command [{command}] got output [{out}] and error [{err}]")
        if err:
            print(f"ERROR: got an error over SSH: [{err}]")
            return False
        if isinstance(mustmatch, str):
            return mustmatch in out
        if isinstance(mustmatch, list) and anycheck:
            # Output has to match at least one element in the list
            return any(_ in out for _ in mustmatch)
        if isinstance(mustmatch, list) and not anycheck:
            # Output has to match all elements in list
            return all(_ in out for _ in mustmatch)
        raise ValueError("check_cluster_conf_ssh: mustmatch must be str or list")

@staticmethod
    def set_test_status(results, test, status):
        results.set_test_status(test, status)

def verify_stonith_in_maintenance(self, results):
        print("TEST: verify_stonith_in_maintenance")
        if self.check_cluster_conf_ssh("crm status | grep stonith-sbd", ["unmanaged", "maintenance"], True):
            print("INFO: stonith-sbd is unmanaged/maintenance")
            self.set_test_status(results, 'verify_stonith_in_maintenance', 'passed')
            return True
        print("ERROR: stonith-sbd is not unmanaged nor in maintenance but should be")
        self.set_test_status(results, 'verify_stonith_in_maintenance', 'failed')
        return False

def verify_node_maintenance(self, results):
        print("TEST: verify_node_maintenance: check cluster node is in maintenance mode")
        if self.check_cluster_conf_ssh("crm status | grep -i node", "maintenance"):
            print("INFO: cluster node set successfully in maintenance mode")
            self.set_test_status(results, 'verify_node_maintenance', 'passed')
            return True
        print("ERROR: cluster node failed to switch to maintenance mode")
        self.set_test_status(results, 'verify_node_maintenance', 'failed')
        return False

def verify_primitive(self, primitive, version, results):
        print(f"TEST: verify_primitive: check primitive [{primitive}] exists")
        matches = [f"{primitive} anything", "binfile=file", "op start timeout=35s",
                   "op monitor timeout=9s interval=13s", "meta target-role=Started"]
        if Version(version) < Version('15'):
            matches.append("op stop timeout=15s")
        else:
            matches.append("op stop timeout=15s on-fail=stop")
        if self.check_cluster_conf_ssh("crm configure show", matches):
            print(f"INFO: primitive [{primitive}] correctly defined in the cluster configuration")
            self.set_test_status(results, 'verify_primitive', 'passed')
            return True
        print(f"ERROR: primitive [{primitive}] missing from cluster configuration")
        self.set_test_status(results, 'verify_primitive', 'failed')
        return False

def verify_primitive_removed(self, primitive, results):
        print(f"TEST: verify_primitive_removed: check primitive [{primitive}] is removed")
        if self.check_cluster_conf_ssh("crm resource status | grep ocf::heartbeat:anything", ''):
            print("INFO: primitive successfully removed")
            self.set_test_status(results, 'verify_primitive_removed', 'passed')
            return True
        print(f"ERROR: primitive [{primitive}] still present in the cluster while checking with SSH")
        self.set_test_status(results, 'verify_primitive_removed', 'failed')
        return False
0707010000000B000081A400000000000000000000000166276CEE0000000E000000000000000000000000000000000000003400000000hawk_test-1713859822.01332c78/requirements-test.txtflake8
pylint
07070100000000000000000000000000000000000000010000000000000000000000000000000000000000000000000000000B00000000TRAILER!!!91 blocks

Places

File _service:obs_scm:hawk_test-1713859822.01332c78.obscpio of Package hawk_test (Revision 473e005e0bbb8224c03f382699ce97f1)

Places