Renovate the MySQL DB system

Author: mdmintz

seleniumbase/fixtures/base_case.py

@@ -75,6 +75,7 @@ def __init__(self, *args, **kwargs):
         self.__last_page_screenshot_png = None
         self.__last_page_url = None
         self.__last_page_source = None
+        self.__skip_reason = None
         self.__added_pytest_html_extra = None
         self.__deferred_assert_count = 0
         self.__deferred_assert_failures = []
@@ -3283,9 +3284,13 @@ def set_time_limit(self, time_limit):
 
     def skip(self, reason=""):
         """ Mark the test as Skipped. """
-        if self.dashboard:
+        if self.dashboard or (self.is_pytest and self.with_db_reporting):
             test_id = self.__get_test_id_2()
             sb_config._results[test_id] = "Skipped"
+            self.__skip_reason = reason
+        elif reason and not self.is_pytest:
+            # Only needed for nosetest db reporting
+            self._nose_skip_reason = reason
         self.skipTest(reason)
 
     ############
@@ -6368,7 +6373,8 @@ def setUp(self, masterqa_mode=False):
                     self._testMethodName)
                 data_payload.env = application.split('.')[0]
                 data_payload.start_time = application.split('.')[1]
-                data_payload.state = constants.State.NOTRUN
+                data_payload.state = constants.State.UNTESTED
+                self.__skip_reason = None
                 self.testcase_manager.insert_testcase_data(data_payload)
                 self.case_start_time = int(time.time() * 1000)
             if self.headless:
@@ -6559,6 +6565,21 @@ def __set_last_page_source(self):
             except Exception:
                 self.__last_page_source = None
 
+    def __get_exception_info(self):
+        exc_message = None
+        if sys.version_info[0] >= 3 and hasattr(self, '_outcome') and (
+                hasattr(self._outcome, 'errors') and self._outcome.errors):
+            try:
+                exc_message = self._outcome.errors[0][1][1]
+            except Exception:
+                exc_message = "(Unknown Exception)"
+        else:
+            try:
+                exc_message = sys.last_value
+            except Exception:
+                exc_message = "(Unknown Exception)"
+        return str(exc_message)
+
     def __insert_test_result(self, state, err):
         from seleniumbase.core.testcase_manager import TestcaseDataPayload
         data_payload = TestcaseDataPayload()
@@ -6577,7 +6598,15 @@ def __insert_test_result(self, state, err):
             elif "Error: " in tb_string:
                 data_payload.message = tb_string.split("Error: ")[-1]
             else:
-                data_payload.message = "Unknown Error: See Stacktrace"
+                data_payload.message = self.__get_exception_info()
+        else:
+            test_id = self.__get_test_id_2()
+            if self.is_pytest and test_id in sb_config._results.keys() and (
+                    sb_config._results[test_id] == "Skipped"):
+                if self.__skip_reason:
+                    data_payload.message = "Skipped:   " + self.__skip_reason
+                else:
+                    data_payload.message = "Skipped:   (no reason given)"
         self.testcase_manager.update_testcase_data(data_payload)
 
     def __add_pytest_html_extra(self):
@@ -7023,9 +7052,16 @@ def tearDown(self):
                     self.display = None
             if self.with_db_reporting:
                 if has_exception:
-                    self.__insert_test_result(constants.State.ERROR, True)
+                    self.__insert_test_result(constants.State.FAILED, True)
                 else:
-                    self.__insert_test_result(constants.State.PASS, False)
+                    test_id = self.__get_test_id_2()
+                    if test_id in sb_config._results.keys() and (
+                            sb_config._results[test_id] == "Skipped"):
+                        self.__insert_test_result(
+                            constants.State.SKIPPED, False)
+                    else:
+                        self.__insert_test_result(
+                            constants.State.PASSED, False)
                 runtime = int(time.time() * 1000) - self.execution_start_time
                 self.testcase_manager.update_execution_data(
                     self.execution_guid, runtime)

seleniumbase/fixtures/constants.py

@@ -268,10 +268,10 @@ class Browser:
 
 
 class State:
-    NOTRUN = "NotRun"
+    PASSED = "Passed"
+    FAILED = "Failed"
+    SKIPPED = "Skipped"
+    UNTESTED = "Untested"
     ERROR = "Error"
-    FAILURE = "Fail"
-    PASS = "Pass"
-    SKIP = "Skip"
     BLOCKED = "Blocked"
     DEPRECATED = "Deprecated"

