以如下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])

我們給這個(gè)接口添加一個(gè)namespace的校驗(yàn)邏輯，驗(yàn)證namespace只包含 大小寫字母、數(shù)字、下劃線、中劃線。

首先，我們需要定義一個(gè)函數(shù)來(lái)檢查kubernetes的namespace是否符合要求

import re


def is_valid_namespace(namespace):
    return bool(re.match(r'^[a-zA-Z0-9_-]+$', namespace))

接下來(lái)，我們需要修改get_deployments函數(shù)，以便在收到帶有無(wú)效namespace的請(qǐng)求時(shí)返回適當(dāng)?shù)腻e(cuò)誤消息:

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])

現(xiàn)在，我們可以編寫針對(duì)這個(gè)接口的測(cè)試用例，確保當(dāng)請(qǐng)求無(wú)效的namespace時(shí)，接口返回400錯(cuò)誤。

假設(shè)當(dāng)前項(xiàng)目的路徑設(shè)計(jì)如下

├── api
│   ├── kafka_api
│   ├── kubernetes_api
│   │   ├── __init__.py
│   ├── __init__.py
├── app.py
├── module

我們接下來(lái)在項(xiàng)目中

• 在api/kubernetes文件夾中，創(chuàng)建一個(gè)名為tests的子文件夾。
• 在api/kubernetes/tests文件夾中，創(chuàng)建一個(gè)名為test_kubernetes_deployments.py的文件。在這個(gè)文件中，您可以編寫針對(duì)api/kubernetes中的功能的測(cè)試用例。

接下來(lái)項(xiàng)目結(jié)構(gòu)應(yīng)該變化成如下所示:

├── api
│   ├── kafka_api
│   ├── kubernetes_api
│   │   ├── __init__.py
│   │   ├── tests
│   │   │   ├── __init__.py
│   │   │   ├── test_kubernetes_deployments.py
│   ├── __init__.py
├── app.py
├── module

為了讓pytest能夠自動(dòng)發(fā)現(xiàn)并運(yùn)行這些測(cè)試用例，您需要在每個(gè)包含測(cè)試用例的文件夾中添加一個(gè)__init__.py文件（如果尚未存在）。在這種情況下，您需要確保api/kubernetes和api/kubernetes/tests文件夾中都有__init__.py文件。

書寫測(cè)試用例，并運(yùn)行。注：也可使用pytest命令行運(yùn)行

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

如果運(yùn)行時(shí)出現(xiàn)了TypeError: __init__() got an unexpected keyword argument 'as_tuple'錯(cuò)誤，請(qǐng)升級(jí)您的flask版本到最新版本