diff options
| author | LV-426 <lv-426@taproot.org.il> | 2015-02-03 16:27:19 +0200 |
|---|---|---|
| committer | LV-426 <lv-426@taproot.org.il> | 2015-02-03 16:27:50 +0200 |
| commit | c3cb91602f90d4fec8a6801203ff78ee10f42a77 (patch) | |
| tree | aa66cad6f8798d0e68f2a55f5b0183a3e6c548d5 /src/server-tests | |
| parent | 972b6a8fbbb1d82d5254abc2c1c8dc94117cd154 (diff) | |
DBO_random_data_generation__domain__CRI() mv + ;tweeking
Diffstat (limited to 'src/server-tests')
| -rw-r--r-- | src/server-tests/test_domain__CRI.py | 58 |
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() |
