|
|
@ -1,38 +1,53 @@ |
|
|
|
import unittest |
|
|
|
import inspect |
|
|
|
import utils.tdata_tool as t |
|
|
|
import basic.constants as B |
|
|
|
import utils.data_const as D |
|
|
|
import test.testtools |
|
|
|
import test.constants |
|
|
|
import basic.program |
|
|
|
import os |
|
|
|
|
|
|
|
class MyTestCase(unittest.TestCase): |
|
|
|
HOME_PATH = test.constants.HOME_PATH |
|
|
|
DATA_PATH = test.constants.DATA_PATH |
|
|
|
OS_SYSTEM = test.constants.OS_SYSTEM |
|
|
|
|
|
|
|
# here you can select single testfunction for developping the tests |
|
|
|
TEST_FUNCTIONS = ["test_tdata", "test_getCsvSpec_data", "test_getCsvSpec_tree", "test_getCsvSpec_key", |
|
|
|
"test_getCsvSpec_conf", "test_extractPattern"] |
|
|
|
#TEST_FUNCTIONS = ["test_getCsvSpec_key"] |
|
|
|
|
|
|
|
def runTest(self): |
|
|
|
self.test_tdata() |
|
|
|
class MyTestCase(unittest.TestCase): |
|
|
|
mymsg = "--------------------------------------------------------------" |
|
|
|
|
|
|
|
|
|
|
|
def xtest_tdata(self): |
|
|
|
job = basic.program.Job("unit") |
|
|
|
args = {"application": "TEST", "application": "ENV01", "modus": "unit", "loglevel": "debug", |
|
|
|
"tool": "job_tool", "tdtyp": "csv", "tdsrc": "implement", "tdname": "firstunit", |
|
|
|
"modus": "unit"} |
|
|
|
job.par.setParameterArgs(args) |
|
|
|
def test_tdata(self): |
|
|
|
global mymsg |
|
|
|
actfunction = str(inspect.currentframe().f_code.co_name) |
|
|
|
cnttest = 0 |
|
|
|
if actfunction not in TEST_FUNCTIONS: |
|
|
|
return |
|
|
|
job = test.testtools.getJob() |
|
|
|
setattr(job.par, "tdtyp", "dir") |
|
|
|
setattr(job.par, "tdsrc", "TC0001") |
|
|
|
setattr(job.par, "tdname", "testspec") |
|
|
|
filename = str(job.conf.confs["paths"]["testdata"]) + "/" + getattr(job.par, "tdsrc") + "/" + getattr(job.par, "tdname") + ".csv" |
|
|
|
tdata = t.readCsv(job.m, filename, None) |
|
|
|
self.assertEqual(len(tdata["testa1"]), 3) |
|
|
|
#tdata = t.readCsv(job.m, filename, None) |
|
|
|
#self.assertEqual(len(tdata["testa1"]), 3) |
|
|
|
setattr(job.par, "tdtyp", "dir") |
|
|
|
setattr(job.par, "tdsrc", "TST001") |
|
|
|
tdata = t.getTestdata() |
|
|
|
self.assertEqual(("steps" in tdata), True) |
|
|
|
#tdata = t.getTestdata() |
|
|
|
#self.assertEqual(("steps" in tdata), True) |
|
|
|
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest) |
|
|
|
|
|
|
|
def test_getCsvSpec_data(self): |
|
|
|
job = basic.program.Job("unit") |
|
|
|
tdata = {} |
|
|
|
args = {"application": "TEST", "application": "ENV01", "modus": "unit", "loglevel": "debug", |
|
|
|
"tdtyp": "csv", "tdsrc": "TC0001", "tdname": "testspec", |
|
|
|
"modus": "unit"} |
|
|
|
job.par.setParameterArgs(args) |
|
|
|
filename = os.path.join(job.conf.confs["paths"]["testdata"], getattr(job.par, "tdsrc"), getattr(job.par, "tdname") + ".csv") |
|
|
|
global mymsg |
|
|
|
actfunction = str(inspect.currentframe().f_code.co_name) |
|
|
|
cnttest = 0 |
|
|
|
if actfunction not in TEST_FUNCTIONS: |
|
|
|
return |
|
|
|
job = test.testtools.getJob() |
|
|
|
# filename = os.path.join(job.conf.confs["paths"]["testdata"], getattr(job.par, "tdsrc"), getattr(job.par, "tdname") + ".csv") |
|
|
|
""" |
|
|
|
a) data : like a table with data-array of key-value-pairs |
|
|
|
a_0 is keyword [option, step, CSV_HEADER_START ] |
|
|
@ -47,6 +62,7 @@ class MyTestCase(unittest.TestCase): |
|
|
|
] |
|
|
|
tdata = t.parseCsvSpec(job.m, specLines, D.CSV_SPECTYPE_DATA) |
|
|
|
self.assertEqual(0, len(tdata)) |
|
|
|
cnttest += 1 |
|
|
|
if "malformated" in tests: |
|
|
|
malformat = "option;arg;;;;;" |
|
|
|
specLines = [ |
|
|
@ -56,6 +72,7 @@ class MyTestCase(unittest.TestCase): |
|
|
|
"#;;;;;;" |
|
|
|
] |
|
|
|
self.assertRaises(Exception, t.parseCsvSpec, (job.m, specLines, D.CSV_SPECTYPE_DATA)) |
|
|
|
cnttest += 1 |
|
|
|
malformat = "step;component;1;arg:val;;;;;" |
|
|
|
specLines = [ |
|
|
|
"step:1;component;1;arg:val;;;", |
|
|
@ -69,6 +86,7 @@ class MyTestCase(unittest.TestCase): |
|
|
|
malformat |
|
|
|
] |
|
|
|
self.assertRaises(Exception, t.parseCsvSpec, (job.m, specLines, D.CSV_SPECTYPE_DATA)) |
|
|
|
cnttest += 1 |
|
|
|
specLines = [ |
|
|
|
"option:par;arg;;;;;", |
|
|
|
"#option:nopar;arg;;;;;", |
|
|
@ -83,6 +101,7 @@ class MyTestCase(unittest.TestCase): |
|
|
|
self.assertEqual(1, len(tdata)) |
|
|
|
print(tdata) |
|
|
|
self.assertIn(D.CSV_BLOCK_OPTION, tdata) |
|
|
|
cnttest += 2 |
|
|
|
if D.CSV_BLOCK_STEP in tests: |
|
|
|
specLines = [ |
|
|
|
"step:1;testa;1;table:_lofts,action:import;;;;;", |
|
|
@ -93,11 +112,13 @@ class MyTestCase(unittest.TestCase): |
|
|
|
self.assertEqual(1, len(tdata)) |
|
|
|
self.assertIn(B.DATA_NODE_STEPS, tdata) |
|
|
|
self.assertIsInstance(tdata[B.DATA_NODE_STEPS], list) |
|
|
|
cnttest += 3 |
|
|
|
for step in tdata[B.DATA_NODE_STEPS]: |
|
|
|
print(step) |
|
|
|
self.assertIn(B.DATA_NODE_COMP, step) |
|
|
|
self.assertIn(B.ATTR_DATA_REF, step) |
|
|
|
self.assertIn(B.ATTR_STEP_ARGS, step) |
|
|
|
cnttest += 3 |
|
|
|
if B.DATA_NODE_TABLES in tests: |
|
|
|
specLines = [ |
|
|
|
"table:testa:lofts;_nr;street;city;zip;state;beds;baths;sqft;type;price;latitude;longitude", |
|
|
@ -109,48 +130,101 @@ class MyTestCase(unittest.TestCase): |
|
|
|
self.assertEqual(1, len(tdata)) |
|
|
|
self.assertIn(B.DATA_NODE_TABLES, tdata) |
|
|
|
self.assertIsInstance(tdata[B.DATA_NODE_TABLES], dict) |
|
|
|
cnttest += 3 |
|
|
|
for k in tdata[B.DATA_NODE_TABLES]["testa"]: |
|
|
|
table = tdata[B.DATA_NODE_TABLES]["testa"][k] |
|
|
|
self.assertIn(B.DATA_NODE_HEADER, table) |
|
|
|
self.assertIn(B.DATA_NODE_DATA, table) |
|
|
|
cnttest += 2 |
|
|
|
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest) |
|
|
|
|
|
|
|
|
|
|
|
def xtest_getCsvSpec_tree(self): |
|
|
|
job = basic.program.Job("unit") |
|
|
|
tdata = {} |
|
|
|
args = {"application": "TEST", "application": "ENV01", "modus": "unit", "loglevel": "debug", |
|
|
|
"tdtyp": "csv", "tdsrc": "TC0001", "tdname": "testspec", |
|
|
|
"modus": "unit"} |
|
|
|
job.par.setParameterArgs(args) |
|
|
|
def test_getCsvSpec_tree(self): |
|
|
|
global mymsg |
|
|
|
actfunction = str(inspect.currentframe().f_code.co_name) |
|
|
|
cnttest = 0 |
|
|
|
if actfunction not in TEST_FUNCTIONS: |
|
|
|
return |
|
|
|
job = test.testtools.getJob() |
|
|
|
"""" |
|
|
|
b) tree : as a tree - the rows must be unique identified by the first column |
|
|
|
a_0 is keyword in CSV_HEADER_START |
|
|
|
a_0 : { .. a_n : { _header : [ fields.. ], _data : { field : value } |
|
|
|
""" |
|
|
|
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest) |
|
|
|
|
|
|
|
|
|
|
|
def xtest_getCsvSpec_key(self): |
|
|
|
job = basic.program.Job("unit") |
|
|
|
tdata = {} |
|
|
|
args = {"application": "TEST", "application": "ENV01", "modus": "unit", "loglevel": "debug", |
|
|
|
"tdtyp": "csv", "tdsrc": "TC0001", "tdname": "testspec", |
|
|
|
"modus": "unit"} |
|
|
|
job.par.setParameterArgs(args) |
|
|
|
def test_getCsvSpec_key(self): |
|
|
|
global mymsg |
|
|
|
actfunction = str(inspect.currentframe().f_code.co_name) |
|
|
|
cnttest = 0 |
|
|
|
if actfunction not in TEST_FUNCTIONS: |
|
|
|
return |
|
|
|
job = test.testtools.getJob() |
|
|
|
"""" |
|
|
|
c) keys : as a tree - the rows must be unique identified by the first column |
|
|
|
a_0 is keyword in CSV_HEADER_START |
|
|
|
a_1 ... a_n is key characterized by header-field like _fk* or _pk* |
|
|
|
a_0 : { .. a_n : { _keys : [ _fpk*.. ] , _header : [ fields.. ], _data : { pk_0 : { ... pk_n : { field : value } |
|
|
|
""" |
|
|
|
tests = ["malformated", "comments", B.DATA_NODE_TABLES] |
|
|
|
if "comments" in tests: |
|
|
|
specLines = [ |
|
|
|
";;;;;;", |
|
|
|
"#;;;;;;" |
|
|
|
] |
|
|
|
tdata = t.parseCsvSpec(job.m, specLines, D.CSV_SPECTYPE_KEYS) |
|
|
|
self.assertEqual(0, len(tdata)) |
|
|
|
cnttest += 1 |
|
|
|
if "malformated" in tests: |
|
|
|
malformat = "table;key;;;;;" |
|
|
|
specLines = [ |
|
|
|
malformat, |
|
|
|
"#;;;;;;" |
|
|
|
] |
|
|
|
self.assertRaises(Exception, t.parseCsvSpec, (job.m, specLines, D.CSV_SPECTYPE_KEYS)) |
|
|
|
cnttest += 1 |
|
|
|
if B.DATA_NODE_TABLES in tests: |
|
|
|
specLines = [ |
|
|
|
"table:capital;key;val;;;;", |
|
|
|
";athens;;;;;", |
|
|
|
";berlin;;;;;", |
|
|
|
";cairo;;;;;" |
|
|
|
] |
|
|
|
tdata = t.parseCsvSpec(job.m, specLines, D.CSV_SPECTYPE_KEYS) |
|
|
|
print(str(tdata)) |
|
|
|
self.assertEqual(1, len(tdata)) |
|
|
|
self.assertEqual(1, len(tdata["_tables"])) |
|
|
|
self.assertEqual(2, len(tdata["_tables"]["capital"])) |
|
|
|
self.assertEqual(3, len(tdata["_tables"]["capital"]["_keys"])) |
|
|
|
cnttest += 4 |
|
|
|
specLines = [ |
|
|
|
"table:capital;key;val;;;;", |
|
|
|
";athens;;;;;", |
|
|
|
";berlin;;;;;", |
|
|
|
"table:country;key;val;;;;", |
|
|
|
";greece;;;;;", |
|
|
|
";germany;;;;;" |
|
|
|
] |
|
|
|
tdata = t.parseCsvSpec(job.m, specLines, D.CSV_SPECTYPE_KEYS) |
|
|
|
#tdata = t.parseCsvSpec(job.m, specLines, D.CSV_SPECTYPE_TREE) |
|
|
|
print(str(tdata)) |
|
|
|
self.assertEqual(1, len(tdata)) |
|
|
|
self.assertNotIn("capital", tdata["_tables"]) |
|
|
|
self.assertEqual(1, len(tdata["_tables"])) |
|
|
|
self.assertEqual(2, len(tdata["_tables"]["country"])) |
|
|
|
self.assertEqual(2, len(tdata["_tables"]["country"]["_keys"])) |
|
|
|
cnttest += 4 |
|
|
|
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest) |
|
|
|
|
|
|
|
|
|
|
|
def xtest_getCsvSpec_conf(self): |
|
|
|
job = basic.program.Job("unit") |
|
|
|
tdata = {} |
|
|
|
args = {"application": "TEST", "application": "ENV01", "modus": "unit", "loglevel": "debug", |
|
|
|
"tdtyp": "csv", "tdsrc": "TC0001", "tdname": "testspec", |
|
|
|
"modus": "unit"} |
|
|
|
job.par.setParameterArgs(args) |
|
|
|
def test_getCsvSpec_conf(self): |
|
|
|
global mymsg |
|
|
|
actfunction = str(inspect.currentframe().f_code.co_name) |
|
|
|
cnttest = 0 |
|
|
|
if actfunction not in TEST_FUNCTIONS: |
|
|
|
return |
|
|
|
job = test.testtools.getJob() |
|
|
|
"""" |
|
|
|
d) conf: |
|
|
|
_header : [ field_0, ... ] |
|
|
@ -167,10 +241,16 @@ class MyTestCase(unittest.TestCase): |
|
|
|
self.assertEqual(1, len(tdata)) |
|
|
|
self.assertNotIn(B.DATA_NODE_TABLES, tdata) |
|
|
|
self.assertIn("lofts", tdata) |
|
|
|
cnttest += 3 |
|
|
|
table = tdata["lofts"] |
|
|
|
self.assertIn(B.DATA_NODE_HEADER, table) |
|
|
|
self.assertIn(B.DATA_NODE_DATA, table) |
|
|
|
self.assertNotIn(B.DATA_NODE_DATA, table) |
|
|
|
cnttest += 2 |
|
|
|
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest) |
|
|
|
|
|
|
|
|
|
|
|
def test_zzz(self): |
|
|
|
print(MyTestCase.mymsg) |
|
|
|
|
|
|
|
if __name__ == '__main__': |
|
|
|
unittest.main() |
|
|
|