python flask框架进行参数校验和单元测试

以如下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版本到最新版本