Merge pull request #457 from opinkerfi/selenium_rework

Selenium rework

Author: palli

adagios/objectbrowser/tests.py

@@ -27,6 +27,7 @@
 import adagios.utils
 import adagios.objectbrowser.forms
 import re
+import adagios.seleniumtests
 
 from adagios.objectbrowser.forms import PynagAutoCompleteField
 
@@ -534,44 +535,46 @@ def test_set_prefix(self):
         field.set_prefix('')
         self.assertEqual('', field.get_prefix())
 
-class SeleniumObjectBrowserTestCase(adagios.utils.SeleniumTestCase):
+class SeleniumObjectBrowserTestCase(adagios.seleniumtests.SeleniumTestCase):
     def test_contacts_loading(self):
         """Test if contacts under configure loads"""
-        self.driver.get(self.live_server_url + "/objectbrowser/#contact-tab_tab")
+        for driver in self.drivers:
+            driver.get(self.live_server_url + "/objectbrowser/#contact-tab_tab")
 
-        wait = WebDriverWait(self.driver, 10)
+            wait = WebDriverWait(driver, 10)
 
-        try:
-            # Get all host rows
-            contact_table_rows = wait.until(
-                EC.presence_of_all_elements_located((
-                    By.XPATH,
-                    "//table[contains(@id, 'contact-table')]/tbody/tr"))
-            )
-        except TimeoutException:
-            self.assertTrue(False, "Timed out waiting for contact table to load")
-
-        self.assertTrue(len(contact_table_rows) > 0,
-                        "No table rows in contact table")
+            try:
+                # Get all host rows
+                contact_table_rows = wait.until(
+                    EC.presence_of_all_elements_located((
+                        By.XPATH,
+                        "//table[contains(@id, 'contact-table')]/tbody/tr"))
+                )
+            except TimeoutException:
+                self.assertTrue(False, "Timed out waiting for contact table to load")
+
+            self.assertTrue(len(contact_table_rows) > 0,
+                            "No table rows in contact table")
 
     def test_hosts_loading(self):
         """Test if hosts under configure loads"""
