x/python-group-based-policy-client
Code Issues Proposed changes

Add Bobcat Support

Browse Source 
Change-Id: I59fffbc1c7c96257bfb3a4ddafe98a46d485cca7
This commit is contained in:
Nisar Khan 2024-05-22 06:38:28 +00:00 committed by Mohd Nisar Khan
parent 70276afb9b
commit e973690178
10 changed files with 622 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
.zuul.yaml
View File

@ -14,42 +14,42 @@
- openstack-tox-pep8:
required-projects:
- name: openstack/requirements
override-checkout: stable/2023.1
override-checkout: stable/2023.2
- openstack-tox-py38:
required-projects:
- name: openstack/requirements
override-checkout: stable/2023.1
override-checkout: stable/2023.2
- openstack-tox-py39:
required-projects:
- name: openstack/requirements
override-checkout: stable/2023.1
override-checkout: stable/2023.2
- openstack-tox-py310:
required-projects:
- name: openstack/requirements
override-checkout: stable/2023.1
override-checkout: stable/2023.2
- openstack-tox-py311:
required-projects:
- name: openstack/requirements
override-checkout: stable/2023.1
override-checkout: stable/2023.2
gate:
jobs:
- openstack-tox-pep8:
required-projects:
- name: openstack/requirements
override-checkout: stable/2023.1
override-checkout: stable/2023.2
- openstack-tox-py38:
required-projects:
- name: openstack/requirements
override-checkout: stable/2023.1
override-checkout: stable/2023.2
- openstack-tox-py39:
required-projects:
- name: openstack/requirements
override-checkout: stable/2023.1
override-checkout: stable/2023.2
- openstack-tox-py310:
required-projects:
- name: openstack/requirements
override-checkout: stable/2023.1
override-checkout: stable/2023.2
- openstack-tox-py311:
required-projects:
- name: openstack/requirements
override-checkout: stable/2023.1
override-checkout: stable/2023.2

6
gbpclient/tests/unit/test_address_scope.py
View File

