| # |
| # Copyright 2018 the original author or authors. |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| # |
| import json |
| from unittest import TestCase, main |
| |
| from bitstring import BitArray |
| from common.kvstore.kv_client import KVPair |
| from common.pon_resource_manager.resource_manager import PONResourceManager |
| from mock import Mock |
| from twisted.internet.defer import inlineCallbacks |
| |
| |
| class TestResourceManager(TestCase): |
| def setUp(self): |
| self._rm = PONResourceManager('xgspon', 'default', |
| '0001c889ee7189fb', 'consul', |
| 'localhost', 8500) |
| self.default_resource_range = { |
| "onu_start_idx": 1, |
| "onu_end_idx": 127, |
| "alloc_id_start_idx": 1024, |
| "alloc_id_end_idx": 65534, |
| "gem_port_id_start_idx": 1024, |
| "gem_port_id_end_idx": 16383, |
| "num_of_pon_port": 16 |
| } |
| |
| def tearDown(self): |
| self._rm = None |
| self.default_resource_range = None |
| |
| @inlineCallbacks |
| def test_init_pon_resource_ranges(self): |
| key = PONResourceManager.PON_RESOURCE_RANGE_CONFIG_PATH.format( |
| 'xgspon', 'default') |
| value = json.dumps(self.default_resource_range).encode('utf-8') |
| output = KVPair(key, value, None) |
| self._rm._kv_store.get = Mock(return_value=(output, None)) |
| |
| yield self._rm.init_pon_resource_ranges() |
| self.assertEqual(self._rm.pon_resource_ranges, |
| self.default_resource_range) |
| |
| self._rm._kv_store.get = Mock(return_value=(None, None)) |
| |
| yield self._rm.init_pon_resource_ranges() |
| self.assertEqual(self._rm.pon_resource_ranges, |
| self.default_resource_range) |
| |
| @inlineCallbacks |
| def test_init_resource_id_pool(self): |
| self._rm._kv_store.get = Mock(return_value=(None, None)) |
| self._rm._kv_store.put = Mock(return_value=None) |
| status = yield self._rm.init_resource_id_pool(0, 'ONU_ID', 1, 127) |
| self.assertTrue(status) |
| status = yield self._rm.init_resource_id_pool( |
| 1, 'ALLOC_ID', 1024, 16383) |
| self.assertTrue(status) |
| status = yield self._rm.init_resource_id_pool( |
| 2, 'GEMPORT_ID', 1023, 65534) |
| self.assertTrue(status) |
| |
| @inlineCallbacks |
| def test_get_resource_id(self): |
| # Get onu id test |
| onu_id_resource = self._rm._format_resource(0, 1, 127) |
| key = self._rm._get_path(0, PONResourceManager.ONU_ID) |
| value = onu_id_resource.encode('utf-8') |
| output = KVPair(key, value, None) |
| self._rm._kv_store.get = Mock(return_value=(output, None)) |
| self._rm._kv_store.put = Mock(return_value=None) |
| result = yield self._rm.get_resource_id(0, 'ONU_ID') |
| self.assertEqual(result, 1) |
| |
| # Get alloc id test |
| alloc_id_resource = self._rm._format_resource(1, 1024, 16383) |
| key = self._rm._get_path(1, PONResourceManager.ALLOC_ID) |
| value = alloc_id_resource.encode('utf-8') |
| output = KVPair(key, value, None) |
| self._rm._kv_store.get = Mock(return_value=(output, None)) |
| result = yield self._rm.get_resource_id(1, 'ALLOC_ID', 1) |
| self.assertEqual(result[0], 1024) |
| result = yield self._rm.get_resource_id(1, 'ALLOC_ID', 4) |
| self.assertEqual(result, [1024, 1025, 1026, 1027]) |
| |
| # Get gemport id test |
| gemport_id_resource = self._rm._format_resource(2, 1023, 65534) |
| key = self._rm._get_path(2, PONResourceManager.GEMPORT_ID) |
| value = gemport_id_resource.encode('utf-8') |
| output = KVPair(key, value, None) |
| self._rm._kv_store.get = Mock(return_value=(output, None)) |
| result = yield self._rm.get_resource_id(2, 'GEMPORT_ID', 1) |
| self.assertEqual(result[0], 1023) |
| result = yield self._rm.get_resource_id(2, 'GEMPORT_ID', 5) |
| self.assertEqual(result, [1023, 1024, 1025, 1026, 1027]) |
| |
| @inlineCallbacks |
| def test_free_resource_id(self): |
| # Free onu id test |
| self._rm._kv_store.put = Mock(return_value=None) |
| onu_id_resource = eval(self._rm._format_resource(0, 1, 127)) |
| onu_id_resource['pool'] = BitArray('0b' + onu_id_resource['pool']) |
| self._rm._generate_next_id(onu_id_resource) |
| onu_id_resource['pool'] = onu_id_resource['pool'].bin |
| key = self._rm._get_path(0, PONResourceManager.ONU_ID) |
| value = json.dumps(onu_id_resource).encode('utf-8') |
| output = KVPair(key, value, None) |
| self._rm._kv_store.get = Mock(return_value=(output, None)) |
| result = yield self._rm.free_resource_id(0, 'ONU_ID', 1) |
| self.assertTrue(result) |
| |
| # Free alloc id test |
| alloc_id_resource = eval(self._rm._format_resource(1, 1024, 16383)) |
| alloc_id_resource['pool'] = BitArray('0b' + alloc_id_resource['pool']) |
| |
| for num in range(5): |
| self._rm._generate_next_id(alloc_id_resource) |
| |
| alloc_id_resource['pool'] = alloc_id_resource['pool'].bin |
| key = self._rm._get_path(0, PONResourceManager.ALLOC_ID) |
| value = json.dumps(alloc_id_resource).encode('utf-8') |
| output = KVPair(key, value, None) |
| self._rm._kv_store.get = Mock(return_value=(output, None)) |
| result = self._rm.free_resource_id(1, 'ALLOC_ID', |
| [1024, 1025, 1026, 1027, 1028]) |
| self.assertTrue(result) |
| |
| # Free gemport id test |
| gemport_id_resource = eval(self._rm._format_resource(2, 1023, 65534)) |
| gemport_id_resource['pool'] = BitArray( |
| '0b' + gemport_id_resource['pool']) |
| |
| for num in range(6): |
| self._rm._generate_next_id(gemport_id_resource) |
| |
| gemport_id_resource['pool'] = gemport_id_resource['pool'].bin |
| key = self._rm._get_path(0, PONResourceManager.GEMPORT_ID) |
| value = json.dumps(gemport_id_resource).encode('utf-8') |
| output = KVPair(key, value, None) |
| self._rm._kv_store.get = Mock(return_value=(output, None)) |
| result = self._rm.free_resource_id(2, 'GEMPORT_ID', |
| [1023, 1024, 1025, 1026, 1027, 1028]) |
| self.assertTrue(result) |
| |
| @inlineCallbacks |
| def test_clear_resource_id_pool(self): |
| self._rm._kv_store.delete = Mock(return_value=None) |
| status = yield self._rm.clear_resource_id_pool(0, 'ONU_ID') |
| self.assertTrue(status) |
| self._rm._kv_store.delete = Mock(return_value="error") |
| status = yield self._rm.clear_resource_id_pool(1, 'ALLOC_ID') |
| self.assertFalse(status) |
| |
| |
| if __name__ == '__main__': |
| main() |