以如下python接口为例:
@statefulsets_api.route('/stateful-sets', methods=['GET'])
@statefulsets_api.route('/namespaces/<namespace>/stateful-sets', methods=['GET'])
def get_deployments(namespace=None):
# If a namespace was specified, list deployments for that namespace only
# Otherwise, list deployments for all namespaces
if namespace:
statefulsets_list = v1.list_namespaced_stateful_set(namespace, watch=False)
else:
statefulsets_list = v1.list_stateful_set_for_all_namespaces(watch=False)
return jsonify([statefulset.to_dict() for statefulset in statefulsets_list.items])
我们给这个接口添加一个namespace的校验逻辑，验证namespace只包含 大小写字母、数字、下划线、中划线。
首先，我们需要定义一个函数来检查kubernetes的namespace是否符合要求
import re
def is_valid_namespace(namespace):
return bool(re.match(r'^[a-zA-Z0-9_-]+$', namespace))
接下来，我们需要修改get_deployments函数，以便在收到带有无效namespace的请求时返回适当的错误消息:
from flask import jsonify, abort
@deployments_api.route('/deployments', methods=['GET'])
@deployments_api.route('/namespaces/<namespace>/deployments', methods=['GET'])
def get_deployments(namespace=None):
if namespace and not is_valid_namespace(namespace):
abort(400, description="Invalid namespace. It should only contain letters, numbers, underscores, and hyphens.")
# If a namespace was specified, list deployments for that namespace only
# Otherwise, list deployments for all namespaces
if namespace:
deployments_list = v1.list_namespaced_deployment(namespace, watch=False)
else:
deployments_list = v1.list_deployment_for_all_namespaces(watch=False)
return jsonify([deployment.to_dict() for deployment in deployments_list.items])
现在，我们可以编写针对这个接口的测试用例，确保当请求无效的namespace时，接口返回400错误。
假设当前项目的路径设计如下
├── api
│ ├── kafka_api
│ ├── kubernetes_api
│ │ ├── __init__.py
│ ├── __init__.py
├── app.py
├── module
我们接下来在项目中
- 在
api/kubernetes文件夹中，创建一个名为tests的子文件夹。 - 在
api/kubernetes/tests文件夹中，创建一个名为test_kubernetes_deployments.py的文件。在这个文件中，您可以编写针对api/kubernetes中的功能的测试用例。
接下来项目结构应该变化成如下所示:
├── api
│ ├── kafka_api
│ ├── kubernetes_api
│ │ ├── __init__.py
│ │ ├── tests
│ │ │ ├── __init__.py
│ │ │ ├── test_kubernetes_deployments.py
│ ├── __init__.py
├── app.py
├── module
为了让pytest能够自动发现并运行这些测试用例，您需要在每个包含测试用例的文件夹中添加一个__init__.py文件（如果尚未存在）。在这种情况下，您需要确保api/kubernetes和api/kubernetes/tests文件夹中都有__init__.py文件。
书写测试用例，并运行。注：也可使用pytest命令行运行
import pytest
from app import app
@pytest.fixture
def client():
with app.test_client() as client:
yield client
def test_get_deployments_invalid_namespace(client):
# Test with an invalid namespace parameter
response = client.get('/api/kubernetes/instances/default/namespaces/invalid@namespace/deployments')
assert response.status_code == 400
decode = response.data.decode('utf-8')
assert "Invalid namespace. It should only contain letters, numbers, underscores, and hyphens." in decode
如果运行时出现了TypeError: __init__() got an unexpected keyword argument 'as_tuple'错误，请升级您的flask版本到最新版本
