Merge branch 'testing_framework' into 'develop'

Testing framework v2

The testing framework should now run tests in parallel and annotate them successfully in SauceLabs

See merge request !1
This commit is contained in:
axl89 2016-08-19 15:49:54 +02:00
commit f8ec880e2a
11 changed files with 274 additions and 196 deletions

0
tests/__init__.py Normal file
View File

View File

@ -26,7 +26,6 @@ class ACL(PandoraWebDriverTestCase):
ACL Propagation test: Creates one group "A" with ACL propagation, then a group "B" son of "A" with no ACL propagation, and finally group "C".
The test asserts if a user with privileges to "A" can see the agent of "B" but no agents of "C".
"""
print "funciono"
group_name_A = gen_random_string(6)
group_name_B = gen_random_string(6)
group_name_C = gen_random_string(6)

View File

@ -42,6 +42,7 @@ class Collections(PandoraWebDriverTestCase):
def test_B_edit_collection(self):
driver = self.driver
self.login()
edit_collection(driver,self.collection_name,new_name=self.new_collection_name,group="Applications",description="Edit collectionPAN11")
@ -54,6 +55,7 @@ class Collections(PandoraWebDriverTestCase):
def test_C_create_text_collection(self):
driver = self.driver
self.login()
create_text_in_collection(driver,self.new_collection_name,"file_collectionPAN11",text_file="test file")
@ -64,6 +66,7 @@ class Collections(PandoraWebDriverTestCase):
def test_D_directory_collection(self):
driver = self.driver
self.login()
create_directory_in_collection(driver,self.new_collection_name,"directory_collectionPAN11")
@ -74,6 +77,7 @@ class Collections(PandoraWebDriverTestCase):
def test_E_delete_collection(self):
driver = self.driver
self.login()
delete_collection(driver,self.new_collection_name)

View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
from include.common_classes_60 import PandoraWebDriverTestCase
from include.common_functions_60 import login, click_menu_element, refresh_N_times_until_find_element, detect_and_pass_all_wizards, is_element_present, logout

View File

@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
from include.common_classes_60 import PandoraWebDriverTestCase
from include.common_functions_60 import login, click_menu_element, detect_and_pass_all_wizards, logout, gen_random_string
from include.common_functions_60 import login, click_menu_element, detect_and_pass_all_wizards, logout, gen_random_string, refresh_N_times_until_find_element
from include.planned_downtime_functions import *
from include.alert_functions import *
from include.module_functions import *
@ -16,13 +16,18 @@ from selenium.webdriver.remote.webelement import WebElement
import unittest, time, re, datetime
class PAN13(PandoraWebDriverTestCase):
class Creation(PandoraWebDriverTestCase):
test_name = u'Planned_downtime'
test_description = u'Planed downtime test'
test_name = u'Planned downtime creation'
test_description = u'Planed downtime creation test'
tickets_associated = []
def avoid_test_A_create_planned_downtime_Quiet(self):
quiet_name = gen_random_string(6)
disabled_agents_name = gen_random_string(6)
disabled_only_alerts_name = gen_random_string(6)
def test_A_create_planned_downtime_Quiet(self):
u"""
Create and search planned downtime quiet
@ -31,135 +36,63 @@ class PAN13(PandoraWebDriverTestCase):
self.login()
detect_and_pass_all_wizards(driver)
planned_name = gen_random_string(6)
planned_name = self.quiet_name
create_planned_downtime(driver,planned_name,"Applications","Quiet","Once",description=planned_name)
time.sleep(10)
search_planned_downtime(driver,planned_name)
element = driver.find_element_by_xpath('//img[@data-title="Running"]')
self.assertIsInstance(element,WebElement)
def avoid_test_B_create_planned_downtime_disabled_agents(self):
element = driver.find_element_by_xpath('//tr[@id="table3-0"]/td[contains(.,"'+planned_name+'")]')
self.assertIsInstance(element,WebElement)
def test_B_create_planned_downtime_disabled_agents(self):
u"""
Create and search planned downtime disabled agents
"""
driver = self.driver
self.login()
planned_name = gen_random_string(6)
planned_name = self.disabled_agents_name
create_planned_downtime(driver,planned_name,"Applications","Disabled Agents","Once",description=planned_name)
time.sleep(10)
search_planned_downtime(driver,planned_name)
element = driver.find_element_by_xpath('//img[@data-title="Running"]')
element = driver.find_element_by_xpath('//tr[@id="table3-0"]/td[contains(.,"'+planned_name+'")]')
self.assertIsInstance(element,WebElement)
def avoid_test_C_create_planned_downtime_disabled_only_alerts(self):
def test_C_create_planned_downtime_disabled_only_alerts(self):
u"""
Create and search planned downtime disabled only alerts
"""
driver = self.driver
self.login()
planned_name = gen_random_string(6)
planned_name = self.disabled_only_alerts_name
create_planned_downtime(driver,planned_name,"Applications","Disabled only Alerts","Once",description=planned_name)
time.sleep(10)
search_planned_downtime(driver,planned_name)
element = driver.find_element_by_xpath('//img[@data-title="Running"]')
element = driver.find_element_by_xpath('//tr[@id="table3-0"]/td[contains(.,"'+planned_name+'")]')
self.assertIsInstance(element,WebElement)
def avoid_test_D_delete_planned_downtime(self):
def test_D_delete_planned_downtime(self):
driver=self.driver
self.login()
planned_name = gen_random_string(6)
downtime_list = [self.disabled_only_alerts_name,self.disabled_agents_name,self.quiet_name]
create_planned_downtime(driver,planned_name,"Applications","Quiet","Once",description=planned_name)
for planned_name in downtime_list:
delete_planned_downtime(driver,planned_name)
element = driver.find_element_by_xpath('//td[contains(.,"Success")]')
self.assertIsInstance(element,WebElement)
delete_planned_downtime(driver,planned_name)
def avoid_test_E_quiet_functionality(self):
driver=self.driver
planned_name = gen_random_string(6)
agent_name_A = gen_random_string(6)
agent_name_B = gen_random_string(6)
module_name_A_A = gen_random_string(6)
module_name_A_B = gen_random_string(6)
module_name_B_A = gen_random_string(6)
module_name_B_B = gen_random_string(6)
create_agent(driver,agent_name_A,ip="127.0.0.1",group="Applications")
create_agent(driver,agent_name_B,ip="127.0.0.1",group="Applications")
create_module('network_server',driver,agent_name=agent_name_A,module_name=module_name_A_A,component_group='Network Management',network_component='Host Alive')
create_module('network_server',driver,agent_name=agent_name_A,module_name=module_name_A_B,component_group='Network Management',network_component='Host Latency')
create_module('network_server',driver,agent_name=agent_name_B,module_name=module_name_B_A,component_group='Network Management',network_component='Host Alive')
create_module('network_server',driver,agent_name=agent_name_B,module_name=module_name_B_B,component_group='Network Management',network_component='Host Alive')
assign_alert_template_to_module(driver,agent_name_A,module_name_A_A,'Critical condition')
assign_alert_template_to_module(driver,agent_name_A,module_name_A_B,'Critical condition')
assign_alert_template_to_module(driver,agent_name_B,module_name_B_A,'Critical condition')
assign_alert_template_to_module(driver,agent_name_B,module_name_B_B,'Critical condition')
#Little hack to allow the planned downtime to be edited
fifteen_seconds_later = datetime.datetime.now() + datetime.timedelta(seconds=15)
fifteen_seconds_later_to_pandora = str(fifteen_seconds_later.hour)+":"+str(fifteen_seconds_later.minute)+":"+str(fifteen_seconds_later.second)
create_planned_downtime(driver,planned_name,"Applications","Quiet","Once",hour_from=fifteen_seconds_later_to_pandora,description=planned_name,agent_module_list=[(agent_name_A,[module_name_A_A]),(agent_name_B,[module_name_B_A])])
#We wait 10 seconds to the planned downtime to start. Since we do not specify a date, default dates are from now to one hour in the future.
time.sleep(40)
#Is the planned downtime running?
search_planned_downtime(driver,planned_name)
element = driver.find_element_by_xpath('//img[@data-title="Running"]')
self.assertIsInstance(element,WebElement)
force_alert_of_module(driver,agent_name_A,module_name_A_A,'Critical condition')
force_alert_of_module(driver,agent_name_A,module_name_A_B,'Critical condition')
force_alert_of_module(driver,agent_name_B,module_name_B_A,'Critical condition')
force_alert_of_module(driver,agent_name_B,module_name_B_B,'Critical condition')
search_events(driver,agent_name=agent_name_A,module_name=module_name_A_A)
event_who_should_not_be_present = driver.find_elements_by_xpath('//tbody/tr[td[3][contains(.,"Alert fired")] and td[4][contains(.,"'+agent_name_A+'")]]')
self.assertEqual(event_who_should_not_be_present,[])
search_events(driver,agent_name=agent_name_A,module_name=module_name_A_B)
event_who_should_be_present = driver.find_elements_by_xpath('//tbody/tr[td[3][contains(.,"Alert fired")] and td[4][contains(.,"'+agent_name_A+'")]]')
self.assertNotEqual(event_who_should_be_present,[])
search_events(driver,agent_name=agent_name_B,module_name=module_name_B_A)
event_who_should_not_be_present_b = driver.find_elements_by_xpath('//tbody/tr[td[3][contains(.,"Alert fired")] and td[4][contains(.,"'+agent_name_B+'")]]')
self.assertEqual(event_who_should_not_be_present_b,[])
search_events(driver,agent_name=agent_name_B,module_name=module_name_B_B)
event_who_should_be_present_b = driver.find_elements_by_xpath('//tbody/tr[td[3][contains(.,"Alert fired")] and td[4][contains(.,"'+agent_name_B+'")]]')
self.assertNotEqual(event_who_should_be_present_b,[])
if __name__ == "__main__":
unittest.main()

