| # Licensed to the Apache Software Foundation (ASF) under one or more |
| # contributor license agreements. See the NOTICE file distributed with |
| # this work for additional information regarding copyright ownership. |
| # The ASF licenses this file to You 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 sys |
| import unittest |
| |
| from libcloud.compute.base import Node |
| from libcloud.compute.types import NodeState |
| from libcloud.loadbalancer.base import Member, Algorithm |
| from libcloud.loadbalancer.drivers.slb import ( |
| SLBDriver, |
| SLBLoadBalancerHttpListener, |
| SLBLoadBalancerHttpsListener, |
| SLBLoadBalancerTcpListener, |
| SLBLoadBalancerUdpListener, |
| ) |
| from libcloud.loadbalancer.types import State |
| from libcloud.test.file_fixtures import LoadBalancerFileFixtures |
| from libcloud.test import MockHttp |
| from libcloud.test.secrets import LB_SLB_PARAMS |
| from libcloud.utils.py3 import httplib |
| |
| |
| class SLBDriverTestCases(unittest.TestCase): |
| region = LB_SLB_PARAMS[2] |
| |
| def setUp(self): |
| SLBMockHttp.test = self |
| SLBDriver.connectionCls.conn_class = SLBMockHttp |
| SLBMockHttp.type = None |
| SLBMockHttp.use_param = "Action" |
| |
| self.driver = SLBDriver(*LB_SLB_PARAMS) |
| |
| def test_list_protocols(self): |
| protocols = self.driver.list_protocols() |
| self.assertEqual(4, len(protocols)) |
| expected = ["tcp", "udp", "http", "https"] |
| diff = set(expected) - set(protocols) |
| self.assertEqual(0, len(diff)) |
| |
| def test_list_balancers(self): |
| balancers = self.driver.list_balancers() |
| |
| self.assertEqual(len(balancers), 1) |
| balancer = balancers[0] |
| self.assertEqual("15229f88562-cn-hangzhou-dg-a01", balancer.id) |
| self.assertEqual("abc", balancer.name) |
| self.assertEqual(State.RUNNING, balancer.state) |
| self.assertEqual("120.27.186.149", balancer.ip) |
| self.assertTrue(balancer.port is None) |
| self.assertEqual(self.driver, balancer.driver) |
| expected_extra = { |
| "create_timestamp": 1452403099000, |
| "address_type": "internet", |
| "region_id": "cn-hangzhou-dg-a01", |
| "region_id_alias": "cn-hangzhou", |
| "create_time": "2016-01-10T13:18Z", |
| "master_zone_id": "cn-hangzhou-d", |
| "slave_zone_id": "cn-hangzhou-b", |
| "network_type": "classic", |
| } |
| self._validate_extras(expected_extra, balancer.extra) |
| |
| def _validate_extras(self, expected, actual): |
| self.assertTrue(actual is not None) |
| for key, value in iter(expected.items()): |
| self.assertTrue(key in actual) |
| self.assertEqual( |
| value, |
| actual[key], |
| ( |
| "extra %(key)s not equal, " |
| 'expected: "%(expected)s", ' |
| 'actual: "%(actual)s"' |
| % {"key": key, "expected": value, "actual": actual[key]} |
| ), |
| ) |
| |
| def test_list_balancers_with_ids(self): |
| SLBMockHttp.type = "list_balancers_ids" |
| self.balancer_ids = ["id1", "id2"] |
| balancers = self.driver.list_balancers(ex_balancer_ids=self.balancer_ids) |
| self.assertTrue(balancers is not None) |
| |
| def test_list_balancers_with_ex_filters(self): |
| SLBMockHttp.type = "list_balancers_filters" |
| self.ex_filters = {"AddressType": "internet"} |
| balancers = self.driver.list_balancers(ex_filters=self.ex_filters) |
| self.assertTrue(balancers is not None) |
| |
| def test_get_balancer(self): |
| SLBMockHttp.type = "get_balancer" |
| balancer = self.driver.get_balancer(balancer_id="tests") |
| |
| self.assertEqual(balancer.id, "15229f88562-cn-hangzhou-dg-a01") |
| self.assertEqual(balancer.name, "abc") |
| self.assertEqual(balancer.state, State.RUNNING) |
| |
| def test_destroy_balancer(self): |
| balancer = self.driver.get_balancer(balancer_id="tests") |
| |
| self.assertTrue(self.driver.destroy_balancer(balancer)) |
| |
| def test_create_balancer(self): |
| self.name = "balancer1" |
| self.port = 80 |
| self.protocol = "http" |
| self.algorithm = Algorithm.WEIGHTED_ROUND_ROBIN |
| self.extra = { |
| "ex_address_type": "internet", |
| "ex_internet_charge_type": "paybytraffic", |
| "ex_bandwidth": 1, |
| "ex_master_zone_id": "cn-hangzhou-d", |
| "ex_slave_zone_id": "cn-hangzhou-b", |
| "StickySession": "on", |
| "HealthCheck": "on", |
| } |
| self.members = [Member("node1", None, None)] |
| |
| balancer = self.driver.create_balancer( |
| name=self.name, |
| port=self.port, |
| protocol=self.protocol, |
| algorithm=self.algorithm, |
| members=self.members, |
| **self.extra, |
| ) |
| |
| self.assertEqual(balancer.name, self.name) |
| self.assertEqual(balancer.port, self.port) |
| self.assertEqual(balancer.state, State.UNKNOWN) |
| |
| def test_create_balancer_no_port_exception(self): |
| self.assertRaises( |
| AttributeError, |
| self.driver.create_balancer, |
| None, |
| None, |
| "http", |
| Algorithm.WEIGHTED_ROUND_ROBIN, |
| None, |
| ) |
| |
| def test_create_balancer_unsupport_protocol_exception(self): |
| self.assertRaises( |
| AttributeError, |
| self.driver.create_balancer, |
| None, |
| 443, |
| "ssl", |
| Algorithm.WEIGHTED_ROUND_ROBIN, |
| None, |
| ) |
| |
| def test_create_balancer_multiple_member_ports_exception(self): |
| members = [Member("m1", "1.2.3.4", 80), Member("m2", "1.2.3.5", 81)] |
| self.assertRaises( |
| AttributeError, |
| self.driver.create_balancer, |
| None, |
| 80, |
| "http", |
| Algorithm.WEIGHTED_ROUND_ROBIN, |
| members, |
| ) |
| |
| def test_create_balancer_bandwidth_value_error(self): |
| self.assertRaises( |
| AttributeError, |
| self.driver.create_balancer, |
| None, |
| 80, |
| "http", |
| Algorithm.WEIGHTED_ROUND_ROBIN, |
| None, |
| ex_bandwidth="NAN", |
| ) |
| |
| def test_create_balancer_paybybandwidth_without_bandwidth_exception(self): |
| self.assertRaises( |
| AttributeError, |
| self.driver.create_balancer, |
| None, |
| 80, |
| "http", |
| Algorithm.WEIGHTED_ROUND_ROBIN, |
| None, |
| ex_internet_charge_type="paybybandwidth", |
| ) |
| |
| def test_balancer_list_members(self): |
| balancer = self.driver.get_balancer(balancer_id="tests") |
| members = balancer.list_members() |
| |
| self.assertEqual(len(members), 1) |
| self.assertEqual(members[0].balancer, balancer) |
| self.assertEqual("i-23tshnsdq", members[0].id) |
| |
| def test_balancer_list_listeners(self): |
| balancer = self.driver.get_balancer(balancer_id="tests") |
| listeners = self.driver.ex_list_listeners(balancer) |
| |
| self.assertEqual(1, len(listeners)) |
| listener = listeners[0] |
| self.assertEqual("80", listener.port) |
| |
| def test_balancer_detach_member(self): |
| self.balancer = self.driver.get_balancer(balancer_id="tests") |
| self.member = Member("i-23tshnsdq", None, None) |
| |
| self.assertTrue(self.balancer.detach_member(self.member)) |
| |
| def test_balancer_attach_compute_node(self): |
| SLBMockHttp.type = "attach_compute_node" |
| self.balancer = self.driver.get_balancer(balancer_id="tests") |
| self.node = Node( |
| id="node1", |
| name="node-name", |
| state=NodeState.RUNNING, |
| public_ips=["1.2.3.4"], |
| private_ips=["4.3.2.1"], |
| driver=self.driver, |
| ) |
| member = self.driver.balancer_attach_compute_node(self.balancer, self.node) |
| self.assertEqual(self.node.id, member.id) |
| self.assertEqual(self.node.public_ips[0], member.ip) |
| self.assertEqual(self.balancer.port, member.port) |
| |
| def test_ex_create_listener(self): |
| SLBMockHttp.type = "create_listener" |
| self.balancer = self.driver.get_balancer(balancer_id="tests") |
| self.backend_port = 80 |
| self.protocol = "http" |
| self.algorithm = Algorithm.WEIGHTED_ROUND_ROBIN |
| self.bandwidth = 1 |
| self.extra = {"StickySession": "off", "HealthCheck": "off"} |
| self.assertTrue( |
| self.driver.ex_create_listener( |
| self.balancer, |
| self.backend_port, |
| self.protocol, |
| self.algorithm, |
| self.bandwidth, |
| **self.extra, |
| ) |
| ) |
| |
| def test_ex_create_listener_override_port(self): |
| SLBMockHttp.type = "create_listener_override_port" |
| self.balancer = self.driver.get_balancer(balancer_id="tests") |
| self.backend_port = 80 |
| self.protocol = "http" |
| self.algorithm = Algorithm.WEIGHTED_ROUND_ROBIN |
| self.bandwidth = 1 |
| self.extra = { |
| "StickySession": "off", |
| "HealthCheck": "off", |
| "ListenerPort": 8080, |
| } |
| self.assertTrue( |
| self.driver.ex_create_listener( |
| self.balancer, |
| self.backend_port, |
| self.protocol, |
| self.algorithm, |
| self.bandwidth, |
| **self.extra, |
| ) |
| ) |
| |
| def test_ex_start_listener(self): |
| SLBMockHttp.type = "start_listener" |
| balancer = self.driver.get_balancer(balancer_id="tests") |
| self.port = 80 |
| self.assertTrue(self.driver.ex_start_listener(balancer, self.port)) |
| |
| def test_ex_stop_listener(self): |
| SLBMockHttp.type = "stop_listener" |
| balancer = self.driver.get_balancer(balancer_id="tests") |
| self.port = 80 |
| self.assertTrue(self.driver.ex_stop_listener(balancer, self.port)) |
| |
| def test_ex_upload_certificate(self): |
| self.name = "cert1" |
| self.cert = "cert-data" |
| self.key = "key-data" |
| certificate = self.driver.ex_upload_certificate(self.name, self.cert, self.key) |
| self.assertEqual(self.name, certificate.name) |
| self.assertEqual("01:DF:AB:CD", certificate.fingerprint) |
| |
| def test_ex_list_certificates(self): |
| certs = self.driver.ex_list_certificates() |
| self.assertEqual(2, len(certs)) |
| cert = certs[0] |
| self.assertEqual("139a00604ad-cn-east-hangzhou-01", cert.id) |
| self.assertEqual("abe", cert.name) |
| self.assertEqual("A:B:E", cert.fingerprint) |
| |
| def test_ex_list_certificates_ids(self): |
| SLBMockHttp.type = "list_certificates_ids" |
| self.cert_ids = ["cert1", "cert2"] |
| certs = self.driver.ex_list_certificates(certificate_ids=self.cert_ids) |
| self.assertTrue(certs is not None) |
| |
| def test_ex_delete_certificate(self): |
| self.cert_id = "cert1" |
| self.assertTrue(self.driver.ex_delete_certificate(self.cert_id)) |
| |
| def test_ex_set_certificate_name(self): |
| self.cert_id = "cert1" |
| self.cert_name = "cert-name" |
| self.assertTrue( |
| self.driver.ex_set_certificate_name(self.cert_id, self.cert_name) |
| ) |
| |
| |
| class SLBMockHttp(MockHttp, unittest.TestCase): |
| fixtures = LoadBalancerFileFixtures("slb") |
| |
| def _DescribeLoadBalancers(self, method, url, body, headers): |
| body = self.fixtures.load("describe_load_balancers.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _list_balancers_ids_DescribeLoadBalancers(self, method, url, body, headers): |
| params = {"LoadBalancerId": ",".join(self.test.balancer_ids)} |
| self.assertUrlContainsQueryParams(url, params) |
| body = self.fixtures.load("describe_load_balancers.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _list_balancers_filters_DescribeLoadBalancers(self, method, url, body, headers): |
| params = {"AddressType": "internet"} |
| self.assertUrlContainsQueryParams(url, params) |
| body = self.fixtures.load("describe_load_balancers.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _get_balancer_DescribeLoadBalancers(self, method, url, body, headers): |
| params = {"LoadBalancerId": "tests"} |
| self.assertUrlContainsQueryParams(url, params) |
| return self._DescribeLoadBalancers(method, url, body, headers) |
| |
| def _DeleteLoadBalancer(self, method, url, body, headers): |
| params = {"LoadBalancerId": "15229f88562-cn-hangzhou-dg-a01"} |
| self.assertUrlContainsQueryParams(url, params) |
| body = self.fixtures.load("delete_load_balancer.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeLoadBalancerAttribute(self, method, url, body, headers): |
| body = self.fixtures.load("describe_load_balancer_attribute.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _CreateLoadBalancer(self, method, url, body, headers): |
| params = {"RegionId": self.test.region, "LoadBalancerName": self.test.name} |
| balancer_keys = { |
| "AddressType": "ex_address_type", |
| "InternetChargeType": "ex_internet_charge_type", |
| "Bandwidth": "ex_bandwidth", |
| "MasterZoneId": "ex_master_zone_id", |
| "SlaveZoneId": "ex_slave_zone_id", |
| } |
| for key in balancer_keys: |
| params[key] = str(self.test.extra[balancer_keys[key]]) |
| |
| self.assertUrlContainsQueryParams(url, params) |
| body = self.fixtures.load("create_load_balancer.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _AddBackendServers(self, method, url, body, headers): |
| _id = self.test.members[0].id |
| self.assertTrue("ServerId" in url and _id in url) |
| self.assertTrue("Weight" in url and "100" in url) |
| body = self.fixtures.load("add_backend_servers.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _CreateLoadBalancerHTTPListener(self, method, url, body, headers): |
| body = self.fixtures.load("create_load_balancer_http_listener.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _StartLoadBalancerListener(self, method, url, body, headers): |
| body = self.fixtures.load("start_load_balancer_listener.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _RemoveBackendServers(self, method, url, body, headers): |
| _id = self.test.member.id |
| servers_json = '["%s"]' % _id |
| params = { |
| "LoadBalancerId": self.test.balancer.id, |
| "BackendServers": servers_json, |
| } |
| self.assertUrlContainsQueryParams(url, params) |
| body = self.fixtures.load("add_backend_servers.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _attach_compute_node_DescribeLoadBalancers(self, method, url, body, headers): |
| body = self.fixtures.load("describe_load_balancers.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _attach_compute_node_AddBackendServers(self, method, url, body, headers): |
| _id = self.test.node.id |
| self.assertTrue("ServerId" in url and _id in url) |
| self.assertTrue("Weight" in url and "100" in url) |
| body = self.fixtures.load("add_backend_servers.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _create_listener_CreateLoadBalancerHTTPListener( |
| self, method, url, body, headers |
| ): |
| params = { |
| "LoadBalancerId": self.test.balancer.id, |
| "RegionId": self.test.region, |
| "ListenerPort": str(self.test.balancer.port), |
| "BackendServerPort": str(self.test.backend_port), |
| "Scheduler": "wrr", |
| "Bandwidth": "1", |
| "StickySession": "off", |
| "HealthCheck": "off", |
| } |
| self.assertUrlContainsQueryParams(url, params) |
| body = self.fixtures.load("create_load_balancer_http_listener.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _create_listener_DescribeLoadBalancers(self, method, url, body, headers): |
| body = self.fixtures.load("describe_load_balancers.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _create_listener_override_port_CreateLoadBalancerHTTPListener( |
| self, method, url, body, headers |
| ): |
| params = { |
| "LoadBalancerId": self.test.balancer.id, |
| "RegionId": self.test.region, |
| "ListenerPort": str(self.test.extra["ListenerPort"]), |
| "BackendServerPort": str(self.test.backend_port), |
| "Scheduler": "wrr", |
| "Bandwidth": "1", |
| "StickySession": "off", |
| "HealthCheck": "off", |
| } |
| self.assertUrlContainsQueryParams(url, params) |
| body = self.fixtures.load("create_load_balancer_http_listener.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _create_listener_override_port_DescribeLoadBalancers( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("describe_load_balancers.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _start_listener_DescribeLoadBalancers(self, method, url, body, headers): |
| return self._DescribeLoadBalancers(method, url, body, headers) |
| |
| def _start_listener_StartLoadBalancerListener(self, method, url, body, headers): |
| params = {"ListenerPort": str(self.test.port)} |
| self.assertUrlContainsQueryParams(url, params) |
| return self._StartLoadBalancerListener(method, url, body, headers) |
| |
| def _stop_listener_DescribeLoadBalancers(self, method, url, body, headers): |
| return self._DescribeLoadBalancers(method, url, body, headers) |
| |
| def _stop_listener_StopLoadBalancerListener(self, method, url, body, headers): |
| params = {"ListenerPort": str(self.test.port)} |
| self.assertUrlContainsQueryParams(url, params) |
| return self._StartLoadBalancerListener(method, url, body, headers) |
| |
| def _UploadServerCertificate(self, method, url, body, headers): |
| body = self.fixtures.load("upload_server_certificate.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DescribeServerCertificates(self, method, url, body, headers): |
| body = self.fixtures.load("describe_server_certificates.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _list_certificates_ids_DescribeServerCertificates( |
| self, method, url, body, headers |
| ): |
| params = { |
| "RegionId": self.test.region, |
| "ServerCertificateId": ",".join(self.test.cert_ids), |
| } |
| self.assertUrlContainsQueryParams(url, params) |
| body = self.fixtures.load("describe_server_certificates.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _DeleteServerCertificate(self, method, url, body, headers): |
| params = { |
| "RegionId": self.test.region, |
| "ServerCertificateId": self.test.cert_id, |
| } |
| self.assertUrlContainsQueryParams(url, params) |
| body = self.fixtures.load("delete_server_certificate.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _SetServerCertificateName(self, method, url, body, headers): |
| params = { |
| "RegionId": self.test.region, |
| "ServerCertificateId": self.test.cert_id, |
| "ServerCertificateName": self.test.cert_name, |
| } |
| self.assertUrlContainsQueryParams(url, params) |
| body = self.fixtures.load("set_server_certificate_name.xml") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| |
| class AssertDictMixin(object): |
| def assert_dict_equals(self, expected, actual): |
| expected_keys = set(expected.keys()) |
| actual_keys = set(actual.keys()) |
| self.assertEqual(len(expected_keys), len(actual_keys)) |
| self.assertEqual(0, len(expected_keys - actual_keys)) |
| for key in expected: |
| self.assertEqual(expected[key], actual[key]) |
| |
| |
| class SLBLoadBalancerHttpListenerTestCase(unittest.TestCase, AssertDictMixin): |
| def setUp(self): |
| self.listener = SLBLoadBalancerHttpListener.create( |
| 80, |
| 8080, |
| Algorithm.WEIGHTED_ROUND_ROBIN, |
| 1, |
| extra={ |
| "StickySession": "on", |
| "StickySessionType": "insert", |
| "HealthCheck": "on", |
| }, |
| ) |
| |
| def test_get_required_params(self): |
| expected = { |
| "Action": "CreateLoadBalancerHTTPListener", |
| "ListenerPort": 80, |
| "BackendServerPort": 8080, |
| "Scheduler": "wrr", |
| "Bandwidth": 1, |
| "StickySession": "on", |
| "HealthCheck": "on", |
| } |
| self.assert_dict_equals(expected, self.listener.get_required_params()) |
| |
| def test_get_optional_params(self): |
| expected = {"StickySessionType": "insert"} |
| self.assert_dict_equals(expected, self.listener.get_optional_params()) |
| |
| def test_repr(self): |
| self.assertTrue("SLBLoadBalancerHttpListener" in str(self.listener)) |
| |
| |
| class SLBLoadBalancerHttpsListenerTestCase(unittest.TestCase, AssertDictMixin): |
| def setUp(self): |
| self.listener = SLBLoadBalancerHttpsListener.create( |
| 80, |
| 8080, |
| Algorithm.WEIGHTED_ROUND_ROBIN, |
| 1, |
| extra={ |
| "StickySession": "on", |
| "StickySessionType": "insert", |
| "HealthCheck": "on", |
| "ServerCertificateId": "fake-cert1", |
| }, |
| ) |
| |
| def test_get_required_params(self): |
| expected = { |
| "Action": "CreateLoadBalancerHTTPSListener", |
| "ListenerPort": 80, |
| "BackendServerPort": 8080, |
| "Scheduler": "wrr", |
| "Bandwidth": 1, |
| "StickySession": "on", |
| "HealthCheck": "on", |
| "ServerCertificateId": "fake-cert1", |
| } |
| self.assert_dict_equals(expected, self.listener.get_required_params()) |
| |
| def test_get_optional_params(self): |
| expected = {"StickySessionType": "insert"} |
| self.assert_dict_equals(expected, self.listener.get_optional_params()) |
| |
| |
| class SLBLoadBalancerTcpListenerTestCase(unittest.TestCase, AssertDictMixin): |
| def setUp(self): |
| self.listener = SLBLoadBalancerTcpListener.create( |
| 80, |
| 8080, |
| Algorithm.WEIGHTED_ROUND_ROBIN, |
| 1, |
| extra={"PersistenceTimeout": 0, "HealthCheckDomain": ""}, |
| ) |
| |
| def test_get_required_params(self): |
| expected = { |
| "Action": "CreateLoadBalancerTCPListener", |
| "ListenerPort": 80, |
| "BackendServerPort": 8080, |
| "Scheduler": "wrr", |
| "Bandwidth": 1, |
| } |
| self.assert_dict_equals(expected, self.listener.get_required_params()) |
| |
| def test_get_optional_params(self): |
| expected = {"PersistenceTimeout": 0, "HealthCheckDomain": ""} |
| self.assert_dict_equals(expected, self.listener.get_optional_params()) |
| |
| |
| class SLBLoadBalancerUdpListenerTestCase(unittest.TestCase, AssertDictMixin): |
| def setUp(self): |
| self.listener = SLBLoadBalancerUdpListener.create( |
| 80, 8080, Algorithm.WEIGHTED_ROUND_ROBIN, 1, extra={"PersistenceTimeout": 0} |
| ) |
| |
| def test_get_required_params(self): |
| expected = { |
| "Action": "CreateLoadBalancerUDPListener", |
| "ListenerPort": 80, |
| "BackendServerPort": 8080, |
| "Scheduler": "wrr", |
| "Bandwidth": 1, |
| } |
| self.assert_dict_equals(expected, self.listener.get_required_params()) |
| |
| def test_get_optional_params(self): |
| expected = {"PersistenceTimeout": 0} |
| self.assert_dict_equals(expected, self.listener.get_optional_params()) |
| |
| |
| if __name__ == "__main__": |
| sys.exit(unittest.main()) |
