FM / v2

/opt/imh-python/lib/python3.9/site-packages/openstack/tests/unit/network/v2/

/opt/imh-python/lib/python3.9/site-packages/openstack/tests/unit/network/v2

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
test_address_scope.py	1621	0644	edit dl rm
test_agent.py	5825	0644	edit dl rm
test_auto_allocated_topology.py	1359	0644	edit dl rm
test_availability_zone.py	1585	0644	edit dl rm
test_extension.py	1683	0644	edit dl rm
test_firewall_group.py	2171	0644	edit dl rm
test_firewall_policy.py	1888	0644	edit dl rm
test_firewall_rule.py	2433	0644	edit dl rm
test_flavor.py	3429	0644	edit dl rm
test_floating_ip.py	4159	0644	edit dl rm
test_health_monitor.py	2309	0644	edit dl rm
test_listener.py	2499	0644	edit dl rm
test_load_balancer.py	2432	0644	edit dl rm
test_metering_label.py	1665	0644	edit dl rm
test_metering_label_rule.py	1830	0644	edit dl rm
test_network.py	5410	0644	edit dl rm
test_network_ip_availability.py	3173	0644	edit dl rm
test_network_segment_range.py	2554	0644	edit dl rm
test_pool.py	3178	0644	edit dl rm
test_pool_member.py	1958	0644	edit dl rm
test_port.py	6638	0644	edit dl rm
test_port_forwarding.py	2461	0644	edit dl rm
test_proxy.py	59907	0644	edit dl rm
test_qos_bandwidth_limit_rule.py	1844	0644	edit dl rm
test_qos_dscp_marking_rule.py	1629	0644	edit dl rm
test_qos_minimum_bandwidth_rule.py	1762	0644	edit dl rm
test_qos_policy.py	1862	0644	edit dl rm
test_qos_rule_type.py	1922	0644	edit dl rm
test_quota.py	4645	0644	edit dl rm
test_rbac_policy.py	1700	0644	edit dl rm
test_router.py	11437	0644	edit dl rm
test_security_group.py	4157	0644	edit dl rm
test_security_group_rule.py	3936	0644	edit dl rm
test_segment.py	1817	0644	edit dl rm
test_service_profile.py	2457	0644	edit dl rm
test_service_provider.py	1479	0644	edit dl rm
test_subnet.py	3152	0644	edit dl rm
test_subnet_pool.py	2659	0644	edit dl rm
test_trunk.py	3581	0644	edit dl rm
test_vpn_service.py	2067	0644	edit dl rm
__init__.py	0	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/openstack/tests/unit/network/v2/test_router.py (11437B)

from unittest import mock

import testtools

from openstack import exceptions
from openstack.network.v2 import router
from openstack.tests.unit import base

IDENTIFIER = 'IDENTIFIER'
EXAMPLE = {
    'admin_state_up': True,
    'availability_zone_hints': ['1'],
    'availability_zones': ['2'],
    'created_at': 'timestamp1',
    'description': '3',
    'distributed': False,
    'external_gateway_info': {'4': 4},
    'flavor_id': '5',
    'ha': False,
    'id': IDENTIFIER,
    'name': '6',
    'revision': 7,
    'routes': ['8'],
    'status': '9',
    'tenant_id': '10',
    'updated_at': 'timestamp2',
}

EXAMPLE_WITH_OPTIONAL = {
    'admin_state_up': False,
    'availability_zone_hints': ['zone-1', 'zone-2'],
    'availability_zones': ['zone-2'],
    'description': 'description',
    'distributed': True,
    'external_gateway_info': {
        'network_id': '1',
        'enable_snat': True,
        'external_fixed_ips': []
    },
    'ha': True,
    'id': IDENTIFIER,
    'name': 'router1',
    'routes': [{
        'nexthop': '172.24.4.20',
        'destination': '10.0.3.1/24'
    }],
    'status': 'ACTIVE',
    'tenant_id': '2',
}

class TestRouter(base.TestCase):

