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.
256 lines
9.8 KiB
256 lines
9.8 KiB
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
|
|
|
|
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"]
|
|
|
|
class MyTestCase(unittest.TestCase):
|
|
mymsg = "--------------------------------------------------------------"
|
|
|
|
|
|
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)
|
|
setattr(job.par, "tdtyp", "dir")
|
|
setattr(job.par, "tdsrc", "TST001")
|
|
#tdata = t.getTestdata()
|
|
#self.assertEqual(("steps" in tdata), True)
|
|
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest)
|
|
|
|
def test_getCsvSpec_data(self):
|
|
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 ]
|
|
a_0 : { a_1 : { f_1 : v_1, .... } # option, step
|
|
a_0 : { .. a_n : { _header : [ .. ], _data : [ rows... ] # table, node
|
|
"""
|
|
tests = ["malformated", "comments", D.CSV_BLOCK_OPTION, D.CSV_BLOCK_STEP, B.DATA_NODE_TABLES]
|
|
if "comments" in tests:
|
|
specLines = [
|
|
";;;;;;",
|
|
"#;;;;;;"
|
|
]
|
|
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 = [
|
|
"option:par;arg;;;;;",
|
|
malformat,
|
|
"#option:nopar;arg;;;;;",
|
|
"#;;;;;;"
|
|
]
|
|
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;;;",
|
|
malformat
|
|
]
|
|
# TODO sortierung nicht ausgwertet
|
|
# self.assertRaises(D.EXCP_MALFORMAT+malformat, t.parseCsvSpec, (job.m, specLines, D.CSV_SPECTYPE_DATA))
|
|
malformat = "step:2;component;1;arg;;;;;"
|
|
specLines = [
|
|
"step:1;component;1;arg:val;;;",
|
|
malformat
|
|
]
|
|
self.assertRaises(Exception, t.parseCsvSpec, (job.m, specLines, D.CSV_SPECTYPE_DATA))
|
|
cnttest += 1
|
|
specLines = [
|
|
"option:par;arg;;;;;",
|
|
"#option:nopar;arg;;;;;",
|
|
"#;;;;;;"
|
|
]
|
|
if D.CSV_BLOCK_OPTION in tests:
|
|
specLines = [
|
|
"option:description;something;;;;;",
|
|
"#;;;;;;"
|
|
]
|
|
tdata = t.parseCsvSpec(job.m, specLines, D.CSV_SPECTYPE_DATA)
|
|
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;;;;;",
|
|
"#;;;;;;"
|
|
]
|
|
tdata = t.parseCsvSpec(job.m, specLines, D.CSV_SPECTYPE_DATA)
|
|
print(tdata)
|
|
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",
|
|
"testa:lofts;1;stra;town;12345;usa;4;1;50;house;111;45;8",
|
|
"#;;;;;;"
|
|
]
|
|
tdata = t.parseCsvSpec(job.m, specLines, B.DATA_NODE_TABLES)
|
|
print(tdata)
|
|
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 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 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 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, ... ]
|
|
{ field_0 : { attr_0 : val_0, .. }, field_1 : { ... }, ... }
|
|
"""
|
|
specLines = [
|
|
"table:lofts;_field;field;type;acceptance;key",
|
|
"lofts;street;a;str;;T:1",
|
|
";city;b;str;;F:1",
|
|
"#;;;;;;"
|
|
]
|
|
tdata = t.parseCsvSpec(job.m, specLines, D.CSV_SPECTYPE_CONF)
|
|
print(tdata)
|
|
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.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()
|
|
|