-        self.driver.get(self.live_server_url + "/objectbrowser")
-
-        wait = WebDriverWait(self.driver, 10)
-
-        try:
-            # Get all host rows
-            host_table_rows = wait.until(
-                EC.presence_of_all_elements_located((
-                    By.XPATH,
-                    "//table[contains(@id, 'host-table')]/tbody/tr"))
-            )
-        except TimeoutException:
-            self.assertTrue(False, "Timed out waiting for host table to load")
-
-        self.assertTrue(len(host_table_rows) > 0,
-                        "No table rows in host-table")
+        for driver in self.drivers:
+            driver.get(self.live_server_url + "/objectbrowser")
+
+            wait = WebDriverWait(driver, 10)
+
+            try:
+                # Get all host rows
+                host_table_rows = wait.until(
+                    EC.presence_of_all_elements_located((
+                        By.XPATH,
+                        "//table[contains(@id, 'host-table')]/tbody/tr"))
+                )
+            except TimeoutException:
+                self.assertTrue(False, "Timed out waiting for host table to load")
+
+            self.assertTrue(len(host_table_rows) > 0,
+                            "No table rows in host-table")
 
 _TEST_SERVICE = """
 define service {

adagios/seleniumtests.py

@@ -0,0 +1,83 @@
+from django.test import LiveServerTestCase
+from django.utils import unittest
+import adagios.settings
+import adagios.utils
+import os
+
+try:
+    from selenium import webdriver
+    from selenium.common.exceptions import WebDriverException
+except ImportError:
+    webdriver = None
+
+def get_remote_webdriver(capabilities=None):
+    """Get remote webdriver. Configured using environment variables
+    to setup where the remote webdriver is. options could be a local
+    install or using the setup at saucelabs"""
+
+    if not capabilities:
+        capabilities = webdriver.DesiredCapabilities.FIREFOX
+
+    # Saucelabs setup,
+    capabilities["build"] = os.environ.get("TRAVIS_BUILD_NUMBER")
+    capabilities["tags"] = [os.environ.get("TRAVIS_PYTHON_VERSION"), "CI"]
+    capabilities["tunnel-identifier"] = os.environ.get("TRAVIS_JOB_NUMBER")
+
+    username = os.environ.get("SAUCE_USERNAME")
+    access_key = os.environ.get("SAUCE_ACCESS_KEY")
+    hub_url = os.environ.get('SAUCE_HUBURL', "ondemand.saucelabs.com/wd/hub")
+
+    if username and access_key:
+        hub_url = "%s:%s@%s" % (username, access_key, hub_url)
+
+    return webdriver.Remote(desired_capabilities=capabilities,
+                            command_executor="http://%s" % hub_url)
+
+class SeleniumTestCase(LiveServerTestCase):
+    environment = None
+
+    @classmethod
+    def setUpClass(cls):
+        if not webdriver:
+            raise unittest.SkipTest("No selenium installed")
+
+        # Tests for pull requests from forks do not get the SAUCE_USERNAME
+        # exposed because of security considerations. Skip selenium tests for
+        # those tests
+        if os.environ.get('TRAVIS_BUILD_NUMBER') and \
+           not os.environ.get('SAUCE_USERNAME'):
+            raise unittest.SkipTest("Travis with no sauce username, skipping")
+
+        super(SeleniumTestCase, cls).setUpClass()
+
+        cls.drivers = []
+
+        try:
+            if 'SELENIUM_REMOTE_TESTS' in os.environ or \
+               'TRAVIS_BUILD_NUMBER' in os.environ:
+                # Fire up remote webdriver
+                remote = get_remote_webdriver()
+                cls.drivers.append(remote)
+            else:
+                # Use the firefox webdriver
+                firefox = webdriver.Firefox()
+                cls.drivers.append(firefox)
+        except WebDriverException as error:
+            raise unittest.SkipTest("Exception in running webdriver, skipping " \
+                         "selenium tests: %s" % str(error))
+
+        cls.nagios_config = adagios.settings.nagios_config
+        cls.environment = adagios.utils.FakeAdagiosEnvironment()
+        cls.environment.create_minimal_environment()
+        cls.environment.configure_livestatus()
+        cls.environment.update_adagios_global_variables()
+        cls.environment.start()
+        cls.livestatus = cls.environment.get_livestatus()
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.environment.terminate()
+        for driver in cls.drivers:
+            driver.quit()
+        super(SeleniumTestCase, cls).tearDownClass()
+

adagios/status/tests.py

@@ -31,6 +31,7 @@
 import adagios.settings
 import adagios.utils
 import simplejson as json
+import adagios.seleniumtests
 
 try:
     from selenium.webdriver.common.by import By
@@ -179,14 +180,15 @@ def test_get(self):
         self.assertTrue('packetloss' in result[0]['metrics'])
 
 
-class SeleniumStatusTestCase(adagios.utils.SeleniumTestCase):
+class SeleniumStatusTestCase(adagios.seleniumtests.SeleniumTestCase):
     def test_network_parents(self):
         """Status Overview, Network Parents should show an integer"""
-        self.driver.get(self.live_server_url + "/status")
+        for driver in self.drivers:
+            driver.get(self.live_server_url + "/status")
 
-        # Second link is Network Parents in overview
-        self.assertEqual(self.driver.find_elements(By.XPATH,
-            "//a[@href='/status/parents']")[1].text.isdigit(), True)
+            # Second link is Network Parents in overview
+            self.assertEqual(driver.find_elements(By.XPATH,
+                "//a[@href='/status/parents']")[1].text.isdigit(), True)
 
     def test_services_select_all(self):
         """Loads services list and tries to select everything
