""" Testcase common functions and classes """
import os
import sys
import subprocess
import shutil
import time
import copy
import importlib
import http.client
import json
import inspect
import urllib
from requests import Session
from selenium import webdriver
from selenium.webdriver.support.ui import Select
from selenium.webdriver.common.keys import Keys
from selenium.common.exceptions import (NoSuchElementException,
StaleElementReferenceException, WebDriverException)
from selenium.webdriver.common.action_chains import ActionChains
import cherrypy
from ppdb import const, config
from ppdb.lib import pglib, util, memlib
from ppdb.lib import dbbuild
js_consts = const.JSConsts()
# This file is assumed to be in ppdb/test
test_root = os.path.abspath(os.path.dirname(__file__))
#~ print("test_root set to %s" % test_root)
#~ DEBUG = False
DEBUG = True
WAIT = 50
TEST_SOCKETPORT = 8008
# Users to allow manual and pytest testing, and a generic user 'tester' for the tests
# using the 'generic_setup' fixture. This dict is used by maketest and dbbuild
TEST_USERS = {
'tester': (424, 'tester#', ()),
'member': (1670, 'member#', ('member',)), # 2022 Life no flock
'breeder': (426, 'breeder#', ('breeder',)), # 1189 secondary
'membrdr': (424, 'membrdr#', ('member', 'breeder')), # 1189 primary
'membrdr2': (792, 'membrdr2#', ('member', 'breeder',)), # 2224 - multiple flocks
'memregsec': (492, 'memregsec#', ('memsec', 'regsec')), # 1359
}
TEST_PMT_DATA = {'class_id': '1', 'pmt_method': 'Cheque', 'pmt_ref': ''}
TEST_PRIVACY_DATA = [1, 3]
TEST_MEM_DATA = {'address_1': 'The House', 'address_2': 'The Road', 'address_3': '',
'post_town': 'BIGTOWN',
'county': 'Devon', 'post_code': 'EX33 8JY', 'country': '',
'non_renewal': '', 'source': '', 'note': ''}
TEST_PERS_DATA = {'active': True, 'title': 'Dr', 'initials': 'HG',
'forename': 'Henry', 'surname': 'Johnson',
'phone': {'number': '01234 567 890', 'phone_type': 'Home',
'comment': ''},
'email': {'address': 'test01@gmail.com', 'comment': ''}}
PPD_PID = None
keep_browser = True
class Fail(Exception):
""" Catch a Fail exception """
def __init__(self, msg=""):
global PPD_PID
global keep_browser
if PPD_PID:
PPD_PID.terminate()
keep_browser = True
print(f"Fail keep_browser: {keep_browser}")
def configure_app(classname='', refresh=False):
""" - Initialise the appropriate dblib using settings in config.
- These tests use a test database so modify the 'database' parameter accordingly.
"""
# Use the test database
dbconf = copy.deepcopy(config.ppdb_conf['pgsql_conf'])
dbname = f"{dbconf['database']}_test"
dbconf['database'] = dbname
dbparm = dbconf['schema']
if refresh:
refresh_test_database(dbname)
pglib.initialise(dbconf, False)
pglib.make_db_connection(dbparm)
# Set a fictitious username for use by the databse trigger functions
with pglib.get_conn() as conn:
with conn.cursor() as curs:
curs.execute("set ppdb.username to 'tester';")
print(f"\nTesting {classname} with PostgreSQL database {dbname}")
def start_ppd(test_pg_pool=False, test_db=True, refresh=False):
""" Start PPD using the CherryPy web server
- Optionally modify the configuration from config to use the test database.
- Optionally refresh the (test) database
"""
global PPD_PID
dbname = ""
if test_db and refresh:
dbname = config.ppdb_conf['pgsql_conf']['database']
if dbname[-5:] != '_test':
dbname = f"{dbname}_test"
refresh_test_database(dbname)
# Start PPD with the specified engine
ppdargs = ('/usr/bin/python', os.path.join(test_root, 'ppd-test.py'),
str(test_pg_pool), str(test_db), str(TEST_SOCKETPORT))
PPD_PID = subprocess.Popen(ppdargs)
return dbname
def stop_ppd():
""" Stop PPD """
PPD_PID.terminate()
def refresh_test_database(dbname):
""" Refresh the PostgreSQL test database 'dbname' """
# Run the refresh module
print("Refreshing the PostgreSQL test database")
build = dbbuild.DbBuild(test=True, quiet=True)
build.connect()
build.build(config.ppdb_conf['pgsql_conf']['schema'],
os.path.join(test_root, dbname + '.tar.gz'))
def start_browser(browser):
"""
Open the browser, set the window size so the form is
visible and set the base URL.
If Firefox hangs with a 'WebDriverException: Message: "Can't load the profile. ...'
then check the python-selenium package.
Return a handle to Firefox
"""
if browser in ('Firefox', 'firefox'):
driver = webdriver.Firefox()
driver.set_window_position(550, 150)
elif browser == 'ffnohead':
print("Starting Firefox headless")
options = webdriver.FirefoxOptions()
options.headless = True
driver = webdriver.Firefox(options=options)
#~ driver.set_window_position(550, 150)
elif browser in ('Chrome', 'chrome'):
options = webdriver.ChromeOptions()
#~ options.add_argument('--ignore-certificate-errors')
driver = webdriver.Chrome(options=options)
driver.set_window_position(500, 100)
elif browser == 'ffnohead':
print("Starting Firefox headless")
options = webdriver.FirefoxOptions()
options.headless = True
driver = webdriver.Firefox(options=options)
else:
print(f"Unsupported browser: '{browser}'")
return None
#~ # Window is sized in proportion to a 1920x1080 diplay so that it should fit a
#~ # 1366x768 screen with the fonts scaled in proportion
driver.set_window_size(1200, 800)
#~ driver.set_window_size(1500, 850)
return driver
def setup_class(cls, test_name, browser, test_db=True, refresh=False):
""" Setup the Pytest test class with either the test or the full database """
print(f"\nRunning file {os.path.split(inspect.stack()[1].filename)[1]}")
if browser:
print(f"\n{util.isots_now()}: Testing {test_name} with PostgreSQL and {browser}")
cls.host = f"http://{config.ppdb_conf['cpsrv_conf']['sockethost']}:{TEST_SOCKETPORT}"
cls.dbname = start_ppd(test_db=test_db, refresh=refresh)
print(f"Running tests against {cls.host}")
if browser:
cls.driver = start_browser(browser)
time.sleep(1)
def teardown_class(cls, keep=False):
""" Tear down the class - stop ppd and the browser """
stop_ppd()
keep = bool(cls._test_reports)
if not keep:
print("closing browser")
cls.driver.quit()
print(f"{util.isots_now()}: Tests done")
def authenticate(host, body):
""" Make a JSON request to authenticate the user using the credentials in 'body' """
req_hdrs = {"Content-type": "application/json", "Accept": "application/json"}
sess = Session()
sess.headers.update(req_hdrs)
print(f"host: {host}")
resp = sess.post(f"{host}/login", json=body)
print("Closing Session")
sess.close()
if resp.status_code != 200:
raise Fail(f"Incorrect status {resp.status_code} from authentication")
def make_request(host, method, path, status, body=None, ctype='json', auth=True):
""" Make a request to the server, each request to a new connection.
This function uses content type application/json to authenticate
The actual request may be either json, html or zip according to the ctype parameter
The 'body' parameter must be a dict, not json
'resp' is a response objetc or a dict if ctype = json (for backward compatibility).
"""
if not body:
body = {}
auth_hdrs = {"Content-type": "application/json", "Accept": "application/json"}
if ctype == 'json':
req_hdrs = auth_hdrs
elif ctype == 'html':
req_hdrs = {"Content-type": "application/x-www-form-urlencoded",
"Accept": "text/html"}
else:
req_hdrs = {"Content-type": "application/x-www-form-urlencoded",
"Accept": "application/zip"}
req_url = f"{host}{path}"
sess = Session()
if auth:
sess.headers.update(auth_hdrs)
resp = sess.post(f"{host}/login", json={"username": "admin",
"passwd": "admin#"})
sess.headers.update(req_hdrs)
resp = sess.request(method, req_url, json=body)
sess.close()
if resp.status_code != status:
if ctype == 'json':
msg = resp.json()
else:
msg = resp
raise Fail(f"Incorrect status from request {req_url}: {resp.status_code}-{msg}")
if ctype == 'json':
return resp.json()
return resp
def add_test_member(cls):
""" Add a test member to the database on today's date and initialise class variables """
body = {'mem_data': TEST_MEM_DATA, 'pers_data': TEST_PERS_DATA,
'pmt_data': TEST_PMT_DATA, 'privacy': TEST_PRIVACY_DATA}
resp = make_request(cls.host, 'POST', "/member", 201, body=body)
if resp['rcode'] != js_consts.XHR_CREATED:
raise Fail("Failed to add test member")
data = resp["data"]
cls.TEST_MEM_NO = data['new_mem']
cls.TEST_PERS_ID = str(data['new_pers'])
cls.TEST_LOGIN = data['username']
# Get the new data and compile some composite fields used by the tests
resp = make_request(cls.host, 'GET', "/test/person?pers_id=" + cls.TEST_PERS_ID, 200)
pers = resp["data"]
cls.TEST_PERS_INFO = pers['name']
cls.TEST_PERS_PHONE = memlib.make_phone_string(pers['phone'][0])
cls.TEST_PERS_EMAIL = memlib.make_email_string(pers['email'][0])
if pers['flocks']:
cls.TEST_PERS_FLOCKS = pers['flocks']
else:
cls.TEST_PERS_FLOCKS = 'Flock: None'
cls.TEST_PERS_DATES = {'created': pers['created'],
'changed': pers['last_changed']}
print(f"Added test member {data['new_mem']}, pripers {data['new_pers']}, "
f"login {data['username']}")
def make_member_expired(cls, member_no, expire):
""" Alter the member's dates by a year """
body = {'member_no': member_no}
if expire:
body['put_action'] = 'expire'
else:
body['put_action'] = 'unexpire'
make_request(cls.host, 'PUT', "/test/member", 200, body=body)
def make_member_current(cls, member_no):
""" Make a member's membership current """
body = {'member_no': member_no, 'put_action': 'renew'}
resp = make_request(cls.host, 'PUT', "/test/member", 200, body=body)
if resp['rcode'] not in (js_consts.XHR_UPDATED, js_consts.XHR_NOERR) :
raise Fail(f"Failed to make member {member_no} current: {resp}")
def add_test_email(cls, person_id, address):
""" Add test email for the person specified """
body = {'person_id': person_id, 'address': address,
'comment': 'Test email'}
resp = make_request(cls.host, 'POST', "/test/email", 201, body=body)
if resp['rcode'] != js_consts.XHR_CREATED:
raise Fail("Failed to add test test email")
print(f"Completed add test email for {cls.TEST_PERS_ID}")
def add_test_flock(cls, person_id, flock_name, prefix=None):
""" Add a test flock for 'person_id' and a tag prefix for the new flock"""
body = {'action': 'flock', 'owner_id': person_id, 'new_name': flock_name}
resp = make_request(cls.host, 'POST', "/flock", 201, body=body)
if resp['rcode'] != js_consts.XHR_CREATED:
raise Fail("Failed to add test flock")
flock_no = resp['data']['flk_no']
ts = resp['data']['ts']
if prefix:
body = {'flock_no': flock_no, 'prefix': prefix}
resp = make_request(cls.host, 'POST', "/flock/prefix", 201, body=body)
if resp['rcode'] != js_consts.XHR_CREATED:
raise Fail("Failed to add test flock prefix")
print(f"Added test flock {flock_no} - {flock_name} for person {person_id}")
return flock_no, ts[:19]
def add_new_flock_owner(cls, owner_data=None):
""" Add a test member to the database for the create flock function """
if not owner_data:
owner_data = {
'pmt_data': {'class_id': 1, 'pmt_method': 'Cheque', 'pmt_ref': ''},
'privacy': [1, 3],
'mem_data': {'address_1': 'The Cottage', 'address_2': 'Winding Lane',
'address_3': '', 'post_town': 'VLLAGE',
'county': 'Derbyshire', 'post_code': 'SK24 9HT', 'country': '',
'non_renewal': '', 'source': '', 'note': ''},
'pers_data': {'active': 1, 'title': 'Ms', 'initials': 'JS',
'forename': 'Jo', 'surname': 'Bloggs',
'phone': {'number': '01634 568 983', 'phone_type': 'Home',
'comment': ''},
'email': {'address': 'test02@gmail.com', 'comment': ''}}}
resp = make_request(cls.host, 'POST', "/member", 201, body=owner_data)
if resp['rcode'] != js_consts.XHR_CREATED:
raise Fail("Failed to add test member")
data = resp["data"]
cls.NF_MEM = data['new_mem']
cls.NF_PERS = data['new_pers']
print(f"Added new 'flock owner' member {data['new_mem']}, pripers {data['new_pers']}, "
f"username {data['username']}")
#~ return data
def add_inspection(cls, body):
""" Add an inspection record for the ear-tag specified
'body' is a dict with ear_tag, sex, dob, insp_date, flock_no and result
Return the inspection id
"""
resp = make_request(cls.host, 'POST', "/test/inspect", 201, body)
if resp['rcode'] != js_consts.XHR_NOERR:
raise Fail(resp['data'])
return resp['data']
def change_flock_owner(cls, flock_no, owner_id):
""" Change the owner of a flock """
body = {"flock_no": flock_no, "person_id": owner_id}
resp = make_request(cls.host, 'POST', "/test/flock", 201, body)
if resp['rcode'] != js_consts.XHR_CREATED:
raise Fail(resp['data'])
return resp['data']
def delete_inspection(cls, insp_id):
""" Delete inspection 'insp_id' """
body = {"insp_id": insp_id}
resp = make_request(cls.host, 'DELETE', "/test/inspect", 200, body)
if resp['rcode'] != js_consts.XHR_NOERR:
raise Fail(resp['data'])
def get_in_region(cls, region):
""" Return lists of the region's county_ids and cntry_codes """
url = urllib.parse.quote("/member/lookup/expand/" + region)
return make_request(cls.host, 'GET', url, 200)["data"]
def get_adjacent_counties(cls, county_id):
""" Return a list of counties adjaceent to county_id """
url = urllib.parse.quote("/member/lookup/adjacent/" + county_id)
return make_request(cls.host, 'GET', url, 200)["data"]
def get_js_areas(cls):
""" Return lists of the available areas with counties sorted the way Javascript does -
by code-point hex value instead of alphabetic sort oder
"""
return make_request(cls.host, 'GET', "/test/areas/jsareas", 200)['data']
def get_users_flock(cls, uname):
""" Return the data for the user 'uname' flock """
flock_data = make_request(cls.host, 'GET', f"/test/flock/user/{uname}", 200)["data"]
return flock_data
def set_cp_domainbase(domainbase):
""" Set domainbase to CherryPy's config """
cherrypy.config['domainbase'] = domainbase
def set_register_code(register_code, regn_no):
""" Set a sheep's register code for the sheep validator tests.
Required to test those codes for which parents of the appropriate code do not
yet exist.
"""
with pglib.get_conn() as conn:
with conn.cursor() as curs:
curs.execute("update sheep set register_code = ? where regn_no = ? "
"returning register_code;", (register_code, regn_no))
row = curs.fetchone()
if not row or row[0] != register_code:
raise Fail(f"Failed to set register code for {regn_no}")
return True
def set_tester_roles(cls, username, roles):
""" Set the user's roles. Called by the test module's generic setup function """
body = {'username': username, 'roles': roles}
make_request(cls.host, 'PUT', "/test/role", 200, body=body)
def iso2str(iso_date):
""" Return a forward slash separated date string from the iso date """
return f"{iso_date[8:10].lstrip('0')}/{iso_date[5:7].lstrip('0')}/{iso_date[0:4]}"
class SelWeb():
""" Class to contain custom methods """
TEST_MEM_NO = ""
TEST_PERS_INFO = ""
TEST_PERS_EMAIL = ""
TEST_PERS_DATES = {}
TEST_PERS_PHONE = ""
TEST_PERS_FLOCKS = ""
TEST_PERS_ID = 0
host = None
driver = None
def pause(self):
""" Pause for half a second """
time.sleep(0.5)
def authenticate(self, uname, passwd):
""" Authenticate with user 'uname' """
print(f"Logging in to {self.host} as user '{uname}'")
self.driver.get(self.host)
self.wait_for_title("SSS - Home")
self.pause()
self.click("sidenav-open-btn")
self.click("sidenav-login-btn")
self.wait_for_text("login-title", "Please log in to continue")
self.set_input_field("uname", uname, True)
self.set_input_field("passwd", passwd, True)
self.click("login_btn")
self.click("sidenav-open-btn")
self.wait_for_text("sidenav-logout-btn", f"Log Out User '{uname}'")
self.click("sidenav-close-btn")
def clear(self, elem):
"""
Used instead of the API clear() because for some reason that does not work
if the element uses onChange instead of onblur
Clear an element by clicking, sending CTRL+a to select the text and sending
DEL to delete the text.
"""
elem.click()
elem.send_keys(Keys.CONTROL + 'a')
elem.send_keys(Keys.DELETE)
def click(self, elem_id):
""" Check the element is visble, then click the element. """
if self.driver.find_element("id", elem_id).is_displayed():
script = f"document.getElementById('{elem_id}').click();"
self.driver.execute_script(script)
self.pause()
self.check_error_dialog()
else:
raise Fail(f"Element id {elem_id} is not visible")
def exists_element_by_id(self, elem_id):
""" Return True if the element exists in the DOM."""
elem = None
try:
elem = self.driver.find_element("id", elem_id)
if elem:
return True
return False
except NoSuchElementException:
return False
except Exception as exc:
raise Fail(f"Unexpected error checking for element {elem_id}") from exc
def exists_element_by_xpath(self, xpath):
""" Return the element if if the element exists in the DOM."""
elem = None
try:
elem = self.driver.find_element("xpath", xpath)
if elem:
return True
return False
except NoSuchElementException:
return False
except Exception as exc:
raise Fail(f"Unexpected error checking for element {xpath}") from exc
def scroll_top(self):
""" Scroll to the top """
self.driver.execute_script("document.getElementById('content').scrollTop = 0")
self.pause()
def scroll_bottom(self):
""" Scroll to the bottom """
jscript = ("var div = document.getElementById('content');"
"div.scrollTop = div.scrollHeight;")
self.driver.execute_script(jscript)
self.pause()
def scroll_into_view(self, cbody, id):
""" Scroll the element into view 'id' is the county/country/region id"""
jscript = (f"let cbody = document.getElementById('{cbody}'); \
let rows = cbody.rows; \
for (let i = 0; i < rows.length; i++) {{ \
let cells = rows[i].cells; \
if (cells[0].innerHTML === {id}) {{ \
rows[i].scrollIntoView(); \
break; }}}} ")
self.driver.execute_script(jscript)
self.pause()
def get_active_element_id(self):
""" Return the id of the element which has focus """
return self.driver.execute_script("return document.activeElement.getAttribute('id')")
def get_attribute(self, elem_id, attrib):
""" Get and return an arbitrary attribute of an element """
return self.driver.find_element("id", elem_id).get_attribute(attrib)
def get_checked_state(self, btn_id):
""" Return the boolean 'checked' attribute of a radio-button """
elem = self.driver.find_element("id", btn_id)
return bool(elem.get_attribute("checked"))
def get_class(self, elem_id):
""" Get and return an element's class """
return self.driver.find_element("id", elem_id).get_attribute("class")
def get_css_value(self, fld_id, rule_name):
""" Return the value of the CSS rule 'rule-name' """
return self.driver.find_element("id", fld_id).value_of_css_property(rule_name)
def get_field_value(self, fld_id):
""" Get and return an input field's value """
elem = self.driver.find_element("id", fld_id)
return elem.get_attribute("value")
def get_hidden_element_text(self, elem):
""" Return the innerHTML (text) from a hidden element """
script = f"return {elem}.innerHTML;"
return self.driver.execute_script(script)
def get_state(self, elem_id):
""" Return the enabled/disabled state ot the element elem_id."""
return self.driver.find_element("id", elem_id).is_enabled()
def get_text(self, elem_id):
""" Get and return an element's text.
- 'truthy' values are returned 'true'/'false' for non-W3C browsers (Chrome)
and the actual value for W3C browsers (FF). Fold them to 'true'/false' here.
- ampersands are returned as HTML entities.
"""
elem = self.driver.find_element("id", elem_id)
text = elem.get_attribute("innerHTML").lstrip()
if text in ("True", "False"):
return text.lower()
return text.replace("&", "&")
def js_set_input_field(self, elem_id, value, leave=False):
""" Use Javascript to set an input field value """
script = f"document.getElementById('{elem_id}').value = '{value}';"
self.driver.execute_script(script)
if leave:
self.driver.find_element("id", elem_id).send_keys(Keys.TAB)
def set_input_field(self, fldid, value, leave=False):
""" Set a value to an input field with id=fldid.
Tab away from the field if 'leave' is True.
Return a handle to the input field
"""
elem = self.driver.find_element("id", fldid)
elem.send_keys(Keys.CONTROL + "a")
elem.send_keys(Keys.DELETE)
if value:
elem.send_keys(value)
if leave:
elem.send_keys(Keys.TAB)
self.check_error_dialog()
return elem
def wait_for_attribute(self, elem_id, attribute, textstr):
""" Poll the 'element' object's 'attribute' every 100ms looking for
'textstr' to match the element's attribute.
Note that an attribute which has been set to 'false', ie removed from the
element will return None
"""
elem = self.driver.find_element("id", elem_id)
actual = ''
for _ in range(WAIT):
try:
actual = elem.get_attribute(attribute)
if textstr == actual:
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
time.sleep(0.1)
else:
raise Fail(f"Timeout: '{textstr}' != '{actual}'")
def wait_for_checked(self, elem_id, checked):
""" Wait for element 'elem_id' to be checked/unchecked """
elem = self.driver.find_element("id", elem_id)
for _ in range(WAIT):
try:
state = elem.get_attribute('checked')
if state is None:
if checked is False:
break
else:
if checked is True:
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
time.sleep(0.1)
else:
raise Fail(f"Timeout: element checked not '{checked}'")
def wait_for_class(self, elem_id, test_class, class_in):
""" Poll the 'element' object's class every 100ms looking for
'test_class' to be in the class if class_in is True, otherwise not in the class.
"""
actual = ''
elem = self.driver.find_element("id", elem_id)
for _ in range(WAIT):
try:
actual = elem.get_attribute('class')
if class_in:
if test_class in actual:
break
else:
if test_class not in actual:
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
time.sleep(0.1)
else:
raise Fail(f"Timeout: class: '{actual}', test_class: '{test_class}'")
def wait_for_element_by_id(self, elem_id):
""" Wait for the element with id='elem_id' to be accessible in the DOM """
elem = None
for _ in range(WAIT):
try:
elem = self.driver.find_element("id", elem_id)
if elem:
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
time.sleep(0.1)
else:
raise Fail(f"Timeout: Element id '{elem_id}' is not accessible")
return elem
def wait_for_element_by_name(self, elem_name):
""" Wait for the element with named='elem_name' to be accessible in the DOM """
elem = None
for _ in range(WAIT):
try:
elem = self.driver.find_element("name", elem_name)
if elem:
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
time.sleep(0.1)
else:
raise Fail(f"Timeout: Element named '{elem_name}' is not accessible")
return elem
def wait_for_element_by_selector(self, selector):
""" Wait for the element with CSS selector 'selector to be accessible in the DOM """
elem = None
for _ in range(WAIT):
try:
elem = self.driver.find_element("css selector", selector)
if elem:
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
time.sleep(0.1)
else:
raise Fail(f"Timeout: Element with selector '{selector}' is not accessible")
return elem
def wait_for_element_by_xpath(self, xpath):
""" Wait for the element with xpath 'xpath' to be accessible in the DOM """
elem = None
for _ in range(WAIT):
try:
elem = self.driver.find_element("xpath", xpath)
if elem:
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
time.sleep(0.1)
else:
raise Fail("Timeout: Element xpath '{xpath}' is not accessible")
return elem
def wait_for_hidden_text_by_id(self, elem_id, textstr):
""" Wait for the text of a hidden element to be set. Has to use JavaScript
because WebDriver deliberately doesn't allow hidden elements to be read.
"""
script = f"return document.getElementById('{elem_id}').innerHTML;"
actual = ''
for _ in range(WAIT):
try:
actual = self.driver.execute_script(script)
if textstr == actual:
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
time.sleep(0.1)
else:
raise Fail(f"Timeout: '{textstr}' != '{actual}'")
def wait_for_hidden_value_by_id(self, elem_id, textstr):
""" Wait for the text of a hidden element to be set. Has to use JavaScript
because WebDriver deliberately doesn't allow hidden elements to be read.
"""
script = f"return document.getElementById('{elem_id}').value;"
actual = ''
for _ in range(WAIT):
try:
actual = self.driver.execute_script(script)
if textstr == actual:
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
time.sleep(0.1)
else:
raise Fail(f"Timeout: '{textstr}' != '{actual}'")
def wait_for_link_text(self, linktext):
""" Wait for the link with 'link text to be accessible in the DOM """
elem = None
for _ in range(WAIT):
try:
elem = self.driver.find_element("link text", linktext)
if elem:
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
time.sleep(0.1)
else:
raise Fail(f"Timeout: Link with text '{linktext}' is not accessible")
return elem
def wait_for_options(self, select_id, textstr):
""" Poll the SELECT element every 100ms looking for
'textstr' to match the concatenated options.
"""
elem = Select(self.driver.find_element("id", select_id))
actual = ''
for _ in range(WAIT):
try:
actual = ",".join([o.text for o in elem.options])
if textstr == actual:
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
time.sleep(0.1)
else:
raise Fail(f"Timeout: '{textstr}' != '{actual}'")
def wait_for_options_in(self, select_id, textstr):
""" Poll the SELECT element every 100ms looking for items in 'textstr' to
match options.
"""
items = textstr.split(',')
opts = ''
elem = Select(self.driver.find_element("id", select_id))
for _ in range(WAIT):
try:
opts = [o.text for o in elem.options]
for item in items:
if item not in opts:
missing = item
break
else:
# All the items were in the options list
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
time.sleep(0.1)
else:
raise Fail(f"Timeout: '{missing}' not in '{opts}'")
def wait_for_tbody(self, tbody, count):
""" Poll the prefixes table waiting for the row count to match 'count'.
"""
for _ in range(WAIT):
try:
rows = tbody.find_elements("tag name", 'tr')
if len(rows) == count:
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
time.sleep(0.1)
else:
raise Fail(f"Timeout: '{len(rows)}' != '{count}'")
def wait_for_selected(self, elem_id, deselect=False):
""" Wait for element 'elem_id' to be selected or de-selected """
elem = self.driver.find_element("id", elem_id)
for _ in range(WAIT):
try:
if not deselect:
if elem.is_selected():
break
else:
if not elem.is_selected():
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
time.sleep(0.1)
else:
if deselect:
raise Fail(f"Timeout: element '{elem_id}' has not been de-selected")
raise Fail(f"Timeout: element '{elem_id}' has not been selected")
def wait_for_staleness(self, element):
""" Wait for an element to become stale - no longer in the DOM """
for _ in range(WAIT):
try:
# Calling any method forces a staleness check
element.is_enabled()
except StaleElementReferenceException:
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
else:
raise Fail("The element never became stale")
def wait_for_state(self, elem_id, enabled):
""" Wait for element 'elem_id' to be enabled/disabled """
elem = self.driver.find_element("id", elem_id)
for _ in range(WAIT):
try:
state = elem.get_attribute('disabled')
if state is None:
if enabled is True:
break
else:
if enabled is False:
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
time.sleep(0.1)
else:
raise Fail(f"Timeout: element state not '{enabled}'")
def wait_for_text(self, elem_id, textstr):
""" Poll the 'element' object's text every 100ms looking for
'textstr' to match the element's text.
"""
actual = ''
for _ in range(WAIT):
try:
actual = self.get_text(elem_id)
if textstr == actual:
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
time.sleep(0.1)
else:
raise Fail(f"Timeout: '{textstr}' != '{actual}'")
def wait_for_text_in(self, elem_id, textstr):
""" Poll the 'element' object's text every 100ms looking for
'textstr' within the element's text.
"""
actual = ''
elem = self.driver.find_element("id", elem_id)
for _ in range(WAIT):
try:
actual = elem.text
if textstr in actual:
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
time.sleep(0.1)
else:
raise Fail(f"Timeout: '{textstr}' not in '{actual}'")
def wait_for_text_starts_with(self, elem_id, textstr):
""" Poll the 'element' object's text every 100ms looking for
'textstr' to match the element's text.
"""
actual = ''
elem = self.driver.find_element("id", elem_id)
for _ in range(WAIT):
try:
actual = elem.text
if actual.startswith(textstr):
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
time.sleep(0.1)
else:
raise Fail(f"Timeout: '{actual}' does not start with '{textstr}'")
def wait_for_text_with_html(self, elem_id, textstr):
""" Poll the 'element' object's text, which is assumed to contain HTML every 100ms
looking for 'textstr' to match the element's text.
"""
actual = ''
elem = self.driver.find_element("id", elem_id)
for _ in range(WAIT):
try:
actual = elem.text
if textstr == actual:
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
time.sleep(0.1)
else:
raise Fail(f"Timeout: '{textstr}' != '{actual}'")
def wait_for_title(self, textstr):
""" Poll the driver.title object's text every 100ms looking for
'textstr' to match the title.
"""
for _ in range(WAIT):
title = self.driver.title
try:
if title == textstr:
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
time.sleep(0.1)
else:
raise Fail("Timeout: '{textstr}' != '{title}'")
def wait_for_value(self, elem_id, textstr):
""" Poll the 'element' object's value every 100ms looking for
'textstr' to match the element's 'value' attribute.
"""
actual = ''
for _ in range(WAIT):
try:
jscript = f"return document.getElementById(\'{elem_id}\').value;"
actual = self.driver.execute_script(jscript)
#~ actual = element.get_attribute("value")
if textstr == actual:
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
time.sleep(0.1)
else:
raise Fail(f"Timeout: '{textstr}' != '{actual}'")
def wait_for_visibility(self, elem_id, visible):
""" Wait for element 'elem_id' to be visible or not """
elem = self.driver.find_element("id", elem_id)
for _ in range(WAIT):
try:
if visible:
if elem.is_displayed():
break
else:
if not elem.is_displayed():
break
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
time.sleep(0.1)
else:
raise Fail(f"Timeout: element '{elem_id}' display does not match '{visible}'")
def check_buttons(self, buttons):
""" Check the existence and state of the buttons. 'buttons' is a list of lists of
button id, exists, state
"""
for btn in buttons:
assert self.exists_element_by_id(btn[0]) == btn[1], \
f"{btn[0]} exists not {btn[1]}"
if btn[1]:
assert self.get_state(btn[0]) == btn[2], f"{btn[0]} state not {btn[2]}"
def check_checkboxes(self, fields, disabled):
""" Check the checkbox options checkboxes """
for fld, value in fields.items():
if value:
assert self.get_attribute(fld, "checked") == 'true'
else:
assert self.get_attribute(fld, "checked") is None
if disabled:
assert self.get_attribute(fld, "disabled") == 'true'
else:
assert self.get_attribute(fld, "disabled") is None
def check_display_fields(self, fields):
""" Check the display-only fields, removing superfluous spaces. """
for fld, value in fields.items():
# This does NOT remove any ' ' characters
assert value[0] == ' '.join(self.get_text(fld).split())
if value[1]:
assert value[1] in self.get_class(fld)
def check_hidden_fields(self, fields):
""" Check the hidden fields """
for fld, value in fields.items():
assert self.get_text(fld) == value
assert self.get_class(fld) == "nodisplay"
def check_info_dialog(self, expected):
""" Check that the dialog is visible with the expected text, and can be dismissed """
self.wait_for_visibility("info_dialog", True)
assert self.get_text("info_dlg_text") == expected
self.click("info_dialog_close")
self.wait_for_visibility("info_dialog", False)
def check_input_fields(self, fields, disabled, pfx=""):
""" Check the input and select data entry fields """
for fld, value in fields.items():
if disabled:
assert self.get_attribute(pfx + fld, "disabled") == 'true'
else:
assert self.get_attribute(pfx + fld, "disabled") is None
if value is not None:
assert self.get_field_value(pfx + fld) == value, f"{self.get_field_value(pfx + fld)} != {value}"
def check_item_lists(self, item, in_list_id, not_list_id):
""" Check that 'item' is in the 'in' list and not in the 'not' list """
sel_list = self.driver.find_element("id", in_list_id)
opts = sel_list.find_elements("tag name", 'option')
for opt in opts:
if item == opt.text:
break
else:
raise self.Fail(f"Item {item} not in the list")
sel_list = self.driver.find_element("id", not_list_id)
opts = sel_list.find_elements("tag name", 'option')
for opt in opts:
if item == opt.text:
raise self.Fail(f"Item {item} is in the list")
def check_messages(self, messages):
""" Check the message fields """
for fld, tests in messages.items():
assert tests[0] in self.get_class(fld)
assert tests[1] == self.get_text(fld)
def check_warning_dialog(self, expected):
""" Check that the dialog is visible with the expected text, and can be dismissed """
self.wait_for_visibility("warning_dialog", True)
assert self.get_text("warning_dlg_text") == expected, \
f"{self.get_text('warning_dlg_text')} != {expected}"
self.click("warning_dialog_close")
self.wait_for_visibility("yesno_dialog", False)
def check_warning_yesno_dialog(self, expected, btn_id, wcls=""):
""" Check that the warning yesno dialog is visible with the expected text,
and can be dismissed """
if btn_id in ('Yes', 'yes'):
btn_id = 'warning_yesno_dlg_yes'
elif btn_id in ('No','no'):
btn_id = 'warning_yesno_dlg_no'
self.wait_for_visibility("warning_yesno_dialog", True)
assert expected in self.get_text("warning_yesno_dlg_text")
if wcls:
assert wcls in self.get_class("warning_yesno_dlg_text")
self.click(btn_id)
self.wait_for_visibility("warning_yesno_dialog", False)
def check_yesno_dialog(self, expected, btn_id):
""" Check that the yesno dialog is visible with the expected text,
and can be dismissed """
if btn_id in ('Yes', 'yes'):
btn_id = 'yesno_dlg_yes'
elif btn_id in ('No','no'):
btn_id = 'yesno_dlg_no'
self.wait_for_visibility("yesno_dialog", True)
assert expected in self.get_text("yesno_dlg_text"), f"expected: {expected}, actual: \
{self.get_text('yesno_dlg_text')}"
self.click(btn_id)
self.wait_for_visibility("yesno_dialog", False)
def check_okcancel_dialog(self, expected, btn_id):
""" Check that the okcancel dialog is visible with the expected text,
and can be dismissed """
if btn_id in ('OK', 'ok'):
btn_id = 'okcancel_dlg_ok'
elif btn_id in ('No','no'):
btn_id = 'okcancel_dlg_cancel'
self.wait_for_visibility("okcancel_dialog", True)
assert expected in self.get_text("okcancel_dlg_text")
self.click(btn_id)
self.wait_for_visibility("okcancel_dialog", False)
def check_error_dialog(self, visible=False):
""" Check that the error dialog is not displayed """
self.pause()
err_dlg = self.driver.find_element("id", "error_dialog")
if visible:
if not err_dlg.is_displayed():
raise Fail("Error dialog is not visible")
else:
if err_dlg.is_displayed():
raise Fail("Error dialog is visible!")
def check_results_dialog(self, title, row, cell_text, buttons, cell=1):
""" Check that the dialog is visible with the expected text.
Check that the buttons are correctly displayed and click the specified button.
'btns' is a list of tuples of data for the left, centre, right and cancel buttons
"""
self.wait_for_visibility("results-view-dialog", True)
assert self.driver.find_element("id", "resultsview-title").text == title
cells = self.driver.find_elements("css selector", f'div.grid_cell[cell-row="{row}"]')
assert cells[cell].text == cell_text, f"{cells[cell].text} != {cell_text}"
for btn in buttons:
id = f"resultsview-dlg-{btn[0]}"
assert self.get_text(id) == btn[1], f"Text {self.get_text(id)} != {btn[1]}"
assert self.get_class(id) == btn[2], f"Class {self.get_class(id)} != {btn[2]}"
cells[cell].click()
for btn in buttons:
if btn[3]:
self.driver.find_element("id", f"resultsview-dlg-{btn[0]}").click()
self.wait_for_visibility("results-view-dialog", False)
#
# Checkbox Macros checking methods
#
def check_cbox_select_macro(self, cbody_id, data, id_pfx='', order=True):
""" Check that the checkbox options of a cbox_select macro is displaying all the
checkboxesand that they are in the correctly checked/unchecked state. if 'order'
is True then the items must be in the same order as 'data', otherwise any order
is allowed - this is to accomodate differences betwen Postgresql and Javascript
sort orders.
'data' is either an array of tuples: (option name, checked state)
"""
labels = [row[0] for row in data]
cbody = self.driver.find_element("id", cbody_id)
pfx = cbody.get_attribute("data-pfx")
rows = cbody.find_elements("css selector", 'tr')
assert len(rows) == len(data), f"{len(rows)} rows, {len(data)} data"
for rix, row in enumerate(cbody.find_elements("css selector", 'tr')):
cb_id = f"{id_pfx}{pfx}{rix}"
for _, _ in enumerate(row.find_elements("tag name", 'td')):
label = self.driver.find_element("css selector", 'label[for="'+ cb_id +'"]')
if order:
assert label.text == data[rix][0], f"{label.text} != {data[rix][0]}"
state = self.get_checked_state(cb_id)
assert state == data[rix][1], f"{cb_id} ({label.text}) state != {data[rix][1]}"
else:
try:
ix = labels.index(label.text)
except ValueError as e:
raise Fail(f"County {label.test} not in data")
assert self.get_checked_state(cb_id) == data[ix][1]
def check_cbox_select_id_macro(self, cbody_id, data, id_pfx='', order=True):
""" Check that the checkbox options table frame body is displaying all the checkboxes
and that they are in the correctly checked/unchecked state.
'data' is an array of tuples: (option id, checked state)
"""
cids = [row[0] for row in data]
cbody = self.driver.find_element("id", cbody_id)
pfx = cbody.get_attribute("data-pfx")
rows = cbody.find_elements("css selector", 'tr')
assert len(rows) == len(data), f"rows: {len(rows)} != data: {len(data)}"
for rix, row in enumerate(cbody.find_elements("css selector", 'tr')):
cb_id = f"{id_pfx}{pfx}{rix}"
elems = row.find_elements("tag name", 'td')
id = elems[0].get_attribute("innerHTML").lstrip()
cnty = elems[1].find_element("tag name", "label").get_attribute("innerHTML")
if order:
assert id == data[rix][0], f"{id} ({cnty}) != {data[rix][0]}, rix: {rix}"
assert self.get_checked_state(cb_id) == data[rix][1]
else:
try:
ix = cids.index(id)
except ValueError as e:
raise Fail(f"County id {id} not in data")
assert self.get_checked_state(cb_id) == data[ix][1]
def cbopts_get_cb_id(self, cbody_id, name):
""" Return the checkbox element corresponding to 'name' """
cbody = self.driver.find_element("id", cbody_id)
for _, row in enumerate(cbody.find_elements("css selector", 'tr')):
cells = row.find_elements("tag name", 'td')
if cells[1].text == name:
return cells[2].find_element("tag name", "input").get_attribute("id")
return ""
def cbopts_get_selected_ids(self, cbody_id, id_pfx=''):
""" Return a list of county ids from the cbox_select_id control """
id_list = []
cbody = self.driver.find_element("id", cbody_id)
pfx = cbody.get_attribute("data-pfx")
for rix, row in enumerate(cbody.find_elements("css selector", 'tr')):
cb_id = f"{id_pfx}{pfx}{rix}"
elems = row.find_elements("tag name", 'td')
if self.get_checked_state(cb_id) == True:
id_list.append(elems[0].get_attribute("innerHTML").lstrip())
return id_list
#
# Item selection lists methods
#
def get_selected_item(self, list_id):
""" Return the currently selected value """
sel_list = Select(self.driver.find_element("id", list_id))
return sel_list.first_selected_option.get_attribute("value")
def select_list_item(self, list_id, item):
""" Select the specified option """
sel_list = self.driver.find_element("id", list_id)
opts = sel_list.find_elements("tag name", 'option')
for opt in opts:
if item == opt.text:
opt.click()
break
time.sleep(0.2)
self.check_error_dialog()
def dblclick_list_item(self, list_id, item):
""" Select the specified option """
sel_list = self.driver.find_element("id", list_id)
opts = sel_list.find_elements("tag name", 'option')
for opt in opts:
if item == opt.text:
actions = ActionChains(self.driver)
actions.double_click(opt)
actions.perform()
break
time.sleep(0.2)
self.check_error_dialog()
def click_button(self, btns_id, btn_val):
""" Click the button specified by btns_is and btn_val (one of >, >>, <, <<) """
btns = {'>': '>', '>>': '>>', '<': '<', '<<': '<<'}
btn_html = btns[btn_val]
btns = self.driver.find_element("id", btns_id)
for btn in btns.find_elements("tag name", "button"):
if btn.get_attribute("innerHTML") == btn_html:
btn.click()
break
time.sleep(0.2)
self.check_error_dialog()
def check_not_selctl_items(self, listid, omitted):
""" Check that a select control item list does not contain the items in 'omitted'. """
sel_list = self.driver.find_element("id", listid)
opts = sel_list.find_elements("tag name", 'option')
for _, opt in enumerate(opts):
if opt.text in omitted:
raise Fail(f"Item {opt.text} is in list {listid}")
def check_selctl_items(self, listid, optcount, expected):
""" Check that a selection control item list has the expected items.
If optcount is -1 then make sure the number of options is the same as or more
then the number of expected values, otherwise check that the number of options
is the same as optcount.
"""
sel_list = self.driver.find_element("id", listid)
opts = sel_list.find_elements("tag name", 'option')
if optcount == -1:
cnt = len(expected)
if len(opts) < cnt:
raise self.Fail(f"Only {cnt} options in the {listid} list")
for i in range(cnt):
assert expected[i] == opts[i].text
else:
assert optcount == len(opts)
for i in range(optcount):
assert expected[i] == opts[i].text
def check_select_fields(self, fields):
""" Check the select field's value, text and state. """
for fld, tests in fields.items():
if tests[2] == 'disabled':
assert self.get_attribute(fld, "disabled") == "true"
else:
assert self.get_attribute(fld, "disabled") is None
assert tests[0] == self.get_field_value(fld)
select = Select(self.driver.find_element("id", fld))
assert tests[1] == select.first_selected_option.text
def check_selctl_btns(self, btns_id, btnvals):
""" Check the movement buttons of a selection control are enabled as expected. """
btns = self.driver.find_element("id", btns_id)
for btn in btns.find_elements("tag name", "input"):
if btn.get_attribute("value") == '>>':
assert btnvals[0] == btn.is_enabled()
elif btn.get_attribute("value") == '>':
assert btnvals[1] == btn.is_enabled()
elif btn.get_attribute("value") == '<':
assert btnvals[2] == btn.is_enabled()
elif btn.get_attribute("value") == '<<':
assert btnvals[3] == btn.is_enabled()
#
# Colour and markings dialogs
#
def colour_select(self, colour):
""" Select 'colour' from the colour dialog """
self.click('colour_sel_btn')
self.wait_for_visibility("colours_dialog", True)
self.driver.find_element("xpath", f"//td[text()='{colour}']").click()
self.pause()
self.click("colours_dlg_submit")
self.wait_for_visibility("colours_dialog", False)
def marking_select(self, markings):
""" Select/de-select markings from the markings dialog """
self.click('marking_sel_btn')
self.wait_for_visibility("markings_dialog", True)
for marking in markings:
self.driver.find_element("xpath", f"//td[text()='{marking}']").click()
self.pause()
self.click("markings_dlg_submit")
self.wait_for_visibility("markings_dialog", False)
def exists_dialog_marking(self, marking):
""" Open the Markings dialog and test for the presence (or not) of 'marking' """
self.click('marking_sel_btn')
self.wait_for_visibility("markings_dialog", True)
try:
elem = self.driver.find_element("xpath", f"//td[text()='{marking}']")
if elem:
exists = True
except NoSuchElementException:
exists = False
except Exception as e:
raise Fail(f"Unexpected error in {inspect.stack()[0][3]}: {e}") from e
self.click("markings_dlg_close")
self.wait_for_visibility("markings_dialog", False)
return exists