summaryrefslogtreecommitdiff
path: root/src/local/domain/cri/test_domain__CRI.py
diff options
context:
space:
mode:
authorLV-426 <lv-426@taproot.org.il>2015-04-06 13:37:07 +0300
committerLV-426 <lv-426@taproot.org.il>2015-04-06 13:40:27 +0300
commit6999761762aa773234de1bb642e91026475ca234 (patch)
tree60b23398b2c3baffabce6a261c89aacfebb6e6d3 /src/local/domain/cri/test_domain__CRI.py
parenteca68e91c1aab13176742ef3d39dc7bd43fc23f0 (diff)
mving cri domain resources under cri.rhizi.net
Diffstat (limited to 'src/local/domain/cri/test_domain__CRI.py')
-rw-r--r--src/local/domain/cri/test_domain__CRI.py58
1 files changed, 0 insertions, 58 deletions
diff --git a/src/local/domain/cri/test_domain__CRI.py b/src/local/domain/cri/test_domain__CRI.py
deleted file mode 100644
index 54131bbd..00000000
--- a/src/local/domain/cri/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 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:Novice|Intermediate|Expert]->(m:Skill)', # [!] expect link type to be proficiency level
- '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()