@@ -197,34 +199,35 @@ def test_services_select_all(self):
             Look for statustable rows
             Assert that all rows are checked"""
 
-        self.driver.get(self.live_server_url + "/status/services")
+        for driver in self.drivers:
+            driver.get(self.live_server_url + "/status/services")
 
-        self.driver.find_element_by_xpath("//input[@class='select_many']").click()
-        self.driver.find_element_by_xpath("//a[@class='select_all']").click()
+            driver.find_element_by_xpath("//input[@class='select_many']").click()
+            driver.find_element_by_xpath("//a[@class='select_all']").click()
 
-        # Get all statustable rows
-        status_table_rows = self.driver.find_element_by_xpath(
-            "//table[contains(@class, 'statustable')]"
-        ).find_elements(By.XPATH, "//tbody/tr[contains(@class, 'mainrow')]")
+            # Get all statustable rows
+            status_table_rows = driver.find_element_by_xpath(
+                "//table[contains(@class, 'statustable')]"
+            ).find_elements(By.XPATH, "//tbody/tr[contains(@class, 'mainrow')]")
 
-        # Sub-select non-selected
-        for row in status_table_rows:
-            self.assertTrue('row_selected' in row.get_attribute('class'),
-                            "Non selected row found after selecting all: " + \
-                            row.text)
+            # Sub-select non-selected
+            for row in status_table_rows:
+                self.assertTrue('row_selected' in row.get_attribute('class'),
+                                "Non selected row found after selecting all: " + \
+                                row.text)
 
     def test_status_overview_top_alert_producers(self):
         """Check the top alert producers part of status overview"""
+        for driver in self.drivers:
+            driver.get(self.live_server_url + "/status")
 
-        self.driver.get(self.live_server_url + "/status")
+            top_alert_table_rows = driver.find_elements(By.XPATH,
+                "//table[@id='top_alert_producers']/tbody/tr"
+            )
 
-        top_alert_table_rows = self.driver.find_elements(By.XPATH,
-            "//table[@id='top_alert_producers']/tbody/tr"
-        )
+            count = 0
+            for row in top_alert_table_rows:
+                if 'display' not in row.get_attribute('style'):
+                    count += 1
 
-        count = 0
-        for row in top_alert_table_rows:
-            if 'display' not in row.get_attribute('style'):
-                count += 1
-
-        self.assertTrue(count <= 3, "Top alert producers returns too many rows")
+            self.assertTrue(count <= 3, "Top alert producers returns too many rows")

adagios/utils.py

@@ -17,19 +17,14 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-import multiprocessing
 import adagios.status.utils
-import time
 import adagios
 import pynag.Model
 import adagios.exceptions
 import adagios.settings
 import os
 import pynag.Utils.misc
-from django.test import LiveServerTestCase
-from django.utils import unittest
-import atexit
-
+from multiprocessing.pool import ThreadPool
 from django.utils.translation import ugettext as _
 
 SELENIUM_DRIVER = None
@@ -56,7 +51,6 @@ def wait_for_service(host_name, service_description, condition='last_check >= 0'
         WaitObject=waitobject
     )
 
-from multiprocessing.pool import ThreadPool
 
 
 class Task(object):
@@ -175,53 +169,3 @@ def terminate(self):
         super(FakeAdagiosEnvironment, self).terminate()
 
 
-class SeleniumTestCase(LiveServerTestCase):
-    driver = None
-    environment = None
-
-    @classmethod
-    def setUpClass(cls):
-        global SELENIUM_DRIVER
-        try:
-            from selenium import webdriver
-        except ImportError:
-            raise unittest.SkipTest("No selenium installed")
-
-        super(SeleniumTestCase, cls).setUpClass()
-
-        cls.nagios_config = adagios.settings.nagios_config
-        cls.environment = adagios.utils.FakeAdagiosEnvironment()
-        cls.environment.create_minimal_environment()
-        cls.environment.configure_livestatus()
-        cls.environment.update_adagios_global_variables()
-        cls.environment.start()
-        cls.livestatus = cls.environment.get_livestatus()
-
-        if not SELENIUM_DRIVER:
-            if 'TRAVIS' in os.environ:
-                capabilities = webdriver.DesiredCapabilities.CHROME
-                capabilities["build"] = os.environ["TRAVIS_BUILD_NUMBER"]
-                capabilities["tags"] = [os.environ["TRAVIS_PYTHON_VERSION"], "CI"]
-                capabilities["tunnel-identifier"] = os.environ["TRAVIS_JOB_NUMBER"]
-                capabilities['platform'] = "Windows 8.1"
-                capabilities['version'] = "31"
-
-                username = os.environ["SAUCE_USERNAME"]
-                access_key = os.environ["SAUCE_ACCESS_KEY"]
-
-                hub_url = "%s:%s@ondemand.saucelabs.com/wd/hub" % (username, access_key)
-                SELENIUM_DRIVER = webdriver.Remote(desired_capabilities=capabilities, command_executor="http://%s" % hub_url)
-            else:
-                SELENIUM_DRIVER = webdriver.Firefox()
-            # Exit browser when all tests are done
-            atexit.register(SELENIUM_DRIVER.quit)
-
-
-
-        cls.driver = SELENIUM_DRIVER
-
-    @classmethod
-    def tearDownClass(cls):
-        cls.environment.terminate()
-        super(SeleniumTestCase, cls).tearDownClass()
-