import unittest from modelverse_state.main import ModelverseState from modelverse_state import status class TestReadOutgoing(unittest.TestCase): def setUp(self): self.mvs = ModelverseState() def test_read_outgoing_node_none(self): b, r = self.mvs.create_node() self.assertEquals(r, status.SUCCESS) l, r = self.mvs.read_outgoing(b) self.assertEquals(len(l), 0) self.assertEquals(set(l), set()) self.assertEquals(r, status.SUCCESS) def test_read_outgoing_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_outgoing(a) self.assertEquals(len(l), 1) self.assertEquals(set(l), set([c])) self.assertEquals(r, status.SUCCESS) l, r = self.mvs.read_outgoing(b) self.assertEquals(len(l), 0) self.assertEquals(set(l), set()) self.assertEquals(r, status.SUCCESS) def test_read_outgoing_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_outgoing(a) self.assertEquals(len(l), 3) self.assertEquals(set(l), set([c, d, e])) self.assertEquals(r, status.SUCCESS) l, r = self.mvs.read_outgoing(b) self.assertEquals(len(l), 0) self.assertEquals(set(l), set()) self.assertEquals(r, status.SUCCESS) def test_read_outgoing_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_outgoing(a) self.assertEquals(len(l), 3) self.assertEquals(set(l), set([c, d, e])) self.assertEquals(r, status.SUCCESS) l, r = self.mvs.read_outgoing(b) self.assertEquals(len(l), 0) self.assertEquals(set(l), set()) self.assertEquals(r, status.SUCCESS) l, r = self.mvs.read_outgoing(f) self.assertEquals(len(l), 0) self.assertEquals(set(l), set()) self.assertEquals(r, status.SUCCESS) def test_read_outgoing_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_outgoing(c) self.assertEquals(len(l), 0) self.assertEquals(set(l), set()) self.assertEquals(r, status.SUCCESS) def test_read_outgoing_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) l, r = self.mvs.read_outgoing(c) self.assertEquals(len(l), 1) self.assertEquals(set(l), set([d])) self.assertEquals(r, status.SUCCESS) l, r = self.mvs.read_outgoing(d) self.assertEquals(len(l), 0) self.assertEquals(set(l), set()) self.assertEquals(r, status.SUCCESS) def test_read_outgoing_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(c, a) self.assertEquals(r, status.SUCCESS) e, r = self.mvs.create_edge(c, b) self.assertEquals(r, status.SUCCESS) f, r = self.mvs.create_edge(c, d) self.assertEquals(r, status.SUCCESS) l, r = self.mvs.read_outgoing(c) self.assertEquals(len(l), 3) self.assertEquals(set(l), set([d, e, f])) self.assertEquals(r, status.SUCCESS) l, r = self.mvs.read_outgoing(d) self.assertEquals(len(l), 0) self.assertEquals(set(l), set()) self.assertEquals(r, status.SUCCESS) l, r = self.mvs.read_outgoing(e) self.assertEquals(len(l), 0) self.assertEquals(set(l), set()) self.assertEquals(r, status.SUCCESS) l, r = self.mvs.read_outgoing(f) self.assertEquals(len(l), 0) self.assertEquals(set(l), set()) self.assertEquals(r, status.SUCCESS) def test_read_outgoing_nodevalue_none(self): b, r = self.mvs.create_nodevalue(1) self.assertEquals(r, status.SUCCESS) l, r = self.mvs.read_outgoing(b) self.assertEquals(len(l), 0) self.assertEquals(set(l), set()) self.assertEquals(r, status.SUCCESS) def test_read_outgoing_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(a, b) self.assertEquals(r, status.SUCCESS) l, r = self.mvs.read_outgoing(a) self.assertEquals(len(l), 1) self.assertEquals(set(l), set([c])) self.assertEquals(r, status.SUCCESS) l, r = self.mvs.read_outgoing(b) self.assertEquals(len(l), 0) self.assertEquals(set(l), set()) self.assertEquals(r, status.SUCCESS) def test_read_outgoing_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(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_outgoing(a) self.assertEquals(len(l), 3) self.assertEquals(set(l), set([c, d, e])) self.assertEquals(r, status.SUCCESS) l, r = self.mvs.read_outgoing(b) self.assertEquals(len(l), 0) self.assertEquals(set(l), set()) self.assertEquals(r, status.SUCCESS) def test_read_outgoing_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(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_nodevalue(1) self.assertEquals(r, status.SUCCESS) l, r = self.mvs.read_outgoing(a) self.assertEquals(len(l), 3) self.assertEquals(set(l), set([c, d, e])) self.assertEquals(r, status.SUCCESS) l, r = self.mvs.read_outgoing(b) self.assertEquals(len(l), 0) self.assertEquals(set(l), set()) self.assertEquals(r, status.SUCCESS) l, r = self.mvs.read_outgoing(f) self.assertEquals(len(l), 0) self.assertEquals(set(l), set()) self.assertEquals(r, status.SUCCESS) def test_read_outgoing_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_outgoing(b) self.assertEquals(l, None) self.assertEquals(r, status.FAIL_RO_UNKNOWN) def test_read_outgoing_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_outgoing(b) self.assertEquals(l, None) self.assertEquals(r, status.FAIL_RO_UNKNOWN) def test_read_outgoing_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_outgoing(c) self.assertEquals(l, None) self.assertEquals(r, status.FAIL_RO_UNKNOWN)