blob: 1e66be2822ce4ffbc5029502565e79d33dc27d97 [file] [log] [blame]
# Copyright 2017-present Open Networking Foundation
#
# 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 unittest
import os, sys
from mock import patch, Mock, MagicMock
test_path=os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
service_dir=os.path.join(test_path, "../../../..")
xos_dir=os.path.join(test_path, "../../..")
if not os.path.exists(os.path.join(test_path, "new_base")):
xos_dir=os.path.join(test_path, "../../../../../../orchestration/xos/xos")
services_dir=os.path.join(xos_dir, "../../xos_services")
# mocking XOS exception, as they're based in Django
class Exceptions:
XOSValidationError = Exception
XOSProgrammingError = Exception
XOSPermissionDenied = Exception
class XOS:
exceptions = Exceptions
class TestRCORDModels(unittest.TestCase):
def setUp(self):
self.sys_path_save = sys.path
sys.path.append(xos_dir)
self.xos = XOS
self.models_decl = Mock()
self.models_decl.RCORDSubscriber_decl = MagicMock
self.models_decl.RCORDSubscriber_decl.save = Mock()
self.models_decl.RCORDSubscriber_decl.objects = Mock()
self.models_decl.RCORDSubscriber_decl.objects.filter.return_value = []
modules = {
'xos.exceptions': self.xos.exceptions,
'models_decl': self.models_decl
}
self.module_patcher = patch.dict('sys.modules', modules)
self.module_patcher.start()
self.volt = Mock()
from models import RCORDSubscriber
self.rcord_subscriber_class = RCORDSubscriber
self.rcord_subscriber = RCORDSubscriber()
self.rcord_subscriber.id = None # this is a new model
self.rcord_subscriber.is_new = True
self.rcord_subscriber.onu_device = "BRCM1234"
self.rcord_subscriber.c_tag = 111
self.rcord_subscriber.s_tag = 222
self.rcord_subscriber.ip_address = "1.1.1.1"
self.rcord_subscriber.mac_address = "00:AA:00:00:00:01"
self.rcord_subscriber.owner.leaf_model.access = "voltha"
self.rcord_subscriber.owner.provider_services = [self.volt]
def tearDown(self):
sys.path = self.sys_path_save
def test_save(self):
self.rcord_subscriber.save()
self.models_decl.RCORDSubscriber_decl.save.assert_called()
def test_validate_ip_address(self):
self.rcord_subscriber.ip_address = "invalid"
with self.assertRaises(Exception) as e:
self.rcord_subscriber.save()
self.assertEqual(e.exception.message, "The ip_address you specified (invalid) is not valid")
self.models_decl.RCORDSubscriber_decl.save.assert_not_called()
def test_validate_mac_address(self):
self.rcord_subscriber.mac_address = "invalid"
with self.assertRaises(Exception) as e:
self.rcord_subscriber.save()
self.assertEqual(e.exception.message, "The mac_address you specified (invalid) is not valid")
self.models_decl.RCORDSubscriber_decl.save.assert_not_called()
def test_valid_onu_device(self):
self.rcord_subscriber.save()
self.models_decl.RCORDSubscriber_decl.save.assert_called()
def test_invalid_onu_device(self):
self.volt.leaf_model.has_access_device.return_value = False
with self.assertRaises(Exception) as e:
self.rcord_subscriber.save()
self.assertEqual(e.exception.message, "The onu_device you specified (BRCM1234) does not exists")
self.models_decl.RCORDSubscriber_decl.save.assert_not_called()
def test_validate_c_tag_pass(self):
"""
check that other subscriber attached to the same ONU don't have the same c_tag
"""
self.models_decl.RCORDSubscriber_decl.objects.filter.return_value = [self.rcord_subscriber]
self.rcord_subscriber.save()
self.models_decl.RCORDSubscriber_decl.save.assert_called()
def test_validate_c_tag_fail(self):
"""
check that other subscriber attached to the same ONU don't have the same c_tag
"""
s = Mock()
s.c_tag = 111
s.onu_device = "BRCM1234"
self.models_decl.RCORDSubscriber_decl.objects.filter.return_value = [s, self.rcord_subscriber]
with self.assertRaises(Exception) as e:
self.rcord_subscriber.save()
self.assertEqual(e.exception.message, "The c_tag you specified (111) has already been used on device BRCM1234")
self.models_decl.RCORDSubscriber_decl.save.assert_not_called()
def _test_validate_c_tag_on_same_s_tag(self):
"""
check that other subscriber using the same s_tag don't have the same c_tag
"""
s = Mock()
s.id = 123
s.c_tag = 111
s.s_tag = 222
s.onu_device = "BRCM1234"
with self.assertRaises(Exception) as e:
self.rcord_subscriber.save()
self.assertEqual(e.exception.message, "The c_tag you specified (111) has already been used by Subscriber with id 123 and the same s_tag: 222")
self.models_decl.RCORDSubscriber_decl.save.assert_not_called()
def test_validate_c_tag_on_update(self):
s = Mock()
s.c_tag = 111
s.onu_device = "BRCM1234"
s.id = 1
self.models_decl.RCORDSubscriber_decl.objects.filter.return_value = [s]
self.rcord_subscriber.is_new = False
self.rcord_subscriber.id = 1
self.rcord_subscriber.save()
self.models_decl.RCORDSubscriber_decl.save.assert_called()
def test_validate_c_tag_on_update_fail(self):
s = Mock()
s.c_tag = 222
s.onu_device = "BRCM1234"
s.id = 2
self.models_decl.RCORDSubscriber_decl.objects.filter.return_value = [s]
self.rcord_subscriber.id = 1
self.rcord_subscriber.is_new = False
self.rcord_subscriber.c_tag = 222
with self.assertRaises(Exception) as e:
self.rcord_subscriber.save()
self.assertEqual(e.exception.message, "The c_tag you specified (222) has already been used on device BRCM1234")
self.models_decl.RCORDSubscriber_decl.save.assert_not_called()
def test_generate_c_tag(self):
s = Mock()
s.c_tag = "111"
s.onu_device = "BRCM1234"
self.models_decl.RCORDSubscriber_decl.objects.filter.return_value = [s]
self.rcord_subscriber.c_tag = None
self.rcord_subscriber.save()
self.models_decl.RCORDSubscriber_decl.save.assert_called()
self.assertNotEquals(self.rcord_subscriber.c_tag, "111")
self.assertGreater(self.rcord_subscriber.c_tag, 16)
self.assertLess(self.rcord_subscriber.c_tag, 4097)
def test_generate_s_tag(self):
self.rcord_subscriber.c_tag = None
self.rcord_subscriber.save()
self.models_decl.RCORDSubscriber_decl.save.assert_called()
self.assertNotEqual(self.rcord_subscriber.s_tag, None)
def test_provisioned_s_stag(self):
self.rcord_subscriber.save()
self.models_decl.RCORDSubscriber_decl.save.assert_called()
self.assertEqual(self.rcord_subscriber.s_tag, 222)
if __name__ == '__main__':
unittest.main()