modelverse/state/test/test_read_incoming.py

import unittest
from modelverse_state.main import ModelverseState

from modelverse_state import status

class TestReadIncoming(unittest.TestCase):
    def setUp(self):
        self.mvs = ModelverseState()

    def test_read_incoming_node_none(self):
        b, r = self.mvs.create_node()
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(b)
        self.assertEquals(len(l), 0)
        self.assertEquals(set(l), set())
        self.assertEquals(r, status.SUCCESS)

    def test_read_incoming_node_one(self):
        a, r = self.mvs.create_node()
        self.assertEquals(r, status.SUCCESS)
        b, r = self.mvs.create_node()
        self.assertEquals(r, status.SUCCESS)
        c, r = self.mvs.create_edge(a, b)
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(a)
        self.assertEquals(len(l), 0)
        self.assertEquals(set(l), set())
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(b)
        self.assertEquals(len(l), 1)
        self.assertEquals(set(l), set([c]))
        self.assertEquals(r, status.SUCCESS)

    def test_read_incoming_node_multi(self):
        a, r = self.mvs.create_node()
        self.assertEquals(r, status.SUCCESS)
        b, r = self.mvs.create_node()
        self.assertEquals(r, status.SUCCESS)
        c, r = self.mvs.create_edge(a, b)
        self.assertEquals(r, status.SUCCESS)
        d, r = self.mvs.create_edge(a, b)
        self.assertEquals(r, status.SUCCESS)
        e, r = self.mvs.create_edge(a, b)
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(a)
        self.assertEquals(len(l), 0)
        self.assertEquals(set(l), set())
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(b)
        self.assertEquals(len(l), 3)
        self.assertEquals(set(l), set([c, d, e]))
        self.assertEquals(r, status.SUCCESS)

    def test_read_incoming_node_multi_others_unaffected(self):
        a, r = self.mvs.create_node()
        self.assertEquals(r, status.SUCCESS)
        b, r = self.mvs.create_node()
        self.assertEquals(r, status.SUCCESS)
        c, r = self.mvs.create_edge(a, b)
        self.assertEquals(r, status.SUCCESS)
        d, r = self.mvs.create_edge(a, b)
        self.assertEquals(r, status.SUCCESS)
        e, r = self.mvs.create_edge(a, b)
        self.assertEquals(r, status.SUCCESS)

        f, r = self.mvs.create_node()
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(a)
        self.assertEquals(len(l), 0)
        self.assertEquals(set(l), set())
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(b)
        self.assertEquals(len(l), 3)
        self.assertEquals(set(l), set([c, d, e]))
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(f)
        self.assertEquals(len(l), 0)
        self.assertEquals(set(l), set())
        self.assertEquals(r, status.SUCCESS)

    def test_read_incoming_edge_none(self):
        a, r = self.mvs.create_node()
        self.assertEquals(r, status.SUCCESS)
        b, r = self.mvs.create_node()
        self.assertEquals(r, status.SUCCESS)
        c, r = self.mvs.create_edge(a, b)
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(c)
        self.assertEquals(len(l), 0)
        self.assertEquals(set(l), set())
        self.assertEquals(r, status.SUCCESS)

    def test_read_incoming_edge_one(self):
        a, r = self.mvs.create_node()
        self.assertEquals(r, status.SUCCESS)
        b, r = self.mvs.create_node()
        self.assertEquals(r, status.SUCCESS)
        c, r = self.mvs.create_edge(a, b)
        self.assertEquals(r, status.SUCCESS)
        d, r = self.mvs.create_edge(c, a)
        self.assertEquals(r, status.SUCCESS)
        e, r = self.mvs.create_edge(a, c)
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(c)
        self.assertEquals(len(l), 1)
        self.assertEquals(set(l), set([e]))
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(d)
        self.assertEquals(len(l), 0)
        self.assertEquals(set(l), set())
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(e)
        self.assertEquals(len(l), 0)
        self.assertEquals(set(l), set())
        self.assertEquals(r, status.SUCCESS)

    def test_read_incoming_edge_multi(self):
        a, r = self.mvs.create_node()
        self.assertEquals(r, status.SUCCESS)
        b, r = self.mvs.create_node()
        self.assertEquals(r, status.SUCCESS)
        c, r = self.mvs.create_edge(a, b)
        self.assertEquals(r, status.SUCCESS)
        d, r = self.mvs.create_edge(a, c)
        self.assertEquals(r, status.SUCCESS)
        e, r = self.mvs.create_edge(b, c)
        self.assertEquals(r, status.SUCCESS)
        f, r = self.mvs.create_edge(d, c)
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(b)
        self.assertEquals(len(l), 1)
        self.assertEquals(set(l), set([c]))
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(c)
        self.assertEquals(len(l), 3)
        self.assertEquals(set(l), set([d, e, f]))
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(d)
        self.assertEquals(len(l), 0)
        self.assertEquals(set(l), set())
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(e)
        self.assertEquals(len(l), 0)
        self.assertEquals(set(l), set())
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(f)
        self.assertEquals(len(l), 0)
        self.assertEquals(set(l), set())
        self.assertEquals(r, status.SUCCESS)

    def test_read_incoming_nodevalue_none(self):
        b, r = self.mvs.create_nodevalue(1)
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(b)
        self.assertEquals(len(l), 0)
        self.assertEquals(set(l), set())
        self.assertEquals(r, status.SUCCESS)

    def test_read_incoming_nodevalue_one(self):
        a, r = self.mvs.create_nodevalue(1)
        self.assertEquals(r, status.SUCCESS)
        b, r = self.mvs.create_node()
        self.assertEquals(r, status.SUCCESS)
        c, r = self.mvs.create_edge(b, a)
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(a)
        self.assertEquals(len(l), 1)
        self.assertEquals(set(l), set([c]))
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(b)
        self.assertEquals(len(l), 0)
        self.assertEquals(set(l), set())
        self.assertEquals(r, status.SUCCESS)

    def test_read_incoming_nodevalue_multi(self):
        a, r = self.mvs.create_nodevalue(1)
        self.assertEquals(r, status.SUCCESS)
        b, r = self.mvs.create_node()
        self.assertEquals(r, status.SUCCESS)
        c, r = self.mvs.create_edge(b, a)
        self.assertEquals(r, status.SUCCESS)
        d, r = self.mvs.create_edge(b, a)
        self.assertEquals(r, status.SUCCESS)
        e, r = self.mvs.create_edge(b, a)
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(a)
        self.assertEquals(len(l), 3)
        self.assertEquals(set(l), set([c, d, e]))
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(b)
        self.assertEquals(len(l), 0)
        self.assertEquals(set(l), set())
        self.assertEquals(r, status.SUCCESS)

    def test_read_incoming_nodevalue_multi_others_unaffected(self):
        a, r = self.mvs.create_nodevalue(1)
        self.assertEquals(r, status.SUCCESS)
        b, r = self.mvs.create_node()
        self.assertEquals(r, status.SUCCESS)
        c, r = self.mvs.create_edge(b, a)
        self.assertEquals(r, status.SUCCESS)
        d, r = self.mvs.create_edge(b, a)
        self.assertEquals(r, status.SUCCESS)
        e, r = self.mvs.create_edge(b, a)
        self.assertEquals(r, status.SUCCESS)

        f, r = self.mvs.create_nodevalue(1)
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(a)
        self.assertEquals(len(l), 3)
        self.assertEquals(set(l), set([c, d, e]))
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(b)
        self.assertEquals(len(l), 0)
        self.assertEquals(set(l), set())
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(f)
        self.assertEquals(len(l), 0)
        self.assertEquals(set(l), set())
        self.assertEquals(r, status.SUCCESS)
    
    def test_read_incoming_node_deleted(self):
        b, r = self.mvs.create_node()
        self.assertEquals(r, status.SUCCESS)
        n, r = self.mvs.delete_node(b)
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(b)
        self.assertEquals(l, None)
        self.assertEquals(r, status.FAIL_RI_UNKNOWN)

    def test_read_incoming_nodevalue_deleted(self):
        b, r = self.mvs.create_nodevalue(1)
        self.assertEquals(r, status.SUCCESS)
        n, r = self.mvs.delete_node(b)
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(b)
        self.assertEquals(l, None)
        self.assertEquals(r, status.FAIL_RI_UNKNOWN)

    def test_read_incoming_edge_deleted(self):
        a, r = self.mvs.create_node()
        self.assertEquals(r, status.SUCCESS)
        b, r = self.mvs.create_node()
        self.assertEquals(r, status.SUCCESS)
        c, r = self.mvs.create_edge(a, b)
        self.assertEquals(r, status.SUCCESS)
        n, r = self.mvs.delete_edge(c)
        self.assertEquals(r, status.SUCCESS)

        l, r = self.mvs.read_incoming(c)
        self.assertEquals(l, None)
        self.assertEquals(r, status.FAIL_RI_UNKNOWN)