zavage-software/wabot
1
0
Fork 0
mirror of https://git.zavage.net/Zavage-Software/wabot.git synced 2026-07-22 13:36:08 -06:00
Code Issues Actions Packages Projects Releases Wiki Activity

build: replace setup.py with uv project, src layout

Browse Source
This commit is contained in:
Mathew Sir Guest the best 2026-07-07 21:23:55 -06:00
parent f3f6d23cc6
commit afa47186cc
11 changed files with 1124 additions and 1262 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
pyproject.toml Normal file
View File

@ -0,0 +1,39 @@
[project]
name = "wabot"
version = "0.2.0"
description = "Stateful Selenium browser automation with sessions that survive the Python process"
readme = "README.md"
authors = [{ name = "Mathew Guest", email = "t3h.zavage@gmail.com" }]
requires-python = ">=3.10"
dependencies = [
"selenium>=4.45,<5",
"platformdirs>=4",
"pillow>=10",
]
[dependency-groups]
dev = [
"pytest>=8",
"pytest-cov>=5",
"ruff>=0.8",
"sphinx>=7",
"sphinx-rtd-theme>=3",
]
[build-system]
requires = ["uv_build>=0.11,<0.12"]
build-backend = "uv_build"
[tool.pytest.ini_options]
addopts = '-m "not integration"'
markers = [
"integration: real-browser tests (run with: uv run pytest -m integration)",
]
testpaths = ["tests"]
[tool.ruff]
line-length = 100
src = ["src", "tests"]
[tool.ruff.lint]
select = ["E", "F", "W", "I", "UP", "B", "SIM"]

44
setup.py
View File

@ -1,44 +0,0 @@
#!/usr/bin/env python
#
# Usage:
#
# First, enable the python environment you want to install to, or if installing
# system-wide then ensure you're logged in with sufficient permissions
# (admin or root to install to system directories)
#
# installation:
#
# $ ./setup.py install
#
# de-installation:
#
# $ pip uninstall <app>
from setuptools import setup
__project__ = 'WaBoT'
__version__ = '0.1.0'
setup(
name = __project__,
version = __version__,
description = '',
author = 'Mathew Guest',
author_email = 't3h.zavage@gmail.com',
# Third-party dependencies; will be automatically installed
install_requires = (
'dill',
'selenium',
),
# Local packages to be installed (our packages)
packages = (
'wabot',
),
# Binaries/Executables to be installed to system
scripts=()
)

1
src/wabot/__init__.py Normal file
View File

@ -0,0 +1 @@
"""wabot — stateful Selenium browser automation with reattachable sessions."""

0
tests/integration/__init__.py Normal file
View File

0
tests/unit/__init__.py Normal file
View File

1084
uv.lock Normal file
View File

File diff suppressed because it is too large Load Diff

4
wabot/__init__.py
View File

@ -1,4 +0,0 @@
from .api import *
from .page import *
from .fields import *

251
wabot/api.py
View File