def test_basic(self):
        sot = router.Router()
        self.assertEqual('router', sot.resource_key)
        self.assertEqual('routers', sot.resources_key)
        self.assertEqual('/routers', sot.base_path)
        self.assertTrue(sot.allow_create)
        self.assertTrue(sot.allow_fetch)
        self.assertTrue(sot.allow_commit)
        self.assertTrue(sot.allow_delete)
        self.assertTrue(sot.allow_list)

def test_make_it(self):
        sot = router.Router(**EXAMPLE)
        self.assertTrue(sot.is_admin_state_up)
        self.assertEqual(EXAMPLE['availability_zone_hints'],
                         sot.availability_zone_hints)
        self.assertEqual(EXAMPLE['availability_zones'],
                         sot.availability_zones)
        self.assertEqual(EXAMPLE['created_at'], sot.created_at)
        self.assertEqual(EXAMPLE['description'], sot.description)
        self.assertFalse(sot.is_distributed)
        self.assertEqual(EXAMPLE['external_gateway_info'],
                         sot.external_gateway_info)
        self.assertEqual(EXAMPLE['flavor_id'], sot.flavor_id)
        self.assertFalse(sot.is_ha)
        self.assertEqual(EXAMPLE['id'], sot.id)
        self.assertEqual(EXAMPLE['name'], sot.name)
        self.assertEqual(EXAMPLE['revision'], sot.revision_number)
        self.assertEqual(EXAMPLE['routes'], sot.routes)
        self.assertEqual(EXAMPLE['status'], sot.status)
        self.assertEqual(EXAMPLE['tenant_id'], sot.project_id)
        self.assertEqual(EXAMPLE['updated_at'], sot.updated_at)

def test_make_it_with_optional(self):
        sot = router.Router(**EXAMPLE_WITH_OPTIONAL)
        self.assertFalse(sot.is_admin_state_up)
        self.assertEqual(EXAMPLE_WITH_OPTIONAL['availability_zone_hints'],
                         sot.availability_zone_hints)
        self.assertEqual(EXAMPLE_WITH_OPTIONAL['availability_zones'],
                         sot.availability_zones)
        self.assertEqual(EXAMPLE_WITH_OPTIONAL['description'],
                         sot.description)
        self.assertTrue(sot.is_distributed)
        self.assertEqual(EXAMPLE_WITH_OPTIONAL['external_gateway_info'],
                         sot.external_gateway_info)
        self.assertTrue(sot.is_ha)
        self.assertEqual(EXAMPLE_WITH_OPTIONAL['id'], sot.id)
        self.assertEqual(EXAMPLE_WITH_OPTIONAL['name'], sot.name)
        self.assertEqual(EXAMPLE_WITH_OPTIONAL['routes'], sot.routes)
        self.assertEqual(EXAMPLE_WITH_OPTIONAL['status'], sot.status)
        self.assertEqual(EXAMPLE_WITH_OPTIONAL['tenant_id'], sot.project_id)

def test_add_interface_subnet(self):
        # Add subnet to a router
        sot = router.Router(**EXAMPLE)
        response = mock.Mock()
        response.body = {"subnet_id": "3", "port_id": "2"}
        response.json = mock.Mock(return_value=response.body)
        response.status_code = 200
        sess = mock.Mock()
        sess.put = mock.Mock(return_value=response)
        body = {"subnet_id": "3"}
        self.assertEqual(response.body, sot.add_interface(sess, **body))

url = 'routers/IDENTIFIER/add_router_interface'
        sess.put.assert_called_with(url,
                                    json=body)

def test_add_interface_port(self):
        # Add port to a router
        sot = router.Router(**EXAMPLE)
        response = mock.Mock()
        response.body = {"subnet_id": "3", "port_id": "3"}
        response.json = mock.Mock(return_value=response.body)
        response.status_code = 200
        sess = mock.Mock()
        sess.put = mock.Mock(return_value=response)

