You can not select more than 25 topics
			Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
		
		
		
		
			
				
					92 lines
				
				3.2 KiB
			
		
		
			
		
	
	
					92 lines
				
				3.2 KiB
			| 
								 
											3 years ago
										 
									 | 
							
								"""
							 | 
						||
| 
								 | 
							
								unit-test
							 | 
						||
| 
								 | 
							
								"""
							 | 
						||
| 
								 | 
							
								import unittest
							 | 
						||
| 
								 | 
							
								import inspect
							 | 
						||
| 
								 | 
							
								import utils.gen_tool
							 | 
						||
| 
								 | 
							
								import basic.program
							 | 
						||
| 
								 | 
							
								import basic.Testserver
							 | 
						||
| 
								 | 
							
								import test.testtools
							 | 
						||
| 
								 | 
							
								import basic.application
							 | 
						||
| 
								 | 
							
								import basic.constants as B
							 | 
						||
| 
								 | 
							
								import utils.path_const as P
							 | 
						||
| 
								 | 
							
								
							 | 
						||
| 
								 | 
							
								# the list of TEST_FUNCTIONS defines which function will be really tested.
							 | 
						||
| 
								 | 
							
								# if you minimize the list you can check the specific test-function
							 | 
						||
| 
								 | 
							
								TEST_FUNCTIONS = ["test_01createTestserver", "test_02getDBSchema", "test_11createDBTables", "test_11syncApplication"]
							 | 
						||
| 
								 | 
							
								TEST_FUNCTIONS = ["test_02getDBSchema"]
							 | 
						||
| 
								 | 
							
								# with this variable you can switch prints on and off
							 | 
						||
| 
								 | 
							
								verbose = False
							 | 
						||
| 
								 | 
							
								
							 | 
						||
| 
								 | 
							
								
							 | 
						||
| 
								 | 
							
								
							 | 
						||
| 
								 | 
							
								class MyTestCase(unittest.TestCase):
							 | 
						||
| 
								 | 
							
								    mymsg = "--------------------------------------------------------------"
							 | 
						||
| 
								 | 
							
								
							 | 
						||
| 
								 | 
							
								    def test_01createTestserver(self):
							 | 
						||
| 
								 | 
							
								        global mymsg
							 | 
						||
| 
								 | 
							
								        actfunction = str(inspect.currentframe().f_code.co_name)
							 | 
						||
| 
								 | 
							
								        cnttest = 0
							 | 
						||
| 
								 | 
							
								        if actfunction not in TEST_FUNCTIONS:
							 | 
						||
| 
								 | 
							
								            return
							 | 
						||
| 
								 | 
							
								        job = test.testtools.getJob()
							 | 
						||
| 
								 | 
							
								        testserver = basic.Testserver.Testserver(job)
							 | 
						||
| 
								 | 
							
								        self.assertIsNotNone(testserver)
							 | 
						||
| 
								 | 
							
								        cnttest += 1
							 | 
						||
| 
								 | 
							
								        if B.TOPIC_NODE_DB in job.conf.confs:
							 | 
						||
| 
								 | 
							
								            self.assertIn(B.TOPIC_NODE_DB, testserver.conf[B.SUBJECT_CONN])
							 | 
						||
| 
								 | 
							
								            self.assertIn(B.ATTR_DB_DATABASE, testserver.conf[B.SUBJECT_CONN][B.TOPIC_NODE_DB])
							 | 
						||
| 
								 | 
							
								        self.assertIn(B.DATA_NODE_DDL, testserver.conf)
							 | 
						||
| 
								 | 
							
								        self.assertIn("application", testserver.conf[B.DATA_NODE_DDL])
							 | 
						||
| 
								 | 
							
								        MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest)
							 | 
						||
| 
								 | 
							
								
							 | 
						||
| 
								 | 
							
								    def test_02getDBSchema(self):
							 | 
						||
| 
								 | 
							
								        global mymsg
							 | 
						||
| 
								 | 
							
								        actfunction = str(inspect.currentframe().f_code.co_name)
							 | 
						||
