-
Notifications
You must be signed in to change notification settings - Fork 15
/
ibis_test_suite.py
138 lines (128 loc) · 6.7 KB
/
ibis_test_suite.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
"""Unit test suite runner."""
import unittest
import os
import sys
from ibis.driver.tests.test_driver import DriverFunctionsTest
from ibis.driver.tests.test_main import MainFunctionsTest
from ibis.inventor.tests.test_workflow_generator \
import WorkflowGeneratorFunctionsTest
from ibis.inventor.tests.test_action_builder import ActionBuilderFunctionsTest
from ibis.inventor.tests.test_dsl_parser import DSLParserTest
from ibis.inventory.tests.test_request_inventory \
import request_inventory_test_suite
from ibis.inventory.tests.test_inventory import InventoryFunctionsTest
from ibis.inventory.tests.test_it_inventory import ITInventoryFunctionsTest
from ibis.inventory.tests.test_cb_inventory import CBInventoryFunctionsTest
from ibis.inventory.tests.test_perf_inventory import PerfInventoryTest
from ibis.inventory.tests.test_automation_ids_inventory \
import AutoInventoryFunctionsTest
from ibis.model.tests.test_shell_action import ShellActionFunctionsTest
from ibis.model.tests.test_sqoop_action import SqoopActionFunctionsTest
from ibis.model.tests.test_hive_action import HiveActionFunctionsTest
from ibis.model.tests.test_freq_ingest import Test_freq_ingest
from ibis.model.tests.test_join_control import JoinActionFunctionsTest
from ibis.model.tests.test_fork_control import ForkActionFunctionsTest
from ibis.model.tests.test_ssh_action import SSHActionFunctionsTest
from ibis.model.tests.test_subwf_action import SubWFActionFunctionsTest
from ibis.model.tests.test_table import tableSuiteTest
from ibis.utilities.tests.test_utilities import UtilitiesFunctionsTest
from ibis.utilities.tests.test_sqoop_helper import SqoopHelperFunctionsTest
from ibis.utilities.tests.test_vizoozie import VizOozieTest
from ibis.utilities.tests.test_it_table_generation \
import it_table_gen_test_suite
from ibis.utilities.tests.test_config_manager import ConfigManagerTest
from ibis.utilities.tests.test_file_parser import FileParserTest
from ibis.utilities.tests.test_sqoop_auth_check import AuthTestTest
from lib.ingest.tests.test_checks_and_balances_export \
import TestChecksBalancesExportManager
from lib.ingest.tests.test_parquet_opt_ddl_time \
import ParquetOptTimeFunctionsTest
from lib.ingest.tests.test_quality_assurance import qa_test_suite
from lib.ingest.tests.test_pre_quality_assurance_export \
import qa_pre_test_suite_export
from lib.ingest.tests.test_quality_assurance_export import qa_test_suite_export
from lib.ingest.tests.test_oozie_ws_helper import oozie_ws_helper_test_suite
from lib.ingest.tests.test_zookeeper_remove_locks \
import ZookeeperLocksFunctionsTest
from lib.ingest.tests.test_impala_utils import ImpalaUtilsFunctionsTest
from lib.ingest.tests.test_import_prep import ImportPrepFunctionsTest
from lib.ingest.tests.test_py_hdfs import PyHDFSTest
def remove_pyc_files():
"""removes .pyc files"""
curr_dir = os.path.dirname(os.path.abspath(__file__))
pyc_files = []
for root, _, filenames in os.walk(curr_dir):
for file_name in filenames:
if len(file_name) >= 4 and file_name[-4:] == '.pyc':
full_file_name = os.path.join(root, file_name)
pyc_files.append(full_file_name)
cnt = 0
for file_name in pyc_files:
try:
os.remove(file_name)
cnt += 1
except OSError:
print 'Could not delete file:{0}'.format(file_name)
print "Deleted {0}/{1} .pyc files".format(cnt, len(pyc_files))
# Main test suite for running all of the test classes
if __name__ == '__main__':
remove_pyc_files()
# List of test classes
test_classes_to_run = [ActionBuilderFunctionsTest, ForkActionFunctionsTest,
HiveActionFunctionsTest, JoinActionFunctionsTest,
ShellActionFunctionsTest, SqoopActionFunctionsTest,
InventoryFunctionsTest, ITInventoryFunctionsTest,
SqoopHelperFunctionsTest, FileParserTest,
UtilitiesFunctionsTest, VizOozieTest,
ParquetOptTimeFunctionsTest,
DriverFunctionsTest, WorkflowGeneratorFunctionsTest,
AutoInventoryFunctionsTest,
CBInventoryFunctionsTest, ConfigManagerTest,
AuthTestTest,
ZookeeperLocksFunctionsTest,
MainFunctionsTest, SubWFActionFunctionsTest,
SSHActionFunctionsTest, ImpalaUtilsFunctionsTest,
TestChecksBalancesExportManager, DSLParserTest,
PyHDFSTest, ImportPrepFunctionsTest,
PerfInventoryTest, Test_freq_ingest]
# test_classes_to_run = []
loader = unittest.TestLoader()
# Create test suite
suites_list = []
for test_class in test_classes_to_run:
suite = loader.loadTestsFromTestCase(test_class)
suites_list.append(suite)
# Add module level test suites here
suites_list += qa_test_suite
suites_list += qa_test_suite_export
suites_list += qa_pre_test_suite_export
suites_list += it_table_gen_test_suite
suites_list += request_inventory_test_suite
suites_list += tableSuiteTest
suites_list += oozie_ws_helper_test_suite
big_suite = unittest.TestSuite(suites_list)
runner = unittest.TextTestRunner()
# Run test suite
results = runner.run(big_suite)
if len(results.errors) != 0 or len(results.failures) != 0:
print 'Unit tests failed!'
print """\n\033[91m
,-------------. \001
( Tests failed! ) .-. \001
`-------------' _ \ \ \001
(_) \ \ \001
O | | \001
o | | \001
. /\---/\ _,---._ | | \001
/^ ^ \,' `. ; \001
( O O ) ; \001
`.=o=__,' \ \001
/ _,--.__ \ \001
/ _ ) ,' `-. `-. \ \001
/ ,' / ,' \ \ \ \ \001
/ / / ,' (,_)(,_) \001
(,; (,,) \001
\033[0m"""
sys.exit(1)
else:
print 'All tests passed.'