SEBA-457 Core xproto cleanup
Change-Id: Ib99680dd81016694094a5c230afdf9dcf2cb307e
diff --git a/lib/xos-synchronizer/xos-synchronizer-tests/test_scheduler.py b/lib/xos-synchronizer/xos-synchronizer-tests/test_scheduler.py
index 0164c5a..0aa5334 100644
--- a/lib/xos-synchronizer/xos-synchronizer-tests/test_scheduler.py
+++ b/lib/xos-synchronizer/xos-synchronizer-tests/test_scheduler.py
@@ -83,36 +83,6 @@
same, t = self.synchronizer.same_object(s, t)
self.assertFalse(same)
- def test_same_object_lst(self):
- s = Slice(pk=5)
- t = ControllerSlice(slice=s)
- u = ControllerSlice(slice=s)
-
- s.controllerslices = mock_enumerator([t, u])
-
- same, et = self.synchronizer.same_object(s.controllerslices, u)
- self.assertTrue(same)
- self.assertEqual(et, event_loop.PROXY_EDGE)
-
- same, et = self.synchronizer.same_object(s.controllerslices, t)
-
- self.assertTrue(same)
- self.assertEqual(et, event_loop.PROXY_EDGE)
-
- def test_same_object_lst_dc(self):
- r = Slice(pk=4)
- s = Slice(pk=5)
- t = ControllerSlice(slice=r)
- u = ControllerSlice(slice=s)
-
- s.controllerslices = mock_enumerator([u])
-
- same, et = self.synchronizer.same_object(s.controllerslices, t)
- self.assertFalse(same)
-
- same, et = self.synchronizer.same_object(s.controllerslices, u)
- self.assertTrue(same)
-
def test_concrete_path_no_model_path(self):
p = Port()
n = NetworkParameter()
@@ -120,60 +90,24 @@
self.assertFalse(verdict)
def test_concrete_no_object_path_adjacent(self):
- p = Instance()
- s1 = Slice()
- s2 = Slice()
- p.slice = s2
- verdict, _ = self.synchronizer.concrete_path_exists(p, s1)
-
+ slice = Slice()
+ site1 = Site()
+ site2 = Site()
+ slice.site = site2
+ verdict, _ = self.synchronizer.concrete_path_exists(slice, site1)
self.assertFalse(verdict)
def test_concrete_object_path_adjacent(self):
- p = Instance()
- s = Slice()
- p.slice = s
- verdict, edge_type = self.synchronizer.concrete_path_exists(p, s)
+ slice = Slice()
+ site = Site()
+ slice.site = site
+ verdict, edge_type = self.synchronizer.concrete_path_exists(slice, site)
self.assertTrue(verdict)
self.assertEqual(edge_type, event_loop.DIRECT_EDGE)
- def test_concrete_object_controller_path_adjacent(self):
- p = Instance()
- q = Instance()
- cs = ControllerSlice()
- cs2 = ControllerSlice()
- s1 = Slice()
- s2 = Slice()
- p.slice = s1
- q.slice = s2
- cs.slice = s1
- s1.controllerslices = mock_enumerator([cs])
- s2.controllerslices = mock_enumerator([])
-
- verdict1, edge_type1 = self.synchronizer.concrete_path_exists(p, cs)
- verdict2, _ = self.synchronizer.concrete_path_exists(q, cs)
- verdict3, _ = self.synchronizer.concrete_path_exists(p, cs2)
-
- self.assertTrue(verdict1)
- self.assertFalse(verdict2)
- self.assertFalse(verdict3)
-
- # TODO(smbaker): This assert was found to be failing. Understand whether the library or the test is at fault.
- #self.assertEqual(edge_type1, event_loop.PROXY_EDGE)
-
- def test_concrete_object_controller_path_distant(self):
- p = Instance()
- s = Slice()
- t = Site()
- ct = ControllerSite()
- ct.site = t
- p.slice = s
- s.site = t
- verdict = self.synchronizer.concrete_path_exists(p, ct)
- self.assertTrue(verdict)
-
def test_concrete_object_path_distant(self):
- p = Instance()
+ p = ComputeServiceInstance()
s = Slice()
t = Site()
p.slice = s
@@ -181,93 +115,14 @@
verdict = self.synchronizer.concrete_path_exists(p, t)
self.assertTrue(verdict)
- def test_concrete_no_object_path_distant(self):
- p = Instance()
- s = Slice()
- s.controllerslice = mock_enumerator([])
-
- t = Site()
- t.controllersite = mock_enumerator([])
-
- ct = ControllerSite()
- ct.site = Site()
- p.slice = s
- s.site = t
-
- verdict, _ = self.synchronizer.concrete_path_exists(p, ct)
- self.assertFalse(verdict)
-
def test_cohorting_independent(self):
i = Image()
p = Slice()
- c = Instance()
- c.slice = None
- c.image = None
+ c = Site()
cohorts = self.synchronizer.compute_dependent_cohorts([i, p, c], False)
self.assertEqual(len(cohorts), 3)
- def test_cohorting_related(self):
- i = Image()
- p = Port()
- c = Instance()
- c.image = i
- s = ControllerSlice()
-
- cohorts = self.synchronizer.compute_dependent_cohorts([i, p, c, s], False)
- self.assertIn([i, c], cohorts)
- self.assertIn([p], cohorts)
- self.assertIn([s], cohorts)
-
- def test_cohorting_related_multi(self):
- i = Image()
- p = Port()
- c = Instance()
- c.image = i
- cs = ControllerSlice()
- s = Slice()
- cs.slice = s
- s.controllerslices = mock_enumerator([cs])
- c.slice = s
-
- cohorts = self.synchronizer.compute_dependent_cohorts([i, p, c, s, cs], False)
-
- big_cohort = max(cohorts, key=len)
- self.assertGreater(big_cohort.index(c), big_cohort.index(i))
- # TODO(smbaker): This assert was found to be failing. Understand whether the library or the test is at fault.
- #self.assertGreater(big_cohort.index(cs), big_cohort.index(s))
- self.assertIn([p], cohorts)
-
- def test_cohorting_related_multi_delete(self):
- i = Image()
- p = Port()
- c = Instance()
- c.image = i
- cs = ControllerSlice()
- s = Slice()
- cs.slice = s
- c.slice = s
-
- cohorts = self.synchronizer.compute_dependent_cohorts([i, p, c, s, cs], True)
-
- big_cohort = max(cohorts, key=len)
- self.assertGreater(big_cohort.index(i), big_cohort.index(c))
- self.assertGreater(big_cohort.index(s), big_cohort.index(cs))
- self.assertIn([p], cohorts)
-
- def test_cohorting_related_delete(self):
- i = Image()
- p = Port()
- c = Instance()
- c.image = i
- s = ControllerSlice()
-
- cohorts = self.synchronizer.compute_dependent_cohorts([i, p, c, s], True)
- self.assertIn([c, i], cohorts)
- self.assertIn([p], cohorts)
- self.assertIn([s], cohorts)
-
-
if __name__ == "__main__":
unittest.main()