Add a tempest test case to verify connection from k8s pod to outside

Change-Id: I948696c846fcc3fd5ac0eb26e4c94e271d66e0d6
Story: 2001585
Task: 23141
This commit is contained in:
Hunt Xu 2018-07-24 16:24:18 +08:00
parent ce6d7e0f64
commit 38be74a7a5
2 changed files with 57 additions and 0 deletions

View File

@ -0,0 +1,24 @@
# Copyright 2018 AWCloud Software Co., Ltd
#
# 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 hashlib
import requests
def main(url='https://docs.openstack.org/qinling/latest/', timeout=10,
*args, **kwargs):
# This function simply returns a sha256 hash of a webpage.
# We use this to verify function pods have access the outside world.
response = requests.get(url, timeout=timeout)
return hashlib.sha256(response.text.encode('utf-8')).hexdigest()

View File

@ -14,7 +14,9 @@
from concurrent import futures
import etcd3gw
import hashlib
import json
import requests
import futurist
from oslo_serialization import jsonutils
@ -451,3 +453,34 @@ class ExecutionsTest(base.BaseQinlingTest):
lower = second_duration * 1.8
self.assertGreaterEqual(upper, first_duration)
self.assertLessEqual(lower, first_duration)
@decorators.idempotent_id('07edf2ff-7544-4f30-b006-fd5302a2a9cc')
def test_python_execution_public_connection(self):
"""Test connections from k8s pod to the outside.
Create a function that reads a webpage on the Internet, to
verify that pods in Kubernetes can connect to the outside.
"""
# Create function
package = self.create_package(name='python/test_python_http_get.py')
function_id = self.create_function(package_path=package)
url = 'https://docs.openstack.org/qinling/latest'
# Gets the page's sha256 outside Qinling
response = requests.get(url, timeout=10)
page_sha256 = hashlib.sha256(response.text.encode('utf-8')).hexdigest()
# Create an execution to get the page's sha256 with Qinling
resp, body = self.client.create_execution(
function_id, input='{"url": "%s"}' % url
)
execution_id = body['id']
self.addCleanup(self.client.delete_resource, 'executions',
execution_id, ignore_notfound=True)
self.assertEqual(201, resp.status)
self.assertEqual('success', body['status'])
result = json.loads(body['result'])
self.assertEqual(page_sha256, result['output'])