| 
								 | 
							
								        cnttest = 0
							 | 
						||
| 
								 | 
							
								        if actfunction not in TEST_FUNCTIONS:
							 | 
						||
| 
								 | 
							
								            return
							 | 
						||
| 
								 | 
							
								        job = test.testtools.getJob()
							 | 
						||
| 
								 | 
							
								        testserver = basic.Testserver.Testserver(job)
							 | 
						||
| 
								 | 
							
								        if B.TOPIC_NODE_DB in job.conf.confs:
							 | 
						||
| 
								 | 
							
								            dbi = basic.toolHandling.getDbTool(job, self, job.conf.confs[B.TOPIC_NODE_DB][B.ATTR_TYPE])
							 | 
						||
| 
								 | 
							
								        else:
							 | 
						||
| 
								 | 
							
								            return "No DB in job-config"
							 | 
						||
| 
								 | 
							
								        sql = testserver.getDBSchema(job, dbi, "application")
							 | 
						||
| 
								 | 
							
								        print(sql+"##")
							 | 
						||
| 
								 | 
							
								        lines = sql.split("\n")
							 | 
						||
| 
								 | 
							
								        self.assertEqual(4, len(lines))
							 | 
						||
| 
								 | 
							
								        self.assertIn("CREATE TABLE", lines[0])
							 | 
						||
| 
								 | 
							
								        self.assertIn("CREATE INDEX", lines[1])
							 | 
						||
| 
								 | 
							
								        sql = testserver.getDBSchema(job, dbi, "ap_project")
							 | 
						||
| 
								 | 
							
								        print(sql+"##")
							 | 
						||
| 
								 | 
							
								        lines = sql.split("\n")
							 | 
						||
| 
								 | 
							
								        self.assertEqual(4, len(lines))
							 | 
						||
| 
								 | 
							
								        self.assertIn("CREATE TABLE", lines[0])
							 | 
						||
| 
								 | 
							
								        self.assertIn("CREATE INDEX", lines[1])
							 | 
						||
| 
								 | 
							
								        sql = testserver.getDBSchema(job, dbi, "ap_component")
							 | 
						||
| 
								 | 
							
								        print(sql+"##")
							 | 
						||
| 
								 | 
							
								        lines = sql.split("\n")
							 | 
						||
| 
								 | 
							
								        self.assertEqual(4, len(lines))
							 | 
						||
| 
								 | 
							
								        self.assertIn("CREATE TABLE", lines[0])
							 | 
						||
| 
								 | 
							
								        self.assertIn("CREATE INDEX", lines[1])
							 | 
						||
| 
								 | 
							
								
							 | 
						||
| 
								 | 
							
								
							 | 
						||
| 
								 | 
							
								    def test_11createDBTables(self):
							 | 
						||
| 
								 | 
							
								        global mymsg
							 | 
						||
| 
								 | 
							
								        actfunction = str(inspect.currentframe().f_code.co_name)
							 | 
						||
| 
								 | 
							
								        cnttest = 0
							 | 
						||
| 
								 | 
							
								        if actfunction not in TEST_FUNCTIONS:
							 | 
						||
| 
								 | 
							
								            return
							 | 
						||
| 
								 | 
							
								        job = test.testtools.getJob()
							 | 
						||
| 
								 | 
							
								        testserver = basic.Testserver.Testserver(job)
							 | 
						||
| 
								 | 
							
								        testserver.createDBTables(job)
							 | 
						||
| 
								 | 
							
								
							 | 
						||
| 
								 | 
							
								    def test_zzz(self):
							 | 
						||
| 
								 | 
							
								        if verbose: print(MyTestCase.mymsg)
							 | 
						||
| 
								 | 
							
								
							 | 
						||
| 
								 | 
							
								
							 | 
						||
| 
								 | 
							
								if __name__ == '__main__':
							 | 
						||
| 
								 | 
							
								    verbose = True
							 | 
						||
| 
								 | 
							
								    unittest.main()
							 |