@ -28,7 +28,7 @@ class TestAddressScopeCreate(
super(TestAddressScopeCreate, self).setUp()
self.new_address_scope = (
test_address_scope.TestCreateAddressScope.new_address_scope)
self.network.create_address_scope = mock.Mock(
self.network_client.create_address_scope = mock.Mock(
return_value=self.new_address_scope)
self.cmd = address_scope.CreateAddressScope(self.app, self.namespace)
@ -47,7 +47,7 @@ class TestAddressScopeCreate(
self.cmd, arglist, verifylist, create_ext)
columns, data = self.cmd.take_action(parsed_args)
self.network.create_address_scope.assert_called_once_with(**{
self.network_client.create_address_scope.assert_called_once_with(**{
'ip_version': self.new_address_scope.ip_version,
'name': self.new_address_scope.name,
})
@ -67,7 +67,7 @@ class TestAddressScopeCreate(
self.cmd, arglist, verifylist, create_ext)
columns, data = self.cmd.take_action(parsed_args)
self.network.create_address_scope.assert_called_once_with(**{
self.network_client.create_address_scope.assert_called_once_with(**{
'ip_version': self.new_address_scope.ip_version,
'apic:distinguished_names': {"VRF": "test1"},
'name': self.new_address_scope.name,

514
gbpclient/tests/unit/test_cli20.py
View File

@ -11,63 +11,246 @@
# under the License.
#
import contextlib
from io import StringIO
import itertools
import sys
from unittest import mock
import urllib.parse as urlparse
import fixtures
from oslo_utils import encodeutils
from oslotest import base
import requests
from gbpclient import gbpshell as shell
from gbpclient.v2_0 import client
from neutronclient.common import constants
from neutronclient.common import exceptions
from neutronclient.tests.unit import test_cli20 as neutron_test_cli20
from neutronclient.tests.unit import test_http
from gbpclient import gbpshell
from gbpclient.v2_0 import client as gbpclient
API_VERSION = "2.0"
TOKEN = test_http.AUTH_TOKEN
ENDURL = test_http.END_URL
REQUEST_ID = 'test_request_id'
from six.moves import StringIO
API_VERSION = neutron_test_cli20.API_VERSION
TOKEN = neutron_test_cli20.TOKEN
ENDURL = neutron_test_cli20.ENDURL
capture_std_streams = neutron_test_cli20.capture_std_streams
end_url = neutron_test_cli20.end_url
@contextlib.contextmanager
def capture_std_streams():
fake_stdout, fake_stderr = StringIO(), StringIO()
stdout, stderr = sys.stdout, sys.stderr
try:
sys.stdout, sys.stderr = fake_stdout, fake_stderr
yield fake_stdout, fake_stderr
finally:
sys.stdout, sys.stderr = stdout, stderr
class ParserException(Exception):
pass
class FakeStdout(neutron_test_cli20.FakeStdout):
class FakeStdout(object):
pass
def __init__(self):
self.content = []
def write(self, text):
self.content.append(text)
def make_string(self):
result = ''
for line in self.content:
result += encodeutils.safe_decode(line, 'utf-8')
return result
class MyResp(neutron_test_cli20.MyResp):
pass
class MyRequest(requests.Request):
def __init__(self, method=None):
self.method = method
class MyApp(neutron_test_cli20.MyApp):
pass
class MyResp(requests.Response):
def __init__(self, status_code, headers=None, reason=None,
request=None, url=None):
self.status_code = status_code
self.headers = headers or {}
self.reason = reason
self.request = request or MyRequest()
self.url = url
class MyUrlComparator(neutron_test_cli20.MyUrlComparator):
pass
class MyApp(object):
def __init__(self, _stdout):
self.stdout = _stdout
class MyComparator(neutron_test_cli20.MyComparator):
pass
def end_url(path, query=None):
_url_str = ENDURL + "/v" + API_VERSION + path
return query and _url_str + "?" + query or _url_str
class CLITestV20Base(neutron_test_cli20.CLITestV20Base):
class MyUrlComparator(object):
def __init__(self, lhs, client):
self.lhs = lhs
self.client = client
shell = gbpshell
client = gbpclient
def __eq__(self, rhs):
lhsp = urlparse.urlparse(self.lhs)
rhsp = urlparse.urlparse(rhs)
lhs_qs = urlparse.parse_qsl(lhsp.query)
rhs_qs = urlparse.parse_qsl(rhsp.query)
return (lhsp.scheme == rhsp.scheme and
lhsp.netloc == rhsp.netloc and
lhsp.path == rhsp.path and
len(lhs_qs) == len(rhs_qs) and
set(lhs_qs) == set(rhs_qs))
def __str__(self):
return self.lhs
def __repr__(self):
return str(self)
class MyComparator(object):
def __init__(self, lhs, client):
self.lhs = lhs
self.client = client
def _com_dict(self, lhs, rhs):
if len(lhs) != len(rhs):
return False
for key, value in lhs.items():
if key not in rhs:
return False
rhs_value = rhs[key]
if not self._com(value, rhs_value):
return False
return True
def _com_list(self, lhs, rhs):
if len(lhs) != len(rhs):
return False
for lhs_value in lhs:
if lhs_value not in rhs:
return False
return True
def _com(self, lhs, rhs):
if lhs is None:
return rhs is None
if isinstance(lhs, dict):
if not isinstance(rhs, dict):
return False
return self._com_dict(lhs, rhs)
if isinstance(lhs, list):
if not isinstance(rhs, list):
return False
return self._com_list(lhs, rhs)
if isinstance(lhs, tuple):
if not isinstance(rhs, tuple):
return False
return self._com_list(lhs, rhs)
return lhs == rhs
def __eq__(self, rhs):
if self.client:
rhs = self.client.deserialize(rhs, 200)
return self._com(self.lhs, rhs)
def __repr__(self):
if self.client:
return self.client.serialize(self.lhs)
return str(self.lhs)
class ContainsKeyValue(object):
"""Checks whether key/value pair(s) are included in a dict parameter.
This class just checks whether specifid key/value pairs passed in
__init__() are included in a dict parameter. The comparison does not
fail even if other key/value pair(s) exists in a target dict.
"""
def __init__(self, expected):
self._expected = expected
def __eq__(self, other):
if not isinstance(other, dict):
return False
for key, value in self._expected.items():
if key not in other:
return False
if other[key] != value:
return False
return True
def __repr__(self):
return ('<%s (expected: %s)>' %
(self.__class__.__name__, self._expected))
class IsA(object):
"""Checks whether the parameter is of specific type."""
def __init__(self, expected_type):
self._expected_type = expected_type
def __eq__(self, other):
return isinstance(other, self._expected_type)
def __repr__(self):
return ('<%s (expected: %s)>' %
(self.__class__.__name__, self._expected_type))
class CLITestV20Base(base.BaseTestCase):
test_id = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
id_field = 'id'
non_admin_status_resources = []
def _find_resourceid(self, client, resource, name_or_id,
cmd_resource=None, parent_id=None):
return name_or_id
def setUp(self, plurals=None):
"""Prepare the test environment."""
super(CLITestV20Base, self).setUp()
self.client = gbpclient.Client(token=TOKEN, endpoint_url=self.endurl)
client.Client.EXTED_PLURALS.update(constants.PLURALS)
if plurals is not None:
client.Client.EXTED_PLURALS.update(plurals)
self.metadata = {'plurals': client.Client.EXTED_PLURALS}
self.endurl = ENDURL
self.fake_stdout = FakeStdout()
self.addCleanup(mock.patch.stopall)
mock.patch('sys.stdout', new=self.fake_stdout).start()
mock.patch('neutronclient.neutron.v2_0.find_resourceid_by_name_or_id',
new=self._find_resourceid).start()
mock.patch('neutronclient.neutron.v2_0.find_resourceid_by_id',
new=self._find_resourceid).start()
self.client = client.Client(token=TOKEN, endpoint_url=self.endurl)
def register_non_admin_status_resource(self, resource_name):
# TODO(amotoki):
# It is recommended to define
# "non_admin_status_resources in each test class rather than
# using register_non_admin_status_resource method.
# If we change self.non_admin_status_resources like this,
# we need to ensure this should be an instance variable
# to avoid changing the class variable.
if (id(self.non_admin_status_resources) ==
id(self.__class__.non_admin_status_resources)):
self.non_admin_status_resources = (self.__class__.
non_admin_status_resources[:])
self.non_admin_status_resources.append(resource_name)
def _test_create_resource(self, resource, cmd, name, myid, args,
position_names, position_values,
@ -106,7 +289,7 @@ class CLITestV20Base(neutron_test_cli20.CLITestV20Base):
) as mock_get_client, mock.patch.object(
self.client.httpclient, "request", return_value=resp
) as mock_request:
gbpshell.run_command(cmd, cmd_parser, args)
shell.run_command(cmd, cmd_parser, args)
self.assert_mock_multiple_calls_with_same_arguments(
mock_get_client, mock.call(), None)
@ -114,7 +297,7 @@ class CLITestV20Base(neutron_test_cli20.CLITestV20Base):
mock_request.assert_called_once_with(
end_url(path), 'POST',
body=mock_body,
headers=neutron_test_cli20.ContainsKeyValue(
headers=ContainsKeyValue(
{'X-Auth-Token': TOKEN}))
_str = self.fake_stdout.make_string()
@ -122,6 +305,275 @@ class CLITestV20Base(neutron_test_cli20.CLITestV20Base):
if name:
self.assertIn(name, _str)
def _test_list_resources(self, resources, cmd, detail=False, tags=(),
fields_1=(), fields_2=(), page_size=None,
sort_key=(), sort_dir=(), response_contents=None,
base_args=None, path=None, cmd_resources=None,
parent_id=None, output_format=None, query=""):
if not cmd_resources:
cmd_resources = resources
if response_contents is None:
contents = [{self.id_field: 'myid1', },
{self.id_field: 'myid2', }, ]
else:
contents = response_contents
reses = {resources: contents}
resstr = self.client.serialize(reses)
# url method body
args = base_args if base_args is not None else []
if detail:
args.append('-D')
if fields_1:
for field in fields_1:
args.append('--fields')
args.append(field)
if tags:
args.append('--')
args.append("--tag")
for tag in tags:
args.append(tag)
tag_query = urlparse.urlencode(
{'tag': encodeutils.safe_encode(tag)})
if query:
query += "&" + tag_query
else:
query = tag_query
if (not tags) and fields_2:
args.append('--')
if fields_2:
args.append("--fields")
for field in fields_2:
args.append(field)
if detail:
query = query and query + '&verbose=True' or 'verbose=True'
for field in itertools.chain(fields_1, fields_2):
if query:
query += "&fields=" + field
else:
query = "fields=" + field
if page_size:
args.append("--page-size")
args.append(str(page_size))
if query:
query += "&limit=%s" % page_size
else:
query = "limit=%s" % page_size
if sort_key:
for key in sort_key:
args.append('--sort-key')
args.append(key)
if query:
query += '&'
query += 'sort_key=%s' % key
if sort_dir:
len_diff = len(sort_key) - len(sort_dir)
if len_diff > 0:
sort_dir = tuple(sort_dir) + ('asc',) * len_diff
elif len_diff < 0:
sort_dir = sort_dir[:len(sort_key)]
for dir in sort_dir:
args.append('--sort-dir')
args.append(dir)
if query:
query += '&'
query += 'sort_dir=%s' % dir
if path is None:
path = getattr(self.client, cmd_resources + "_path")
if parent_id:
path = path % parent_id
if output_format:
args.append('-f')
args.append(output_format)
cmd_parser = cmd.get_parser("list_" + cmd_resources)
resp = (MyResp(200), resstr)
with mock.patch.object(cmd, "get_client",
return_value=self.client) as mock_get_client, \
mock.patch.object(self.client.httpclient, "request",
return_value=resp) as mock_request:
shell.run_command(cmd, cmd_parser, args)
self.assert_mock_multiple_calls_with_same_arguments(
mock_get_client, mock.call(), None)
mock_request.assert_called_once_with(
MyUrlComparator(end_url(path, query), self.client),
'GET',
body=None,
headers=ContainsKeyValue({'X-Auth-Token': TOKEN}))
_str = self.fake_stdout.make_string()
if response_contents is None:
self.assertIn('myid1', _str)
return _str
def _test_list_resources_with_pagination(self, resources, cmd,
base_args=None,
cmd_resources=None,
parent_id=None, query=""):
if not cmd_resources:
cmd_resources = resources
path = getattr(self.client, cmd_resources + "_path")
if parent_id:
path = path % parent_id
fake_query = "marker=myid2&limit=2"
reses1 = {resources: [{'id': 'myid1', },
{'id': 'myid2', }],
'%s_links' % resources: [{'href': end_url(path, fake_query),
'rel': 'next'}]}
reses2 = {resources: [{'id': 'myid3', },
{'id': 'myid4', }]}
resstr1 = self.client.serialize(reses1)
resstr2 = self.client.serialize(reses2)
cmd_parser = cmd.get_parser("list_" + cmd_resources)
args = base_args if base_args is not None else []
mock_request_calls = [
mock.call(
end_url(path, query), 'GET',
body=None,
headers=ContainsKeyValue({'X-Auth-Token': TOKEN})),
mock.call(
MyUrlComparator(end_url(path, fake_query),
self.client), 'GET',
body=None,
headers=ContainsKeyValue({'X-Auth-Token': TOKEN}))]
mock_request_resp = [(MyResp(200), resstr1), (MyResp(200), resstr2)]
with mock.patch.object(cmd, "get_client",
return_value=self.client) as mock_get_client, \
mock.patch.object(self.client.httpclient,
"request") as mock_request:
mock_request.side_effect = mock_request_resp
shell.run_command(cmd, cmd_parser, args)
self.assert_mock_multiple_calls_with_same_arguments(
mock_get_client, mock.call(), None)
self.assertEqual(2, mock_request.call_count)
mock_request.assert_has_calls(mock_request_calls)
def _test_update_resource(self, resource, cmd, myid, args, extrafields,
cmd_resource=None, parent_id=None):
if not cmd_resource:
cmd_resource = resource
body = {resource: extrafields}
path = getattr(self.client, cmd_resource + "_path")
if parent_id:
path = path % (parent_id, myid)
else:
path = path % myid
mock_body = MyComparator(body, self.client)
cmd_parser = cmd.get_parser("update_" + cmd_resource)
resp = (MyResp(204), None)
with mock.patch.object(cmd, "get_client",
return_value=self.client) as mock_get_client, \
mock.patch.object(self.client.httpclient, "request",
return_value=resp) as mock_request:
shell.run_command(cmd, cmd_parser, args)
self.assert_mock_multiple_calls_with_same_arguments(
mock_get_client, mock.call(), None)
mock_request.assert_called_once_with(
MyUrlComparator(end_url(path), self.client),
'PUT',
body=mock_body,
headers=ContainsKeyValue({'X-Auth-Token': TOKEN}))
_str = self.fake_stdout.make_string()
self.assertIn(myid, _str)
def _test_show_resource(self, resource, cmd, myid, args, fields=(),
cmd_resource=None, parent_id=None):
if not cmd_resource:
cmd_resource = resource
query = "&".join(["fields=%s" % field for field in fields])
expected_res = {resource:
{self.id_field: myid,
'name': 'myname', }, }
resstr = self.client.serialize(expected_res)
path = getattr(self.client, cmd_resource + "_path")
if parent_id:
path = path % (parent_id, myid)
else:
path = path % myid
cmd_parser = cmd.get_parser("show_" + cmd_resource)
resp = (MyResp(200), resstr)
with mock.patch.object(cmd, "get_client",
return_value=self.client) as mock_get_client, \
mock.patch.object(self.client.httpclient, "request",
return_value=resp) as mock_request:
shell.run_command(cmd, cmd_parser, args)
self.assert_mock_multiple_calls_with_same_arguments(
mock_get_client, mock.call(), None)
mock_request.assert_called_once_with(
end_url(path, query), 'GET',
body=None,
headers=ContainsKeyValue({'X-Auth-Token': TOKEN}))
_str = self.fake_stdout.make_string()
self.assertIn(myid, _str)
self.assertIn('myname', _str)
def _test_set_path_and_delete(self, path, parent_id, myid,
mock_request_calls, mock_request_returns,
delete_fail=False):
return_val = 404 if delete_fail else 204
if parent_id:
path = path % (parent_id, myid)
else:
path = path % (myid)
mock_request_returns.append((MyResp(return_val), None))
mock_request_calls.append(mock.call(
end_url(path), 'DELETE',
body=None,
headers=ContainsKeyValue({'X-Auth-Token': TOKEN})))
def _test_delete_resource(self, resource, cmd, myid, args,
cmd_resource=None, parent_id=None,
extra_id=None, delete_fail=False):
mock_request_calls = []
mock_request_returns = []
if not cmd_resource:
cmd_resource = resource
path = getattr(self.client, cmd_resource + "_path")
self._test_set_path_and_delete(path, parent_id, myid,
mock_request_calls,
mock_request_returns)
# extra_id is used to test for bulk_delete
if extra_id:
self._test_set_path_and_delete(path, parent_id, extra_id,
mock_request_calls,
mock_request_returns,
delete_fail)
cmd_parser = cmd.get_parser("delete_" + cmd_resource)
with mock.patch.object(cmd, "get_client",
return_value=self.client) as mock_get_client, \
mock.patch.object(self.client.httpclient,
"request") as mock_request:
mock_request.side_effect = mock_request_returns
shell.run_command(cmd, cmd_parser, args)
self.assert_mock_multiple_calls_with_same_arguments(
mock_get_client, mock.call(), None)
mock_request.assert_has_calls(mock_request_calls)
_str = self.fake_stdout.make_string()
self.assertIn(myid, _str)
if extra_id:
self.assertIn(extra_id, _str)
def assert_mock_multiple_calls_with_same_arguments(
self, mocked_method, expected_call, count):
if count is None:
self.assertLessEqual(1, mocked_method.call_count)
else:
self.assertEqual(count, mocked_method.call_count)
mocked_method.assert_has_calls(
[expected_call] * mocked_method.call_count)
def check_parser_ext(self, cmd, args, verify_args, ext):
cmd_parser = self.cmd.get_parser('check_parser')
cmd_parser = ext.get_parser(cmd_parser)
@ -157,7 +609,7 @@ class CLITestV20ExceptionHandler(CLITestV20Base):
'detail': error_detail}}
e = self.assertRaises(expected_exception,
gbpclient.exception_handler_v20,
client.exception_handler_v20,
status_code, error_content)
self.assertEqual(status_code, e.status_code)
@ -251,7 +703,7 @@ class CLITestV20ExceptionHandler(CLITestV20Base):
mock_request.assert_called_once_with(
end_url('/test'), 'GET',
body=None,
headers=neutron_test_cli20.ContainsKeyValue(
headers=ContainsKeyValue(
{'X-Auth-Token': 'token'}))
# NB: ConnectionFailed has no explicit status_code, so this
# tests that there is a fallback defined.

78
gbpclient/tests/unit/test_cli20_purge.py
View File

@ -10,11 +10,13 @@
# License for the specific language governing permissions and limitations
# under the License.
#
import sys
from neutronclient.tests.unit import test_cli20_purge
from gbpclient.tests.unit import test_cli20
from neutronclient.neutron.v2_0 import purge
class CLITestV20Purge(test_cli20_purge.CLITestV20Purge):
class CLITestV20Purge(test_cli20.CLITestV20Base):
def setUp(self):
super(CLITestV20Purge, self).setUp()
self.resource_types = ['policy_target', 'policy_target_group',
@ -24,3 +26,75 @@ class CLITestV20Purge(test_cli20_purge.CLITestV20Purge):
'policy_classifier', 'policy_action',
'network_service_policy',
'application_policy_group']
def _generate_resources_dict(self, value=0):
resources_dict = {}
resources_dict['true'] = value
for resource_type in self.resource_types:
resources_dict[resource_type] = value
return resources_dict
def _verify_suffix(self, resources, message):
for resource, value in resources.items():
if value > 0:
suffix = list('%(value)d %(resource)s' %
{'value': value, 'resource': resource})
if value != 1:
suffix.append('s')
suffix = ''.join(suffix)
self.assertIn(suffix, message)
else:
self.assertNotIn(resource, message)
def _verify_message(self, message, deleted, failed):
message = message.split('.')
success_prefix = "Deleted "
failure_prefix = "The following resources could not be deleted: "
if not deleted['true']:
for msg in message:
self.assertNotIn(success_prefix, msg)
message = message[0]
if not failed['true']:
expected = 'Tenant has no supported resources'
self.assertEqual(expected, message)
else:
self.assertIn(failure_prefix, message)
self._verify_suffix(failed, message)
else:
resources_deleted = message[0]
self.assertIn(success_prefix, resources_deleted)
self._verify_suffix(deleted, resources_deleted)
if failed['true']:
resources_failed = message[1]
self.assertIn(failure_prefix, resources_failed)
self._verify_suffix(failed, resources_failed)
else:
for msg in message:
self.assertNotIn(failure_prefix, msg)
def _verify_result(self, my_purge, deleted, failed):
message = my_purge._build_message(deleted, failed, failed['true'])
self._verify_message(message, deleted, failed)
def test_build_message(self):
my_purge = purge.Purge(test_cli20.MyApp(sys.stdout), None)
# Verify message when tenant has no supported resources
deleted = self._generate_resources_dict()
failed = self._generate_resources_dict()
self._verify_result(my_purge, deleted, failed)
# Verify message when tenant has supported resources,
# and they are all deleteable
deleted = self._generate_resources_dict(1)
self._verify_result(my_purge, deleted, failed)
# Verify message when tenant has supported resources,
# and some are not deleteable
failed = self._generate_resources_dict(1)
self._verify_result(my_purge, deleted, failed)
# Verify message when tenant has supported resources,
# and all are not deleteable
deleted = self._generate_resources_dict()
self._verify_result(my_purge, deleted, failed)

21
gbpclient/tests/unit/test_network.py
View File

@ -27,7 +27,7 @@ class TestNetworkCreate(test_network.TestNetwork, test_cli20.CLITestV20Base):
def setUp(self):
super(TestNetworkCreate, self).setUp()
self.network.create_network = mock.Mock(
self.network_client.create_network = mock.Mock(
return_value=self._network)
self.cmd = network.CreateNetwork(self.app, self.namespace)
@ -62,7 +62,7 @@ class TestNetworkCreate(test_network.TestNetwork, test_cli20.CLITestV20Base):
self.cmd, arglist, verifylist, create_ext)
columns, data = self.cmd.take_action(parsed_args)
self.network.create_network.assert_called_once_with(**{
self.network_client.create_network.assert_called_once_with(**{
'admin_state_up': True,
'name': self._network.name,
'router:external': True,
@ -118,7 +118,7 @@ class TestNetworkCreate(test_network.TestNetwork, test_cli20.CLITestV20Base):
self.cmd, arglist, verifylist, create_ext)
columns, data = self.cmd.take_action(parsed_args)
self.network.create_network.assert_called_once_with(**{
self.network_client.create_network.assert_called_once_with(**{
'admin_state_up': True,
'name': self._network.name,
'router:external': True,
@ -159,7 +159,7 @@ class TestNetworkCreate(test_network.TestNetwork, test_cli20.CLITestV20Base):
self.cmd, arglist, verifylist, create_ext)
columns, data = self.cmd.take_action(parsed_args)
self.network.create_network.assert_called_once_with(**{
self.network_client.create_network.assert_called_once_with(**{
'admin_state_up': True,
'name': self._network.name,
'apic:extra_consumed_contracts': [],
@ -188,7 +188,7 @@ class TestNetworkCreate(test_network.TestNetwork, test_cli20.CLITestV20Base):
self.cmd, arglist, verifylist, create_ext)
columns, data = self.cmd.take_action(parsed_args)
self.network.create_network.assert_called_once_with(**{
self.network_client.create_network.assert_called_once_with(**{
'admin_state_up': True,
'name': self._network.name,
'router:external': True,
@ -207,8 +207,9 @@ class TestNetworkSet(test_network.TestNetwork, test_cli20.CLITestV20Base):
def setUp(self):
super(TestNetworkSet, self).setUp()
self.network.update_network = mock.Mock(return_value=None)
self.network.find_network = mock.Mock(return_value=self._network)
self.network_client.update_network = mock.Mock(return_value=None)
self.network_client.find_network = mock.Mock(
return_value=self._network)
self.cmd = network.SetNetwork(self.app, self.namespace)
def test_set_no_options(self):
@ -237,7 +238,7 @@ class TestNetworkSet(test_network.TestNetwork, test_cli20.CLITestV20Base):
self.cmd, arglist, verifylist, set_ext)
result = self.cmd.take_action(parsed_args)
self.assertFalse(self.network.update_network.called)
self.assertFalse(self.network_client.update_network.called)
self.assertIsNone(result)
def test_set_all_valid_options(self):
@ -301,7 +302,7 @@ class TestNetworkSet(test_network.TestNetwork, test_cli20.CLITestV20Base):
'apic:no_nat_cidrs': ['10.10.10.0/24'],
}
self.network.update_network.assert_called_once_with(
self.network_client.update_network.assert_called_once_with(
self._network, **attrs)
self.assertIsNone(result)
@ -342,6 +343,6 @@ class TestNetworkSet(test_network.TestNetwork, test_cli20.CLITestV20Base):
'apic:no_nat_cidrs': [],
}
self.network.update_network.assert_called_once_with(
self.network_client.update_network.assert_called_once_with(
self._network, **attrs)
self.assertIsNone(result)

18
gbpclient/tests/unit/test_port.py
View File

@ -26,7 +26,7 @@ class TestPortCreate(test_port.TestPort, test_cli20.CLITestV20Base):
_port = test_port.TestCreatePort._port
extension_details = (
network_fakes.FakeExtension.create_one_extension()
network_fakes.create_one_extension()
)
def setUp(self):
@ -36,8 +36,8 @@ class TestPortCreate(test_port.TestPort, test_cli20.CLITestV20Base):
fake_net = network_fakes.create_one_network({
'id': self._port.network_id,
})
self.network.find_network = mock.Mock(return_value=fake_net)
self.network.create_port = mock.Mock(
self.network_client.find_network = mock.Mock(return_value=fake_net)
self.network_client.create_port = mock.Mock(
return_value=self._port)
self.cmd = port.CreatePort(self.app, self.namespace)
@ -56,7 +56,7 @@ class TestPortCreate(test_port.TestPort, test_cli20.CLITestV20Base):
self.cmd, arglist, verifylist, create_ext)
columns, data = self.cmd.take_action(parsed_args)
self.network.create_port.assert_called_once_with(**{
self.network_client.create_port.assert_called_once_with(**{
'admin_state_up': True,
'name': self._port.name,
'network_id': self._port.network_id,
@ -79,7 +79,7 @@ class TestPortCreate(test_port.TestPort, test_cli20.CLITestV20Base):
self.cmd, arglist, verifylist, create_ext)
columns, data = self.cmd.take_action(parsed_args)
self.network.create_port.assert_called_once_with(**{
self.network_client.create_port.assert_called_once_with(**{
'admin_state_up': True,
'name': self._port.name,
'apic:erspan_config': [{"dest_ip": "10.0.0.0",
@ -97,8 +97,8 @@ class TestPortSet(test_port.TestPort, test_cli20.CLITestV20Base):
def setUp(self):
super(TestPortSet, self).setUp()
self.network.update_port = mock.Mock(return_value=None)
self.network.find_port = mock.Mock(return_value=self._port)
self.network_client.update_port = mock.Mock(return_value=None)
self.network_client.find_port = mock.Mock(return_value=self._port)
self.cmd = port.SetPort(self.app, self.namespace)
def test_set_no_options(self):
@ -115,7 +115,7 @@ class TestPortSet(test_port.TestPort, test_cli20.CLITestV20Base):
self.cmd, arglist, verifylist, set_ext)
result = self.cmd.take_action(parsed_args)
self.assertFalse(self.network.update_port.called)
self.assertFalse(self.network_client.update_port.called)
self.assertIsNone(result)
def test_set_all_valid_options(self):
@ -140,6 +140,6 @@ class TestPortSet(test_port.TestPort, test_cli20.CLITestV20Base):
"direction": "in"}],
}
self.network.update_port.assert_called_once_with(
self.network_client.update_port.assert_called_once_with(
self._port, **attrs)
self.assertIsNone(result)

27
gbpclient/tests/unit/test_router.py
View File

@ -26,7 +26,8 @@ class TestRouterCreate(test_router.TestRouter, test_cli20.CLITestV20Base):
def setUp(self):
super(TestRouterCreate, self).setUp()
self.new_router = test_router.TestCreateRouter.new_router
self.network.create_router = mock.Mock(return_value=self.new_router)
self.network_client.create_router = mock.Mock(
return_value=self.new_router)
self.cmd = router.CreateRouter(self.app, self.namespace)
def test_create_default_options(self):
@ -43,7 +44,7 @@ class TestRouterCreate(test_router.TestRouter, test_cli20.CLITestV20Base):
self.cmd, arglist, verifylist, create_ext)
columns, data = self.cmd.take_action(parsed_args)
self.network.create_router.assert_called_once_with(**{
self.network_client.create_router.assert_called_once_with(**{
'admin_state_up': True,
'name': self.new_router.name,
})
@ -64,7 +65,7 @@ class TestRouterCreate(test_router.TestRouter, test_cli20.CLITestV20Base):
self.cmd, arglist, verifylist, create_ext)
columns, data = self.cmd.take_action(parsed_args)
self.network.create_router.assert_called_once_with(**{
self.network_client.create_router.assert_called_once_with(**{
'admin_state_up': True,
'name': self.new_router.name,
'apic:external_provided_contracts': ['ptest1'],
@ -82,12 +83,14 @@ class TestRouterSet(test_router.TestRouter, test_cli20.CLITestV20Base):
def setUp(self):
super(TestRouterSet, self).setUp()
self.network.router_add_gateway = mock.Mock()
self.network.update_router = mock.Mock(return_value=None)
self.network.set_tags = mock.Mock(return_value=None)
self.network.find_router = mock.Mock(return_value=self._router)
self.network.find_network = mock.Mock(return_value=self._network)
self.network.find_subnet = mock.Mock(return_value=self._subnet)
self.network_client.router_add_gateway = mock.Mock()
self.network_client.update_router = mock.Mock(return_value=None)
self.network_client.set_tags = mock.Mock(return_value=None)
self.network_client.find_router = mock.Mock(
return_value=self._router)
self.network_client.find_network = mock.Mock(
return_value=self._network)
self.network_client.find_subnet = mock.Mock(return_value=self._subnet)
self.cmd = router.SetRouter(self.app, self.namespace)
def test_set_no_options(self):
@ -102,8 +105,8 @@ class TestRouterSet(test_router.TestRouter, test_cli20.CLITestV20Base):
self.cmd, arglist, verifylist, set_ext)
result = self.cmd.take_action(parsed_args)
self.assertFalse(self.network.update_router.called)
self.assertFalse(self.network.set_tags.called)
self.assertFalse(self.network_client.update_router.called)
self.assertFalse(self.network_client.set_tags.called)
self.assertIsNone(result)
def test_set_all_valid_options(self):
@ -126,6 +129,6 @@ class TestRouterSet(test_router.TestRouter, test_cli20.CLITestV20Base):
'apic:external_provided_contracts': ['ptest1', 'ptest11'],
'apic:external_consumed_contracts': ['ctest1', 'ctest11'],
}
self.network.update_router.assert_called_once_with(
self.network_client.update_router.assert_called_once_with(
self._router, **attrs)
self.assertIsNone(result)

19
gbpclient/tests/unit/test_subnet.py
View File

@ -36,8 +36,10 @@ class TestSubnetCreate(test_subnet.TestSubnet, test_cli20.CLITestV20Base):
'id': self._subnet.network_id,
}
)
self.network.create_subnet = mock.Mock(return_value=self._subnet)
self.network.find_network = mock.Mock(return_value=self._network)
self.network_client.create_subnet = mock.Mock(
return_value=self._subnet)
self.network_client.find_network = mock.Mock(
return_value=self._network)
self.cmd = subnet.CreateSubnet(self.app, self.namespace)
def test_create_default_options(self):
@ -62,7 +64,7 @@ class TestSubnetCreate(test_subnet.TestSubnet, test_cli20.CLITestV20Base):
self.cmd, arglist, verifylist, create_ext)
columns, data = self.cmd.take_action(parsed_args)
self.network.create_subnet.assert_called_once_with(**{
self.network_client.create_subnet.assert_called_once_with(**{
'ip_version': 4,
'cidr': '10.10.10.0/24',
'name': self._subnet.name,
@ -98,7 +100,7 @@ class TestSubnetCreate(test_subnet.TestSubnet, test_cli20.CLITestV20Base):
self.cmd, arglist, verifylist, create_ext)
columns, data = self.cmd.take_action(parsed_args)
self.network.create_subnet.assert_called_once_with(**{
self.network_client.create_subnet.assert_called_once_with(**{
'ip_version': 4,
'cidr': '10.10.10.0/24',
'name': self._subnet.name,
@ -121,8 +123,8 @@ class TestSubnetSet(test_subnet.TestSubnet, test_cli20.CLITestV20Base):
def setUp(self):
super(TestSubnetSet, self).setUp()
self.network.update_subnet = mock.Mock(return_value=None)
self.network.find_subnet = mock.Mock(return_value=self._subnet)
self.network_client.update_subnet = mock.Mock(return_value=None)
self.network_client.find_subnet = mock.Mock(return_value=self._subnet)
self.cmd = subnet.SetSubnet(self.app, self.namespace)
def test_set_no_options(self):
@ -139,7 +141,7 @@ class TestSubnetSet(test_subnet.TestSubnet, test_cli20.CLITestV20Base):
self.cmd, arglist, verifylist, set_ext)
result = self.cmd.take_action(parsed_args)
self.assertFalse(self.network.update_subnet.called)
self.assertFalse(self.network_client.update_subnet.called)
self.assertIsNone(result)
def test_set_all_valid_options(self):
@ -171,5 +173,6 @@ class TestSubnetSet(test_subnet.TestSubnet, test_cli20.CLITestV20Base):
'apic:shared_between_vrfs': True,
'apic:router_gw_ip_pool': True
}
self.network.update_subnet.assert_called_with(self._subnet, **attrs)
self.network_client.update_subnet.assert_called_with(
self._subnet, **attrs)
self.assertIsNone(result)

2
test-requirements.txt
View File

@ -16,3 +16,5 @@ stestr>=2.0.0 # Apache-2.0
testrepository>=0.0.18 # Apache-2.0/BSD
testtools>=2.2.0 # MIT
testscenarios>=0.4 # Apache-2.0/BSD
requests_mock
osprofiler

4
tox.ini
View File

@ -1,5 +1,5 @@
[tox]
envlist = py36,py37,py27,py38,pep8
envlist = py38,pep8,py39,py310
minversion = 2.3.2
skipsdist = True
ignore_basepython_conflict = True
@ -13,7 +13,7 @@ setenv = VIRTUAL_ENV={envdir}
usedevelop = True
install_command = pip install {opts} {packages}
deps =
-c{env:TOX_CONSTRAINTS_FILE:https://releases.openstack.org/constraints/upper/2023.1}
-c{env:TOX_CONSTRAINTS_FILE:https://releases.openstack.org/constraints/upper/2023.2}
-r{toxinidir}/requirements.txt
-r{toxinidir}/test-requirements.txt
commands = stestr run {posargs}