@ -1,251 +0,0 @@
from .create_browser import *
import appdirs
import logging
# import pickle
import dill as pickle
import selenium.common.exceptions
import selenium.webdriver
import sys
import time
import traceback
import os
USER_AGENT = "Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 6.0; Trident/5.0)"
REFUSE_AFTER_EXCEPTION = True
EXECUTOR_PORT = 4444
REMOTE_EXECUTOR = 'http://127.0.0.1:%s/wd/hub'
# PICKLE_FILENAME = '/tmp/nhsnwebdriverdump'
# PICKLE_FILENAME = os.path.join(
# appdirs.user_data_dir('wabot'),
# 'saved_browser_instances.pickle'
# )
LOGGER = logging.getLogger('wabot')
DEFAULT_WEBDRIVER_TYPE = 'firefox1'
class BrowserProxy:
def __init__(
self,
session_name='webdriver',
pickle_filename=None,
phantom=False,
webdriver_type=None # remote_chromium2
):
"""
BrowserProxy wraps a selenium webdriver instance and provides utility
functions for automation webpages.
"""
LOGGER.info('requesting selenium browser instance (%s): instance_name = %s', webdriver_type, session_name)
# if pickle_filename is None:
# pickle_filename = PICKLE_FILENAME
# self._pickle_filename = pickle_filename
if webdriver_type is None:
webdriver_type = 'firefox1'
assert webdriver_type in (
'firefox1',
'firefox2',
'chromium2',
'remote_chromium',
'remote_chrome',
'phantomjs',
'remote_firefox'
), 'webdriver_type must be firefox1, firefox2, chromium2, remote_chromium2, or phantomjs'
try:
self.driver_type = webdriver_type
if phantom:
pass
# driver_type = "phantomjs"
self.driver = self.get_driver(webdriver_type, session_name)
if not self.driver:
LOGGER.error('failed to get selenium webdriver')
self.good = False
return
except Exception as ex:
print('caught exception at BrowserProxy().__init__')
print(type(ex), ex)
raise
# self.page = nhsn_lo.pages.Login(self)
# self.good = True
def get_page(self):
return type(self.page)
def set_page(self, page):
LOGGER.trace('switching page to %s' % page)
newpage = page(self)
if hasattr(newpage, 'verify'):
rc = newpage.verify()
if not rc:
LOGGER.error('failed to verify page: %s' % page)
# input('failure')
return False
else:
LOGGER.trace('verified page: %s' % page)
self.page = newpage
return True
def switch_after(self, fn=None, page=None):
if not fn:
return lambda fn: self.switch_after(fn, page = page)
def wrapper(*args, **kwargs):
self.set_page(page)
return fn(*args, **kwargs)
return wrapper
def __getattr__(self, name):
"""
Access method on current page.
When object.method is referenced, python invokes
object.__getattribute__ which checks the instance
for a matching property, If none is found, python
invokes object.__getattr__, which is overridden here.
If the attribute is not found in the main class, this
will search the page object for the method. This
allows invoking page methods through the proxy class.
nhsn.foo() invokes nhsn.page.foo()
"""
# todo(mg) would this be a good spot to inject page verification?
# idea: maybe we could decorate method with error checking and
# error collection? method can return value, rc and this can
# queue actions and error check them as it goes? At the end,
# the user would be able to pull out the errors and check them
if REFUSE_AFTER_EXCEPTION and not self.good:
LOGGER.warn("broken state - refusing to invoke page action: %s" % (name))
return
try:
method = getattr(self.page, name)
return method
except AttributeError as ex:
raise
def __getitem__(self, key):
"""
Access element on the current page.
"""
return self.page[key]
def perform(self, meth, *args, **kwargs):
"""
Calls self.page.method and catches exceptions.
"""
# input('<breakpoint> at perform')
if REFUSE_AFTER_EXCEPTION and not self.good:
LOGGER.warn("broken state - refusing to invoke page action: %s" % (meth))
return
try:
x = getattr(self.page, meth)
except AttributeError as ex:
LOGGER.error("Failed to invoke page action. "\
"page = %s, method = %s" %\
(self.page.__class__, meth))
return None
# Hooks:
# x = self.page.detect_logged_out()
# y = self.page.detected_logged_in()
# print('logged out/in: ', x, y)
LOGGER.trace("invoking %s.%s" % (self.page.__class__.__name__, meth))
try:
return x(*args, **kwargs)
except selenium.common.exceptions.NoSuchWindowException:
self.good = False
except Exception as ex:
print("caught exception: %s" % (type(ex)))
exc_type, exc_value, exc_traceback = sys.exc_info()
lines = traceback.format_exception(exc_type, exc_value,
exc_traceback)
for line in lines:
print(line, end = "")
self.good = False
def reset_good_status(self):
self.good = True
def get_driver(self, browser, session_name='webdriver'):
"""
Returns selenium objects of different types
The combinations of selenium{1,2} {chrome,firefox} are supported.
In the case of selenium 1 (selenium server), the host is a constant
defined at the top of this file.
Args:
browser: The selenium webdriver requested.
If a chromium2 webdriver is requested, selenium
will try to use chromium as the browser. If
firefox2 is requested, selenium will try to use
firefox. Chromium is very fast but it is designed
to be fast for an interactive user which means it
is fast at the cost of processing power and
ram. Firefox is about as fast but uses less
resources.
Available browser drivers are:
chromium2, chromium1, firefox2, firefox1
The associated webdriver has to be installed
and runnable on the system.
Returns:
The selenium webdriver handle.
"""
LOGGER.debug('requesting selenium browser instance: type = %s' % (browser))
driver = None
browser_factory = CreateBrowser()
if browser == 'chromium2': # Selenium 2 - Chrome
driver = self._create_driver_chromium()
elif (
browser == 'remote_chromium'
or browser == 'remote_chrome'
):
driver = browser_factory._create_driver_remote_chromium(session_name)
elif browser == 'remote_firefox': # Selenium 1 - Firefox
driver = browser_factory._create_driver_remote_firefox(session_name)
# TODO(MG) need to redo/rename below. selenium2/ not remote, uses native browser
# webdrviers instead of javascript
elif browser == 'chromium1': # Selenium 1 - Chrome without working user agent switch
driver = self._create_driver_chromium1()
elif browser == 'firefox2': # Selenium 2 - Firefox
driver = self._create_driver_firefox2()
elif browser == 'phantomjs':
driver = self._create_driver_phantomjs()
else:
LOGGER.error(
'an attempt was made to request an '\
'unsupported (by this product) selenium '\
'webdriver; refusing. requested = %s'\
% (browser)
)
driver.implicitly_wait(10)
return driver

