summaryrefslogtreecommitdiff
path: root/src/server-tests
diff options
context:
space:
mode:
authorLV-426 <lv-426@taproot.org.il>2015-02-03 16:27:19 +0200
committerLV-426 <lv-426@taproot.org.il>2015-02-03 16:27:50 +0200
commitc3cb91602f90d4fec8a6801203ff78ee10f42a77 (patch)
treeaa66cad6f8798d0e68f2a55f5b0183a3e6c548d5 /src/server-tests
parent972b6a8fbbb1d82d5254abc2c1c8dc94117cd154 (diff)
DBO_random_data_generation__domain__CRI() mv + ;tweeking
Diffstat (limited to 'src/server-tests')
-rw-r--r--src/server-tests/test_domain__CRI.py58
1 files changed, 0 insertions, 58 deletions
diff --git a/src/server-tests/test_domain__CRI.py b/src/server-tests/test_domain__CRI.py
deleted file mode 100644
index b76ecd0e..00000000
--- a/src/server-tests/test_domain__CRI.py
+++ /dev/null
@@ -1,58 +0,0 @@
-import inspect
-import logging
-import tempfile
-import unittest
-
-import db_controller as dbc
-from db_op import DBO_cypher_query
-from domain.cri.domain_cri import DBO_random_data_generation__domain__CRI
-from rz_server import Config
-
-
-class Test_Domain_CRI(unittest.TestCase):
-
- @classmethod
- def setUpClass(self):
- cfg = Config.init_from_file('res/etc/rhizi-server.conf')
- self.db_ctl = dbc.DB_Controller(cfg)
- self.log = logging.getLogger('rhizi')
- self.log.addHandler(logging.StreamHandler())
-
- def test_random_data_generation__domain__CRI(self, export_as_csv=True):
- """
- test:
- - random CRI domain data generation
- - DB dump in cvs format: person x skill x skill-level
- """
- op = DBO_random_data_generation__domain__CRI()
- self.db_ctl.exec_op(op)
-
- if False == export_as_csv:
- return
-
- q_arr = ['match (n: Person)',
- 'with n',
- 'match (n)-[r:Knows]->(m:Skill)',
- 'return n.name, collect({skill_name: m.name, skill_level: r.proficiency})'
- ]
- op = DBO_cypher_query(q_arr);
-
- cur_f_name = inspect.stack()[0][3]
- def q_process_result_set():
- with tempfile.NamedTemporaryFile(prefix='rz_%s' % (cur_f_name), dir='/tmp', suffix='.csv', delete=False) as f_out:
- for _, _, r_set in op:
- for row in r_set:
- person_name, skill_dict_set = row # person to {sname: skill, s:pro: skill_level{ dict set
- cvs_line_arr = [person_name]
- for skill_dict in skill_dict_set:
- skill_name = skill_dict['skill_name']
- skill_level = skill_dict['skill_level']
- cvs_line_arr += [skill_name, skill_level]
- f_out.write(','.join(cvs_line_arr) + '\n')
- f_out.write('\n')
-
- op.process_result_set = q_process_result_set
- self.db_ctl.exec_op(op)
-
-if __name__ == "__main__":
- unittest.main()