festival.edmaven.com:80

Linux b747.shapiroconsult.com 3.10.0-1160.119.1.vz7.224.4 #1 SMP Mon Sep 30 15:36:27 MSK 2024 x86_64
Apache
173.231.212.80

test_proxy.py

# 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 uuid
from unittest import mock

from openstack.load_balancer.v2 import _proxy
from openstack.load_balancer.v2 import amphora
from openstack.load_balancer.v2 import availability_zone
from openstack.load_balancer.v2 import availability_zone_profile
from openstack.load_balancer.v2 import flavor
from openstack.load_balancer.v2 import flavor_profile
from openstack.load_balancer.v2 import health_monitor
from openstack.load_balancer.v2 import l7_policy
from openstack.load_balancer.v2 import l7_rule
from openstack.load_balancer.v2 import listener
from openstack.load_balancer.v2 import load_balancer as lb
from openstack.load_balancer.v2 import member
from openstack.load_balancer.v2 import pool
from openstack.load_balancer.v2 import provider
from openstack.load_balancer.v2 import quota
from openstack import proxy as proxy_base
from openstack.tests.unit import test_proxy_base

class TestLoadBalancerProxy(test_proxy_base.TestProxyBase):

LB_ID = uuid.uuid4()
    LISTENER_ID = uuid.uuid4()
    POOL_ID = uuid.uuid4()
    L7_POLICY_ID = uuid.uuid4()
    AMPHORA = 'amphora'
    AMPHORA_ID = uuid.uuid4()

def setUp(self):
        super(TestLoadBalancerProxy, self).setUp()
        self.proxy = _proxy.Proxy(self.session)

def test_load_balancers(self):
        self.verify_list(self.proxy.load_balancers,
                         lb.LoadBalancer)

def test_load_balancer_get(self):
        self.verify_get(self.proxy.get_load_balancer,
                        lb.LoadBalancer)

def test_load_balancer_stats_get(self):
        self.verify_get(self.proxy.get_load_balancer_statistics,
                        lb.LoadBalancerStats,
                        value=[self.LB_ID],
                        expected_args=[lb.LoadBalancerStats],
                        expected_kwargs={'lb_id': self.LB_ID,
                                         'requires_id': False})

def test_load_balancer_create(self):
        self.verify_create(self.proxy.create_load_balancer,
                           lb.LoadBalancer)

@mock.patch.object(proxy_base.Proxy, '_get_resource')
    def test_load_balancer_delete_non_cascade(self, mock_get_resource):
        fake_load_balancer = mock.Mock()
        fake_load_balancer.id = "load_balancer_id"
        mock_get_resource.return_value = fake_load_balancer
        self._verify2("openstack.proxy.Proxy._delete",
                      self.proxy.delete_load_balancer,
                      method_args=["resource_or_id", True,
                                   False],
                      expected_args=[lb.LoadBalancer,
                                     fake_load_balancer],
                      expected_kwargs={"ignore_missing": True})
        self.assertFalse(fake_load_balancer.cascade)
        mock_get_resource.assert_called_once_with(lb.LoadBalancer,
                                                  "resource_or_id")

@mock.patch.object(proxy_base.Proxy, '_get_resource')
    def test_load_balancer_delete_cascade(self, mock_get_resource):
        fake_load_balancer = mock.Mock()
        fake_load_balancer.id = "load_balancer_id"
        mock_get_resource.return_value = fake_load_balancer
        self._verify2("openstack.proxy.Proxy._delete",
                      self.proxy.delete_load_balancer,
                      method_args=["resource_or_id", True,
                                   True],
                      expected_args=[lb.LoadBalancer,
                                     fake_load_balancer],
                      expected_kwargs={"ignore_missing": True})
        self.assertTrue(fake_load_balancer.cascade)
        mock_get_resource.assert_called_once_with(lb.LoadBalancer,
                                                  "resource_or_id")

def test_load_balancer_find(self):
        self.verify_find(self.proxy.find_load_balancer,
                         lb.LoadBalancer)

def test_load_balancer_update(self):
        self.verify_update(self.proxy.update_load_balancer,
                           lb.LoadBalancer)

def test_load_balancer_failover(self):
        self.verify_update(self.proxy.failover_load_balancer,
                           lb.LoadBalancerFailover,
                           value=[self.LB_ID],
                           expected_args=[],
                           expected_kwargs={'lb_id': self.LB_ID})

def test_listeners(self):
        self.verify_list(self.proxy.listeners,
                         listener.Listener)

def test_listener_get(self):
        self.verify_get(self.proxy.get_listener,
                        listener.Listener)