316
wabot/create_browser.py
View File

@ -1,316 +0,0 @@
import appdirs
import datetime
import logging
import os
import selenium
import pickle
LOGGER = logging.getLogger('wabot')
USER_AGENT = "Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 6.0; Trident/5.0)"
REFUSE_AFTER_EXCEPTION = True
EXECUTOR_PORT = 4444
REMOTE_EXECUTOR = 'http://127.0.0.1:%s/wd/hub'
class PickledBrowserReference:
"""
Structure for saving a webdriver instance. Also includes the timestamp
to help invalidate old references.
"""
def __init__(self):
self.browser_ref = None
self.timestamp = None
# Factory Creator - instantiate Selenium1 and Selenium2 webdriver instances.
class CreateBrowser:
"""
Creates and instantiates selenium webbrowser instances.
Two strategies: remote via selenium server or local.
The advantage of using remote is the browser instance can stay
open after the process closes. This allows you to disconnect
and re-connect to the same browser instance with different
processes.
"""
def __init__(self, pickle_filename=None):
self._pickle_filename = None
if pickle_filename is not None:
self.pickle_filename = pickle_filename
self._has_cleaned = False
@property
def pickle_filename(self):
"""
Reference to the user-data-dir pickle target filename, for
saving browser references.
"""
if self._pickle_filename is None:
self._pickle_filename = os.path.join(
appdirs.user_data_dir('wabot'),
'saved_browser_instances.pickle'
)
return self._pickle_filename
else:
return self._pickle_filename
@pickle_filename.setter
def pickle_filename(self, value):
self._pickle_filename = value
def _clean_old_pickles(self):
if self._has_cleaned is True:
LOGGER.debug('not cleaning pickles twice')
return
p = self.pickle_filename
try:
fp = open(p, 'rb')
pickles = pickle.load(fp)
if not pickles:
raise Exception
LOGGER.info('cleaning any old pickle refs')
for pickle_ref in pickles:
driver_ref = pickles[pickle_ref]
ts = driver_ref.timestamp
if datetime.datetime.now() - driver_ref.timestamp >= datetime.timedelta(days=3):
LOGGER.info('deleting old pickled driver ref: %s at %s', pickle_ref, ts)
pickles.pop(final_name)
self._has_cleaned = True
except (FileNotFoundError, IOError) as ex:
LOGGER.error('unable to load pickles: no pickled drivers found')
except Exception as ex:
print(ex)
pass
def _load_from_pickle(self, final_name):
p = self.pickle_filename
# Definitely no browser instance already, we must instantiate
if not os.path.exists(p):
LOGGER.debug('no pickled file for saved browser instances (nothing saved yet)')
return
# There MAY be an open browser or an invalidated reference to a once-open browser
if os.path.exists(p):
LOGGER.debug('found pickled file for saved browser instances: %s', p)
# First, see if existing session_name browser instance exists
fp = None
self._clean_old_pickles()
try:
fp = open(p, 'rb')
# drivers = pickle.load(fp)
pickles = pickle.load(fp)
if not pickles:
raise Exception
for idx, saved_instance in enumerate(pickles):
LOGGER.debug('found saved browser instances (%s): %s', idx, saved_instance)
driver_ref = pickles.get(final_name)
if not driver_ref:
raise Exception
# if datetime.datetime.now() - driver_ref.timestamp >= datetime.timedelta(days=3):
# pickles.pop(final_name)
driver = driver_ref.browser_ref
LOGGER.debug('connected to pickled webdriver instance: %s', final_name)
url = driver.current_url # throw error if driver isn't reliable anymore
LOGGER.info('webdriver instance is ready')
# self.driver = driver
return driver
except (FileNotFoundError, IOError) as ex:
LOGGER.error('unable to connect to existing webdriver: no pickled drivers found')
except Exception as ex:
print(ex)
self.driver = None
def _save_to_pickle(self, final_name, driver_ref):
p = self.pickle_filename
pickles = {}
# Definitely no browser instance already, we must instantiate
if not os.path.exists(p):
LOGGER.debug('no pickled file for saved browser instances (nothing saved yet)')
# Create usr-app-dir if doesn't exist
dirname = os.path.dirname(p)
if not os.path.isdir(dirname):
os.mkdir(dirname)
# There MAY be an open browser or an invalidated reference to a once-open browser
if os.path.exists(p):
LOGGER.debug('found existing pickle file for saved browser instances: %s', p)
# First, see if existing session_name browser instance exists
self._clean_old_pickles()
fp = open(p, 'rb')
pickles = pickle.load(fp)
if not pickles:
raise Exception
LOGGER.debug('found saved browser instances while saving: %s', list(pickles.keys()))
fp.close()
# Save to pickle - creating file if necessary
pkl = PickledBrowserReference()
pkl.browser_ref = driver_ref
pkl.timestamp = datetime.datetime.now()
pickles[final_name] = pkl
import pprint;
print('pickles:')
pprint.pprint(pickles)
# pickle file must be created
fp = open(p, 'wb')
# drivers[final_name] = driver
LOGGER.info('saving browser instance to pickle: %s', final_name)
pickle.dump(pickles, fp)
fp.close()
def _create_driver_remote_chromium(self, session_name):
"""
Creates and returns Selenium1 chromium instance.
"""
p = self.pickle_filename
# e.g.: remote-chromium-webdriver
final_name = '{}-{}'.format('remote-chromium', session_name)
driver = None
driver = self._load_from_pickle(final_name)
if driver is not None:
LOGGER.info('webdriver instance is ready (from pickle)')
return driver
# At this point, need to instantiate a new browser instance
sel_host = REMOTE_EXECUTOR % (EXECUTOR_PORT)
LOGGER.info('instantianting new browser instance (remote_chromium)')
LOGGER.info('remote selenium: %s', sel_host)
opt = selenium.webdriver.chrome.options.Options()
opt.add_argument("--user-agent=" + USER_AGENT)
opt.add_argument("--kiosk-printing")
opt.add_argument("--focus-existing-tab-on-open=false")
driver = selenium.webdriver.Remote(
command_executor=sel_host,
desired_capabilities = opt.to_capabilities()
)
self._save_to_pickle(final_name, driver)
# self.driver = driver
return driver
# Pickle impl. is duped here
def _create_driver_remote_firefox(self, session_name):
"""
Creates and returns Selenium1 firefox instance.
"""
final_name = '{}-{}'.format('remote-firefox', session_name)
driver = None
driver = self._load_from_pickle(final_name)
if driver is not None:
LOGGER.info('webdriver instance is ready (from pickle)')
return driver
# At this point, need to instantiate a new browser instance
sel_host = REMOTE_EXECUTOR % (EXECUTOR_PORT)
LOGGER.info('instantianting new browser instance (remote_firefox)')
LOGGER.info('remote selenium: %s', sel_host)
profile = selenium.webdriver.FirefoxProfile()
profile.set_preference("general.useragent.override", USER_AGENT)
driver = selenium.webdriver.Remote(
# SELENIUM1_SERVER_PATH,
sel_host,
selenium.webdriver.DesiredCapabilities.FIREFOX.copy(),
browser_profile = profile
)
self._save_to_pickle(final_name, driver)
return driver
def _create_driver_chromium2(self):
opt = selenium.webdriver.chrome.options.Options()
opt.add_argument("--user-agent=" + USER_AGENT)
opt.add_argument("--kiosk-printing")
driver = selenium.webdriver.Chrome(chrome_options = opt)
self.driver = driver
return driver
def _create_driver_chromium1(self):
# Selenium 1 - Chrome without working user agent switch
# These two methods of creation ChromeOptions are equivalent objects
options = selenium.webdriver.ChromeOptions()
options.add_argument("--user-agent=" + USER_AGENT)
driver = selenium.webdriver.Remote(desired_capabilities = options.to_capabilities())
driver = selenium.webdriver.Remote(SELENIUM1_SERVER_PATH,
selenium.webdriver.DesiredCapabilities.CHROME.copy())
return driver
def _create_driver_firefox2(self):
# tmp = selenium.webdriver.FirefoxProfile()
# tmp = None
profile = None
# filename = "/tmp/firefox_profile"
# try:
# fp = open(filename, "rb")
# profile = pickle.load(fp)
# except:
# pass
if not profile:
profile = selenium.webdriver.FirefoxProfile(profile_directory = "/home/mathew/firefox_prof")
profile.set_preference("general.useragent.override", USER_AGENT )
profile.set_preference("browser.helperApps.neverAsk.saveToDisk", "text/csv");
profile.set_preference("network.http.redirection-limit", "0" )
# profile.set_preference("javascript.enabled", False )
# profile.set_preference("print.always_print_silent", True)
profile.set_preference("print.print_to_file", True)
profile.set_preference("print.print_to_filename", "/tmp/print.pdf")
profile.update_preferences()
profile.set_preference("network.http.redirection-limit", "0" )
# with open("/tmp/firefox_profile", "wb") as fp:
# pickle.dump(profile, fp, pickle.HIGHEST_PROTOCOL)
# driver = selenium.webdriver.Firefox()
driver = selenium.webdriver.Firefox(profile)
return driver
def _create_driver_phantomjs(self):
# Note(MG): Selenium support for PhantomJS has been deprecated, please use headless
# driver = selenium.webdriver.PhantomJS()
# return driver
opt = selenium.webdriver.chrome.options.Options()
opt.add_argument("--user-agent=" + USER_AGENT)
opt.add_argument("--kiosk-printing")
opt.add_argument("--headless")
driver = selenium.webdriver.Chrome(chrome_options = opt)
driver.set_window_size(838, 907)
self.driver = driver
return driver

