test_framework
/
SeleniumBase
mirror of https://github.com/seleniumbase/SeleniumBase.git
3
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #163 from seleniumbase/mysql-deploy 

MySQL Deploy
This commit is contained in:
Michael Mintz 2018-04-17 03:35:09 -04:00 committed by GitHub
parent 6af020b5bf 32289fc6e7
commit 8ca1165997
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 153 additions and 325 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
.travis.yml
View File

@ -2,9 +2,16 @@ language: python
sudo: false
python:
- "2.7"
- "3.6"
addons:
firefox: latest
chrome: stable
before_install:
- "sudo mysql -e 'CREATE DATABASE IF NOT EXISTS test_db;'"
- "sudo mysql -h 127.0.0.1 -u root test_db < seleniumbase/core/create_db_tables.sql"
- "mysqladmin -u root password test"
# - "mysqladmin -u root -p'old_password' password new_password"
- "sudo service mysql restart"
install:
- "pip install --upgrade pip"
- "pip install -r requirements.txt --upgrade"
@ -12,14 +19,16 @@ install:
- "sudo rm -f /etc/boto.cfg"
before_script:
- "flake8 seleniumbase/*.py && flake8 seleniumbase/*/*.py && flake8 seleniumbase/*/*/*.py && flake8 seleniumbase/*/*/*/*.py"
# - "export DISPLAY=:99.0 && sh -e /etc/init.d/xvfb start"
- "wget https://chromedriver.storage.googleapis.com/2.37/chromedriver_linux64.zip && unzip chromedriver_linux64.zip && sudo cp chromedriver /usr/local/bin/ && sudo chmod +x /usr/local/bin/chromedriver"
- "wget https://github.com/mozilla/geckodriver/releases/download/v0.20.0/geckodriver-v0.20.0-linux64.tar.gz -O /tmp/geckodriver.tar.gz && tar -C /opt -xzf /tmp/geckodriver.tar.gz && sudo chmod 755 /opt/geckodriver && sudo ln -fs /opt/geckodriver /usr/bin/geckodriver && sudo ln -fs /opt/geckodriver /usr/local/bin/geckodriver"
# - "wget https://bitbucket.org/ariya/phantomjs/downloads/phantomjs-2.1.1-linux-x86_64.tar.bz2 && tar -xvf ./phantomjs-2.1.1-linux-x86_64.tar.bz2 && export PATH=$PWD/phantomjs-2.1.1-linux-x86_64/bin:$PATH"
script:
- "pytest examples/my_first_test.py --browser=chrome -s --headless"
- "nosetests examples/boilerplates/boilerplate_test.py --headless"
- "pytest examples/my_first_test.py --browser=firefox -s --headless"
- "pytest examples/github_test.py --browser=chrome -s --headless"
- "pytest examples/my_first_test.py --browser=chrome -s --headless --with-db_reporting"
- "nosetests examples/boilerplates/boilerplate_test.py --browser=chrome --headless"
- "pytest examples/my_first_test.py --browser=firefox -s --headless --with-db_reporting"
- "pytest examples/github_test.py --browser=firefox -s --headless --with-db_reporting --demo_mode --demo_sleep=0.2"
- "sudo mysql --password=test -e 'select test_address,browser,state,start_time,runtime from test_db.test_run_data'"
after_script:
- "sudo mysql -e 'DROP DATABASE test_db;'"
notifications:
email: false

48
README.md
View File

