blob: 3de65c1ddd9b9bb79dd64fc8b3b93f23f3561d14 [file] [log] [blame]
Matteo Scandolo85f54982018-05-18 11:33:35 -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 json
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
43class TestSyncOLTDevice(unittest.TestCase):
44
45 def setUp(self):
46 global DeferredException
47
48 self.sys_path_save = sys.path
49 sys.path.append(xos_dir)
50 sys.path.append(os.path.join(xos_dir, 'synchronizers', 'new_base'))
51
52 # Setting up the config module
53 from xosconfig import Config
54 config = os.path.join(test_path, "../model_policies/test_config.yaml")
55 Config.clear()
56 Config.init(config, "synchronizer-config-schema.yaml")
57 # END Setting up the config module
58
59 from synchronizers.new_base.mock_modelaccessor_build import build_mock_modelaccessor
60 # build_mock_modelaccessor(xos_dir, services_dir, [get_models_fn("olt-service", "volt.xproto")])
61
62 # FIXME this is to get jenkins to pass the tests, somehow it is running tests in a different order
63 # and apparently it is not overriding the generated model accessor
64 build_mock_modelaccessor(xos_dir, services_dir, [get_models_fn("olt-service", "volt.xproto"),
65 get_models_fn("vsg", "vsg.xproto"),
66 get_models_fn("../profiles/rcord", "rcord.xproto")])
67 import synchronizers.new_base.modelaccessor
68 from onu_event import ONUEventStep, model_accessor
69
70 # import all class names to globals
71 for (k, v) in model_accessor.all_model_classes.items():
72 globals()[k] = v
73
74 self.log = Mock()
75
76 self.event_step = ONUEventStep(self.log)
77
78 self.event = Mock()
79 self.event.value = json.dumps({
Matteo Scandolo7db85372018-05-22 15:26:55 -070080 'status': 'activated',
Matteo Scandolo85f54982018-05-18 11:33:35 -070081 'serial_number': 'BRCM1234',
82 'uni_port_of_id': 'of:00100101',
83 'of_dpid': 'of:109299321'
84 })
85
86 self.onu = Mock()
87 self.onu.serial_number = "BRCM1234"
88 self.onu.pon_port.olt_device.volt_service.id = 1
89
90 self.service = Mock(id=1)
Matteo Scandolo7db85372018-05-22 15:26:55 -070091 self.service.subscriber_services = []
Matteo Scandolo85f54982018-05-18 11:33:35 -070092
93 self.oss = Mock()
94 self.oss.kind = "OSS"
95 self.oss.leaf_model = Mock()
96
97 def tearDown(self):
Matteo Scandolo7db85372018-05-22 15:26:55 -070098 self.service.subscriber_services = []
Matteo Scandolo85f54982018-05-18 11:33:35 -070099
100 def test_missing_onu(self):
101 with patch.object(ONUDevice.objects, "get_items") as onu_device_mock:
102 onu_device_mock.side_effect = IndexError("No ONU")
103
104 with self.assertRaises(Exception) as e:
105 self.event_step.process_event(self.event)
106
107 self.assertEqual(e.exception.message, "No ONUDevice with serial_number %s is present in XOS" % self.onu.serial_number)
108
109 def test_do_nothing(self):
110 with patch.object(ONUDevice.objects, "get_items") as onu_device_mock , \
111 patch.object(Service.objects, "get_items") as service_mock, \
112 patch.object(self.log, "info") as logInfo:
113
114 onu_device_mock.return_value = [self.onu]
115 service_mock.return_value = [self.service]
116
117 self.event_step.process_event(self.event)
118
119 logInfo.assert_called_with("Not processing events as no OSS service is present (is it a provider of vOLT?")
120
121 def test_call_oss(self):
Matteo Scandolo7db85372018-05-22 15:26:55 -0700122 self.service.subscriber_services = [self.oss]
Matteo Scandolo85f54982018-05-18 11:33:35 -0700123
124 with patch.object(ONUDevice.objects, "get_items") as onu_device_mock , \
125 patch.object(Service.objects, "get_items") as service_mock, \
126 patch.object(self.oss.leaf_model, "validate_onu") as validate_onu:
127
128 onu_device_mock.return_value = [self.onu]
129
130 service_mock.return_value = [self.service]
131
132 self.event_step.process_event(self.event)
133
134 validate_onu.assert_called_with(json.loads(self.event.value))