def test_listener_stats_get(self):
        self.verify_get(self.proxy.get_listener_statistics,
                        listener.ListenerStats,
                        value=[self.LISTENER_ID],
                        expected_args=[listener.ListenerStats],
                        expected_kwargs={'listener_id': self.LISTENER_ID,
                                         'requires_id': False})

def test_listener_create(self):
        self.verify_create(self.proxy.create_listener,
                           listener.Listener)

def test_listener_delete(self):
        self.verify_delete(self.proxy.delete_listener,
                           listener.Listener, True)

def test_listener_find(self):
        self.verify_find(self.proxy.find_listener,
                         listener.Listener)

def test_listener_update(self):
        self.verify_update(self.proxy.update_listener,
                           listener.Listener)

def test_pools(self):
        self.verify_list(self.proxy.pools,
                         pool.Pool)

def test_pool_get(self):
        self.verify_get(self.proxy.get_pool,
                        pool.Pool)

def test_pool_create(self):
        self.verify_create(self.proxy.create_pool,
                           pool.Pool)

def test_pool_delete(self):
        self.verify_delete(self.proxy.delete_pool,
                           pool.Pool, True)

def test_pool_find(self):
        self.verify_find(self.proxy.find_pool,
                         pool.Pool)

def test_pool_update(self):
        self.verify_update(self.proxy.update_pool,
                           pool.Pool)

def test_members(self):
        self.verify_list(self.proxy.members,
                         member.Member,
                         method_kwargs={'pool': self.POOL_ID},
                         expected_kwargs={'pool_id': self.POOL_ID})

def test_member_get(self):
        self.verify_get(self.proxy.get_member,
                        member.Member,
                        method_kwargs={'pool': self.POOL_ID},
                        expected_kwargs={'pool_id': self.POOL_ID})

def test_member_create(self):
        self.verify_create(self.proxy.create_member,
                           member.Member,
                           method_kwargs={'pool': self.POOL_ID},
                           expected_kwargs={'pool_id': self.POOL_ID})

def test_member_delete(self):
        self.verify_delete(self.proxy.delete_member,
                           member.Member,
                           True,
                           method_kwargs={'pool': self.POOL_ID},
                           expected_kwargs={'pool_id': self.POOL_ID})

def test_member_find(self):
        self._verify2('openstack.proxy.Proxy._find',
                      self.proxy.find_member,
                      method_args=["MEMBER", self.POOL_ID],
                      expected_args=[member.Member, "MEMBER"],
                      expected_kwargs={"pool_id": self.POOL_ID,
                                       "ignore_missing": True})

def test_member_update(self):
        self._verify2('openstack.proxy.Proxy._update',
                      self.proxy.update_member,
                      method_args=["MEMBER", self.POOL_ID],
                      expected_args=[member.Member, "MEMBER"],
                      expected_kwargs={"pool_id": self.POOL_ID})

def test_health_monitors(self):
        self.verify_list(self.proxy.health_monitors,
                         health_monitor.HealthMonitor)

def test_health_monitor_get(self):
        self.verify_get(self.proxy.get_health_monitor,
                        health_monitor.HealthMonitor)

def test_health_monitor_create(self):
        self.verify_create(self.proxy.create_health_monitor,
                           health_monitor.HealthMonitor)

def test_health_monitor_delete(self):
        self.verify_delete(self.proxy.delete_health_monitor,
                           health_monitor.HealthMonitor, True)

def test_health_monitor_find(self):
        self.verify_find(self.proxy.find_health_monitor,
                         health_monitor.HealthMonitor)

def test_health_monitor_update(self):
        self.verify_update(self.proxy.update_health_monitor,
                           health_monitor.HealthMonitor)

def test_l7_policies(self):
        self.verify_list(self.proxy.l7_policies,
                         l7_policy.L7Policy)

def test_l7_policy_get(self):
        self.verify_get(self.proxy.get_l7_policy,
                        l7_policy.L7Policy)

def test_l7_policy_create(self):
        self.verify_create(self.proxy.create_l7_policy,
                           l7_policy.L7Policy)

def test_l7_policy_delete(self):
        self.verify_delete(self.proxy.delete_l7_policy,
                           l7_policy.L7Policy, True)

def test_l7_policy_find(self):
        self.verify_find(self.proxy.find_l7_policy,
                         l7_policy.L7Policy)

def test_l7_policy_update(self):
        self.verify_update(self.proxy.update_l7_policy,
                           l7_policy.L7Policy)

