Revert "Merge branch 'testing_framework' into 'develop'

This reverts commit f8ec880e2a, reversing
changes made to dcb18bcbd2.
This commit is contained in:
axl89 2016-08-22 11:17:02 +02:00
parent cf2f97d43b
commit 3a03b7e46c
11 changed files with 193 additions and 271 deletions

View File

View File

@ -26,6 +26,7 @@ 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,7 +42,6 @@ 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")
@ -55,7 +54,6 @@ 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")
@ -66,7 +64,6 @@ 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")
@ -77,7 +74,6 @@ class Collections(PandoraWebDriverTestCase):
def test_E_delete_collection(self):
driver = self.driver
self.login()
delete_collection(driver,self.new_collection_name)

View File

@ -1,3 +1,4 @@
# -*- 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, refresh_N_times_until_find_element, detect_and_pass_all_wizards, gen_random_string
from include.common_functions_60 import login, click_menu_element, refresh_N_times_until_find_element, detect_and_pass_all_wizards
from include.agent_functions import create_agent
from selenium import webdriver
from selenium.webdriver.common.by import By
@ -11,21 +11,18 @@ from include.module_functions import *
import unittest, time, re
class SimpleICMPTest(PandoraWebDriverTestCase):
class PAN3(PandoraWebDriverTestCase):
test_name = u'Simple ICMP Test'
test_name = u'PAN_3'
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,agent_name,ip="127.0.0.1")
create_agent(driver,"PAN3_agent",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")
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")
driver.find_element_by_xpath('//*[@id="menu_tab"]//a[contains(@href,"ver_agente")]').click()

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, refresh_N_times_until_find_element
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 *
@ -16,18 +16,13 @@ from selenium.webdriver.remote.webelement import WebElement
import unittest, time, re, datetime
class Creation(PandoraWebDriverTestCase):
class PAN13(PandoraWebDriverTestCase):
test_name = u'Planned downtime creation'
test_description = u'Planed downtime creation test'
test_name = u'Planned_downtime'
test_description = u'Planed downtime test'
tickets_associated = []
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):
def avoid_test_A_create_planned_downtime_Quiet(self):
u"""
Create and search planned downtime quiet
@ -36,63 +31,135 @@ class Creation(PandoraWebDriverTestCase):
self.login()
detect_and_pass_all_wizards(driver)
planned_name = self.quiet_name
planned_name = gen_random_string(6)
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('//tr[@id="table3-0"]/td[contains(.,"'+planned_name+'")]')
self.assertIsInstance(element,WebElement)
def test_B_create_planned_downtime_disabled_agents(self):
element = driver.find_element_by_xpath('//img[@data-title="Running"]')
self.assertIsInstance(element,WebElement)
def avoid_test_B_create_planned_downtime_disabled_agents(self):
u"""
Create and search planned downtime disabled agents
"""
driver = self.driver
self.login()
planned_name = self.disabled_agents_name
planned_name = gen_random_string(6)
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('//tr[@id="table3-0"]/td[contains(.,"'+planned_name+'")]')
element = driver.find_element_by_xpath('//img[@data-title="Running"]')
self.assertIsInstance(element,WebElement)
def test_C_create_planned_downtime_disabled_only_alerts(self):
def avoid_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 = self.disabled_only_alerts_name
planned_name = gen_random_string(6)
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('//tr[@id="table3-0"]/td[contains(.,"'+planned_name+'")]')
element = driver.find_element_by_xpath('//img[@data-title="Running"]')
self.assertIsInstance(element,WebElement)
def test_D_delete_planned_downtime(self):
def avoid_test_D_delete_planned_downtime(self):
driver=self.driver
self.login()
downtime_list = [self.disabled_only_alerts_name,self.disabled_agents_name,self.quiet_name]
planned_name = gen_random_string(6)
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)
create_planned_downtime(driver,planned_name,"Applications","Quiet","Once",description=planned_name)
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,7 +48,6 @@ 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")
@ -68,7 +67,6 @@ class Policies(PandoraWebDriverTestCase):
driver = self.driver
self.login()
create_policy(driver,policy_name,"Applications",description="Policy for test")
@ -95,7 +93,6 @@ 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

@ -1,100 +0,0 @@
# -*- 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

@ -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,28 +36,34 @@ 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)
def setUp(self):
if self.is_development != False:
self.driver = webdriver.Firefox()
self.base_url = os.getenv('DEVELOPMENT_URL')
if cls.is_development != False:
cls.driver = webdriver.Firefox()
cls.base_url = os.getenv('DEVELOPMENT_URL')
else:
#Start VM in Sauce Labs
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/"
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.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):
@ -82,21 +88,16 @@ class PandoraWebDriverTestCase(TestCase):
finally: self.accept_next_alert = True
def tearDown(self):
self.driver.quit()
ok = self.currentResult.wasSuccessful()
errors = self.currentResult.errors
failures = self.currentResult.failures
skipped = self.currentResult.skipped
tack = datetime.now()
diff = tack - self.time_started
self.time_elapsed = diff.seconds
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
self.assertEqual([], self.verificationErrors)
super(PandoraWebDriverTestCase, self).tearDown()
def login(self,user="admin",passwd="pandora",pandora_url=None):
print u"Logging in"
driver = self.driver
@ -108,19 +109,10 @@ 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
@ -135,4 +127,3 @@ class PandoraWebDriverTestCase(TestCase):
driver.get(pandora_url+"/pandora_console/index.php")
refresh_N_times_until_find_element(driver,2,"nick")

View File

@ -34,6 +34,26 @@ 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+'")]')
@ -44,7 +64,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 (i<=n):
while (1<=n):
try:
element = WebDriverWait(driver, refresh_time).until(EC.presence_of_element_located((how, element_text)))
return element
@ -54,6 +74,17 @@ 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):
@ -79,7 +110,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,107 +3,48 @@ 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]
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)
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
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)
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')
#Run Enterprise tests
is_enterprise = '1' == getenv('ENTERPRISE', False)
if is_enterprise:
num_threads = 2
else:
num_threads = 3
a = ArticaTestLoader()
#Network server tests
a = TestLoader()
tests = a.discover(start_dir='console',pattern='*.py')
print str(tests.countTestCases())+" tests found"
c = ArticaTestResult()
tests.run(c)
concurrent_suite = testtools.ConcurrentStreamTestSuite(lambda: (split_suite_into_chunks(num_threads, tests)))
result = TracingStreamResult()
#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
try:
result.startTestRun()
finally:
concurrent_suite.run(result)
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
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:
if (len(c.failures)+len(c.errors)) != 0:
sys.exit(1)
else:
sys.exit(0)