110
wabot/fields.py
View File

@ -1,110 +0,0 @@
import logging
import random
import selenium
import time
LOGGER = logging.getLogger('wabot')
class PageObject:
"""
Wrapper around page element that provides an object orientated interface
to elements. Can be sublcassed to integrate more complicated instruction.
"""
def __init__(self, page, accessors=None, name=None):
self.page = page
self.driver = page.driver
self.accessors = accessors
self.name = name
# self.el = None
def __getattr__(self, name):
# if not self.el:
# raise AttributeError
try:
return getattr(self.el, name)
except AttributeError as ex:
raise
def selenium_element(self, accessors=None):
"""
Creates and returns a selenium webelement for
interfacing with the page.
"""
if not accessors:
accessors = self.accessors
accessor = accessors
by = accessor[0]
value = accessor[1]
el = self.page.driver.find_element(by=by, value=value)
return el
def click(self):
return self.page.click(self.el)
def click_and_go(self):
time.sleep(random.uniform(3,6))
return self.page.click_and_go(self.el)
# def get_value(self):
# def set_value(self, value):
# def custom, e.g. zoom(self, x, y, dx, dy)
class TextField(PageObject):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
el = self.selenium_element(kwargs['accessors'])
self.el = el
def get_value(self):
return self.page.get_el_value(self.el)
def set_value(self, value):
LOGGER.info('[%s]->set_text (%s)' % (self.name, value))
time.sleep(random.uniform(3,5))
return self.page.set_el_value(self.el, value)
class SelectField(PageObject):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
el = self.selenium_element(kwargs['accessors'])
dropdown = selenium.webdriver.support.ui.Select(el)
self.el = el
self.dropdown = dropdown
def get_value(self):
return self.page.get_select_value(self.dropdown)
def set_value(self, value):
LOGGER.info('[%s]->set_select (%s)' % (self.name, value))
time.sleep(random.uniform(6,11))
return self.page.set_select_value(self.dropdown, value)
def select_by_index(self, idx):
return self.dropdown.select_by_index(idx)
class CheckField(PageObject):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
el = self.selenium_element(kwargs['accessors'])
self.el = el
def get_checked(self, ignore=False):
return self.page.get_checkbox_value(self.el, ignore)
def set_checked(self, checked):
LOGGER.info('[%s]->set_checked (%s)' % (self.name, checked))
time.sleep(random.uniform(2,3))
return self.page.set_checkbox(self.el, checked)
class NullField(PageObject):
def __init__(self, el, *args, **kwargs):
self.el = el
def __getattr__(self, attr):
raise Exception
def __bool__(self):
return False

