blob: 80a644bfae929d04eebd743946f27dd223dda0dc [file] [log] [blame]
Matteo Scandoloea529092018-09-11 16:36:39 -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
Matteo Scandoloea529092018-09-11 16:36:39 -070021test_path=os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
Matteo Scandoloea529092018-09-11 16:36:39 -070022
23
Matteo Scandoloea529092018-09-11 16:36:39 -070024class TestAttHelpers(unittest.TestCase):
25
26 def setUp(self):
27
28 self.sys_path_save = sys.path
Matteo Scandoloea529092018-09-11 16:36:39 -070029
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
Scott Baker0b4ad932018-11-26 15:02:13 -080037 from multistructlog import create_logger
38 self.log = create_logger(Config().get('logging'))
39
Scott Baker71d20472019-02-01 12:05:35 -080040 from xossynchronizer.mock_modelaccessor_build import mock_modelaccessor_config
41 mock_modelaccessor_config(test_path, [("att-workflow-driver", "att-workflow-driver.xproto"),
42 ("olt-service", "volt.xproto"),
Matteo Scandoload6c8942019-02-14 13:02:08 -080043 ("rcord", "rcord.xproto")])
Matteo Scandoloea529092018-09-11 16:36:39 -070044
Scott Baker71d20472019-02-01 12:05:35 -080045 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 AttHelpers
Matteo Scandoloea529092018-09-11 16:36:39 -070052
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 = AttHelpers
Scott Baker71d20472019-02-01 12:05:35 -080058 self.model_accessor = model_accessor
Matteo Scandoloea529092018-09-11 16:36:39 -070059
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.att_si = AttWorkflowDriverServiceInstance(
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 = AttWorkflowDriverWhiteListEntry(
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(AttWorkflowDriverWhiteListEntry.objects, "get_items") as whitelist_mock:
97 whitelist_mock.return_value = []
98
Scott Baker71d20472019-02-01 12:05:35 -080099 [res, message] = self.helpers.validate_onu(self.model_accessor, self.log, self.att_si)
Matteo Scandoloea529092018-09-11 16:36:39 -0700100
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(AttWorkflowDriverWhiteListEntry.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
Scott Baker71d20472019-02-01 12:05:35 -0800111 [res, message] = self.helpers.validate_onu(self.model_accessor, self.log, self.att_si)
Matteo Scandoloea529092018-09-11 16:36:39 -0700112
113 self.assertFalse(res)
114 self.assertEqual(message, "ONU activated in wrong location")
115
116 def test_wrong_location_device(self):
117 self.att_si.of_dpid = 666
118 with patch.object(AttWorkflowDriverWhiteListEntry.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
Scott Baker71d20472019-02-01 12:05:35 -0800123 [res, message] = self.helpers.validate_onu(self.model_accessor, self.log, self.att_si)
Matteo Scandoloea529092018-09-11 16:36:39 -0700124
125 self.assertFalse(res)
126 self.assertEqual(message, "ONU activated in wrong location")
127
128 def test_deferred_validation(self):
129 with patch.object(AttWorkflowDriverWhiteListEntry.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:
Scott Baker71d20472019-02-01 12:05:35 -0800135 self.helpers.validate_onu(self.model_accessor, self.log, self.att_si)
Matteo Scandoloea529092018-09-11 16:36:39 -0700136
137 self.assertEqual(e.exception.message, "ONU device %s is not know to XOS yet" % self.att_si.serial_number)
138
139 def test_validating_onu(self):
140 with patch.object(AttWorkflowDriverWhiteListEntry.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
Scott Baker71d20472019-02-01 12:05:35 -0800145 [res, message] = self.helpers.validate_onu(self.model_accessor, self.log, self.att_si)
Matteo Scandoloea529092018-09-11 16:36:39 -0700146
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(AttWorkflowDriverWhiteListEntry.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
Scott Baker71d20472019-02-01 12:05:35 -0800157 [res, message] = self.helpers.validate_onu(self.model_accessor, self.log, self.att_si)
Matteo Scandoloea529092018-09-11 16:36:39 -0700158
159 self.assertTrue(res)
160 self.assertEqual(message, "ONU has been validated")
161
162if __name__ == '__main__':
163 unittest.main()