body = {"port_id": "3"}
        self.assertEqual(response.body, sot.add_interface(sess, **body))

url = 'routers/IDENTIFIER/add_router_interface'
        sess.put.assert_called_with(url,
                                    json=body)

def test_remove_interface_subnet(self):
        # Remove subnet from a router
        sot = router.Router(**EXAMPLE)
        response = mock.Mock()
        response.body = {"subnet_id": "3", "port_id": "2"}
        response.json = mock.Mock(return_value=response.body)
        response.status_code = 200
        sess = mock.Mock()
        sess.put = mock.Mock(return_value=response)
        body = {"subnet_id": "3"}
        self.assertEqual(response.body, sot.remove_interface(sess, **body))

url = 'routers/IDENTIFIER/remove_router_interface'
        sess.put.assert_called_with(url,
                                    json=body)

def test_remove_interface_port(self):
        # Remove port from a router
        sot = router.Router(**EXAMPLE)
        response = mock.Mock()
        response.body = {"subnet_id": "3", "port_id": "3"}
        response.json = mock.Mock(return_value=response.body)
        response.status_code = 200
        sess = mock.Mock()
        sess.put = mock.Mock(return_value=response)
        body = {"network_id": 3, "enable_snat": True}
        self.assertEqual(response.body, sot.remove_interface(sess, **body))

url = 'routers/IDENTIFIER/remove_router_interface'
        sess.put.assert_called_with(url,
                                    json=body)

def test_add_interface_4xx(self):
        # Neutron may return 4xx, we have to raise if that happens
        sot = router.Router(**EXAMPLE)
        response = mock.Mock()
        msg = '.*borked'
        response.body = {'NeutronError': {'message': msg}}
        response.json = mock.Mock(return_value=response.body)
        response.ok = False
        response.status_code = 409
        response.headers = {'content-type': 'application/json'}
        sess = mock.Mock()
        sess.put = mock.Mock(return_value=response)
        body = {'subnet_id': '3'}
        with testtools.ExpectedException(exceptions.ConflictException, msg):
            sot.add_interface(sess, **body)

def test_remove_interface_4xx(self):
        # Neutron may return 4xx for example if a router interface has
        # extra routes referring to it as a nexthop
        sot = router.Router(**EXAMPLE)
        response = mock.Mock()
        msg = '.*borked'
        response.body = {'NeutronError': {'message': msg}}
        response.json = mock.Mock(return_value=response.body)
        response.ok = False
        response.status_code = 409
        response.headers = {'content-type': 'application/json'}
        sess = mock.Mock()
        sess.put = mock.Mock(return_value=response)
        body = {'subnet_id': '3'}
        with testtools.ExpectedException(exceptions.ConflictException, msg):
            sot.remove_interface(sess, **body)

def test_add_extra_routes(self):
        r = router.Router(**EXAMPLE)
        response = mock.Mock()
        response.headers = {}
        json_body = {'router': {}}
        response.body = json_body
        response.json = mock.Mock(return_value=response.body)
        response.status_code = 200
        sess = mock.Mock()
        sess.put = mock.Mock(return_value=response)
        ret = r.add_extra_routes(sess, json_body)
        self.assertIsInstance(ret, router.Router)
        self.assertIsInstance(ret.routes, list)
        url = 'routers/IDENTIFIER/add_extraroutes'
        sess.put.assert_called_with(url, json=json_body)

def test_remove_extra_routes(self):
        r = router.Router(**EXAMPLE)
        response = mock.Mock()
        response.headers = {}
        json_body = {'router': {}}
        response.body = json_body
        response.json = mock.Mock(return_value=response.body)
        response.status_code = 200
        sess = mock.Mock()
        sess.put = mock.Mock(return_value=response)
        ret = r.remove_extra_routes(sess, json_body)
        self.assertIsInstance(ret, router.Router)
        self.assertIsInstance(ret.routes, list)
        url = 'routers/IDENTIFIER/remove_extraroutes'
        sess.put.assert_called_with(url, json=json_body)