def test_l7_rules(self):
        self.verify_list(self.proxy.l7_rules,
                         l7_rule.L7Rule,
                         method_kwargs={'l7_policy': self.L7_POLICY_ID},
                         expected_kwargs={'l7policy_id': self.L7_POLICY_ID})

def test_l7_rule_get(self):
        self.verify_get(self.proxy.get_l7_rule,
                        l7_rule.L7Rule,
                        method_kwargs={'l7_policy': self.L7_POLICY_ID},
                        expected_kwargs={'l7policy_id': self.L7_POLICY_ID})

def test_l7_rule_create(self):
        self.verify_create(self.proxy.create_l7_rule,
                           l7_rule.L7Rule,
                           method_kwargs={'l7_policy': self.L7_POLICY_ID},
                           expected_kwargs={'l7policy_id': self.L7_POLICY_ID})

def test_l7_rule_delete(self):
        self.verify_delete(self.proxy.delete_l7_rule,
                           l7_rule.L7Rule,
                           True,
                           method_kwargs={'l7_policy': self.L7_POLICY_ID},
                           expected_kwargs={'l7policy_id': self.L7_POLICY_ID})

def test_l7_rule_find(self):
        self._verify2('openstack.proxy.Proxy._find',
                      self.proxy.find_l7_rule,
                      method_args=["RULE", self.L7_POLICY_ID],
                      expected_args=[l7_rule.L7Rule, "RULE"],
                      expected_kwargs={"l7policy_id": self.L7_POLICY_ID,
                                       "ignore_missing": True})

def test_l7_rule_update(self):
        self._verify2('openstack.proxy.Proxy._update',
                      self.proxy.update_l7_rule,
                      method_args=["RULE", self.L7_POLICY_ID],
                      expected_args=[l7_rule.L7Rule, "RULE"],
                      expected_kwargs={"l7policy_id": self.L7_POLICY_ID})

def test_quotas(self):
        self.verify_list(self.proxy.quotas, quota.Quota)

def test_quota_get(self):
        self.verify_get(self.proxy.get_quota, quota.Quota)

def test_quota_update(self):
        self.verify_update(self.proxy.update_quota, quota.Quota)

def test_quota_default_get(self):
        self._verify2("openstack.proxy.Proxy._get",
                      self.proxy.get_quota_default,
                      expected_args=[quota.QuotaDefault],
                      expected_kwargs={'requires_id': False})

def test_quota_delete(self):
        self.verify_delete(self.proxy.delete_quota, quota.Quota, False)

def test_quota_delete_ignore(self):
        self.verify_delete(self.proxy.delete_quota, quota.Quota, True)

def test_providers(self):
        self.verify_list(self.proxy.providers, provider.Provider)

def test_provider_flavor_capabilities(self):
        self.verify_list(self.proxy.provider_flavor_capabilities,
                         provider.ProviderFlavorCapabilities,
                         method_args=[self.AMPHORA],
                         expected_kwargs={'provider': self.AMPHORA})

def test_flavor_profiles(self):
        self.verify_list(self.proxy.flavor_profiles,
                         flavor_profile.FlavorProfile)

def test_flavor_profile_get(self):
        self.verify_get(self.proxy.get_flavor_profile,
                        flavor_profile.FlavorProfile)

def test_flavor_profile_create(self):
        self.verify_create(self.proxy.create_flavor_profile,
                           flavor_profile.FlavorProfile)

def test_flavor_profile_delete(self):
        self.verify_delete(self.proxy.delete_flavor_profile,
                           flavor_profile.FlavorProfile, True)

def test_flavor_profile_find(self):
        self.verify_find(self.proxy.find_flavor_profile,
                         flavor_profile.FlavorProfile)

def test_flavor_profile_update(self):
        self.verify_update(self.proxy.update_flavor_profile,
                           flavor_profile.FlavorProfile)

def test_flavors(self):
        self.verify_list(self.proxy.flavors, flavor.Flavor)

def test_flavor_get(self):
        self.verify_get(self.proxy.get_flavor, flavor.Flavor)

def test_flavor_create(self):
        self.verify_create(self.proxy.create_flavor, flavor.Flavor)

def test_flavor_delete(self):
        self.verify_delete(self.proxy.delete_flavor, flavor.Flavor, True)

def test_flavor_find(self):
        self.verify_find(self.proxy.find_flavor, flavor.Flavor)

def test_flavor_update(self):
        self.verify_update(self.proxy.update_flavor, flavor.Flavor)

def test_amphorae(self):
        self.verify_list(self.proxy.amphorae, amphora.Amphora)

def test_amphora_get(self):
        self.verify_get(self.proxy.get_amphora, amphora.Amphora)

