MCPcopy Create free account
hub / github.com/apache/kafka / SystemTestEnv

Class SystemTestEnv

system_test/system_test_env.py:30–136  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

28from utils import system_test_utils
29
30class SystemTestEnv():
31
32 # private:
33 _cwdFullPath = os.getcwd()
34 _thisScriptFullPathName = os.path.realpath(__file__)
35 _thisScriptBaseDir = os.path.abspath(os.path.join(os.path.dirname(sys.argv[0])))
36
37 # public:
38 SYSTEM_TEST_BASE_DIR = os.path.abspath(_thisScriptBaseDir)
39 SYSTEM_TEST_UTIL_DIR = os.path.abspath(SYSTEM_TEST_BASE_DIR + "/utils")
40 SYSTEM_TEST_SUITE_SUFFIX = "_testsuite"
41 SYSTEM_TEST_CASE_PREFIX = "testcase_"
42 SYSTEM_TEST_MODULE_EXT = ".py"
43 CLUSTER_CONFIG_FILENAME = "cluster_config.json"
44 CLUSTER_CONFIG_PATHNAME = os.path.abspath(SYSTEM_TEST_BASE_DIR + "/" + CLUSTER_CONFIG_FILENAME)
45 METRICS_FILENAME = "metrics.json"
46 METRICS_PATHNAME = os.path.abspath(SYSTEM_TEST_BASE_DIR + "/" + METRICS_FILENAME)
47 TESTCASE_TO_RUN_FILENAME = "testcase_to_run.json"
48 TESTCASE_TO_RUN_PATHNAME = os.path.abspath(SYSTEM_TEST_BASE_DIR + "/" + TESTCASE_TO_RUN_FILENAME)
49 TESTCASE_TO_SKIP_FILENAME = "testcase_to_skip.json"
50 TESTCASE_TO_SKIP_PATHNAME = os.path.abspath(SYSTEM_TEST_BASE_DIR + "/" + TESTCASE_TO_SKIP_FILENAME)
51
52 clusterEntityConfigDictList = [] # cluster entity config for current level
53 clusterEntityConfigDictListInSystemTestLevel = [] # cluster entity config defined in system level
54 clusterEntityConfigDictListLastFoundInTestSuite = [] # cluster entity config last found in testsuite level
55 clusterEntityConfigDictListLastFoundInTestCase = [] # cluster entity config last found in testcase level
56
57 systemTestResultsList = []
58 testCaseToRunListDict = {}
59 testCaseToSkipListDict = {}
60
61 printTestDescriptionsOnly = False
62 doNotValidateRemoteHost = False
63
64 def __init__(self):
65 "Create an object with this system test session environment"
66
67 # load the system level cluster config
68 system_test_utils.load_cluster_config(self.CLUSTER_CONFIG_PATHNAME, self.clusterEntityConfigDictList)
69
70 # save the system level cluster config
71 self.clusterEntityConfigDictListInSystemTestLevel = copy.deepcopy(self.clusterEntityConfigDictList)
72
73 # retrieve testcases to run from testcase_to_run.json
74 try:
75 testcaseToRunFileContent = open(self.TESTCASE_TO_RUN_PATHNAME, "r").read()
76 testcaseToRunData = json.loads(testcaseToRunFileContent)
77 for testClassName, caseList in testcaseToRunData.items():
78 self.testCaseToRunListDict[testClassName] = caseList
79 except:
80 pass
81
82 # retrieve testcases to skip from testcase_to_skip.json
83 try:
84 testcaseToSkipFileContent = open(self.TESTCASE_TO_SKIP_PATHNAME, "r").read()
85 testcaseToSkipData = json.loads(testcaseToSkipFileContent)
86 for testClassName, caseList in testcaseToSkipData.items():
87 self.testCaseToSkipListDict[testClassName] = caseList

Callers 1

mainFunction · 0.90

Calls 1

joinMethod · 0.80

Tested by 1

mainFunction · 0.72