seleniumbase/fixtures/errors.py

@@ -1,8 +1,12 @@
 """
-SeleniumBase MySQL plugin-related exceptions.
+SeleniumBase MySQL-related exceptions.
+
+This feature is DEPRECATED!
+Use self.skip() for skipping tests!
+
 Raising one of these in a test will cause the
 test-state to be logged appropriately in the DB
-for tests that use the SeleniumBase MySQL plugin.
+for tests that use the SeleniumBase MySQL option.
 """
 
 

seleniumbase/plugins/db_reporting_plugin.py

@@ -30,6 +30,8 @@ def __init__(self):
         self.application = None
         self.testcase_manager = None
         self.error_handled = False
+        self._result_set = False
+        self._test = None
 
     def options(self, parser, env):
         super(DBReporting, self).options(parser, env=env)
@@ -77,61 +79,85 @@ def startTest(self, test):
         application = ApplicationManager.generate_application_string(test)
         data_payload.env = application.split('.')[0]
         data_payload.start_time = application.split('.')[1]
-        data_payload.state = constants.State.NOTRUN
+        data_payload.state = constants.State.UNTESTED
         self.testcase_manager.insert_testcase_data(data_payload)
         self.case_start_time = int(time.time() * 1000)
         # Make the testcase guid available to other plugins
         test.testcase_guid = self.testcase_guid
+        self._test = test
+        self._test._nose_skip_reason = None
 
     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 test run, we want to
+            update the DB row with the total execution time. """
         runtime = int(time.time() * 1000) - self.execution_start_time
-        self.testcase_manager.update_execution_data(self.execution_guid,
-                                                    runtime)
+        self.testcase_manager.update_execution_data(
+            self.execution_guid, runtime)
+
+    def afterTest(self, test):
+        if not self._result_set:
+            err = None
+            try:
+                err = self._test._nose_skip_reason
+                if err:
+                    err = "Skipped:   " + str(err)
+                    err = (err, err)
+            except Exception:
+                pass
+            if not err:
+                err = "Skipped:   (no reason given)"
+                err = (err, err)
+            self.__insert_test_result(constants.State.SKIPPED, self._test, err)
 
     def addSuccess(self, test, capt):
         """
-        After test completion, we want to record testcase run information.
+        After each test success, record testcase run information.
         """
-        self.__insert_test_result(constants.State.PASS, test)
+        self.__insert_test_result(constants.State.PASSED, test)
+        self._result_set = True
+
+    def addFailure(self, test, err, capt=None, tbinfo=None):
+        """
+        After each test failure, record testcase run information.
+        """
+        self.__insert_test_result(constants.State.FAILED, test, err)
+        self._result_set = True
 
     def addError(self, test, err, capt=None):
         """
-        After a test error, we want to record testcase run information.
+        After each test error, record testcase run information.
+        (Test errors should be treated the same as test failures.)
         """
-        self.__insert_test_result(constants.State.ERROR, test, err)
+        self.__insert_test_result(constants.State.FAILED, test, err)
+        self._result_set = True
 
     def handleError(self, test, err, capt=None):
         """
-        After a test error, we want to record testcase run information.
+        After each test error, record testcase run information.
         "Error" also encompasses any states other than Pass or Fail, so we
         check for those first.
         """
         if err[0] == errors.BlockedTest:
             self.__insert_test_result(constants.State.BLOCKED, test, err)
+            self._result_set = True
             self.error_handled = True
             raise SkipTest(err[1])
             return True
 
         elif err[0] == errors.DeprecatedTest:
             self.__insert_test_result(constants.State.DEPRECATED, test, err)
+            self._result_set = True
             self.error_handled = True
             raise SkipTest(err[1])
             return True
 
         elif err[0] == errors.SkipTest:
-            self.__insert_test_result(constants.State.SKIP, test, err)
+            self.__insert_test_result(constants.State.SKIPPED, test, err)
+            self._result_set = True
             self.error_handled = True
             raise SkipTest(err[1])
             return True
 
-    def addFailure(self, test, err, capt=None, tbinfo=None):
-        """
-        After a test failure, we want to record testcase run information.
-        """
-        self.__insert_test_result(constants.State.FAILURE, test, err)
-
     def __insert_test_result(self, state, test, err=None):
         data_payload = TestcaseDataPayload()
         data_payload.runtime = int(time.time() * 1000) - self.case_start_time