blob: da77395f3a89621e0b02c4b9d1cbdb9c690db886 [file] [log] [blame]
Matteo Scandolof6337eb2018-04-05 15:58:37 -07001# Copyright 2017-present Open Networking Foundation
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14
15import unittest
16from mock import patch, call, Mock, PropertyMock
17import requests_mock
18
19import os, sys
20
21# Hack to load synchronizer framework
22test_path=os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
23xos_dir=os.path.join(test_path, "../../..")
24if not os.path.exists(os.path.join(test_path, "new_base")):
25 xos_dir=os.path.join(test_path, "../../../../../../orchestration/xos/xos")
26 services_dir = os.path.join(xos_dir, "../../xos_services")
27sys.path.append(xos_dir)
28sys.path.append(os.path.join(xos_dir, 'synchronizers', 'new_base'))
29# END Hack to load synchronizer framework
30
31# generate model from xproto
32def get_models_fn(service_name, xproto_name):
33 name = os.path.join(service_name, "xos", xproto_name)
34 if os.path.exists(os.path.join(services_dir, name)):
35 return name
36 else:
37 name = os.path.join(service_name, "xos", "synchronizer", "models", xproto_name)
38 if os.path.exists(os.path.join(services_dir, name)):
39 return name
40 raise Exception("Unable to find service=%s xproto=%s" % (service_name, xproto_name))
41# END generate model from xproto
42
43def mock_get_westbound_service_instance_properties(prop):
44 return prop
45
Matteo Scandolod44ca992018-05-17 15:02:10 -070046class TestSyncVOLTServiceInstance(unittest.TestCase):
Matteo Scandolof6337eb2018-04-05 15:58:37 -070047 def setUp(self):
48 global DeferredException
49
50 self.sys_path_save = sys.path
51 sys.path.append(xos_dir)
52 sys.path.append(os.path.join(xos_dir, 'synchronizers', 'new_base'))
53
54 # Setting up the config module
55 from xosconfig import Config
56 config = os.path.join(test_path, "../model_policies/test_config.yaml")
57 Config.clear()
58 Config.init(config, "synchronizer-config-schema.yaml")
59 # END Setting up the config module
60
61 from synchronizers.new_base.syncstep import DeferredException
62 from synchronizers.new_base.mock_modelaccessor_build import build_mock_modelaccessor
Matteo Scandolo19466a02018-05-16 17:43:39 -070063 # build_mock_modelaccessor(xos_dir, services_dir, [get_models_fn("olt-service", "volt.xproto")])
64
65 # FIXME this is to get jenkins to pass the tests, somehow it is running tests in a different order
66 # and apparently it is not overriding the generated model accessor
67 build_mock_modelaccessor(xos_dir, services_dir, [get_models_fn("olt-service", "volt.xproto"),
68 get_models_fn("vsg", "vsg.xproto"),
69 get_models_fn("../profiles/rcord", "rcord.xproto")])
Matteo Scandolof6337eb2018-04-05 15:58:37 -070070 import synchronizers.new_base.modelaccessor
71 from sync_volt_service_instance import SyncVOLTServiceInstance, model_accessor
72
73 # import all class names to globals
74 for (k, v) in model_accessor.all_model_classes.items():
75 globals()[k] = v
76
77 self.sync_step = SyncVOLTServiceInstance
78
79 # create a mock service instance
80 o = Mock()
81 o.id = 1
82 o.owner_id = "volt_service"
83 o.tologdict.return_value = {}
84
85 volt_service = Mock()
Luca Preteca974c82018-05-01 18:06:16 -070086 volt_service.onos_voltha_url = "onos_voltha_url"
87 volt_service.onos_voltha_port = 4321
88 volt_service.onos_voltha_user = "onos_voltha_user"
89 volt_service.onos_voltha_pass = "onos_voltha_pass"
Matteo Scandolof6337eb2018-04-05 15:58:37 -070090
91 si = Mock()
92 si.get_westbound_service_instance_properties = mock_get_westbound_service_instance_properties
93
Matteo Scandoloe2cb8a42018-05-18 16:30:06 -070094 onu_device = Mock()
95 onu_device.name = "BRCM1234"
96 onu_device.pon_port.olt_device.dp_id = None
97 onu_device.pon_port.olt_device.name = "Test OLT Device"
Matteo Scandolof6337eb2018-04-05 15:58:37 -070098
99 self.o = o
100 self.si = si
Matteo Scandoloe2cb8a42018-05-18 16:30:06 -0700101 self.onu_device = onu_device
Matteo Scandolof6337eb2018-04-05 15:58:37 -0700102 self.volt_service = volt_service
103
104 def tearDown(self):
105 self.o = None
106 sys.path = self.sys_path_save
107
Matteo Scandolof6337eb2018-04-05 15:58:37 -0700108 @requests_mock.Mocker()
109 def test_do_not_sync(self, m):
Matteo Scandoloe2cb8a42018-05-18 16:30:06 -0700110 self.onu_device.pon_port.olt_device.dp_id = None
Matteo Scandolof6337eb2018-04-05 15:58:37 -0700111
112 with patch.object(ServiceInstance.objects, "get") as service_instance_mock, \
Matteo Scandoloe2cb8a42018-05-18 16:30:06 -0700113 patch.object(ONUDevice.objects, "get") as onu_device_mock, \
Matteo Scandolof6337eb2018-04-05 15:58:37 -0700114 patch.object(VOLTService.objects, "get") as olt_service_mock:
115 service_instance_mock.return_value = self.si
Matteo Scandoloe2cb8a42018-05-18 16:30:06 -0700116 onu_device_mock.return_value = self.onu_device
Matteo Scandolof6337eb2018-04-05 15:58:37 -0700117 olt_service_mock.return_value = self.volt_service
118
119 with self.assertRaises(DeferredException) as e:
120 self.sync_step().sync_record(self.o)
121
122 self.assertFalse(m.called)
123 self.assertEqual(e.exception.message, "Waiting for OLTDevice Test OLT Device to be synchronized")
124
125 @requests_mock.Mocker()
126 def test_do_sync(self, m):
Luca Preteca974c82018-05-01 18:06:16 -0700127 m.post("http://onos_voltha_url:4321/onos/olt/oltapp/of:dp_id/uni_port_id/c_tag", status_code=200, json={})
Matteo Scandolof6337eb2018-04-05 15:58:37 -0700128
Matteo Scandoloe2cb8a42018-05-18 16:30:06 -0700129 self.onu_device.pon_port.olt_device.dp_id = "of:dp_id"
Matteo Scandolof6337eb2018-04-05 15:58:37 -0700130
131 with patch.object(ServiceInstance.objects, "get") as service_instance_mock, \
Matteo Scandoloe2cb8a42018-05-18 16:30:06 -0700132 patch.object(ONUDevice.objects, "get") as onu_device_mock, \
Matteo Scandolof6337eb2018-04-05 15:58:37 -0700133 patch.object(VOLTService.objects, "get") as olt_service_mock:
134 service_instance_mock.return_value = self.si
Matteo Scandoloe2cb8a42018-05-18 16:30:06 -0700135 onu_device_mock.return_value = self.onu_device
Matteo Scandolof6337eb2018-04-05 15:58:37 -0700136 olt_service_mock.return_value = self.volt_service
137
138 self.sync_step().sync_record(self.o)
139 self.assertTrue(m.called)
140
141 @requests_mock.Mocker()
142 def test_do_sync_fail(self, m):
Luca Preteca974c82018-05-01 18:06:16 -0700143 m.post("http://onos_voltha_url:4321/onos/olt/oltapp/of:dp_id/uni_port_id/c_tag", status_code=500, text="Mock Error")
Matteo Scandolof6337eb2018-04-05 15:58:37 -0700144
Matteo Scandoloe2cb8a42018-05-18 16:30:06 -0700145 self.onu_device.pon_port.olt_device.dp_id = "of:dp_id"
Matteo Scandolof6337eb2018-04-05 15:58:37 -0700146
147 with patch.object(ServiceInstance.objects, "get") as service_instance_mock, \
Matteo Scandoloe2cb8a42018-05-18 16:30:06 -0700148 patch.object(ONUDevice.objects, "get") as onu_device_mock, \
Matteo Scandolof6337eb2018-04-05 15:58:37 -0700149 patch.object(VOLTService.objects, "get") as olt_service_mock:
150 service_instance_mock.return_value = self.si
Matteo Scandoloe2cb8a42018-05-18 16:30:06 -0700151 onu_device_mock.return_value = self.onu_device
Matteo Scandolof6337eb2018-04-05 15:58:37 -0700152 olt_service_mock.return_value = self.volt_service
153
154 with self.assertRaises(Exception) as e:
155 self.sync_step().sync_record(self.o)
Luca Preteca974c82018-05-01 18:06:16 -0700156 self.assertTrue(m.called)
157 self.assertEqual(e.exception.message, "Failed to add subscriber in onos voltha: Mock Error")
Matteo Scandolof6337eb2018-04-05 15:58:37 -0700158
159if __name__ == "__main__":
Luca Preteca974c82018-05-01 18:06:16 -0700160 unittest.main()