@ -1,5 +1,4 @@
## Automated Web-UI testing reimagined.
## Automated testing made fast, easy, and reliable.
<img src="https://cdn2.hubspot.net/hubfs/100006/images/laptop_logo.png" title="SeleniumBase" height="160">
@ -610,7 +609,7 @@ Let's say you have a test that sends an email, and now you want to check that th
from seleniumbase.fixtures.email_manager import EmailManager, EmailException
num_email_results = 0
email_subject = "This is the subject to search for (maybe include a timestamp)"
email_manager = EmailManager("[YOUR SELENIUM GMAIL EMAIL ADDRESS]") # the password for this is elsewhere (in the library) because this is a default email account
email_manager = EmailManager("{YOUR SELENIUM GMAIL ACCOUNT EMAIL ADDRESS}") # the password for this would be stored in seleniumbase/config/settings.py
try:
html_text = email_manager.search(SUBJECT="%s" % email_subject, timeout=300)
num_email_results = len(html_text)
@ -622,49 +621,6 @@ self.assertTrue(num_email_results) # true if not zero
Now you can parse through the email if you're looking for specific text or want to navigate to a link listed there.
#### Database Powers:
Let's say you have a test that needs to access the database. First make sure you already have a table ready. Then try this example:
```python
from seleniumbase.core.mysql import DatabaseManager
def write_data_to_db(self, theId, theValue, theUrl):
db = DatabaseManager()
query = """INSERT INTO myTable(theId,theValue,theUrl)
VALUES (%(theId)s,%(theValue)s,%(theUrl)s)"""
db.execute_query_and_close(query, {"theId":theId,
"theValue":theValue,
"theUrl":theUrl})
```
Access credentials are stored in [settings.py](https://github.com/seleniumbase/SeleniumBase/blob/master/seleniumbase/config/settings.py) for your convenience (you have to add them first).
The following example below (taken from the Delayed Data Manager) shows how data can be pulled from the database.
```python
import logging
from seleniumbase.core.mysql import DatabaseManager
def get_delayed_test_data(self, testcase_address, done=0):
""" Returns a list of rows """
db = DatabaseManager()
query = """SELECT guid,testcaseAddress,insertedAt,expectedResult,done
FROM delayedTestData
WHERE testcaseAddress=%(testcase_address)s
AND done=%(done)s"""
data = db.fetchall_query_and_close(query, {"testcase_address":testcase_address, "done":done})
if data:
return data
else:
logging.debug("Could not find any rows in delayedTestData.")
logging.debug("DB Query = " + query % {"testcase_address":testcase_address, "done":done})
return []
```
Now you know how to pull data from your MySQL DB.
Delayed Data usage example: If you scheduled an email to go out 3 hours from now and you wanted to check that the email gets received (but you don't want your test sitting idle for 3 hours) you can store the email credentials as a unique time-stamp for the email subject in the DB (along with a time for when it's safe for the email to be searched for) and then a later-running test can do the checking after the right amount of time has passed.
### ![http://seleniumbase.com](https://cdn2.hubspot.net/hubfs/100006/images/super_logo_tiny.png "SeleniumBase") Wrap-Up
**Congratulations** on learning how to use **SeleniumBase**!

17
examples/github_test.py
View File

@ -1,15 +1,24 @@
from seleniumbase import BaseCase
import time
class GitHubTests(BaseCase):
# Selenium can trigger GitHub's abuse detection mechanism:
# "You have triggered an abuse detection mechanism."
# "Please wait a few minutes before you try again."
# To avoid this, slow down Selenium actions.
def slow_click(self, css_selector):
time.sleep(1)
self.click(css_selector)
def test_github(self):
self.open("https://github.com/")
self.update_text("input.header-search-input", "SeleniumBase\n")
self.click('a[href="/seleniumbase/SeleniumBase"]')
self.slow_click('a[href="/seleniumbase/SeleniumBase"]')
self.assert_element("div.repository-content")
self.assert_text("SeleniumBase", "h1")
self.click('a[title="seleniumbase"]')
self.click('a[title="fixtures"]')
self.click('a[title="base_case.py"]')
self.slow_click('a[title="seleniumbase"]')
self.slow_click('a[title="fixtures"]')
self.slow_click('a[title="base_case.py"]')
self.assert_text("Code", "nav a.selected")

11
integrations/google_cloud/ReadMe.md
View File

@ -168,18 +168,13 @@ If you have a web application that you want to test, you'll be able to create Se
#### Step 26. Create the necessary tables in your MySQL database/schema
* Run a SQL script in your MySQL database/schema using [testcaserepository.sql](https://raw.githubusercontent.com/seleniumbase/SeleniumBase/master/seleniumbase/core/testcaserepository.sql)
* Run the [create_db_tables.sql](https://raw.githubusercontent.com/seleniumbase/SeleniumBase/master/seleniumbase/core/create_db_tables.sql) script in your MySQL database/schema to create all the required DB tables.
#### Step 27. Have your local clone of SeleniumBase connect to your MySQL Instance
#### Step 27. Have your local clone of SeleniumBase connect to your MySQL DB Instance
* Update the MySQL connection details in your [settings.py](https://github.com/seleniumbase/SeleniumBase/blob/master/seleniumbase/config/settings.py) file to use the credentials that you saved in Step 21.
* If you haven't already installed the MySQL-Python connector, run the following command below:
```bash
pip install MySQL-python==1.2.5
```
#### Step 28. Have your SeleniumBase Jenkins jobs use your MySQL Instance
#### Step 28. Have your SeleniumBase Jenkins jobs use your MySQL DB Instance
* For the "Execute shell", use the following as your updated "Command":

3
requirements.txt
View File

@ -10,9 +10,10 @@ six==1.10.0
flake8==3.5.0
requests==2.18.4
beautifulsoup4==4.6.0
mysqlclient==1.3.12
unittest2==1.1.0
chardet==3.0.4
boto==2.48.0
ipdb==0.11
pyvirtualdisplay==0.2.1
PyVirtualDisplay==0.2.1
-e .

30
seleniumbase/core/create_db_tables.sql Executable file
View File

@ -0,0 +1,30 @@
# Creates test_db tables for using SeleniumBase with MySQL
# test_run_data table
# -----------------------------------
CREATE TABLE `test_run_data` (
`guid` varchar(64) NOT NULL DEFAULT '',
`test_address` varchar(255) DEFAULT NULL,
`env` varchar(64) DEFAULT NULL,
`start_time` varchar(64) DEFAULT NULL,
`execution_guid` varchar(64) DEFAULT NULL,
`runtime` int(11),
`state` varchar(64) DEFAULT NULL,
`browser` varchar(64) DEFAULT NULL,
`message` text,
`stack_trace` text,
`retry_count` int(11) DEFAULT '0',
`exception_map_guid` varchar(64) DEFAULT NULL,
`log_url` text,
PRIMARY KEY (`guid`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
# test_execution table
# -----------------------------------
CREATE TABLE `test_execution` (
`guid` varchar(64) NOT NULL DEFAULT '',
`total_execution_time` int(11),
`username` varchar(255) DEFAULT NULL,
`execution_start` bigint(20) DEFAULT '0',
PRIMARY KEY (`guid`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

22
seleniumbase/core/mysql.py
View File

@ -1,27 +1,25 @@
"""
Wrapper for MySQL functions to make life easier
Due to compatibility issues, might only work for Python 2.7 right now
Wrapper for MySQL DB functions to make life easier.
"""
import time
from seleniumbase.core import mysql_conf as conf
class DatabaseManager():
"""
This class wraps database functions for easy use.
It connects to the testcase database.
This class wraps MySQL database methods for easy use.
"""
def __init__(self, database_env='test', conf_creds=None):
"""
Gets database information from mysql_conf.py and creates a connection.
"""
import mysql_conf as conf # This had problems when using Python 3
import MySQLdb
db_server, db_user, db_pass, db_schema = \
conf.APP_CREDS[conf.Apps.TESTCASE_REPOSITORY][database_env]
retry_count = 3
backoff = 1.2 # Time to wait (in seconds) between retries
backoff = 1.2 # Time to wait (in seconds) between retries.
count = 0
while count < retry_count:
try:
@ -38,27 +36,27 @@ class DatabaseManager():
if retry_count == 3:
raise Exception("Unable to connect to Database after 3 retries.")
def fetchall_query_and_close(self, query, values):
def query_fetch_all(self, query, values):
"""
Executes a query, gets all the values and then closes up the connection
Executes a db query, gets all the values, and closes the connection.
"""
self.cursor.execute(query, values)
retval = self.cursor.fetchall()
self.__close_db()
return retval
def fetchone_query_and_close(self, query, values):
def query_fetch_one(self, query, values):
"""
Executes a query, gets the first value, and closes up the connection
Executes a db query, gets the first value, and closes the connection.
"""
self.cursor.execute(query, values)
retval = self.cursor.fetchone()
self.__close_db()
return retval
def execute_query_and_close(self, query, values):
def execute_query(self, query, values):
"""
Executes a query and closes the connection
Executes a query to the test_db and closes the connection afterwards.
"""
retval = self.cursor.execute(query, values)
self.__close_db()

3
seleniumbase/core/mysql_conf.py
View File

@ -1,6 +1,5 @@
"""
This file contains database credentials for the various databases
that the tests need to access
This file organizes connection details to the Testcase Database.
"""
from seleniumbase.config import settings

88
seleniumbase/core/testcase_manager.py
View File

@ -1,48 +1,41 @@
"""
Testcase database related methods
"""
from seleniumbase.core.mysql import DatabaseManager
class TestcaseManager:
"""
Helper for Testcase related DB stuff
"""
def __init__(self, database_env):
self.database_env = database_env
def insert_execution_data(self, execution_query_payload):
""" Inserts an execution into the database.
Returns the execution guid. """
""" Inserts a test execution row into the database.
Returns the execution guid.
"execution_start_time" is defined by milliseconds since the Epoch.
(See https://currentmillis.com to convert that to a real date.) """
query = """INSERT INTO execution
(guid, executionStart, totalExecutionTime, username)
query = """INSERT INTO test_execution
(guid, execution_start, total_execution_time, username)
VALUES (%(guid)s,%(execution_start_time)s,
%(total_execution_time)s,%(username)s)"""
DatabaseManager(self.database_env).execute_query_and_close(
DatabaseManager(self.database_env).execute_query(
query,
execution_query_payload.get_params())
return execution_query_payload.guid
def update_execution_data(self, execution_guid, execution_time):
"""updates an existing execution in the database"""
query = """UPDATE execution
SET totalExecutionTime=%(execution_time)s
""" Updates an existing test execution row in the database. """
query = """UPDATE test_execution
SET total_execution_time=%(execution_time)s
WHERE guid=%(execution_guid)s """
DatabaseManager(self.database_env).execute_query_and_close(
DatabaseManager(self.database_env).execute_query(
query,
{"execution_guid": execution_guid,
"execution_time": execution_time})
def insert_testcase_data(self, testcase_run_payload):
"""inserts all data for the test case, returns the new row guid"""
query = """INSERT INTO testcaseRunData
(guid, browser, state, execution_guid, env, start_time,
testcaseAddress, runtime, retryCount, message, stackTrace)
""" Inserts all data for the test in the DB. Returns new row guid. """
query = """INSERT INTO test_run_data(
guid, browser, state, execution_guid, env, start_time,
test_address, runtime, retry_count, message, stack_trace)
VALUES (
%(guid)s,
%(browser)s,
@ -50,39 +43,35 @@ class TestcaseManager:
%(execution_guid)s,
%(env)s,
%(start_time)s,
%(testcaseAddress)s,
%(test_address)s,
%(runtime)s,
%(retryCount)s,
%(retry_count)s,
%(message)s,
%(stackTrace)s) """
DatabaseManager(self.database_env).execute_query_and_close(
%(stack_trace)s) """
DatabaseManager(self.database_env).execute_query(
query, testcase_run_payload.get_params())
def update_testcase_data(self, testcase_payload):
"""updates an existing testcase run in the database"""
query = """UPDATE testcaseRunData SET
runtime=%(runtime)s,
state=%(state)s,
retryCount=%(retryCount)s,
stackTrace=%(stackTrace)s,
message=%(message)s
WHERE guid=%(guid)s """
DatabaseManager(self.database_env).execute_query_and_close(
""" Updates an existing test run in the database. """
query = """UPDATE test_run_data SET
runtime=%(runtime)s,
state=%(state)s,
retry_count=%(retry_count)s,
stack_trace=%(stack_trace)s,
message=%(message)s
WHERE guid=%(guid)s """
DatabaseManager(self.database_env).execute_query(
query, testcase_payload.get_params())
def update_testcase_log_url(self, testcase_payload):
"""updates an existing testcase run's logging URL in the database"""
query = """UPDATE testcaseRunData
SET logURL=%(logURL)s
query = """UPDATE test_run_data
SET log_url=%(log_url)s
WHERE guid=%(guid)s """
DatabaseManager(self.database_env).execute_query_and_close(
DatabaseManager(self.database_env).execute_query(
query, testcase_payload.get_params())
class ExecutionQueryPayload:
""" Helper class for containing the execution query data """
def __init__(self):
self.execution_start_time = None
self.total_execution_time = -1
@ -90,7 +79,6 @@ class ExecutionQueryPayload:
self.guid = None
def get_params(self):
""" Returns a params object for use with the pool """
return {
"execution_start_time": self.execution_start_time,
"total_execution_time": self.total_execution_time,
@ -100,10 +88,9 @@ class ExecutionQueryPayload:
class TestcaseDataPayload:
""" Helper class for containing all the testcase query data """
def __init__(self):
self.guid = None
self.testcaseAddress = None
self.test_address = None
self.browser = None
self.state = None
self.execution_guid = None
@ -113,21 +100,20 @@ class TestcaseDataPayload:
self.retry_count = 0
self.stack_trace = None
self.message = None
self.logURL = None
self.log_url = None
def get_params(self):
""" Returns a params object for use with the pool """
return {
"guid": self.guid,
"testcaseAddress": self.testcaseAddress,
"test_address": self.test_address,
"browser": self.browser,
"state": self.state,
"execution_guid": self.execution_guid,
"env": self.env,
"start_time": self.start_time,
"runtime": self.runtime,
"retryCount": self.retry_count,
"stackTrace": self.stack_trace,
"retry_count": self.retry_count,
"stack_trace": self.stack_trace,
"message": self.message,
"logURL": self.logURL
"log_url": self.log_url
}

41
seleniumbase/core/testcaserepository.sql
View File

@ -1,41 +0,0 @@
# table delayedTestData
# -----------------------------------
CREATE TABLE `delayedTestData` (
`guid` varchar(64) NOT NULL DEFAULT '',
`testcaseAddress` varchar(255) NOT NULL DEFAULT '',
`insertedAt` bigint(20) NOT NULL,
`expectedResult` text,
`done` tinyint(1) DEFAULT '0',
`expiresAt` bigint(20) DEFAULT NULL,
PRIMARY KEY (`guid`),
UNIQUE KEY `uuid` (`guid`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
# table execution
# -----------------------------------
CREATE TABLE `execution` (
`guid` varchar(64) NOT NULL DEFAULT '',
`totalExecutionTime` int(11),
`username` varchar(255) DEFAULT NULL,
`executionStart` bigint(20) DEFAULT '0',
PRIMARY KEY (`guid`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
# table testcaseRunData
# -----------------------------------
CREATE TABLE `testcaseRunData` (
`guid` varchar(64) NOT NULL DEFAULT '',
`testcaseAddress` varchar(255) DEFAULT NULL,
`env` varchar(64) DEFAULT NULL,
`start_time` varchar(64) DEFAULT NULL,
`execution_guid` varchar(64) DEFAULT NULL,
`runtime` int(11),
`state` varchar(64) DEFAULT NULL,
`browser` varchar(64) DEFAULT NULL,
`message` text,
`stackTrace` text,
`retryCount` int(11) DEFAULT '0',
`exceptionMap_guid` varchar(64) DEFAULT NULL,
`logURL` text,
PRIMARY KEY (`guid`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

31
seleniumbase/fixtures/base_case.py
View File

@ -2062,7 +2062,7 @@ class BaseCase(unittest.TestCase):
data_payload.browser = self.browser
else:
data_payload.browser = "N/A"
data_payload.testcaseAddress = test_id
data_payload.test_address = test_id
application = ApplicationManager.generate_application_string(
self._testMethodName)
data_payload.env = application.split('.')[0]
@ -2176,7 +2176,34 @@ class BaseCase(unittest.TestCase):
test_id = "%s.%s.%s" % (self.__class__.__module__,
self.__class__.__name__,
self._testMethodName)
if self.with_selenium:
try:
with_selenium = self.with_selenium
except Exception:
sub_class_name = str(
self.__class__.__bases__[0]).split('.')[-1].split("'")[0]
sub_file_name = str(self.__class__.__bases__[0]).split('.')[-2]
sub_file_name = sub_file_name + ".py"
class_name = str(self.__class__).split('.')[-1].split("'")[0]
file_name = str(self.__class__).split('.')[-2] + ".py"
class_name_used = sub_class_name
file_name_used = sub_file_name
if sub_class_name == "BaseCase":
class_name_used = class_name
file_name_used = file_name
fix_setup = "super(%s, self).setUp()" % class_name_used
fix_teardown = "super(%s, self).tearDown()" % class_name_used
message = ("You're overriding SeleniumBase's BaseCase setUp() "
"method with your own setUp() method, which breaks "
"SeleniumBase. You can fix this by going to your "
"%s class located in your %s file and adding the "
"following line of code AT THE BEGINNING of your "
"setUp() method:\n%s\n\nAlso make sure "
"you have added the following line of code AT THE "
"END of your tearDown() method:\n%s\n"
% (class_name_used, file_name_used,
fix_setup, fix_teardown))
raise Exception(message)
if with_selenium:
# Save a screenshot if logging is on when an exception occurs
if has_exception:
self._add_pytest_html_extra()

141
seleniumbase/fixtures/delayed_data_manager.py
View File

@ -1,141 +0,0 @@
import json
import logging
import time
import uuid
from seleniumbase.core.mysql import DatabaseManager
DEFAULT_EXPIRATION = 1000 * 60 * 60 * 48
class DelayedTestStorage:
""" The database-calling methods of the Delayed Test Framework """
@classmethod
def get_delayed_test_data(self, testcase_address, done=0):
""" This method queries the delayedTestData table in the DB and
then returns a list of rows with the matching parameters.
:param testcase_address: The ID (address) of the test case.
:param done: (0 for test not done or 1 for test done)
:returns: A list of rows found with the matching testcase_address.
"""
db = DatabaseManager()
query = """SELECT guid,testcaseAddress,insertedAt,expectedResult,done
FROM delayedTestData
WHERE testcaseAddress=%(testcase_address)s
AND done=%(done)s"""
data = db.fetchall_query_and_close(
query, {"testcase_address": testcase_address,
"done": done})
if data:
return data
else:
logging.debug("Could not find any rows in delayedTestData.")
logging.debug("DB Query = " + query %
{"testcase_address": testcase_address, "done": done})
return []
@classmethod
def insert_delayed_test_data(self, guid_, testcase_address,
expected_result, done=0,
expires_at=DEFAULT_EXPIRATION):
""" This method inserts rows into the delayedTestData table
in the DB based on the given parameters where
inserted_at (Date format) is automatically set in this method.
:param guid_: The guid that is provided by the test case.
(Format: str(uuid.uuid4()))
:param testcase_address: The ID (address) of the test case.
:param expected_result: The result string of persistent data
that will be stored in the DB.
:param done: (0 for test not done or 1 for test done)
:returns: True (when no exceptions or errors occur)
"""
inserted_at = int(time.time() * 1000)
db = DatabaseManager()
query = """INSERT INTO delayedTestData(
guid,testcaseAddress,insertedAt,
expectedResult,done,expiresAt)
VALUES (%(guid)s,%(testcaseAddress)s,%(inserted_at)s,
%(expected_result)s,%(done)s,%(expires_at)s)"""
db.execute_query_and_close(
query, {"guid": guid_,
"testcaseAddress": testcase_address,
"inserted_at": inserted_at,
"expected_result": expected_result,
"done": done,
"expires_at": inserted_at + expires_at})
return True
@classmethod
def set_delayed_test_to_done(self, guid_):
""" This method updates the delayedTestData table in the DB
to set the test with the selected guid to done.
:param guid_: The guid that is provided by the test case.
(Format: str(uuid.uuid4()))
:returns: True (when no exceptions or errors occur)
"""
db = DatabaseManager()
query = """UPDATE delayedTestData
SET done=TRUE
WHERE guid=%(guid)s
AND done=FALSE"""
db.execute_query_and_close(query, {"guid": guid_})
return True
class DelayedTestAssistant:
""" Some methods for assisting tests (that don't call the DB directly) """
@classmethod
def get_delayed_results(self, test_id, seconds, set_done=True):
"""
This method gets the delayed_test_data and sets the applicable rows
in the DB to done.
The results is a list of dicts where each list item contains
item[0] = guid
item[1] = testcaseAddress
item[2] = seconds from epoch
item[3] = expected results dict encoded in json
:param test_id: the self.id() of the test
:param seconds: the wait period until the data can be checked
:returns: the results for a specific test where enough time has passed
"""
delayed_test_data = DelayedTestStorage.get_delayed_test_data(
testcase_address=test_id)
now = int(time.time() * 1000)
results_to_check = []
if delayed_test_data is None:
return results_to_check
for item in delayed_test_data:
if item[2] < now - (seconds * 1000):
results_to_check.append(item)
if set_done:
DelayedTestStorage.set_delayed_test_to_done(item[0])
return results_to_check
@classmethod
def store_delayed_data(self, test_id, expected_result_dict,
expires_at=DEFAULT_EXPIRATION):
"""
Loads the dictionary of information into the delayed test database
:param test_id: the self.id() of the test
:param expected_result_dict: a dictionary of what's to be checked later
"""
expected_result_json = json.JSONEncoder().encode(expected_result_dict)
DelayedTestStorage.insert_delayed_test_data(str(uuid.uuid4()),
test_id,
expected_result_json,
0,
expires_at)
@classmethod
def set_test_done(self, test_guid):
""" This method calls set_delayed_test_to_done to set a
row in the db to done.
:param test_guid: The guid that is provided by the test.
(Format: str(uuid.uuid4()))
:returns: True (when no exceptions or errors occur)
"""
DelayedTestStorage.set_delayed_test_to_done(test_guid)
return True

17
seleniumbase/plugins/db_reporting_plugin.py
View File

@ -1,6 +1,5 @@
"""
This is the Database test reporting plugin for
recording all test run data in the database.
This plugin is for recording test results in the Testcase Database.
"""
import getpass
@ -19,7 +18,7 @@ from seleniumbase.fixtures import errors
class DBReporting(Plugin):
"""
The plugin for reporting test results in the database.
This plugin records test results in the Testcase Database.
"""
name = 'db_reporting' # Usage: --with-db_reporting
@ -48,8 +47,8 @@ class DBReporting(Plugin):
self.testcase_manager = TestcaseManager(self.options.database_env)
def begin(self):
"""At the start of the run, we want to record the test
execution information in the database."""
""" At the start of the run, we want to record the test
execution information in the database. """
exec_payload = ExecutionQueryPayload()
exec_payload.execution_start_time = int(time.time() * 1000)
self.execution_start_time = exec_payload.execution_start_time
@ -58,7 +57,7 @@ class DBReporting(Plugin):
self.testcase_manager.insert_execution_data(exec_payload)
def startTest(self, test):
"""At the start of the test, set the testcase details."""
""" At the start of the test, set the testcase details. """
data_payload = TestcaseDataPayload()
self.testcase_guid = str(uuid.uuid4())
data_payload.guid = self.testcase_guid
@ -67,7 +66,7 @@ class DBReporting(Plugin):
data_payload.browser = test.browser
else:
data_payload.browser = "N/A"
data_payload.testcaseAddress = test.id()
data_payload.test_address = test.id()
application = ApplicationManager.generate_application_string(test)
data_payload.env = application.split('.')[0]
data_payload.start_time = application.split('.')[1]
@ -78,8 +77,8 @@ class DBReporting(Plugin):
test.testcase_guid = self.testcase_guid
def finalize(self, result):
"""At the end of the run, we want to
update the DB row with the execution time."""
""" At the end of the run, we want to
update the DB row with the execution time. """
runtime = int(time.time() * 1000) - self.execution_start_time
self.testcase_manager.update_execution_data(self.execution_guid,
runtime)

2
seleniumbase/plugins/s3_logging_plugin.py
View File

@ -47,5 +47,5 @@ class S3Logging(Plugin):
self.testcase_manager = TestcaseManager(self.options.database_env)
data_payload = TestcaseDataPayload()
data_payload.guid = test.test.testcase_guid
data_payload.logURL = index_file
data_payload.log_url = index_file
self.testcase_manager.update_testcase_log_url(data_payload)

5
setup.py
View File

@ -7,7 +7,7 @@ from setuptools import setup, find_packages # noqa
setup(
name='seleniumbase',
version='1.8.7',
version='1.9.0',
description='Web Automation & Testing Framework - http://seleniumbase.com',
long_description='Web Automation and Testing Framework - seleniumbase.com',
platforms='Mac * Windows * Linux * Docker',
@ -29,11 +29,12 @@ setup(
'flake8==3.5.0',
'requests==2.18.4',
'beautifulsoup4==4.6.0',
'mysqlclient==1.3.12',
'unittest2==1.1.0',
'chardet==3.0.4',
'boto==2.48.0',
'ipdb==0.11',
'pyvirtualdisplay==0.2.1',
'PyVirtualDisplay==0.2.1',
],
packages=['seleniumbase',
'seleniumbase.core',