<commit_before>import unittest
import logging
import db_controller as dbc

from rhizi_server import Config
from neo4j_test_util import rand_id
from neo4j_test_util import flush_db
from neo4j_util import Neo4JException

from model.graph import Attr_Diff
from model.graph import Topo_Diff
from model.model import Link

class TestDBController(unittest.TestCase):

    db_ctl = None
    log = None

    n_map = { 'Skill': [{'name': '<NAME>', 'id': 'skill_00' },
                        {'name': 'Judo', 'id': 'skill_01' }
                        ],

              'Person': [{'name': 'Bob', 'id': 'person_00', 'age': 128 },
                         {'name': 'Alice', 'id': 'person_01', 'age': 256 }
                         ]
            }

    l_map = { 'Knows' : [{'__src': 'person_00', '__dst': 'skill_00'},
                         {'__src': 'person_00', '__dst': 'skill_01'}] }

    @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 setUp(self):
        flush_db(self.db_ctl) # remove once embedded DB test mode is supported
        self.db_ctl.exec_op(dbc.DBO_add_node_set(self.n_map))
        self.db_ctl.exec_op(dbc.DBO_add_link_set(self.l_map))

    def test_db_op_statement_iteration(self):
        s_arr = ['create (b:Book {title: \'foo\'}) return b',
                 'match (n) return n', ]

        op = dbc.DB_op()
        op.add_statement(s_arr[0])
        op.add_statement(s_arr[1])

        i = 0
        for _, s, r in op:
            # access: second tuple item -> REST-form 'statement' key
            self.assertEqual(s_arr[i], s['statement'])
            self.assertEqual(None, r)
            i = i + 1

        self.db_ctl.exec_op(op)

        i = 0
        for _, s, r_set in op:
            # access: second tuple item -> REST-form 'statement' key
            self.assertNotEqual(None, r_set)
            for x in r_set:
                pass
            i = i + 1

    def test_add_node_set(self):
        n_map = { 'T_test_add_node_set': [{'id': rand_id()}, {'id': rand_id()}] }
        op = dbc.DBO_add_node_set(n_map)

        self.assertEqual(len(op.statement_set), 1)  # assert a single statement is issued

        id_set = self.db_ctl.exec_op(op)
        self.assertEqual(len(id_set), 2)

    def test_add_link_set(self):
        src_id = rand_id()
        dst_id_0 = rand_id()
        dst_id_1 = rand_id()
        n_map = { 'T_test_add_node_set': [{'id': src_id },
                                          {'id': dst_id_0 },
                                          {'id': dst_id_1 }] }
        self.db_ctl.exec_op(dbc.DBO_add_node_set(n_map))

        l_map = { 'T_test_add_link_set' : [{'__src': src_id, '__dst': dst_id_0},
                                           {'__src': src_id, '__dst': dst_id_1}] }

        op = dbc.DBO_add_link_set(l_map)
        self.assertEqual(len(op.statement_set), 2)  # no support yet for parameterized statements for link creation

        l_set = self.db_ctl.exec_op(op)
        self.assertEqual(len(l_set), 2)

    def test_match_node_set_by_type(self):
        op = dbc.DBO_match_node_id_set(filter_type='Person')
        id_set = self.db_ctl.exec_op(op)
        self.assertEqual(len(id_set), 2)

        op = dbc.DBO_match_node_id_set(filter_type='Nan_Type')
        id_set = self.db_ctl.exec_op(op)
        self.assertEqual(len(id_set), 0)

    def test_match_node_set_by_attribute(self):
        fam = { 'name': ['Bob', u'Judo'], 'age': [128] }
        n_set = self.db_ctl.exec_op(dbc.DBO_match_node_id_set(filter_attr_map=fam))
        self.assertEqual(len(n_set), 1)

        fam = { 'age': [128, 256, 404] }
        n_set = self.db_ctl.exec_op(dbc.DBO_match_node_id_set(filter_attr_map=fam))
        self.assertEqual(len(n_set), 2)

    def test_match_node_set_by_DB_id(self):
        pass  # TODO

    def test_match_node_set_by_id_attribute(self):
        n_set = self.db_ctl.exec_op(dbc.DBO_match_node_set_by_id_attribute(['skill_00', 'person_01']))
        self.assertEqual(len(n_set), 2)

    def test_match_link_set_by_type(self):
        op = dbc.DBO_match_link_id_set(filter_type='Knows')
        id_set = self.db_ctl.exec_op(op)
        self.assertEqual(len(id_set), 2)

        op = dbc.DBO_match_link_id_set(filter_type='Nan_Type')
        id_set = self.db_ctl.exec_op(op)
        self.assertEqual(len(id_set), 0)

    def test_load_link_set(self):
        l_ptr = Link.link_ptr(src_id='person_00', dst_id='skill_00')
        op = dbc.DBO_load_link_set.init_from_link_ptr(l_ptr)
        l_set = self.db_ctl.exec_op(op)
        self.assertEqual(len(l_set), 1)

        l_ptr = Link.link_ptr(src_id='person_00')
        op = dbc.DBO_load_link_set.init_from_link_ptr(l_ptr)
        l_set = self.db_ctl.exec_op(op)
        self.assertEqual(len(l_set), 2)

        l_ptr = Link.link_ptr(dst_id='skill_00')
        op = dbc.DBO_load_link_set.init_from_link_ptr(l_ptr)
        l_set = self.db_ctl.exec_op(op)
        self.assertEqual(len(l_set), 1)

        # load sets
        l_ptr_set = [Link.link_ptr(s,d) for (s,d) in [('person_00', 'skill_00'), ('person_00', 'skill_01')]]
        op = dbc.DBO_load_link_set.init_from_link_ptr_set(l_ptr_set)
        l_set = self.db_ctl.exec_op(op)
        self.assertEqual(len(l_set), 2)
    
    def test_load_node_set_by_DB_id(self):
        """
        test node DB id life cycle
        """<commit_msg>acknowledge duplicate links may be returned in response<commit_after>import unittest
