+ def test_pushdb_no_deploy_scenario(self):
+ self._test_pushdb_missing_env('DEPLOY_SCENARIO')
+
+ def test_pushdb_no_node_name(self):
+ self._test_pushdb_missing_env('NODE_NAME')
+
+ def test_pushdb_no_build_tag(self):
+ self._test_pushdb_missing_env('BUILD_TAG')
+
+ @mock.patch('requests.post')
+ def test_pushdb_bad_start_time(self, mock_function=None):
+ self.test.start_time = "1"
+ self.assertEqual(
+ self.test.push_to_db(),
+ testcase.TestCase.EX_PUSH_TO_DB_ERROR)
+ mock_function.assert_not_called()
+
+ @mock.patch('requests.post')
+ def test_pushdb_bad_end_time(self, mock_function=None):
+ self.test.stop_time = "2"
+ self.assertEqual(
+ self.test.push_to_db(),
+ testcase.TestCase.EX_PUSH_TO_DB_ERROR)
+ mock_function.assert_not_called()
+
+ def _get_data(self):
+ return {
+ "build_tag": os.environ['BUILD_TAG'],
+ "case_name": self._case_name,
+ "criteria": 'PASS' if self.test.is_successful(
+ ) == self.test.EX_OK else 'FAIL',
+ "details": self.test.details,
+ "installer": os.environ['INSTALLER_TYPE'],
+ "pod_name": os.environ['NODE_NAME'],
+ "project_name": self.test.project_name,
+ "scenario": os.environ['DEPLOY_SCENARIO'],
+ "start_date": datetime.fromtimestamp(
+ self.test.start_time).strftime('%Y-%m-%d %H:%M:%S'),
+ "stop_date": datetime.fromtimestamp(
+ self.test.stop_time).strftime('%Y-%m-%d %H:%M:%S'),
+ "version": "master"}
+
+ @mock.patch('requests.post')
+ def _test_pushdb_version(self, mock_function=None, **kwargs):
+ payload = self._get_data()
+ payload["version"] = kwargs.get("version", "unknown")
+ self.assertEqual(self.test.push_to_db(), testcase.TestCase.EX_OK)
+ mock_function.assert_called_once_with(
+ os.environ['TEST_DB_URL'],
+ data=json.dumps(payload, sort_keys=True),
+ headers=self._headers)
+
+ def test_pushdb_daily_job(self):
+ self._test_pushdb_version(version="master")
+
+ def test_pushdb_weekly_job(self):
+ os.environ['BUILD_TAG'] = 'foo-weekly-master-bar'
+ self._test_pushdb_version(version="master")
+
+ def test_pushdb_random_build_tag(self):
+ os.environ['BUILD_TAG'] = 'whatever'
+ self._test_pushdb_version(version="unknown")
+
+ @mock.patch('requests.post', return_value=mock.Mock(
+ raise_for_status=mock.Mock(
+ side_effect=requests.exceptions.HTTPError)))
+ def test_pushdb_http_errors(self, mock_function=None):
+ self.assertEqual(
+ self.test.push_to_db(),
+ testcase.TestCase.EX_PUSH_TO_DB_ERROR)