blob: e46cbc9152f8ba6fc37aab630cac506a4851d8cc [file] [log] [blame]
Andy Bavier561f3e52019-03-15 10:46:05 -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
21test_path=os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
22
23
24class TestTtHelpers(unittest.TestCase):
25
26 def setUp(self):
27
28 self.sys_path_save = sys.path
29
30 # Setting up the config module
31 from xosconfig import Config
32 config = os.path.join(test_path, "test_config.yaml")
33 Config.clear()
34 Config.init(config, "synchronizer-config-schema.yaml")
35 # END Setting up the config module
36
37 from multistructlog import create_logger
38 self.log = create_logger(Config().get('logging'))
39
40 from xossynchronizer.mock_modelaccessor_build import mock_modelaccessor_config
41 mock_modelaccessor_config(test_path, [("tt-workflow-driver", "tt-workflow-driver.xproto"),
42 ("olt-service", "volt.xproto"),
43 ("rcord", "rcord.xproto")])
44
45 import xossynchronizer.modelaccessor
46 import mock_modelaccessor
47 reload(mock_modelaccessor) # in case nose2 loaded it in a previous test
48 reload(xossynchronizer.modelaccessor) # in case nose2 loaded it in a previous test
49
50 from xossynchronizer.modelaccessor import model_accessor
51 from helpers import TtHelpers
52
53 # import all class names to globals
54 for (k, v) in model_accessor.all_model_classes.items():
55 globals()[k] = v
56
57 self.helpers = TtHelpers
58 self.model_accessor = model_accessor
59
60 self._volt = VOLTService()
61 self._volt.id = 1
62
63 self.volt = Service()
64 self.volt.id = 1
65 self.volt.name = "vOLT"
66 self.volt.leaf_model = self._volt
67
68 self.pon_port = PONPort()
69 self.pon_port.port_no = 1234
70
71 self.onu = ONUDevice()
72 self.onu.pon_port = self.pon_port
73 self.onu.serial_number = "BRCM1234"
74
75 self.tt_si = TtWorkflowDriverServiceInstance(
76 serial_number="BRCM1234",
77 owner=self.volt,
78 owner_id=self.volt.id,
79 of_dpid="of:1234"
80 )
81
82 self.whitelist_entry = TtWorkflowDriverWhiteListEntry(
83 serial_number="BRCM1234",
84 owner=self.volt,
85 owner_id=self.volt.id,
86 pon_port_id=1234,
87 device_id="of:1234"
88 )
89
90
91 def tearDown(self):
92 sys.path = self.sys_path_save
93
94 def test_not_in_whitelist(self):
95
96 with patch.object(TtWorkflowDriverWhiteListEntry.objects, "get_items") as whitelist_mock:
97 whitelist_mock.return_value = []
98
99 [res, message] = self.helpers.validate_onu(self.model_accessor, self.log, self.tt_si)
100
101 self.assertFalse(res)
102 self.assertEqual(message, "ONU not found in whitelist")
103
104 def test_wrong_location_port(self):
105 self.pon_port.port_no = 666
106 with patch.object(TtWorkflowDriverWhiteListEntry.objects, "get_items") as whitelist_mock, \
107 patch.object(ONUDevice.objects, "get_items") as onu_mock:
108 whitelist_mock.return_value = [self.whitelist_entry]
109 onu_mock.return_value = [self.onu]
110
111 [res, message] = self.helpers.validate_onu(self.model_accessor, self.log, self.tt_si)
112
113 self.assertFalse(res)
114 self.assertEqual(message, "ONU activated in wrong location")
115
116 def test_wrong_location_device(self):
117 self.tt_si.of_dpid = 666
118 with patch.object(TtWorkflowDriverWhiteListEntry.objects, "get_items") as whitelist_mock, \
119 patch.object(ONUDevice.objects, "get_items") as onu_mock:
120 whitelist_mock.return_value = [self.whitelist_entry]
121 onu_mock.return_value = [self.onu]
122
123 [res, message] = self.helpers.validate_onu(self.model_accessor, self.log, self.tt_si)
124
125 self.assertFalse(res)
126 self.assertEqual(message, "ONU activated in wrong location")
127
128 def test_deferred_validation(self):
129 with patch.object(TtWorkflowDriverWhiteListEntry.objects, "get_items") as whitelist_mock, \
130 patch.object(ONUDevice.objects, "get_items") as onu_mock:
131 whitelist_mock.return_value = [self.whitelist_entry]
132 onu_mock.return_value = []
133
134 with self.assertRaises(Exception) as e:
135 self.helpers.validate_onu(self.model_accessor, self.log, self.tt_si)
136
137 self.assertEqual(e.exception.message, "ONU device %s is not know to XOS yet" % self.tt_si.serial_number)
138
139 def test_validating_onu(self):
140 with patch.object(TtWorkflowDriverWhiteListEntry.objects, "get_items") as whitelist_mock, \
141 patch.object(ONUDevice.objects, "get_items") as onu_mock:
142 whitelist_mock.return_value = [self.whitelist_entry]
143 onu_mock.return_value = [self.onu]
144
145 [res, message] = self.helpers.validate_onu(self.model_accessor, self.log, self.tt_si)
146
147 self.assertTrue(res)
148 self.assertEqual(message, "ONU has been validated")
149
150 def test_validating_onu_lowercase(self):
151 self.whitelist_entry.serial_number = "brcm1234"
152 with patch.object(TtWorkflowDriverWhiteListEntry.objects, "get_items") as whitelist_mock, \
153 patch.object(ONUDevice.objects, "get_items") as onu_mock:
154 whitelist_mock.return_value = [self.whitelist_entry]
155 onu_mock.return_value = [self.onu]
156
157 [res, message] = self.helpers.validate_onu(self.model_accessor, self.log, self.tt_si)
158
159 self.assertTrue(res)
160 self.assertEqual(message, "ONU has been validated")
161
162if __name__ == '__main__':
163 unittest.main()