View File

@ -48,6 +48,7 @@ class Policies(PandoraWebDriverTestCase):
Add network server module to previous policy
"""
driver = self.driver
self.login()
add_module_policy(driver,self.policy_name,"network_server",driver,module_name=self.network_server_module_name,component_group="Network Management",network_component="Host Alive")
@ -67,6 +68,7 @@ class Policies(PandoraWebDriverTestCase):
driver = self.driver
self.login()
create_policy(driver,policy_name,"Applications",description="Policy for test")
@ -93,6 +95,7 @@ class Policies(PandoraWebDriverTestCase):
module_name_2 = gen_random_string(6)
driver = self.driver
self.login()
create_agent(driver,agent_name_1,description="First agent by test")
create_agent(driver,agent_name_2,description="Second agent 2 by test")

View File

@ -0,0 +1,100 @@
# -*- coding: utf-8 -*-
from include.common_classes_60 import PandoraWebDriverTestCase
from include.common_functions_60 import login, click_menu_element, detect_and_pass_all_wizards, logout, gen_random_string
from include.planned_downtime_functions import *
from include.alert_functions import *
from include.module_functions import *
from include.agent_functions import *
from include.event_functions import *
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.ui import Select
from selenium.common.exceptions import NoSuchElementException
from selenium.common.exceptions import NoAlertPresentException
from selenium.webdriver.remote.webelement import WebElement
import unittest, time, re, datetime
class QuietTest(PandoraWebDriverTestCase):
test_name = u'Planned downtime quiet type test'
test_description = u'Planed downtime quiet functionality '
tickets_associated = []
def test_A_quiet_functionality(self):
driver=self.driver
self.login()
planned_name = gen_random_string(6)
agent_name_A = gen_random_string(6)
agent_name_B = gen_random_string(6)
module_name_A_A = gen_random_string(6)
module_name_A_B = gen_random_string(6)
module_name_B_A = gen_random_string(6)
module_name_B_B = gen_random_string(6)
create_agent(driver,agent_name_A,ip="127.0.0.1",group="Applications")
create_agent(driver,agent_name_B,ip="127.0.0.1",group="Applications")
create_module('network_server',driver,agent_name=agent_name_A,module_name=module_name_A_A,component_group='Network Management',network_component='Host Alive')
create_module('network_server',driver,agent_name=agent_name_A,module_name=module_name_A_B,component_group='Network Management',network_component='Host Latency')
create_module('network_server',driver,agent_name=agent_name_B,module_name=module_name_B_A,component_group='Network Management',network_component='Host Alive')
create_module('network_server',driver,agent_name=agent_name_B,module_name=module_name_B_B,component_group='Network Management',network_component='Host Alive')
assign_alert_template_to_module(driver,agent_name_A,module_name_A_A,'Critical condition')
assign_alert_template_to_module(driver,agent_name_A,module_name_A_B,'Critical condition')
assign_alert_template_to_module(driver,agent_name_B,module_name_B_A,'Critical condition')
assign_alert_template_to_module(driver,agent_name_B,module_name_B_B,'Critical condition')
#Little hack to allow the planned downtime to be edited
fifteen_seconds_later = datetime.datetime.now() + datetime.timedelta(seconds=15)
fifteen_seconds_later_to_pandora = str(fifteen_seconds_later.hour)+":"+str(fifteen_seconds_later.minute)+":"+str(fifteen_seconds_later.second)
create_planned_downtime(driver,planned_name,"Applications","Quiet","Once",hour_from=fifteen_seconds_later_to_pandora,description=planned_name,agent_module_list=[(agent_name_A,[module_name_A_A]),(agent_name_B,[module_name_B_A])])
#We wait 10 seconds to the planned downtime to start. Since we do not specify a date, default dates are from now to one hour in the future.
time.sleep(40)
#Is the planned downtime running?
search_planned_downtime(driver,planned_name)
element = driver.find_element_by_xpath('//img[@data-title="Running"]')
self.assertIsInstance(element,WebElement)
force_alert_of_module(driver,agent_name_A,module_name_A_A,'Critical condition')
force_alert_of_module(driver,agent_name_A,module_name_A_B,'Critical condition')
force_alert_of_module(driver,agent_name_B,module_name_B_A,'Critical condition')
force_alert_of_module(driver,agent_name_B,module_name_B_B,'Critical condition')
search_events(driver,agent_name=agent_name_A,module_name=module_name_A_A)
event_who_should_not_be_present = driver.find_elements_by_xpath('//tbody/tr[td[3][contains(.,"Alert fired")] and td[4][contains(.,"'+agent_name_A+'")]]')
self.assertEqual(event_who_should_not_be_present,[])
search_events(driver,agent_name=agent_name_A,module_name=module_name_A_B)
event_who_should_be_present = driver.find_elements_by_xpath('//tbody/tr[td[3][contains(.,"Alert fired")] and td[4][contains(.,"'+agent_name_A+'")]]')
self.assertNotEqual(event_who_should_be_present,[])
search_events(driver,agent_name=agent_name_B,module_name=module_name_B_A)
event_who_should_not_be_present_b = driver.find_elements_by_xpath('//tbody/tr[td[3][contains(.,"Alert fired")] and td[4][contains(.,"'+agent_name_B+'")]]')
self.assertEqual(event_who_should_not_be_present_b,[])
search_events(driver,agent_name=agent_name_B,module_name=module_name_B_B)
event_who_should_be_present_b = driver.find_elements_by_xpath('//tbody/tr[td[3][contains(.,"Alert fired")] and td[4][contains(.,"'+agent_name_B+'")]]')
self.assertNotEqual(event_who_should_be_present_b,[])
if __name__ == "__main__":
unittest.main()

View File

@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
from include.common_classes_60 import PandoraWebDriverTestCase
from include.common_functions_60 import login, click_menu_element, refresh_N_times_until_find_element, detect_and_pass_all_wizards
from include.common_functions_60 import login, click_menu_element, refresh_N_times_until_find_element, detect_and_pass_all_wizards, gen_random_string
from include.agent_functions import create_agent
from selenium import webdriver
from selenium.webdriver.common.by import By
@ -11,18 +11,21 @@ from include.module_functions import *
import unittest, time, re
class PAN3(PandoraWebDriverTestCase):
class SimpleICMPTest(PandoraWebDriverTestCase):
test_name = u'PAN_3'
test_name = u'Simple ICMP Test'
test_description = u'Creates a simple ICMP check against localhost and checks the result is 1'
tickets_associated = []
def test_pan3(self):
driver = self.driver
agent_name = gen_random_string(6)
module_name = gen_random_string(6)
self.login()
create_agent(driver,"PAN3_agent",ip="127.0.0.1")
create_agent(driver,agent_name,ip="127.0.0.1")
create_module("network_server",driver,agent_name="PAN3_agent",module_name="PAN3_module",component_group="Network Management",network_component="Host Alive",ip="127.0.0.1")
create_module("network_server",driver,agent_name=agent_name,module_name=module_name,component_group="Network Management",network_component="Host Alive",ip="127.0.0.1")
driver.find_element_by_xpath('//*[@id="menu_tab"]//a[contains(@href,"ver_agente")]').click()

View File

@ -22,13 +22,13 @@ class ArticaTestResult(TestResult):
class PandoraWebDriverTestCase(TestCase):
test_name = u'' #Name of the test.
test_description = u'' #Description of the test
time_started = None
time_elapsed = None #Total time of the test
tickets_associated = []
sauce_username = environ["SAUCE_USERNAME"]
sauce_access_key = environ["SAUCE_ACCESS_KEY"]
sauce_client = None
sauce_labs_job_id = None
currentResult = None # holds last result object passed to run method
desired_cap = {
'tunnel-identifier': environ["TRAVIS_JOB_NUMBER"],
@ -36,34 +36,28 @@ class PandoraWebDriverTestCase(TestCase):
'browserName': "firefox",
'version': "46",
}
def run(self,result=None,*args,**kwargs):
self.currentResult = result # remember result for use in tearDown
unittest.TestCase.run(self, result,*args,**kwargs) # call superclass run method
@classmethod
def setUpClass(cls):
cls.is_development = os.getenv('DEVELOPMENT', False)
cls.is_enterprise = os.getenv('ENTERPRISE', False)
if cls.is_development != False:
cls.driver = webdriver.Firefox()
cls.base_url = os.getenv('DEVELOPMENT_URL')
def setUp(self):
if self.is_development != False:
self.driver = webdriver.Firefox()
self.base_url = os.getenv('DEVELOPMENT_URL')
else:
#Start VM in Sauce Labs
cls.driver = webdriver.Remote(command_executor='http://'+cls.sauce_username+':'+cls.sauce_access_key+'@ondemand.saucelabs.com:80/wd/hub',desired_capabilities=cls.desired_cap)
cls.sauce_labs_job_id = cls.driver.session_id
cls.base_url = "http://127.0.0.1/"
@classmethod
def tearDownClass(cls):
cls.driver.quit()
def setUp(self):
self.time_started = datetime.now()
self.driver = webdriver.Remote(command_executor='http://'+self.sauce_username+':'+self.sauce_access_key+'@ondemand.saucelabs.com:80/wd/hub',desired_capabilities=self.desired_cap)
self.sauce_labs_job_id = self.driver.session_id
self.base_url = "http://127.0.0.1/"
self.driver.implicitly_wait(30)
self.verificationErrors = []
self.accept_next_alert = True
#self.is_development = self.is_development
#TODO Print test name
print "Starting test"
super(PandoraWebDriverTestCase, self).setUp()
def is_element_present(self, how, what):
@ -88,16 +82,21 @@ class PandoraWebDriverTestCase(TestCase):
finally: self.accept_next_alert = True
def tearDown(self):
tack = datetime.now()
diff = tack - self.time_started
self.time_elapsed = diff.seconds
self.driver.quit()
ok = self.currentResult.wasSuccessful()
errors = self.currentResult.errors
failures = self.currentResult.failures
skipped = self.currentResult.skipped
self.assertEqual([], self.verificationErrors)
if not self.is_development:
sauce_client = SauceClient(environ["SAUCE_USERNAME"], environ["SAUCE_ACCESS_KEY"])
sauce_client.jobs.update_job(self.sauce_labs_job_id, passed=ok==True,tags=[environ["TRAVIS_BRANCH"],self.id()],build_num=environ["TRAVIS_JOB_NUMBER"],name=str(environ["TRAVIS_COMMIT"])+"_"+str(self.id().split('.')[1])+"_"+str(self.id().split('.')[2]))
#self.assertEqual([], self.verificationErrors) #TODO Review if this line is actually needed
super(PandoraWebDriverTestCase, self).tearDown()
def login(self,user="admin",passwd="pandora",pandora_url=None):
print u"Logging in"
driver = self.driver
@ -109,10 +108,19 @@ class PandoraWebDriverTestCase(TestCase):
driver.find_element_by_id("nick").send_keys(user)
driver.find_element_by_id("pass").clear()
driver.find_element_by_id("pass").send_keys(passwd)
#Hack
driver.add_cookie({'name': 'clippy', 'value': 'deleted'})
driver.add_cookie({'name': 'clippy_is_annoying', 'value': '1'})
driver.find_element_by_id("submit-login_button").click()
"""
def test_ZZZZZZZZZZZ(self):
#The hackiest way to end the driver in the LAST test (all credits to python unittest library for sorting tests alphabetically! :D)
self.driver.quit()
"""
def logout(self,pandora_url=None):
print u"Logging out"
driver = self.driver
@ -127,3 +135,4 @@ class PandoraWebDriverTestCase(TestCase):
driver.get(pandora_url+"/pandora_console/index.php")
refresh_N_times_until_find_element(driver,2,"nick")

View File

@ -34,26 +34,6 @@ def gen_random_string(size,preffix=None):
return random_string
def login(driver,user="admin",passwd="pandora",pandora_url=None):
print u"Logging in"
try:
print driver.base_url
except Exception as e:
print "se casco"
print e
print u"yep"
if pandora_url is None:
pandora_url = driver.base_url
print "Pandora url is "+pandora_url
driver.get(pandora_url+"/pandora_console/index.php")
driver.find_element_by_id("nick").clear()
driver.find_element_by_id("nick").send_keys(user)
driver.find_element_by_id("pass").clear()
driver.find_element_by_id("pass").send_keys(passwd)
driver.find_element_by_id("submit-login_button").click()
def get_menu_element(driver,menu_item_text):
return driver.find_element_by_xpath('//div[@class="menu"]//a[contains(.,"'+menu_item_text+'")]')
@ -64,7 +44,7 @@ def refresh_N_times_until_find_element(driver,n,element_text,how=By.ID,refresh_t
from selenium.common.exceptions import TimeoutException
i = 1
while (1<=n):
while (i<=n):
try:
element = WebDriverWait(driver, refresh_time).until(EC.presence_of_element_located((how, element_text)))
return element
@ -74,17 +54,6 @@ def refresh_N_times_until_find_element(driver,n,element_text,how=By.ID,refresh_t
raise TimeoutException("Element %s not found" % (element_text))
def logout(driver,url):
print u"Logging out"
if url[-1] != '/':
driver.find_element_by_xpath('//div[@id="container"]//a[@href="'+url+'/pandora_console/index.php?bye=bye"]').click()
else:
driver.find_element_by_xpath('//div[@id="container"]//a[@href="'+url+'pandora_console/index.php?bye=bye"]').click()
driver.get(url+"/pandora_console/index.php")
refresh_N_times_until_find_element(driver,2,"nick")
#Pass Wizards
def detect_and_pass_pandorin(driver):
@ -110,7 +79,7 @@ def detect_and_pass_newsletter_wizard(driver):
def detect_and_pass_all_wizards(driver):
driver.implicitly_wait(2) #Optimisation workaround for skipping wizards quickly
detect_and_pass_pandorin(driver)
#detect_and_pass_pandorin(driver)
detect_and_pass_initial_wizard(driver)
detect_and_pass_newsletter_wizard(driver)
driver.implicitly_wait(30)

View File

@ -3,48 +3,107 @@ from unittest import *
from console.include.common_functions_60 import *
from console.include.common_classes_60 import *
from sauceclient import SauceClient
import testtools
from os import environ, getenv
import subprocess, time, sys
def get_test_file(test_list):
#return [test[0].split(' ')[0].split('.')[0].split('<')[1] for test in test_list]
return [test[0].test_name for test in test_list]
#Run Enterprise tests
is_enterprise = '1' == getenv('ENTERPRISE', False)
class ArticaTestSuite(TestSuite):
def __init__(self,*args,**kwargs):
super(ArticaTestSuite,self).__init__(*args,**kwargs)
def run(self,*args,**kwargs):
#print "Running "+str(self.countTestCases())+" tests."
#print "Tests are: "+str(self._tests)
super(ArticaTestSuite,self).run(*args,**kwargs)
a = TestLoader()
tests = a.discover(start_dir='console',pattern='*.py')
c = ArticaTestResult()
tests.run(c)
#Update Saouce Labs jobs
sauce_client = SauceClient(environ["SAUCE_USERNAME"], environ["SAUCE_ACCESS_KEY"])
for test,error_msg in c.failures+c.errors:
try:
sauce_client.jobs.update_job(test.sauce_labs_job_id, passed=False,tags=[environ["TRAVIS_BRANCH"],test.id()],build_num=environ["TRAVIS_JOB_NUMBER"],name=str(environ["TRAVIS_COMMIT"])+"_"+str(test.id().split('.')[1]))
except:
print "Could not annotate Sauce Labs job #%s" % str(test)
next
for test,error_msg in c.success+c.skipped:
try:
sauce_client.jobs.update_job(test.sauce_labs_job_id, passed=True,tags=[environ["TRAVIS_BRANCH"],test.id()],build_num=environ["TRAVIS_JOB_NUMBER"],name=str(environ["TRAVIS_COMMIT"])+"_"+str(test.id().split('.')[1]))
except:
print "Could not annotate Sauce Labs job #%s" % str(test)
next
class ArticaTestLoader(TestLoader):
def __init__(self,*args,**kwargs):
self.suiteClass = ArticaTestSuite
super(ArticaTestLoader, self).__init__(*args,**kwargs)
""" Splits a Test Suite so that no more than 'n' threads will execute the tests """
def split_suite_into_chunks(num_threads, suite):
# Compute num_threads such that the number of threads does not exceed the value passed to the function
# Keep num_threads to a reasonable number of threads
if num_threads < 0: num_threads = 1
if num_threads > 8: num_threads = 8
num_tests = suite.countTestCases()
s = []
s_tmp = ArticaTestSuite()
n = round(num_tests / num_threads)
for case in suite:
if n <= 0 and s_tmp.countTestCases() > 0:
s.append([s_tmp, None])
num_threads -= 1
num_tests -= s_tmp.countTestCases()
s_tmp = ArticaTestSuite()
n = round(num_tests / num_threads)
s_tmp.addTest(case)
n -= 1
if s_tmp.countTestCases() > 0:
if s_tmp.countTestCases() > 0: s.append([s_tmp, None])
num_tests -= s_tmp.countTestCases()
if num_tests != 0: print("Error: num_tests should be 0 but is %s!" % num_tests)
return s
print "Tests failed: %s" % c.failures
print "Tests succeeded: %s" % c.success
print "Tests skipped: %s" % c.skipped
print "Tests with errors: %s" % c.errors
class TracingStreamResult(testtools.StreamResult):
failures = []
success = []
skipped = []
errors = []
def status(self, test_status, test_id, *args, **kwargs):
if test_status=='inprogress':
print "Running test "+str(test_id)
if (len(c.failures)+len(c.errors)) != 0:
elif test_status=='xfail' or test_status=='fail' or test_status=='exists':
print "Test "+str(test_id)+" has failed"
self.failures.append(test_id)
elif test_status=='uxsuccess' or test_status=='success':
print "Test "+str(test_id)+" has succeeded"
self.success.append(test_id)
elif test_status=='exists':
print "Test "+str(test_id)+" has failed (already existed)"
self.errors.append(test_id)
elif test_status=='skip':
print "Test "+str(test_id)+" was skipped"
self.skipped.append('test_id')
is_enterprise = '1' == getenv('ENTERPRISE', False)
if is_enterprise:
num_threads = 2
else:
num_threads = 3
a = ArticaTestLoader()
#Network server tests
tests = a.discover(start_dir='console',pattern='*.py')
print str(tests.countTestCases())+" tests found"
concurrent_suite = testtools.ConcurrentStreamTestSuite(lambda: (split_suite_into_chunks(num_threads, tests)))
result = TracingStreamResult()
try:
result.startTestRun()
finally:
concurrent_suite.run(result)
print "SUMMARY"
print "======="
print "Tests failed: %s" % result.failures
print "Tests succeeded: %s" % result.success
print "Tests skipped: %s" % result.skipped
print "Tests with errors: %s" % result.errors
print "======="
if (len(result.failures)+len(result.errors)) != 0:
sys.exit(1)
else:
sys.exit(0)