def test_amphora_find(self):
        self.verify_find(self.proxy.find_amphora, amphora.Amphora)

def test_amphora_configure(self):
        self.verify_update(self.proxy.configure_amphora,
                           amphora.AmphoraConfig,
                           value=[self.AMPHORA_ID],
                           expected_args=[],
                           expected_kwargs={'amphora_id': self.AMPHORA_ID})

def test_amphora_failover(self):
        self.verify_update(self.proxy.failover_amphora,
                           amphora.AmphoraFailover,
                           value=[self.AMPHORA_ID],
                           expected_args=[],
                           expected_kwargs={'amphora_id': self.AMPHORA_ID})

def test_availability_zone_profiles(self):
        self.verify_list(self.proxy.availability_zone_profiles,
                         availability_zone_profile.AvailabilityZoneProfile)

def test_availability_zone_profile_get(self):
        self.verify_get(self.proxy.get_availability_zone_profile,
                        availability_zone_profile.AvailabilityZoneProfile)

def test_availability_zone_profile_create(self):
        self.verify_create(self.proxy.create_availability_zone_profile,
                           availability_zone_profile.AvailabilityZoneProfile)

def test_availability_zone_profile_delete(self):
        self.verify_delete(self.proxy.delete_availability_zone_profile,
                           availability_zone_profile.AvailabilityZoneProfile,
                           True)

def test_availability_zone_profile_find(self):
        self.verify_find(self.proxy.find_availability_zone_profile,
                         availability_zone_profile.AvailabilityZoneProfile)

def test_availability_zone_profile_update(self):
        self.verify_update(self.proxy.update_availability_zone_profile,
                           availability_zone_profile.AvailabilityZoneProfile)

def test_availability_zones(self):
        self.verify_list(self.proxy.availability_zones,
                         availability_zone.AvailabilityZone)

def test_availability_zone_get(self):
        self.verify_get(self.proxy.get_availability_zone,
                        availability_zone.AvailabilityZone)

def test_availability_zone_create(self):
        self.verify_create(self.proxy.create_availability_zone,
                           availability_zone.AvailabilityZone)

def test_availability_zone_delete(self):
        self.verify_delete(self.proxy.delete_availability_zone,
                           availability_zone.AvailabilityZone, True)

def test_availability_zone_find(self):
        self.verify_find(self.proxy.find_availability_zone,
                         availability_zone.AvailabilityZone)

def test_availability_zone_update(self):
        self.verify_update(self.proxy.update_availability_zone,
                           availability_zone.AvailabilityZone)

Name	Type	Size	Owner/Group	Permission	Last Modified	Actions
.	dir	-	root/root	0755	2025-01-07 04:51:46
..	dir	-	root/root	0755	2025-01-07 04:51:46
__pycache__	dir	-	root/root	0755	2025-01-07 04:51:46
__init__.py	application/x-empty	0 B	root/root	0644	2025-01-06 04:11:25
test_amphora.py	text/x-python	5.46 KB	root/root	0644	2025-01-06 04:11:25
test_availability_zone.py	text/x-python	2.41 KB	root/root	0644	2025-01-06 04:11:25
test_availability_zone_profile.py	text/x-python	2.28 KB	root/root	0644	2025-01-06 04:11:25
test_flavor.py	text/x-python	2.14 KB	root/root	0644	2025-01-06 04:11:25
test_flavor_profile.py	text/x-python	2.05 KB	root/root	0644	2025-01-06 04:11:25
test_health_monitor.py	text/x-python	3.98 KB	root/root	0644	2025-01-06 04:11:25
test_l7policy.py	text/x-python	3.93 KB	root/root	0644	2025-01-06 04:11:25
test_l7rule.py	text/x-python	3.41 KB	root/root	0644	2025-01-06 04:11:25
test_listener.py	text/x-python	7.42 KB	root/root	0644	2025-01-06 04:11:25
test_load_balancer.py	text/x-python	8.12 KB	root/root	0644	2025-01-06 04:11:25
test_member.py	text/x-python	3.31 KB	root/root	0644	2025-01-06 04:11:25
test_pool.py	text/x-python	4.66 KB	root/root	0644	2025-01-06 04:11:25
test_provider.py	text/x-python	2.69 KB	root/root	0644	2025-01-06 04:11:25
test_proxy.py	text/x-python	17.61 KB	root/root	0644	2025-01-06 04:11:25
test_quota.py	text/x-python	2.93 KB	root/root	0644	2025-01-06 04:11:25
test_version.py	text/x-python	1.38 KB	root/root	0644	2025-01-06 04:11:25