x/vmware-nsxlib
Code Issues Proposed changes
vmware-nsxlib/vmware_nsxlib/tests/unit/v3/policy/test_lb_resources.py
Erica Liu c9fabb82ee Support getting realtime LBS usage in policy 
In current implementation, the LBS get_usage function
returns the number of current realized resources. This could
potentially break the l4 autoscaling logic. Thus we need to
get the realtime usage of LBS, which includes the resources in
progress

Change-Id: I5454a572cd751d2dd2a51135122d902031a0eaf3
2019-05-22 13:54:11 -07:00

1312 lines
53 KiB
Python
Raw Blame History

# Copyright 2017 VMware, Inc.
# All Rights Reserved
#
# 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 mock
from vmware_nsxlib.tests.unit.v3.policy import test_resources
from vmware_nsxlib.v3.policy import lb_defs
TEST_TENANT = 'test'
class TestPolicyLBClientSSLProfileApi(test_resources.NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyLBClientSSLProfileApi, self).setUp()
self.resourceApi = self.policy_lib.load_balancer.client_ssl_profile
def test_create_with_id(self):
name = 'd1'
description = 'desc'
obj_id = '111'
protocols = ['TLS_V1_1']
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = self.resourceApi.create_or_overwrite(
name,
client_ssl_profile_id=obj_id,
description=description,
protocols=protocols,
tenant=TEST_TENANT)
expected_def = lb_defs.LBClientSslProfileDef(
client_ssl_profile_id=obj_id,
name=name,
description=description,
protocols=protocols,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(obj_id, result)
def test_create_without_id(self):
name = 'd1'
description = 'desc'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = self.resourceApi.create_or_overwrite(
name, description=description,
tenant=TEST_TENANT)
expected_def = lb_defs.LBClientSslProfileDef(
client_ssl_profile_id=mock.ANY,
name=name,
description=description,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertIsNotNone(result)
def test_delete(self):
obj_id = '111'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(obj_id, tenant=TEST_TENANT)
expected_def = lb_defs.LBClientSslProfileDef(
client_ssl_profile_id=obj_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get(self):
obj_id = '111'
with mock.patch.object(self.policy_api, "get",
return_value={'id': obj_id}) as api_call:
result = self.resourceApi.get(obj_id, tenant=TEST_TENANT)
expected_def = lb_defs.LBClientSslProfileDef(
client_ssl_profile_id=obj_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(obj_id, result['id'])
def test_get_by_name(self):
name = 'd1'
with mock.patch.object(
self.policy_api, "list",
return_value={'results': [{'display_name': name}]}) as api_call:
obj = self.resourceApi.get_by_name(name, tenant=TEST_TENANT)
self.assertIsNotNone(obj)
expected_def = lb_defs.LBClientSslProfileDef(
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_list(self):
with mock.patch.object(self.policy_api, "list",
return_value={'results': []}) as api_call:
result = self.resourceApi.list(tenant=TEST_TENANT)
expected_def = lb_defs.LBClientSslProfileDef(
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual([], result)
def test_update(self):
obj_id = '111'
name = 'new name'
description = 'new desc'
with self.mock_get(obj_id, name), \
self.mock_create_update() as update_call:
self.resourceApi.update(obj_id,
name=name,
description=description,
tenant=TEST_TENANT)
expected_def = lb_defs.LBClientSslProfileDef(
client_ssl_profile_id=obj_id,
name=name,
description=description,
tenant=TEST_TENANT)
self.assert_called_with_def(update_call, expected_def)
class TestPolicyLBCookiePersistenceProfile(
test_resources.NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyLBCookiePersistenceProfile, self).setUp()
self.resourceApi = (
self.policy_lib.load_balancer.lb_cookie_persistence_profile)
def test_create_with_id(self):
name = 'd1'
description = 'desc'
obj_id = '111'
cookie_garble = 'test_garble'
cookie_name = 'test_name'
cookie_mode = 'INSERT'
cookie_path = 'path'
cookie_time = 'time'
persistence_shared = False
with self.mock_create_update() as api_call:
result = self.resourceApi.create_or_overwrite(
name,
persistence_profile_id=obj_id,
description=description,
cookie_name=cookie_name,
cookie_garble=cookie_garble,
cookie_mode=cookie_mode,
cookie_path=cookie_path,
cookie_time=cookie_time,
persistence_shared=persistence_shared,
tenant=TEST_TENANT)
expected_def = (
lb_defs.LBCookiePersistenceProfileDef(
persistence_profile_id=obj_id,
name=name,
description=description,
cookie_name=cookie_name,
cookie_garble=cookie_garble,
cookie_mode=cookie_mode,
cookie_path=cookie_path,
cookie_time=cookie_time,
persistence_shared=persistence_shared,
tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(obj_id, result)
def test_create_without_id(self):
name = 'd1'
description = 'desc'
with self.mock_create_update() as api_call:
result = self.resourceApi.create_or_overwrite(
name, description=description,
tenant=TEST_TENANT)
expected_def = (
lb_defs.LBCookiePersistenceProfileDef(
persistence_profile_id=mock.ANY,
name=name,
description=description,
tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
self.assertIsNotNone(result)
def test_delete(self):
obj_id = '111'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(obj_id, tenant=TEST_TENANT)
expected_def = (
lb_defs.LBCookiePersistenceProfileDef(
persistence_profile_id=obj_id,
tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
def test_get(self):
obj_id = '111'
with mock.patch.object(self.policy_api, "get",
return_value={'id': obj_id}) as api_call:
result = self.resourceApi.get(obj_id, tenant=TEST_TENANT)
expected_def = (
lb_defs.LBCookiePersistenceProfileDef(
persistence_profile_id=obj_id,
tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(obj_id, result['id'])
def test_get_by_name(self):
name = 'd1'
with mock.patch.object(
self.policy_api, "list",
return_value={'results': [{'display_name': name}]}) as api_call:
obj = self.resourceApi.get_by_name(name, tenant=TEST_TENANT)
self.assertIsNotNone(obj)
expected_def = (
lb_defs.LBCookiePersistenceProfileDef(
tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
def test_list(self):
with mock.patch.object(self.policy_api, "list",
return_value={'results': []}) as api_call:
result = self.resourceApi.list(tenant=TEST_TENANT)
expected_def = (
lb_defs.LBCookiePersistenceProfileDef(
tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
self.assertEqual([], result)
def test_update(self):
obj_id = '111'
name = 'new name'
description = 'new desc'
cookie_garble = 'test_garble'
cookie_name = 'test_name'
cookie_mode = 'INSERT'
cookie_path = 'path'
cookie_time = 'time'
persistence_shared = False
with self.mock_get(obj_id, name), \
self.mock_create_update() as update_call:
self.resourceApi.update(obj_id,
name=name,
description=description,
cookie_name=cookie_name,
cookie_garble=cookie_garble,
cookie_mode=cookie_mode,
cookie_path=cookie_path,
cookie_time=cookie_time,
persistence_shared=persistence_shared,
tenant=TEST_TENANT)
expected_def = (
lb_defs.LBCookiePersistenceProfileDef(
persistence_profile_id=obj_id,
name=name,
description=description,
cookie_name=cookie_name,
cookie_garble=cookie_garble,
cookie_mode=cookie_mode,
cookie_path=cookie_path,
cookie_time=cookie_time,
persistence_shared=persistence_shared,
tenant=TEST_TENANT))
self.assert_called_with_def(update_call, expected_def)
class TestPolicyLBSourceIpProfileApi(test_resources.NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyLBSourceIpProfileApi, self).setUp()
self.resourceApi = (
self.policy_lib.load_balancer.lb_source_ip_persistence_profile)
def test_create_with_id(self):
name = 'd1'
description = 'desc'
obj_id = '111'
ha = 'ha'
persistence_shared = True
purge = 'purge'
timeout = 100
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = self.resourceApi.create_or_overwrite(
name,
persistence_profile_id=obj_id,
description=description,
ha_persistence_mirroring_enabled=ha,
persistence_shared=persistence_shared,
purge=purge,
timeout=timeout,
tenant=TEST_TENANT)
expected_def = (
lb_defs.LBSourceIpPersistenceProfileDef(
persistence_profile_id=obj_id,
name=name,
description=description,
ha_persistence_mirroring_enabled=ha,
persistence_shared=persistence_shared,
purge=purge,
timeout=timeout,
tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(obj_id, result)
def test_create_without_id(self):
name = 'd1'
description = 'desc'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = self.resourceApi.create_or_overwrite(
name, description=description,
tenant=TEST_TENANT)
expected_def = (
lb_defs.LBSourceIpPersistenceProfileDef(
persistence_profile_id=mock.ANY,
name=name,
description=description,
tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
self.assertIsNotNone(result)
def test_delete(self):
obj_id = '111'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(obj_id, tenant=TEST_TENANT)
expected_def = (
lb_defs.LBSourceIpPersistenceProfileDef(
persistence_profile_id=obj_id,
tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
def test_get(self):
obj_id = '111'
with mock.patch.object(self.policy_api, "get",
return_value={'id': obj_id}) as api_call:
result = self.resourceApi.get(obj_id, tenant=TEST_TENANT)
expected_def = (
lb_defs.LBSourceIpPersistenceProfileDef(
persistence_profile_id=obj_id,
tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(obj_id, result['id'])
def test_get_by_name(self):
name = 'd1'
with mock.patch.object(
self.policy_api, "list",
return_value={'results': [{'display_name': name}]}) as api_call:
obj = self.resourceApi.get_by_name(name, tenant=TEST_TENANT)
self.assertIsNotNone(obj)
expected_def = (
lb_defs.LBSourceIpPersistenceProfileDef(
tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
def test_list(self):
with mock.patch.object(self.policy_api, "list",
return_value={'results': []}) as api_call:
result = self.resourceApi.list(tenant=TEST_TENANT)
expected_def = (
lb_defs.LBSourceIpPersistenceProfileDef(
tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
self.assertEqual([], result)
def test_update(self):
obj_id = '111'
name = 'new name'
description = 'new desc'
ha = False
persistence_shared = False
purge = 'no purge'
timeout = 101
with self.mock_get(obj_id, name), \
self.mock_create_update() as update_call:
self.resourceApi.update(obj_id,
name=name,
description=description,
ha_persistence_mirroring_enabled=ha,
persistence_shared=persistence_shared,
purge=purge,
timeout=timeout,
tenant=TEST_TENANT)
expected_def = (
lb_defs.LBSourceIpPersistenceProfileDef(
persistence_profile_id=obj_id,
name=name,
description=description,
ha_persistence_mirroring_enabled=ha,
persistence_shared=persistence_shared,
purge=purge,
timeout=timeout,
tenant=TEST_TENANT))
self.assert_called_with_def(update_call, expected_def)
class TestPolicyLBApplicationProfile(test_resources.NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyLBApplicationProfile, self).setUp()
self.resourceApi = self.policy_lib.load_balancer.lb_http_profile
def test_create_with_id(self):
name = 'd1'
description = 'desc'
obj_id = '111'
http_redirect_to_https = False
http_redirect_to = "sample-url"
idle_timeout = 100
ntlm = False
request_body_size = 1025
request_header_size = 10
response_header_size = 10
response_timeout = 10
x_forwarded_for = 'INSERT'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = self.resourceApi.create_or_overwrite(
name,
lb_app_profile_id=obj_id,
description=description,
http_redirect_to_https=http_redirect_to_https,
http_redirect_to=http_redirect_to,
idle_timeout=idle_timeout,
ntlm=ntlm,
request_body_size=request_body_size,
request_header_size=request_header_size,
response_header_size=response_header_size,
response_timeout=response_timeout,
x_forwarded_for=x_forwarded_for,
tenant=TEST_TENANT)
expected_def = (
lb_defs.LBHttpProfileDef(
lb_app_profile_id=obj_id,
name=name,
description=description,
http_redirect_to_https=http_redirect_to_https,
http_redirect_to=http_redirect_to,
idle_timeout=idle_timeout,
ntlm=ntlm,
request_body_size=request_body_size,
request_header_size=request_header_size,
response_header_size=response_header_size,
response_timeout=response_timeout,
x_forwarded_for=x_forwarded_for,
tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(obj_id, result)
def test_create_without_id(self):
name = 'd1'
description = 'desc'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = self.resourceApi.create_or_overwrite(
name, description=description,
tenant=TEST_TENANT)
expected_def = (
lb_defs.LBHttpProfileDef(
lb_app_profile_id=mock.ANY,
name=name,
description=description,
tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
self.assertIsNotNone(result)
def test_delete(self):
obj_id = '111'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(obj_id, tenant=TEST_TENANT)
expected_def = (
lb_defs.LBHttpProfileDef(
lb_app_profile_id=obj_id,
tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
def test_get(self):
obj_id = '111'
with mock.patch.object(self.policy_api, "get",
return_value={'id': obj_id}) as api_call:
result = self.resourceApi.get(obj_id, tenant=TEST_TENANT)
expected_def = (
lb_defs.LBHttpProfileDef(
lb_app_profile_id=obj_id,
tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(obj_id, result['id'])
def test_get_by_name(self):
name = 'd1'
with mock.patch.object(
self.policy_api, "list",
return_value={'results': [{'display_name': name}]}) as api_call:
obj = self.resourceApi.get_by_name(name, tenant=TEST_TENANT)
self.assertIsNotNone(obj)
expected_def = (
lb_defs.LBHttpProfileDef(tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
def test_list(self):
with mock.patch.object(self.policy_api, "list",
return_value={'results': []}) as api_call:
result = self.resourceApi.list(tenant=TEST_TENANT)
expected_def = (
lb_defs.LBHttpProfileDef(tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
self.assertEqual([], result)
def test_update(self):
obj_id = '111'
name = 'new name'
description = 'new desc'
with self.mock_get(obj_id, name), \
self.mock_create_update() as update_call:
self.resourceApi.update(obj_id,
name=name,
description=description,
tenant=TEST_TENANT)
expected_def = (
lb_defs.LBHttpProfileDef(
lb_app_profile_id=obj_id,
name=name,
description=description,
tenant=TEST_TENANT))
self.assert_called_with_def(update_call, expected_def)
class TestPolicyLBService(test_resources.NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyLBService, self).setUp()
self.resourceApi = self.policy_lib.load_balancer.lb_service
def test_create_with_id(self):
name = 'd1'
description = 'desc'
obj_id = '111'
size = 'SMALL'
connectivity_path = 'path'
with self.mock_create_update() as api_call:
result = self.resourceApi.create_or_overwrite(
name,
lb_service_id=obj_id,
description=description,
size=size,
connectivity_path=connectivity_path,
tenant=TEST_TENANT)
expected_def = (
lb_defs.LBServiceDef(
lb_service_id=obj_id,
name=name,
description=description,
size=size,
connectivity_path=connectivity_path,
tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(obj_id, result)
def test_create_without_id(self):
name = 'd1'
description = 'desc'
with self.mock_create_update() as api_call:
result = self.resourceApi.create_or_overwrite(
name, description=description,
tenant=TEST_TENANT)
expected_def = (
lb_defs.LBServiceDef(lb_service_id=mock.ANY,
name=name,
description=description,
tenant=TEST_TENANT))
self.assert_called_with_def(api_call, expected_def)
self.assertIsNotNone(result)
def test_delete(self):
obj_id = '111'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(obj_id, tenant=TEST_TENANT)
expected_def = lb_defs.LBServiceDef(
lb_service_id=obj_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get(self):
obj_id = '111'
with mock.patch.object(self.policy_api, "get",
return_value={'id': obj_id}) as api_call:
result = self.resourceApi.get(obj_id, tenant=TEST_TENANT)
expected_def = lb_defs.LBServiceDef(
lb_service_id=obj_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(obj_id, result['id'])
def test_get_by_name(self):
name = 'd1'
with mock.patch.object(
self.policy_api, "list",
return_value={'results': [{'display_name': name}]}) as api_call:
obj = self.resourceApi.get_by_name(name, tenant=TEST_TENANT)
self.assertIsNotNone(obj)
expected_def = lb_defs.LBServiceDef(
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_list(self):
with mock.patch.object(self.policy_api, "list",
return_value={'results': []}) as api_call:
result = self.resourceApi.list(tenant=TEST_TENANT)
expected_def = lb_defs.LBServiceDef(
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual([], result)
def test_update(self):
obj_id = '111'
name = 'new name'
description = 'new desc'
size = 'SMALL'
connectivity_path = 'path'
with self.mock_get(obj_id, name), \
self.mock_create_update() as update_call:
self.resourceApi.update(obj_id,
name=name,
description=description,
tenant=TEST_TENANT,
size=size,
connectivity_path=connectivity_path)
expected_def = lb_defs.LBServiceDef(
lb_service_id=obj_id,
name=name,
description=description,
tenant=TEST_TENANT,
size=size,
connectivity_path=connectivity_path)
self.assert_called_with_def(update_call, expected_def)
def test_get_status(self):
obj_id = '111'
with mock.patch.object(self.policy_api, "get") as api_call:
self.resourceApi.get_status(obj_id, tenant=TEST_TENANT)
expected_def = lb_defs.LBServiceStatusDef(
lb_service_id=obj_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get_statistics(self):
obj_id = '111'
with mock.patch.object(self.policy_api, "get") as api_call:
self.resourceApi.get_statistics(obj_id, tenant=TEST_TENANT)
expected_def = lb_defs.LBServiceStatisticsDef(
lb_service_id=obj_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get_virtual_server_status(self):
obj_id = '111'
vs_id = '222'
with mock.patch.object(self.policy_api, "get") as api_call:
self.resourceApi.get_virtual_server_status(
obj_id, vs_id, tenant=TEST_TENANT)
expected_def = lb_defs.LBVirtualServerStatusDef(
lb_service_id=obj_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get_usage(self):
lbs_id = 'test_vs'
with mock.patch.object(self.policy_api, "get") as api_call:
self.resourceApi.get_usage(
lbs_id, realtime=True, tenant=TEST_TENANT)
expected_def = lb_defs.LBServiceUsageDef(
lb_service_id=lbs_id,
realtime=True,
tenant=TEST_TENANT)
expected_path = '%s/lb-services/%s/service-usage?source=realtime'
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(expected_def.path_pattern, expected_path)
class TestPolicyLBVirtualServer(test_resources.NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyLBVirtualServer, self).setUp()
self.resourceApi = self.policy_lib.load_balancer.virtual_server
def test_create_with_id(self):
name = 'd1'
description = 'desc'
obj_id = '111'
waf_profile_id = 'waf'
waf_profile_path = self.policy_lib.waf_profile.get_path(
profile_id=waf_profile_id, tenant=TEST_TENANT)
waf_profile_binding = lb_defs.WAFProfileBindingDef(
waf_profile_path=waf_profile_path)
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = self.resourceApi.create_or_overwrite(
name,
virtual_server_id=obj_id,
waf_profile_binding=waf_profile_binding,
description=description,
tenant=TEST_TENANT)
expected_def = lb_defs.LBVirtualServerDef(
virtual_server_id=obj_id, name=name, description=description,
waf_profile_binding=waf_profile_binding,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(obj_id, result)
def test_create_without_id(self):
name = 'd1'
description = 'desc'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = self.resourceApi.create_or_overwrite(
name, description=description,
tenant=TEST_TENANT)
expected_def = lb_defs.LBVirtualServerDef(
virtual_server_id=mock.ANY, name=name, description=description,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertIsNotNone(result)
def test_delete(self):
obj_id = '111'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(obj_id, tenant=TEST_TENANT)
expected_def = lb_defs.LBVirtualServerDef(
virtual_server_id=obj_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get(self):
obj_id = '111'
with mock.patch.object(self.policy_api, "get",
return_value={'id': obj_id}) as api_call:
result = self.resourceApi.get(obj_id, tenant=TEST_TENANT)
expected_def = lb_defs.LBVirtualServerDef(
virtual_server_id=obj_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(obj_id, result['id'])
def test_get_by_name(self):
name = 'd1'
with mock.patch.object(
self.policy_api, "list",
return_value={'results': [{'display_name': name}]}) as api_call:
obj = self.resourceApi.get_by_name(name, tenant=TEST_TENANT)
self.assertIsNotNone(obj)
expected_def = lb_defs.LBVirtualServerDef(
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_list(self):
with mock.patch.object(self.policy_api, "list",
return_value={'results': []}) as api_call:
result = self.resourceApi.list(tenant=TEST_TENANT)
expected_def = lb_defs.LBVirtualServerDef(
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual([], result)
def test_update(self):
obj_id = '111'
name = 'new name'
description = 'new desc'
vs_name = 'name-name'
with self.mock_get(obj_id, vs_name), \
self.mock_create_update() as update_call:
self.resourceApi.update(obj_id,
name=name,
description=description,
tenant=TEST_TENANT)
expected_def = lb_defs.LBVirtualServerDef(
virtual_server_id=obj_id, name=name,
description=description,
tenant=TEST_TENANT)
self.assert_called_with_def(update_call, expected_def)
def test_add_lb_rule(self):
vs_obj_id = '111'
vs_name = 'name-name'
rule_actions = 'test1'
rule_match_conditions = 'test2'
rule_name = 'dummy_rule'
rule_match_strategy = 'test3'
rule_phase = 'test4'
with self.mock_get(vs_obj_id, vs_name), \
self.mock_create_update() as update_call:
self.resourceApi.add_lb_rule(
vs_obj_id, actions=rule_actions, name=rule_name,
match_conditions=rule_match_conditions,
match_strategy=rule_match_strategy, phase=rule_phase)
lb_rule = lb_defs.LBRuleDef(
rule_actions, rule_match_conditions, rule_name,
rule_match_strategy, rule_phase)
expected_def = lb_defs.LBVirtualServerDef(
virtual_server_id=vs_obj_id,
name=vs_name,
rules=[lb_rule])
self.assert_called_with_def(update_call, expected_def)
def test_add_lb_rule_first(self):
vs_obj_id = '111'
vs_name = 'name-name'
rule_actions = 'test1'
rule_match_conditions = 'test2'
rule_name = 'dummy_rule'
rule_match_strategy = 'test3'
rule_phase = 'test4'
with self.mock_get(vs_obj_id, vs_name,
rules=[{'display_name': 'xx'},
{'display_name': 'yy'}]), \
self.mock_create_update() as update_call:
self.resourceApi.add_lb_rule(
vs_obj_id, actions=rule_actions, name=rule_name,
match_conditions=rule_match_conditions,
match_strategy=rule_match_strategy, phase=rule_phase,
position=0)
lb_rule = lb_defs.LBRuleDef(
rule_actions, rule_match_conditions, rule_name,
rule_match_strategy, rule_phase)
expected_def = lb_defs.LBVirtualServerDef(
virtual_server_id=vs_obj_id,
name=vs_name,
rules=[lb_rule,
{'display_name': 'xx'},
{'display_name': 'yy'}])
self.assert_called_with_def(update_call, expected_def)
def test_add_lb_rule_last(self):
vs_obj_id = '111'
vs_name = 'name-name'
rule_actions = 'test1'
rule_match_conditions = 'test2'
rule_name = 'dummy_rule'
rule_match_strategy = 'test3'
rule_phase = 'test4'
with self.mock_get(vs_obj_id, vs_name,
rules=[{'display_name': 'xx'},
{'display_name': 'yy'}]), \
self.mock_create_update() as update_call:
self.resourceApi.add_lb_rule(
vs_obj_id, actions=rule_actions, name=rule_name,
match_conditions=rule_match_conditions,
match_strategy=rule_match_strategy, phase=rule_phase)
lb_rule = lb_defs.LBRuleDef(
rule_actions, rule_match_conditions, rule_name,
rule_match_strategy, rule_phase)
expected_def = lb_defs.LBVirtualServerDef(
virtual_server_id=vs_obj_id,
name=vs_name,
rules=[{'display_name': 'xx'},
{'display_name': 'yy'},
lb_rule])
self.assert_called_with_def(update_call, expected_def)
def test_add_lb_rule_last_over(self):
vs_obj_id = '111'
vs_name = 'name-name'
rule_actions = 'test1'
rule_match_conditions = 'test2'
rule_name = 'dummy_rule'
rule_match_strategy = 'test3'
rule_phase = 'test4'
with self.mock_get(vs_obj_id, vs_name,
rules=[{'display_name': 'xx'},
{'display_name': 'yy'}]), \
self.mock_create_update() as update_call:
self.resourceApi.add_lb_rule(
vs_obj_id, actions=rule_actions, name=rule_name,
match_conditions=rule_match_conditions,
match_strategy=rule_match_strategy, phase=rule_phase,
position=999)
lb_rule = lb_defs.LBRuleDef(
rule_actions, rule_match_conditions, rule_name,
rule_match_strategy, rule_phase)
expected_def = lb_defs.LBVirtualServerDef(
virtual_server_id=vs_obj_id, name=vs_name,
rules=[{'display_name': 'xx'},
{'display_name': 'yy'},
lb_rule])
self.assert_called_with_def(update_call, expected_def)
def test_add_lb_rule_mid(self):
vs_obj_id = '111'
vs_name = 'name-name'
rule_actions = 'test1'
rule_match_conditions = 'test2'
rule_name = 'dummy_rule'
rule_match_strategy = 'test3'
rule_phase = 'test4'
with self.mock_get(vs_obj_id, vs_name,
rules=[{'display_name': 'xx'},
{'display_name': 'yy'}]), \
self.mock_create_update() as update_call:
self.resourceApi.add_lb_rule(
vs_obj_id, actions=rule_actions, name=rule_name,
match_conditions=rule_match_conditions,
match_strategy=rule_match_strategy, phase=rule_phase,
position=1)
lb_rule = lb_defs.LBRuleDef(
rule_actions, rule_match_conditions, rule_name,
rule_match_strategy, rule_phase)
expected_def = lb_defs.LBVirtualServerDef(
name=vs_name,
virtual_server_id=vs_obj_id,
rules=[{'display_name': 'xx'},
lb_rule,
{'display_name': 'yy'}])
self.assert_called_with_def(update_call, expected_def)
def test_update_lb_rule(self):
vs_obj_id = '111'
vs_name = 'name-name'
with self.mock_get(
vs_obj_id, vs_name,
rules=[{'display_name': 'xx', 'actions': '11'},
{'display_name': 'yy'}]), \
self.mock_create_update() as update_call:
self.resourceApi.update_lb_rule(vs_obj_id, 'xx', actions='22')
expected_def = lb_defs.LBVirtualServerDef(
name=vs_name,
virtual_server_id=vs_obj_id,
rules=[{'display_name': 'xx', 'actions': '22'},
{'display_name': 'yy'}])
self.assert_called_with_def(update_call, expected_def)
def test_update_lb_rule_position(self):
vs_obj_id = '111'
vs_name = 'name-name'
with self.mock_get(
vs_obj_id, vs_name,
rules=[{'display_name': 'xx', 'actions': '11'},
{'display_name': 'yy'}]), \
self.mock_create_update() as update_call:
self.resourceApi.update_lb_rule(vs_obj_id, 'xx', actions='22',
position=1)
expected_def = lb_defs.LBVirtualServerDef(
virtual_server_id=vs_obj_id,
name=vs_name,
rules=[{'display_name': 'yy'},
{'display_name': 'xx', 'actions': '22'}])
self.assert_called_with_def(update_call, expected_def)
def test_remove_lb_rule(self):
vs_obj_id = '111'
vs_name = 'name-name'
with self.mock_get(vs_obj_id, vs_name,
rules=[{'display_name': 'xx'},
{'display_name': 'yy'}]), \
self.mock_create_update() as update_call:
self.resourceApi.remove_lb_rule(vs_obj_id, 'xx')
expected_def = lb_defs.LBVirtualServerDef(
virtual_server_id=vs_obj_id,
name=vs_name,
rules=[{'display_name': 'yy'}])
self.assert_called_with_def(update_call, expected_def)
class TestPolicyLBPoolApi(test_resources.NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyLBPoolApi, self).setUp()
self.resourceApi = self.policy_lib.load_balancer.lb_pool
def test_create_with_id(self):
name = 'd1'
description = 'desc'
obj_id = '111'
members = [
lb_defs.LBPoolMemberDef(ip_address='10.0.0.1')]
algorithm = 'algo'
active_monitor_paths = 'path1'
member_group = 'group1'
snat_translation = False
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = self.resourceApi.create_or_overwrite(
name,
lb_pool_id=obj_id,
description=description,
members=members,
active_monitor_paths=active_monitor_paths,
algorithm=algorithm,
member_group=member_group,
snat_translation=snat_translation,
tenant=TEST_TENANT)
expected_def = lb_defs.LBPoolDef(
lb_pool_id=obj_id,
name=name,
description=description,
members=members,
active_monitor_paths=active_monitor_paths,
algorithm=algorithm,
member_group=member_group,
snat_translation=snat_translation,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(obj_id, result)
def test_create_without_id(self):
name = 'd1'
description = 'desc'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = self.resourceApi.create_or_overwrite(
name, description=description,
tenant=TEST_TENANT)
expected_def = lb_defs.LBPoolDef(
lb_pool_id=mock.ANY,
name=name,
description=description,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertIsNotNone(result)
def test_delete(self):
obj_id = '111'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(obj_id, tenant=TEST_TENANT)
expected_def = lb_defs.LBPoolDef(
lb_pool_id=obj_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get(self):
obj_id = '111'
with mock.patch.object(self.policy_api, "get",
return_value={'id': obj_id}) as api_call:
result = self.resourceApi.get(obj_id, tenant=TEST_TENANT)
expected_def = lb_defs.LBPoolDef(
lb_pool_id=obj_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(obj_id, result['id'])
def test_get_by_name(self):
name = 'd1'
with mock.patch.object(
self.policy_api, "list",
return_value={'results': [{'display_name': name}]}) as api_call:
obj = self.resourceApi.get_by_name(name, tenant=TEST_TENANT)
self.assertIsNotNone(obj)
expected_def = lb_defs.LBPoolDef(
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_list(self):
with mock.patch.object(self.policy_api, "list",
return_value={'results': []}) as api_call:
result = self.resourceApi.list(tenant=TEST_TENANT)
expected_def = lb_defs.LBPoolDef(
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual([], result)
def test_update(self):
obj_id = '111'
name = 'new name'
description = 'new desc'
members = [{'ip_address': '10.0.0.1'}]
algorithm = 'algo'
active_monitor_paths = ['path1']
member_group = 'group1'
snat_translation = False
with mock.patch.object(self.policy_api, "get",
return_value={'id': obj_id}), \
mock.patch.object(self.policy_api,
"create_or_update") as update_call:
self.resourceApi.update(obj_id,
name=name,
description=description,
members=members,
active_monitor_paths=active_monitor_paths,
algorithm=algorithm,
member_group=member_group,
snat_translation=snat_translation,
tenant=TEST_TENANT)
expected_def = lb_defs.LBPoolDef(
lb_pool_id=obj_id,
name=name,
description=description,
members=members,
active_monitor_paths=active_monitor_paths,
algorithm=algorithm,
member_group=member_group,
snat_translation=snat_translation,
tenant=TEST_TENANT)
self.assert_called_with_def(update_call, expected_def)
def test_add_monitor_to_pool(self):
obj_id = '111'
active_monitor_paths = ['path1']
with mock.patch.object(self.policy_api, "get",
return_value={'id': obj_id}), \
mock.patch.object(self.policy_api,
"create_or_update") as update_call:
self.resourceApi.add_monitor_to_pool(
obj_id,
active_monitor_paths,
tenant=TEST_TENANT)
expected_def = lb_defs.LBPoolDef(
lb_pool_id=obj_id,
active_monitor_paths=active_monitor_paths,
tenant=TEST_TENANT)
self.assert_called_with_def(update_call, expected_def)
def test_remove_monitor_from_pool(self):
obj_id = '111'
removed_monitor_path = 'path1'
stay_monitor_path = 'path2'
active_monitors = [removed_monitor_path, stay_monitor_path]
with mock.patch.object(
self.policy_api, "get", return_value={
'id': obj_id, 'active_monitor_paths': active_monitors}), \
mock.patch.object(self.policy_api,
"create_or_update") as update_call:
self.resourceApi.remove_monitor_from_pool(
obj_id,
removed_monitor_path,
tenant=TEST_TENANT)
expected_def = lb_defs.LBPoolDef(
lb_pool_id=obj_id,
active_monitor_paths=[stay_monitor_path],
tenant=TEST_TENANT)
self.assert_called_with_def(update_call, expected_def)
def test_create_pool_member_and_add_to_pool(self):
obj_id = '111'
ip_address = '1.1.1.1'
with mock.patch.object(self.policy_api, "get",
return_value={'id': obj_id}), \
mock.patch.object(self.policy_api,
"create_or_update") as update_call:
self.resourceApi.create_pool_member_and_add_to_pool(
obj_id, ip_address,
tenant=TEST_TENANT)
mem_def = lb_defs.LBPoolMemberDef(ip_address)
expected_def = lb_defs.LBPoolDef(
lb_pool_id=obj_id,
members=[mem_def],
tenant=TEST_TENANT)
self.assert_called_with_def(update_call, expected_def)
def test_update_pool_member(self):
obj_id = '111'
ip_address = '1.1.1.1'
port = '80'
new_name = 'mem1'
member = {'ip_address': ip_address, 'port': port}
with mock.patch.object(self.policy_api, "get",
return_value={'id': obj_id,
'members': [member]}), \
mock.patch.object(self.policy_api,
"create_or_update") as update_call:
self.resourceApi.update_pool_member(
obj_id, ip_address, port=port, display_name=new_name,
tenant=TEST_TENANT)
member['display_name'] = new_name
expected_def = lb_defs.LBPoolDef(
lb_pool_id=obj_id,
members=[member],
tenant=TEST_TENANT)
self.assert_called_with_def(update_call, expected_def)
class TestPolicyLBMonitorProfileHttpApi(test_resources.NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyLBMonitorProfileHttpApi, self).setUp()
self.resourceApi = (
self.policy_lib.load_balancer.lb_monitor_profile_http)
self.obj_def = lb_defs.LBHttpMonitorProfileDef
def test_create_with_id(self):
name = 'd1'
obj_id = '111'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = self.resourceApi.create_or_overwrite(
lb_monitor_profile_id=obj_id,
name=name,
tenant=TEST_TENANT)
expected_def = self.obj_def(
lb_monitor_profile_id=obj_id,
name=name,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(obj_id, result)
def test_create_without_id(self):
name = 'd1'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
result = self.resourceApi.create_or_overwrite(
name=name,
tenant=TEST_TENANT)
expected_def = self.obj_def(
lb_monitor_profile_id=mock.ANY,
name=name,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertIsNotNone(result)
def test_delete(self):
obj_id = '111'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(obj_id, tenant=TEST_TENANT)
expected_def = self.obj_def(
lb_monitor_profile_id=obj_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_get(self):
obj_id = '111'
with mock.patch.object(self.policy_api, "get",
return_value={'id': obj_id}) as api_call:
result = self.resourceApi.get(obj_id, tenant=TEST_TENANT)
expected_def = self.obj_def(
lb_monitor_profile_id=obj_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual(obj_id, result['id'])
def test_get_by_name(self):
name = 'd1'
with mock.patch.object(
self.policy_api, "list",
return_value={'results': [{'display_name': name}]}) as api_call:
obj = self.resourceApi.get_by_name(name, tenant=TEST_TENANT)
self.assertIsNotNone(obj)
expected_def = self.obj_def(
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
def test_list(self):
with mock.patch.object(self.policy_api, "list",
return_value={'results': []}) as api_call:
result = self.resourceApi.list(tenant=TEST_TENANT)
expected_def = self.obj_def(
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual([], result)
def test_update(self):
obj_id = '111'
name = 'new name'
with self.mock_get(obj_id, name), \
self.mock_create_update() as update_call:
self.resourceApi.update(obj_id,
name=name,
tenant=TEST_TENANT)
expected_def = self.obj_def(
lb_monitor_profile_id=obj_id,
name=name,
tenant=TEST_TENANT)
self.assert_called_with_def(update_call, expected_def)
class TestPolicyLBMonitorProfileHttpsApi(TestPolicyLBMonitorProfileHttpApi):
def setUp(self, *args, **kwargs):
super(TestPolicyLBMonitorProfileHttpsApi, self).setUp()
self.resourceApi = (
self.policy_lib.load_balancer.lb_monitor_profile_https)
self.obj_def = lb_defs.LBHttpsMonitorProfileDef
class TestPolicyLBMonitorProfileUdpApi(TestPolicyLBMonitorProfileHttpApi):
def setUp(self, *args, **kwargs):
super(TestPolicyLBMonitorProfileUdpApi, self).setUp()
self.resourceApi = (
self.policy_lib.load_balancer.lb_monitor_profile_udp)
self.obj_def = lb_defs.LBUdpMonitorProfileDef
class TestPolicyLBMonitorProfileIcmpApi(TestPolicyLBMonitorProfileHttpApi):
def setUp(self, *args, **kwargs):
super(TestPolicyLBMonitorProfileIcmpApi, self).setUp()
self.resourceApi = (
self.policy_lib.load_balancer.lb_monitor_profile_icmp)
self.obj_def = lb_defs.LBIcmpMonitorProfileDef
class TestPolicyLBMonitorProfileTcpApi(TestPolicyLBMonitorProfileHttpApi):
def setUp(self, *args, **kwargs):
super(TestPolicyLBMonitorProfileTcpApi, self).setUp()
self.resourceApi = (
self.policy_lib.load_balancer.lb_monitor_profile_tcp)
self.obj_def = lb_defs.LBTcpMonitorProfileDef
View Git Blame Copy Permalink