537
wabot/page.py
View File

@ -1,537 +0,0 @@
import enum
import random
import selenium.webdriver
import time
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import selenium.common.exceptions
from .fields import *
import selenium.webdriver.support.ui
from selenium.webdriver.common.by import By
from PIL import Image
import os
import inspect
import logging
LOGGER = logging.getLogger('wabot')
ENABLE_GOTO_CLINIC_SELECT_OPTIMIZATION = True
ALERT_TIMEOUT = 3
class OBJ_T(enum.Enum):
ELEMENT = 1
ELEMENT_ARRAY = 2
SELECT = 3
CUSTOM = 4
class RC(enum.Enum):
FAILURE = 0
SUCCESS = 1
EVENT_WITHIN_21_DAYS = 2
DUPLICATE_EVENT = 3
NO_FACILITY = 4
MISSING_REPORTING_PLAN = 5
class Page:
"""
Provides ancillary utility methods such as finding
elements, interacting with forms, and taking screenshots.
"""
def __init__(self, parent):
self.parent = parent
self.driver = parent.driver
LOGGER.info("Loaded Page() '%s'", self.__class__.__name__)
def find_element_locators(self, key):
"""
Returns the generator data for an element, i.e. what
class to use and the arguments to construct it with.
This is found/stored in the static class definition,
page.elements = {key: (accessors)}
"""
class_hierarchy = inspect.getmro(type(self))
locators = None
for cls in class_hierarchy:
if not hasattr(cls, 'elements'): # no elements defined here...
continue
locators = cls.elements.get(key)
if not locators: # our element isn't defined here
continue
return locators # return first found match
def get_proxy(self, key):
locators = self.find_element_locators(key)
if not locators:
LOGGER.warn('element not found: %s' % (key))
return NullField(None, page=self)
obj_type = locators[0]
accessors = locators[1]
# if not el:
# LOGGER.warn('failed to make proxy object: %s' % key)
# return NullField(None, page=self)
if obj_type == 'el':
obj = TextField(page=self, accessors=accessors, name=key)
return obj
elif obj_type == 'select':
obj = SelectField(page=self, accessors=accessors, name=key)
return obj
elif obj_type == 'checkbox':
obj = CheckField(page=self, accessors=accessors, name=key)
return obj
elif obj_type == 'els':
el = self.find_element(key)
return self.find_element(key)
elif isinstance(obj_type, type):
# by = accessors[0]
# value = accessors[1]
# el = self.driver.find_element(by=by, value=value)
obj = obj_type(idelement=accessors[1], page=self, accessors=accessors,
name=key)
return obj
else:
LOGGER.error('failed to create page element type: %s' % (obj_type))
LOGGER.error('requested unknown page element: %s' % key)
return
def __getitem__(self, key):
# input('__getitem__ %s' % key)
# return self.find_element(key) # DEPRECATED
return self.get_proxy(key)
def find_element(self, key):
"""
Accesses element on page
"""
locators = self.find_element_locators(key)
if not locators:
LOGGER.error('failed to find page element: %s' % (key))
return False
try:
it = iter(locators)
type_ = next(it)
if type_ == 'custom':
cls = next(it)
locators = []
for locator in it:
locators.append(locator)
obj = cls(self, OBJ_T.CUSTOM, locators)
# print(obj)
return obj
for locator in it:
# print('locator =', locator)
if type_ == 'el' or type_ == 'checkbox':
try:
by = locator[0]
value = locator[1]
el = self.driver.find_element(by=by, value=value)
LOGGER.trace('found single element (%s) by %s = %s', key, by, value)
return el
except Exception as ex:
LOGGER.warn('failed to find single element (%s) by %s = %s', key, by, value)
continue
elif type_ == 'els':
try:
by = locator[0]
value = locator[1]
els = self.driver.find_elements(by=by, value=value)
LOGGER.trace('found elements (%s) by %s = %s', key, by, value)
return els
except Exception as ex:
LOGGER.warn('failed to find any elements (%s) by %s = %s', key, by, value)
continue
elif type_ == 'select':
try:
by = locator[0]
value = locator[1]
el = self.driver.find_element(by=by, value=value)
LOGGER.trace('found dropdown element (%s) by %s = %s', key, by, value)
dropdown = selenium.webdriver.support.ui.Select(el)
return dropdown
except Exception as ex:
LOGGER.warn('failed to find dropdown (%s) by %s = %s', key, by, value)
continue
else:
LOGGER.error('unable to find element (%s): unknown type = %s' % (key, type_))
return
except Exception as ex:
print(ex)
def verify(self):
# The page proxy api calls PageObject.verify() if it exists, which
# should call super().verify(). It is necessary for every class in
# the hierarchy to have a callable verify(). Every class in the
# hierarchy should implement verify method or subclass from here.
return True
def click_and_go(self, el):
rc = self.click(el)
if not rc:
return rc
self.accept_alert()
LOGGER.debug('waiting for page to load...')
rc = self._wait_for_element_to_go_stale(el)
if not rc:
LOGGER.error('failed: timed out waiting for page load')
return False
return rc
def _wait_for_element_to_go_stale(self, el):
try:
wait = selenium.webdriver.support.ui.WebDriverWait(self.driver, 10)
wait.until(lambda driver: self.is_element_stale(el))
return True
except selenium.common.exceptions.TimeoutException as ex:
LOGGER.error('failed: timed out waiting for page load')
return False
def click(self, el):
"""
Clicks element at a random coordinate based on the size of the
element.
By default, selenium clicks elements at pos(0, 0). NHSN records
where things are clicked at.
"""
if not el:
LOGGER.warn("refusing to click null element")
return False
try:
size = el.size
except selenium.common.exceptions.StaleElementReferenceException as ex:
LOGGER.error('failed to click element: stale reference')
return False
# Use a guassian distribution to click more often towards the center
width = size["width"]
if width > 4:
x = random.gauss((width/2), (width/7))
if x < 0: x = 1
elif x > width: x = width - 1
else:
i = 0
while i < 10:
try:
if el.is_displayed() and el.is_enabled():
el.click()
return True
except selenium.common.exceptions.ElementNotVisibleException as ex:
LOGGER.error('failed to click element: not visible')
return False
except selenium.common.exceptions.StaleElementReferenceException as ex:
LOGGER.error('failed to click element: stale reference')
return False
time.sleep(.2)
i = i+1
return False
height = size["height"]
if height > 4:
y = random.gauss((height/2), (height/7))
if y < 0: y = 1
elif y > height: y = height -1
else:
el.click()
return
LOGGER.debug( "clicking %s (dim: x = %s, y = %s) at %d, %d" %
(self.get_el_identifier(el), size["width"], size["height"], x,y) )
i = 0
n = 20
while i < n:
if el.is_displayed() and el.is_enabled():
break
if i == n:
raise Exception("unable to click element")
time.sleep(.5)
i = i + 1
# return el.click()
x = int(x)
y = int(y)
# print('size is', width, height)
# print('clicking through selenium actions at', x, y)
actions = selenium.webdriver.ActionChains(self.driver)
actions.move_to_element_with_offset(el, x, y)
actions.click()
try:
actions.perform()
except Exception as ex: # type is selenium timeout... not sure
print(ex)
LOGGER.error("%s" % (ex))
return False
return True
def save_screenshot(self, filename):
LOGGER.info("saving page screenshot: %s" % (filename))
# Chromium2 screenshot only captures viewable area,
# selenium is waiting on WebDriver which is waiting
# on chromium. Doesn't look like it will be fixed soon.
# more info:
# https://code.google.com/p/chromedriver/issues/detail?id=294
#
# For now, the workaround is stitch it together for chromium.
# Firefox2 works fine but has it's own problems, hence the
# chromium stitch.
if self.parent.driver_type == "chromium2":
self.chrome_take_full_page_screenshot(filename)
else:
self.driver.get_screenshot_as_file(filename)
def chrome_take_full_page_screenshot(self, file):
self.driver.maximize_window()
# Global.scroll_to_zero()
time.sleep(0.2)
# Log.Debug("Starting chrome full page screenshot workaround ...")
total_width = self.driver.execute_script("return document.body.offsetWidth")
total_height = self.driver.execute_script("return document.body.parentNode.scrollHeight")
viewport_width = self.driver.execute_script("return document.body.clientWidth")
viewport_height = self.driver.execute_script("return window.innerHeight")
# Log.Debug("Total: ({0}, {1}), Viewport: ({2},{3})".format(total_width, total_height,viewport_width,viewport_height))
rectangles = []
i = 0
while i < total_height:
ii = 0
top_height = i + viewport_height
if top_height > total_height:
top_height = total_height
while ii < total_width:
top_width = ii + viewport_width
if top_width > total_width:
top_width = total_width
# Log.Debug("Appending rectangle ({0},{1},{2},{3})".format(ii,i,top_width,top_height))
rectangles.append((ii,i,top_width,top_height))
ii = ii + viewport_width
i = i + viewport_height
stitched_image = Image.new('RGB', (total_width, total_height))
previous = None
part = 0
for rectangle in rectangles:
if not previous is None:
self.driver.execute_script("window.scrollTo({0}, {1})".format(rectangle[0], rectangle[1]))
# Log.Debug("Scrolled To ({0},{1})".format(rectangle[0], rectangle[1]))
time.sleep(0.2)
tmp_path = "/tmp/"
file_name = "{0}scroll_{1}_part_{2}.png".format(tmp_path, 1, part)
# file_name = "/tmp/screen.png"
# Log.Debug("Capturing {0} ...".format(file_name))
self.driver.get_screenshot_as_file(file_name)
screenshot = Image.open(file_name)
# offset = (rectangle[0], rectangle[1])
if rectangle[1] + viewport_height > total_height:
offset = (rectangle[0], total_height - viewport_height)
else:
offset = (rectangle[0], rectangle[1])
# Log.Debug("Adding to stitched image with offset ({0}, {1})".format(offset[0],offset[1]))
stitched_image.paste(screenshot, offset)
del screenshot
os.remove(file_name)
part = part + 1
previous = rectangle
stitched_image.save(file)
# Log.Debug("Finishing chrome full page screenshot workaround ...")
return True
def get_el_value(self, el):
if not el:
return None
val = el.get_attribute("value")
# LOGGER.debug('peeked at field <%s>, value = %s'
# % (self.get_el_identifier(el), val))
return val
def get_el_text(self, el):
if not el:
return None
return el.text
def set_el_value(self, el, value, js=False, slow_type=False):
if not el:
return False
prev_val = self.get_el_value(el)
if value is None:
el.clear()
return
el.clear()
if js:
# doesn't work
rc = self.driver.execute_script("arguments[0].setAttribute('value', '%s');arguments[0].onchange();" % (value), el)
else:
if slow_type:
try:
for ch in value:
el.send_keys(ch)
time.sleep(1)
except Exception as ex:
LOGGER.error("failed to send keys, element in unknown state!!: %s" % (ex))
return False
else:
try:
el.send_keys(value)
except Exception as ex:
LOGGER.error("failed to send keys, element in unknown state!!: %s" % (ex))
return False
el_value = self.get_el_value(el)
if str(el_value) != str(value):
print(type(el_value), type(value))
print("values didn't match.", value, el_value)
return False
# LOGGER.info('set field <%s> -> %s; previous = %s'
# % (self.get_el_identifier(el), el_value, prev_val))
return True
def get_select_value(self, select):
if not select:
LOGGER.error("tried to get select value of NULL element")
return None
try:
value = select.first_selected_option.get_attribute("value")
except selenium.common.exceptions.NoSuchElementException as ex:
value = None
return value
def set_select_value(self, select, value=None, text=None):
if not select:
return False
# LOGGER.trace("setting select value (%s) for (%s)"\
# % (value, self.get_el_identifier(select._el)))
if value:
try:
select.select_by_value(str(value))
return True
except Exception as ex:
return False
elif text:
select.select_by_visible_text(text)
return False
def set_checkbox(self, el, checked):
if not el:
return False
is_enabled = el.is_enabled()
is_selected = el.is_selected()
# print("enabled, selected, check:", is_enabled, is_selected, checked)
if not is_enabled:
return False
if checked:
if not is_selected:
# LOGGER.trace("checking unchecked box (%s)" % (self.get_el_identifier(el)))
self.click(el)
else:
if is_selected:
# LOGGER.trace("unchecking checked box (%s)" % (self.get_el_identifier(el)))
self.click(el)
val = self.get_checkbox_value(el)
if val != checked:
pass
# False != None
# print('checkbox not checked properly?', val, checked)
# input('checkbox not checked properly?')
# todo(mathew guest) assert checked
return True
def get_checkbox_value(self, el, ignore_disabled=False):
"""
Returns the checked status of a checkbox element. True if enabled
and checked, False if disabled or unchecked.
"""
if not el:
return None
return (ignore_disabled or el.is_enabled()) and el.is_selected()
def get_el_identifier(self, el):
"""
Returns a quick identifier to refer an element by.
"""
id_ = el.get_attribute("id")
name = el.get_attribute("name")
class_ = el.get_attribute("style")
if id_:
return id_
if name:
return name
if class_:
return class_
def accept_alert(self, accept=True, timeout=ALERT_TIMEOUT):
"""
Looks for and tries to accept a javascript alert if it exists.
returns bool: whether or not an alert was accepted.
There is a timeout penalty when calling this method if there is no alert.
"""
try:
selenium.webdriver.support.ui.WebDriverWait(self.driver, timeout).\
until(selenium.webdriver.support.expected_conditions.alert_is_present(),
'Timed out waiting alert' )
alert = self.driver.switch_to_alert()
text = alert.text
if accept:
alert.accept()
else:
alert.dismiss()
LOGGER.trace('caught js alert: %s' % text)
return text
except selenium.common.exceptions.TimeoutException:
LOGGER.trace('no js alert present')
return False
def is_element_stale(self, webelement):
"""
Checks if a webelement is stale.
@param webelement: A selenium webdriver webelement
"""
try:
webelement.tag_name
except selenium.common.exceptions.StaleElementReferenceException:
return True
except NameError:
pass
except:
pass
return False