Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self,
log_bucket=settings.S3_LOG_BUCKET,
bucket_url=settings.S3_BUCKET_URL,
selenium_access_key=settings.S3_SELENIUM_ACCESS_KEY,
selenium_secret_key=settings.S3_SELENIUM_SECRET_KEY):
self.conn = S3Connection(selenium_access_key,
selenium_secret_key)
self.bucket = self.conn.get_bucket(log_bucket)
self.bucket_url = bucket_url
def wait_for_angularjs(driver, timeout=settings.LARGE_TIMEOUT, **kwargs):
if not settings.WAIT_FOR_ANGULARJS:
return
NG_WRAPPER = '%(prefix)s' \
'var $elm=document.querySelector(' \
'\'[data-ng-app],[ng-app],.ng-scope\')||document;' \
'if(window.angular && angular.getTestability){' \
'angular.getTestability($elm).whenStable(%(handler)s)' \
'}else{' \
'var $inj;try{$inj=angular.element($elm).injector()||' \
'angular.injector([\'ng\'])}catch(ex){' \
'$inj=angular.injector([\'ng\'])};$inj.get=$inj.get||' \
'$inj;$inj.get(\'$browser\').' \
'notifyWhenNoOutstandingRequests(%(handler)s)}' \
'%(suffix)s'
def_pre = 'var cb=arguments[arguments.length-1];if(window.angular){'
prefix = kwargs.pop('prefix', def_pre)
def is_obfuscated(string):
# Based on settings, determines if a string has already been obfuscated.
# Obfuscated strings have a common predefined start token and end token.
start_token = settings.OBFUSCATION_START_TOKEN
end_token = settings.OBFUSCATION_END_TOKEN
return (string.startswith(start_token) and string.endswith(end_token))
def post_message(driver, message, msg_dur, style="info"):
""" A helper method to post a message on the screen with Messenger.
(Should only be called from post_message() in base_case.py) """
if not msg_dur:
msg_dur = settings.DEFAULT_MESSAGE_DURATION
msg_dur = float(msg_dur)
message = re.escape(message)
message = escape_quotes_if_needed(message)
messenger_script = ('''Messenger().post({message: "%s", type: "%s", '''
'''hideAfter: %s, hideOnNavigate: true});'''
% (message, style, msg_dur))
try:
driver.execute_script(messenger_script)
except Exception:
activate_messenger(driver)
set_messenger_theme(driver)
try:
driver.execute_script(messenger_script)
except Exception:
time.sleep(0.2)
activate_messenger(driver)
instructions = instructions.replace(
'duration: 0,', 'duration: %s,' % interval)
if not is_bootstrap_activated(driver):
activate_bootstrap(driver)
if len(tour_steps[name]) > 1:
try:
if "element: " in tour_steps[name][1]:
selector = re.search(
r"[\S\s]+element: '([\S\s]+)',[\S\s]+title: '",
tour_steps[name][1]).group(1)
selector = selector.replace('\\', '').replace(':first', '')
page_actions.wait_for_element_present(
driver, selector, by=By.CSS_SELECTOR,
timeout=settings.SMALL_TIMEOUT)
else:
selector = "html"
except Exception:
js_utils.post_messenger_error_message(
driver, "Tour Error: {'%s'} was not found!" % selector,
msg_dur)
raise Exception(
"Tour Error: {'%s'} was not found! "
"Exiting due to failure on first tour step!"
"" % selector)
driver.execute_script(instructions)
tour_on = True
while tour_on:
try:
time.sleep(0.01)
DEPRECATED soon. (I haven't tested this code since 2014)
This plugin allows you to receive test notifications through HipChat.
Mentions only occur during normal business hours. (Can be changed)
By default, only failure notifications will be sent.
"""
import os
import requests
import logging
import datetime
from nose.plugins import Plugin
from seleniumbase.config import settings
HIPCHAT_URL = 'https://api.hipchat.com/v1/rooms/message'
HIPCHAT_AUTH_TOKEN = settings.HIPCHAT_AUTH_TOKEN
class HipchatReporting(Plugin):
'''
Usage: --with-hipchat_reporting --hipchat_room_id=[HIPCHAT ROOM ID]
--hipchat_owner_to_mention=[HIPCHAT @NAME]
'''
name = 'hipchat_reporting'
def __init__(self):
super(HipchatReporting, self).__init__()
self.hipchat_room_id = None
self.hipchat_owner_to_mention = None
self.hipchat_notify_on_success = False
self.build_url = os.environ.get('BUILD_URL')
self.successes = []
def wait_for_link_text_visible(self, link_text,
timeout=settings.LARGE_TIMEOUT):
if self.timeout_multiplier and timeout == settings.LARGE_TIMEOUT:
timeout = self.__get_new_timeout(timeout)
return self.wait_for_element_visible(
link_text, by=By.LINK_TEXT, timeout=timeout)
settings.MASTERQA_MAX_IDLE_TIME_BEFORE_QUIT = (
override_settings[key])
elif key == "TOTP_KEY":
settings.TOTP_KEY = override_settings[key]
elif key == "DB_HOST":
settings.DB_HOST = override_settings[key]
elif key == "DB_USERNAME":
settings.DB_USERNAME = override_settings[key]
elif key == "DB_PASSWORD":
settings.DB_PASSWORD = override_settings[key]
elif key == "DB_SCHEMA":
settings.DB_SCHEMA = override_settings[key]
elif key == "S3_LOG_BUCKET":
settings.S3_LOG_BUCKET = override_settings[key]
elif key == "S3_BUCKET_URL":
settings.S3_BUCKET_URL = override_settings[key]
elif key == "S3_SELENIUM_ACCESS_KEY":
settings.S3_SELENIUM_ACCESS_KEY = override_settings[key]
elif key == "S3_SELENIUM_SECRET_KEY":
settings.S3_SELENIUM_SECRET_KEY = override_settings[key]
elif key == "ENCRYPTION_KEY":
settings.ENCRYPTION_KEY = override_settings[key]
elif key == "OBFUSCATION_START_TOKEN":
settings.OBFUSCATION_START_TOKEN = override_settings[key]
elif key == "OBFUSCATION_END_TOKEN":
settings.OBFUSCATION_END_TOKEN = override_settings[key]
else:
continue
if num_settings == 0:
raise Exception("Unable to parse the settings file!")
def open(self, url):
self.__last_page_load_url = None
self.driver.get(url)
if settings.WAIT_FOR_RSC_ON_PAGE_LOADS:
self.wait_for_ready_state_complete()
self.__demo_mode_pause_if_active()