openstack/kuryr-tempest-plugin
Code Issues Proposed changes

Merge "Testing curl to the service of type LoadBalancer"

Browse Source
This commit is contained in:
Zuul 2018-06-05 12:54:49 +00:00 committed by Gerrit Code Review
commit 7d5e376cc6
2 changed files with 31 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
kuryr_tempest_plugin/tests/scenario/base.py
View File

@ -223,13 +223,13 @@ class BaseKuryrScenarioTest(manager.NetworkScenarioTest):
raise lib_exc.ServerFault()
@classmethod
def create_setup_for_service_test(cls, pod_num=2):
def create_setup_for_service_test(cls, pod_num=2, spec_type="ClusterIP"):
for i in range(pod_num):
pod_name, pod = cls.create_pod(
labels={"app": 'pod-label'}, image='celebdor/kuryr-demo')
cls.addClassResourceCleanup(cls.delete_pod, pod_name)
service_name, service_obj = cls.create_service(
pod_label=pod.metadata.labels)
pod_label=pod.metadata.labels, spec_type=spec_type)
cls.service_ip = cls.get_service_ip(service_name)
cls.wait_service_status(
cls.service_ip, CONF.kuryr_kubernetes.lb_build_timeout)

29
kuryr_tempest_plugin/tests/scenario/test_service.py
View File

@ -68,3 +68,32 @@ class TestServiceScenario(base.BaseKuryrScenarioTest):
LOG.error("Curl the service IP %s failed" % self.service_ip)
raise lib_exc.UnexpectedResponseCode()
self.assertNotEqual(cmp(cmd_output_list[0], cmd_output_list[1]), '0')
class TestLoadBalancerServiceScenario(base.BaseKuryrScenarioTest):
@classmethod
def skip_checks(cls):
super(TestLoadBalancerServiceScenario, cls).skip_checks()
if not CONF.network_feature_enabled.floating_ips:
raise cls.skipException("Floating ips are not available")
@classmethod
def resource_setup(cls):
super(TestLoadBalancerServiceScenario, cls).resource_setup()
cls.create_setup_for_service_test(spec_type="LoadBalancer")
@decorators.idempotent_id('bddf5441-1244-449d-a175-b5fdcfc2a1a9')
def test_lb_service_curl(self):
cmd_output_list = list()
LOG.info("Trying to curl the service IP %s" % self.service_ip)
cmd = "curl {dst_ip}".format(dst_ip=self.service_ip)
for i in range(2):
try:
cmd_output_list.append(
subprocess.check_output(shlex.split(cmd)))
except subprocess.CalledProcessError:
LOG.error("Checking output of curl to the service IP %s "
"failed" % self.service_ip)
raise lib_exc.UnexpectedResponseCode()
self.assertNotEqual(cmp(cmd_output_list[0], cmd_output_list[1]), '0')