def test_add_router_gateway(self):
        # Add gateway to a router
        sot = router.Router(**EXAMPLE_WITH_OPTIONAL)
        response = mock.Mock()
        response.body = {"network_id": "3", "enable_snat": True}
        response.json = mock.Mock(return_value=response.body)
        sess = mock.Mock()
        sess.put = mock.Mock(return_value=response)
        body = {"network_id": 3, "enable_snat": True}
        self.assertEqual(response.body, sot.add_gateway(sess, **body))

url = 'routers/IDENTIFIER/add_gateway_router'
        sess.put.assert_called_with(url,
                                    json=body)

def test_remove_router_gateway(self):
        # Remove gateway to a router
        sot = router.Router(**EXAMPLE_WITH_OPTIONAL)
        response = mock.Mock()
        response.body = {"network_id": "3", "enable_snat": True}
        response.json = mock.Mock(return_value=response.body)
        sess = mock.Mock()
        sess.put = mock.Mock(return_value=response)
        body = {"network_id": 3, "enable_snat": True}
        self.assertEqual(response.body, sot.remove_gateway(sess, **body))

url = 'routers/IDENTIFIER/remove_gateway_router'
        sess.put.assert_called_with(url,
                                    json=body)

class TestL3AgentRouters(base.TestCase):

def test_basic(self):
        sot = router.L3AgentRouter()
        self.assertEqual('router', sot.resource_key)
        self.assertEqual('routers', sot.resources_key)
        self.assertEqual('/agents/%(agent_id)s/l3-routers', sot.base_path)
        self.assertEqual('l3-router', sot.resource_name)
        self.assertFalse(sot.allow_create)
        self.assertTrue(sot.allow_retrieve)
        self.assertFalse(sot.allow_commit)
        self.assertFalse(sot.allow_delete)
        self.assertTrue(sot.allow_list)