import logging
import db_controller as dbc

from rhizi_server import Config
from neo4j_test_util import rand_id
from neo4j_test_util import flush_db
from neo4j_util import Neo4JException

from model.graph import Attr_Diff
from model.graph import Topo_Diff
from model.model import Link

class TestDBController(unittest.TestCase):

    db_ctl = None
    log = None

    n_map = { 'Skill': [{'name': '<NAME>', 'id': 'skill_00' },
                        {'name': 'Judo', 'id': 'skill_01' }
                        ],

              'Person': [{'name': 'Bob', 'id': 'person_00', 'age': 128 },
                         {'name': 'Alice', 'id': 'person_01', 'age': 256 }
                         ]
            }

    l_map = { 'Knows' : [{'__src': 'person_00', '__dst': 'skill_00'},
                         {'__src': 'person_00', '__dst': 'skill_01'}] }

    @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 setUp(self):
        flush_db(self.db_ctl) # remove once embedded DB test mode is supported
        self.db_ctl.exec_op(dbc.DBO_add_node_set(self.n_map))
        self.db_ctl.exec_op(dbc.DBO_add_link_set(self.l_map))

    def test_db_op_statement_iteration(self):
        s_arr = ['create (b:Book {title: \'foo\'}) return b',
                 'match (n) return n', ]

        op = dbc.DB_op()
        op.add_statement(s_arr[0])
        op.add_statement(s_arr[1])

        i = 0
        for _, s, r in op:
            # access: second tuple item -> REST-form 'statement' key
            self.assertEqual(s_arr[i], s['statement'])
            self.assertEqual(None, r)
            i = i + 1

        self.db_ctl.exec_op(op)

        i = 0
        for _, s, r_set in op:
            # access: second tuple item -> REST-form 'statement' key
            self.assertNotEqual(None, r_set)
            for x in r_set:
                pass
            i = i + 1

    def test_add_node_set(self):
        n_map = { 'T_test_add_node_set': [{'id': rand_id()}, {'id': rand_id()}] }
        op = dbc.DBO_add_node_set(n_map)

        self.assertEqual(len(op.statement_set), 1)  # assert a single statement is issued

        id_set = self.db_ctl.exec_op(op)
        self.assertEqual(len(id_set), 2)

    def test_add_link_set(self):
        src_id = rand_id()
        dst_id_0 = rand_id()
        dst_id_1 = rand_id()
        n_map = { 'T_test_add_node_set': [{'id': src_id },
                                          {'id': dst_id_0 },
                                          {'id': dst_id_1 }] }
        self.db_ctl.exec_op(dbc.DBO_add_node_set(n_map))

        l_map = { 'T_test_add_link_set' : [{'__src': src_id, '__dst': dst_id_0},
                                           {'__src': src_id, '__dst': dst_id_1}] }

        op = dbc.DBO_add_link_set(l_map)
        self.assertEqual(len(op.statement_set), 2)  # no support yet for parameterized statements for link creation

        l_set = self.db_ctl.exec_op(op)
        self.assertEqual(len(l_set), 2)

    def test_match_node_set_by_type(self):
        op = dbc.DBO_match_node_id_set(filter_type='Person')
        id_set = self.db_ctl.exec_op(op)
        self.assertEqual(len(id_set), 2)

        op = dbc.DBO_match_node_id_set(filter_type='Nan_Type')
        id_set = self.db_ctl.exec_op(op)
        self.assertEqual(len(id_set), 0)

    def test_match_node_set_by_attribute(self):
        fam = { 'name': ['Bob', u'Judo'], 'age': [128] }
        n_set = self.db_ctl.exec_op(dbc.DBO_match_node_id_set(filter_attr_map=fam))
        self.assertEqual(len(n_set), 1)

        fam = { 'age': [128, 256, 404] }
        n_set = self.db_ctl.exec_op(dbc.DBO_match_node_id_set(filter_attr_map=fam))
        self.assertEqual(len(n_set), 2)

    def test_match_node_set_by_DB_id(self):
        pass  # TODO

    def test_match_node_set_by_id_attribute(self):
        n_set = self.db_ctl.exec_op(dbc.DBO_match_node_set_by_id_attribute(['skill_00', 'person_01']))
        self.assertEqual(len(n_set), 2)

    def test_match_link_set_by_type(self):
        op = dbc.DBO_match_link_id_set(filter_type='Knows')
        id_set = self.db_ctl.exec_op(op)
        self.assertEqual(len(id_set), 2)

        op = dbc.DBO_match_link_id_set(filter_type='Nan_Type')
        id_set = self.db_ctl.exec_op(op)
        self.assertEqual(len(id_set), 0)

    def test_load_link_set(self):
        l_ptr = Link.link_ptr(src_id='person_00', dst_id='skill_00')
        op = dbc.DBO_load_link_set.init_from_link_ptr(l_ptr)
        l_set = self.db_ctl.exec_op(op)
        self.assertEqual(len(l_set), 1)

        l_ptr = Link.link_ptr(src_id='person_00')
        op = dbc.DBO_load_link_set.init_from_link_ptr(l_ptr)
        l_set = self.db_ctl.exec_op(op)
        self.assertEqual(len(l_set), 2)

        l_ptr = Link.link_ptr(dst_id='skill_00')
        op = dbc.DBO_load_link_set.init_from_link_ptr(l_ptr)
        l_set = self.db_ctl.exec_op(op)
        self.assertEqual(len(l_set), 1)

        # load by l_ptr sets
        l_ptr_set = [Link.link_ptr(s, d) for (s, d) in [('person_00', 'skill_00'), ('person_00', 'skill_01')]]
        op = dbc.DBO_load_link_set.init_from_link_ptr_set(l_ptr_set)
        l_set = self.db_ctl.exec_op(op)
        self.assertEqual(len(l_set), 2)

        # this should return the same link twice
        l_ptr_set = [Link.link_ptr(s, d) for (s, d) in [('person_00', 'skill_00'), ('person_00', 'skill_01')]]
        l_ptr_set.append(Link.link_ptr(dst_id='skill_00'))
        op = dbc.DBO_load_link_set.init_from_link_ptr_set(l_ptr_set)
        l_set = self.db_ctl.exec_op(op)
        self.assertEqual(len(l_set), 3)

    def test_load_node_set_by_DB_id(self):
        """
        test node DB id life cycle
        """
{'fraction_non_alphanumeric': 0.12789482848685357, 'fraction_numerical': 0.015758314469789307, 'mean_word_length': 2.4392215568862277, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7193623', 'n_tokens_mistral': 4364, 'n_tokens_neox': 4177, 'n_words': 714}