blob: 20835a0a7b6b0132963a450bac09426c75db9c1e [file] [log] [blame]
Scott Bakerb404b492017-12-01 13:01:10 -08001
2# Copyright 2017-present Open Networking Foundation
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15
16
17import unittest
18from mock import patch, call, Mock, PropertyMock
Scott Bakerb404b492017-12-01 13:01:10 -080019
20import os, sys
21
22test_path=os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
23service_dir=os.path.join(test_path, "../../../..")
24xos_dir=os.path.join(test_path, "../../..")
25if not os.path.exists(os.path.join(test_path, "new_base")):
26 xos_dir=os.path.join(test_path, "../../../../../../orchestration/xos/xos")
27 services_dir=os.path.join(xos_dir, "../../xos_services")
28
Scott Baker27e4fcb2018-01-09 10:24:52 -080029# While transitioning from static to dynamic load, the path to find neighboring xproto files has changed. So check
30# both possible locations...
31def get_models_fn(service_name, xproto_name):
32 name = os.path.join(service_name, "xos", xproto_name)
33 if os.path.exists(os.path.join(services_dir, name)):
34 return name
35 else:
36 name = os.path.join(service_name, "xos", "synchronizer", "models", xproto_name)
37 if os.path.exists(os.path.join(services_dir, name)):
38 return name
39 raise Exception("Unable to find service=%s xproto=%s" % (service_name, xproto_name))
40
Matteo Scandolod2458012018-03-01 13:40:36 -080041class TestModelPolicyVOLTServiceInstance(unittest.TestCase):
Scott Bakerb404b492017-12-01 13:01:10 -080042 def setUp(self):
Matteo Scandolod2458012018-03-01 13:40:36 -080043 global VOLTServiceInstancePolicy, MockObjectList
Scott Bakerb404b492017-12-01 13:01:10 -080044
45 self.sys_path_save = sys.path
46 sys.path.append(xos_dir)
47 sys.path.append(os.path.join(xos_dir, 'synchronizers', 'new_base'))
48
49 config = os.path.join(test_path, "test_config.yaml")
50 from xosconfig import Config
51 Config.clear()
52 Config.init(config, 'synchronizer-config-schema.yaml')
53
54 from synchronizers.new_base.mock_modelaccessor_build import build_mock_modelaccessor
Scott Baker27e4fcb2018-01-09 10:24:52 -080055 build_mock_modelaccessor(xos_dir, services_dir, [get_models_fn("olt-service", "volt.xproto"),
56 get_models_fn("vsg", "vsg.xproto"),
57 get_models_fn("../profiles/rcord", "rcord.xproto")])
Scott Bakerb404b492017-12-01 13:01:10 -080058
59 import synchronizers.new_base.modelaccessor
Matteo Scandolod2458012018-03-01 13:40:36 -080060 import model_policy_voltserviceinstance
61 from model_policy_voltserviceinstance import VOLTServiceInstancePolicy, model_accessor
Scott Bakerb404b492017-12-01 13:01:10 -080062
63 from mock_modelaccessor import MockObjectList
64
65 # import all class names to globals
66 for (k, v) in model_accessor.all_model_classes.items():
67 globals()[k] = v
68
69 # Some of the functions we call have side-effects. For example, creating a VSGServiceInstance may lead to creation of
70 # tags. Ideally, this wouldn't happen, but it does. So make sure we reset the world.
71 model_accessor.reset_all_object_stores()
72
Matteo Scandolod2458012018-03-01 13:40:36 -080073 self.policy = VOLTServiceInstancePolicy()
Matteo Scandolo2360fd92018-05-29 17:27:51 -070074 self.si = Mock()
Scott Bakerb404b492017-12-01 13:01:10 -080075
76 def tearDown(self):
77 sys.path = self.sys_path_save
78
79 def test_handle_create(self):
Matteo Scandolo2360fd92018-05-29 17:27:51 -070080 with patch.object(VOLTServiceInstancePolicy, "create_eastbound_instance") as create_eastbound_instance, \
81 patch.object(VOLTServiceInstancePolicy, "associate_onu_device") as associate_onu_device:
Scott Bakerb404b492017-12-01 13:01:10 -080082
Matteo Scandolo2360fd92018-05-29 17:27:51 -070083 self.policy.handle_create(self.si)
84 create_eastbound_instance.assert_called_with(self.si)
85 associate_onu_device.assert_called_with(self.si)
Scott Bakerb404b492017-12-01 13:01:10 -080086
87 def test_create_vsg(self):
Matteo Scandolo89f16482018-03-12 16:12:53 -070088 with patch.object(ServiceInstanceLink, "save", autospec=True) as save_link, \
Matteo Scandolo2360fd92018-05-29 17:27:51 -070089 patch.object(VSGServiceInstance, "save", autospec=True) as save_vsg:
Matteo Scandolo89f16482018-03-12 16:12:53 -070090
91 link = Mock()
92 link.provider_service.get_service_instance_class_name.return_value = "VSGServiceInstance"
93
94 si = Mock()
Matteo Scandolo89f16482018-03-12 16:12:53 -070095 si.subscribed_links.all.return_value = []
96 si.owner.subscribed_dependencies.all.return_value = [link]
97
98 self.policy.create_eastbound_instance(si)
Scott Bakerb404b492017-12-01 13:01:10 -080099
100 # Should have created a vsg
101
102 self.assertEqual(save_vsg.call_count, 1)
103 vsg = save_vsg.call_args[0][0]
Scott Bakerb404b492017-12-01 13:01:10 -0800104
105 # Should have created a link from OLT to vsg
106
107 self.assertEqual(save_link.call_count, 1)
108 link = save_link.call_args[0][0]
109 self.assertEqual(link.provider_service_instance, vsg)
Matteo Scandolo89f16482018-03-12 16:12:53 -0700110 self.assertEqual(link.subscriber_service_instance, si)
Scott Bakerb404b492017-12-01 13:01:10 -0800111
Matteo Scandolo2360fd92018-05-29 17:27:51 -0700112 def test_associate_onu(self):
113 with patch.object(ServiceInstance.objects, "get") as get_si, \
114 patch.object(ONUDevice.objects, "get") as get_onu:
115
116 mock_si = Mock()
117 mock_si.get_westbound_service_instance_properties.return_value = "BRCM1234"
118 get_si.return_value = mock_si
119
120 mock_onu = Mock()
121 mock_onu.id = 12
122 get_onu.return_value = mock_onu
123
124 self.policy.associate_onu_device(self.si)
125
126 self.assertEqual(self.si.onu_device_id, mock_onu.id)
127 self.si.save.assert_called()
128
Scott Bakerb404b492017-12-01 13:01:10 -0800129 def test_handle_delete(self):
Matteo Scandolo2360fd92018-05-29 17:27:51 -0700130 self.policy.handle_delete(self.si)
Scott Bakerb404b492017-12-01 13:01:10 -0800131 # handle delete does nothing, and should trivially succeed
132
Scott Bakerb404b492017-12-01 13:01:10 -0800133if __name__ == '__main__':
134 unittest.main()
135