# 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. from unittest import mock import testtools from openstack import exceptions from openstack.network.v2 import router from openstack.tests.unit import base IDENTIFIER = 'IDENTIFIER' EXAMPLE = { 'admin_state_up': True, 'availability_zone_hints': ['1'], 'availability_zones': ['2'], 'created_at': 'timestamp1', 'description': '3', 'distributed': False, 'external_gateway_info': {'4': 4}, 'flavor_id': '5', 'ha': False, 'id': IDENTIFIER, 'name': '6', 'revision': 7, 'routes': ['8'], 'status': '9', 'tenant_id': '10', 'updated_at': 'timestamp2', } EXAMPLE_WITH_OPTIONAL = { 'admin_state_up': False, 'availability_zone_hints': ['zone-1', 'zone-2'], 'availability_zones': ['zone-2'], 'description': 'description', 'distributed': True, 'external_gateway_info': { 'network_id': '1', 'enable_snat': True, 'external_fixed_ips': [] }, 'ha': True, 'id': IDENTIFIER, 'name': 'router1', 'routes': [{ 'nexthop': '172.24.4.20', 'destination': '10.0.3.1/24' }], 'status': 'ACTIVE', 'tenant_id': '2', } class TestRouter(base.TestCase): def test_basic(self): sot = router.Router() self.assertEqual('router', sot.resource_key) self.assertEqual('routers', sot.resources_key) self.assertEqual('/routers', sot.base_path) self.assertTrue(sot.allow_create) self.assertTrue(sot.allow_fetch) self.assertTrue(sot.allow_commit) self.assertTrue(sot.allow_delete) self.assertTrue(sot.allow_list) def test_make_it(self): sot = router.Router(**EXAMPLE) self.assertTrue(sot.is_admin_state_up) self.assertEqual(EXAMPLE['availability_zone_hints'], sot.availability_zone_hints) self.assertEqual(EXAMPLE['availability_zones'], sot.availability_zones) self.assertEqual(EXAMPLE['created_at'], sot.created_at) self.assertEqual(EXAMPLE['description'], sot.description) self.assertFalse(sot.is_distributed) self.assertEqual(EXAMPLE['external_gateway_info'], sot.external_gateway_info) self.assertEqual(EXAMPLE['flavor_id'], sot.flavor_id) self.assertFalse(sot.is_ha) self.assertEqual(EXAMPLE['id'], sot.id) self.assertEqual(EXAMPLE['name'], sot.name) self.assertEqual(EXAMPLE['revision'], sot.revision_number) self.assertEqual(EXAMPLE['routes'], sot.routes) self.assertEqual(EXAMPLE['status'], sot.status) self.assertEqual(EXAMPLE['tenant_id'], sot.project_id) self.assertEqual(EXAMPLE['updated_at'], sot.updated_at) def test_make_it_with_optional(self): sot = router.Router(**EXAMPLE_WITH_OPTIONAL) self.assertFalse(sot.is_admin_state_up) self.assertEqual(EXAMPLE_WITH_OPTIONAL['availability_zone_hints'], sot.availability_zone_hints) self.assertEqual(EXAMPLE_WITH_OPTIONAL['availability_zones'], sot.availability_zones) self.assertEqual(EXAMPLE_WITH_OPTIONAL['description'], sot.description) self.assertTrue(sot.is_distributed) self.assertEqual(EXAMPLE_WITH_OPTIONAL['external_gateway_info'], sot.external_gateway_info) self.assertTrue(sot.is_ha) self.assertEqual(EXAMPLE_WITH_OPTIONAL['id'], sot.id) self.assertEqual(EXAMPLE_WITH_OPTIONAL['name'], sot.name) self.assertEqual(EXAMPLE_WITH_OPTIONAL['routes'], sot.routes) self.assertEqual(EXAMPLE_WITH_OPTIONAL['status'], sot.status) self.assertEqual(EXAMPLE_WITH_OPTIONAL['tenant_id'], sot.project_id) def test_add_interface_subnet(self): # Add subnet to a router sot = router.Router(**EXAMPLE) response = mock.Mock() response.body = {"subnet_id": "3", "port_id": "2"} response.json = mock.Mock(return_value=response.body) response.status_code = 200 sess = mock.Mock() sess.put = mock.Mock(return_value=response) body = {"subnet_id": "3"} self.assertEqual(response.body, sot.add_interface(sess, **body)) url = 'routers/IDENTIFIER/add_router_interface' sess.put.assert_called_with(url, json=body) def test_add_interface_port(self): # Add port to a router sot = router.Router(**EXAMPLE) response = mock.Mock() response.body = {"subnet_id": "3", "port_id": "3"} response.json = mock.Mock(return_value=response.body) response.status_code = 200 sess = mock.Mock() sess.put = mock.Mock(return_value=response) body = {"port_id": "3"} self.assertEqual(response.body, sot.add_interface(sess, **body)) url = 'routers/IDENTIFIER/add_router_interface' sess.put.assert_called_with(url, json=body) def test_remove_interface_subnet(self): # Remove subnet from a router sot = router.Router(**EXAMPLE) response = mock.Mock() response.body = {"subnet_id": "3", "port_id": "2"} response.json = mock.Mock(return_value=response.body) response.status_code = 200 sess = mock.Mock() sess.put = mock.Mock(return_value=response) body = {"subnet_id": "3"} self.assertEqual(response.body, sot.remove_interface(sess, **body)) url = 'routers/IDENTIFIER/remove_router_interface' sess.put.assert_called_with(url, json=body) def test_remove_interface_port(self): # Remove port from a router sot = router.Router(**EXAMPLE) response = mock.Mock() response.body = {"subnet_id": "3", "port_id": "3"} response.json = mock.Mock(return_value=response.body) response.status_code = 200 sess = mock.Mock() sess.put = mock.Mock(return_value=response) body = {"network_id": 3, "enable_snat": True} self.assertEqual(response.body, sot.remove_interface(sess, **body)) url = 'routers/IDENTIFIER/remove_router_interface' sess.put.assert_called_with(url, json=body) def test_add_interface_4xx(self): # Neutron may return 4xx, we have to raise if that happens sot = router.Router(**EXAMPLE) response = mock.Mock() msg = '.*borked' response.body = {'NeutronError': {'message': msg}} response.json = mock.Mock(return_value=response.body) response.ok = False response.status_code = 409 response.headers = {'content-type': 'application/json'} sess = mock.Mock() sess.put = mock.Mock(return_value=response) body = {'subnet_id': '3'} with testtools.ExpectedException(exceptions.ConflictException, msg): sot.add_interface(sess, **body) def test_remove_interface_4xx(self): # Neutron may return 4xx for example if a router interface has # extra routes referring to it as a nexthop sot = router.Router(**EXAMPLE) response = mock.Mock() msg = '.*borked' response.body = {'NeutronError': {'message': msg}} response.json = mock.Mock(return_value=response.body) response.ok = False response.status_code = 409 response.headers = {'content-type': 'application/json'} sess = mock.Mock() sess.put = mock.Mock(return_value=response) body = {'subnet_id': '3'} with testtools.ExpectedException(exceptions.ConflictException, msg): sot.remove_interface(sess, **body) def test_add_extra_routes(self): r = router.Router(**EXAMPLE) response = mock.Mock() response.headers = {} json_body = {'router': {}} response.body = json_body response.json = mock.Mock(return_value=response.body) response.status_code = 200 sess = mock.Mock() sess.put = mock.Mock(return_value=response) ret = r.add_extra_routes(sess, json_body) self.assertIsInstance(ret, router.Router) self.assertIsInstance(ret.routes, list) url = 'routers/IDENTIFIER/add_extraroutes' sess.put.assert_called_with(url, json=json_body) def test_remove_extra_routes(self): r = router.Router(**EXAMPLE) response = mock.Mock() response.headers = {} json_body = {'router': {}} response.body = json_body response.json = mock.Mock(return_value=response.body) response.status_code = 200 sess = mock.Mock() sess.put = mock.Mock(return_value=response) ret = r.remove_extra_routes(sess, json_body) self.assertIsInstance(ret, router.Router) self.assertIsInstance(ret.routes, list) url = 'routers/IDENTIFIER/remove_extraroutes' sess.put.assert_called_with(url, json=json_body) def test_add_router_gateway(self): # Add gateway to a router sot = router.Router(**EXAMPLE_WITH_OPTIONAL) response = mock.Mock() response.body = {"network_id": "3", "enable_snat": True} response.json = mock.Mock(return_value=response.body) sess = mock.Mock() sess.put = mock.Mock(return_value=response) body = {"network_id": 3, "enable_snat": True} self.assertEqual(response.body, sot.add_gateway(sess, **body)) url = 'routers/IDENTIFIER/add_gateway_router' sess.put.assert_called_with(url, json=body) def test_remove_router_gateway(self): # Remove gateway to a router sot = router.Router(**EXAMPLE_WITH_OPTIONAL) response = mock.Mock() response.body = {"network_id": "3", "enable_snat": True} response.json = mock.Mock(return_value=response.body) sess = mock.Mock() sess.put = mock.Mock(return_value=response) body = {"network_id": 3, "enable_snat": True} self.assertEqual(response.body, sot.remove_gateway(sess, **body)) url = 'routers/IDENTIFIER/remove_gateway_router' sess.put.assert_called_with(url, json=body) class TestL3AgentRouters(base.TestCase): def test_basic(self): sot = router.L3AgentRouter() self.assertEqual('router', sot.resource_key) self.assertEqual('routers', sot.resources_key) self.assertEqual('/agents/%(agent_id)s/l3-routers', sot.base_path) self.assertEqual('l3-router', sot.resource_name) self.assertFalse(sot.allow_create) self.assertTrue(sot.allow_retrieve) self.assertFalse(sot.allow_commit) self.assertFalse(sot.allow_delete) self.assertTrue(sot.allow_list)