blob: aba57aab1d0912f1c5d84093e808b14c343e38ff [file] [log] [blame]
Matteo Scandolo53d61192018-06-01 16:11:14 -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
Matteo Scandolo53d61192018-06-01 16:11:14 -070021test_path=os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
Matteo Scandolo53d61192018-06-01 16:11:14 -070022
23class TestSyncVOLTServiceInstance(unittest.TestCase):
24 def setUp(self):
25
26 self.sys_path_save = sys.path
Matteo Scandolo53d61192018-06-01 16:11:14 -070027
28 # Setting up the config module
29 from xosconfig import Config
Matteo Scandolof7ebb112018-09-18 16:17:22 -070030 config = os.path.join(test_path, "../test_config.yaml")
Matteo Scandolo53d61192018-06-01 16:11:14 -070031 Config.clear()
32 Config.init(config, "synchronizer-config-schema.yaml")
33 # END Setting up the config module
34
Scott Baker47b47302019-01-30 16:55:07 -080035 from xossynchronizer.mock_modelaccessor_build import mock_modelaccessor_config
36 mock_modelaccessor_config(test_path, [("olt-service", "volt.xproto"),
Matteo Scandolo35207b72019-05-10 08:46:48 -070037 ("rcord", "rcord.xproto")])
Matteo Scandolo53d61192018-06-01 16:11:14 -070038
Scott Baker47b47302019-01-30 16:55:07 -080039 import xossynchronizer.modelaccessor
40 reload(xossynchronizer.modelaccessor) # in case nose2 loaded it in a previous test
41
42 from xossynchronizer.modelaccessor import model_accessor
43 self.model_accessor = model_accessor
44
45 from xossynchronizer.steps.syncstep import DeferredException
Matteo Scandolo53d61192018-06-01 16:11:14 -070046 from sync_onu_device import SyncONUDevice, model_accessor
47
48 # import all class names to globals
49 for (k, v) in model_accessor.all_model_classes.items():
50 globals()[k] = v
51
52 self.sync_step = SyncONUDevice
53
54 volt_service = Mock()
55 volt_service.voltha_url = "voltha_url"
56 volt_service.voltha_port = 1234
57 volt_service.voltha_user = "voltha_user"
58 volt_service.voltha_pass = "voltha_pass"
59
60 self.o = Mock()
61 self.o.device_id = "test_id"
62 self.o.pon_port.olt_device.volt_service = volt_service
63
64 def tearDown(self):
65 self.o = None
66 sys.path = self.sys_path_save
67
68 @requests_mock.Mocker()
Matteo Scandoloe3daaa22018-06-19 14:36:31 -070069 def test_enable(self, m):
70 m.post("http://voltha_url:1234/api/v1/devices/test_id/enable")
71
Matteo Scandolo53d61192018-06-01 16:11:14 -070072 self.o.admin_state = "ENABLED"
Scott Baker47b47302019-01-30 16:55:07 -080073 self.sync_step(model_accessor=self.model_accessor).sync_record(self.o)
Matteo Scandoloe3daaa22018-06-19 14:36:31 -070074 self.assertTrue(m.called)
Matteo Scandolo53d61192018-06-01 16:11:14 -070075
76 @requests_mock.Mocker()
77 def test_disable(self, m):
78 m.post("http://voltha_url:1234/api/v1/devices/test_id/disable")
79
80 self.o.admin_state = "DISABLED"
Scott Baker47b47302019-01-30 16:55:07 -080081 self.sync_step(model_accessor=self.model_accessor).sync_record(self.o)
Matteo Scandolo53d61192018-06-01 16:11:14 -070082 self.assertTrue(m.called)
83
84 @requests_mock.Mocker()
Andy Baviercd940192019-05-15 12:59:35 -070085 def test_admin_disabled(self, m):
86 m.post("http://voltha_url:1234/api/v1/devices/test_id/disable")
87
88 self.o.admin_state = "ADMIN_DISABLED"
89 self.sync_step(model_accessor=self.model_accessor).sync_record(self.o)
90 self.assertTrue(m.called)
91
92 @requests_mock.Mocker()
Matteo Scandolo53d61192018-06-01 16:11:14 -070093 def test_disable_fail(self, m):
94 m.post("http://voltha_url:1234/api/v1/devices/test_id/disable", status_code=500, text="Mock Error")
95
96 self.o.admin_state = "DISABLED"
97
98 with self.assertRaises(Exception) as e:
Scott Baker47b47302019-01-30 16:55:07 -080099 self.sync_step(model_accessor=self.model_accessor).sync_record(self.o)
Matteo Scandolo53d61192018-06-01 16:11:14 -0700100 self.assertTrue(m.called)
101 self.assertEqual(e.exception.message, "Failed to disable ONU device: Mock Error")
102
103if __name__ == "__main__":
104 unittest.main()