@allure.title("验证周期MA批量同步") def test_schedule_ma_multiple(self): ma_count = 2 ma_list = [] vin_list = ["L6T79T2E1NP004452"] ecu_cmd_list = generate_ecu_cmd_list( [{"model": 0, "ecu": "VGM", "service": "22", "interval": 1, "did": "F1AE"}, {"model": 1, "ecu": "VGM", "service": "1902", "interval": 1, "input_parameter": "FF"}]) for i in range(ma_count): ma_list.append(create_schedule_or_event_ma("schedule", "2.0多任务周期同步" + str(randint(0, 100)), vin_list, ecu_cmd_list)) # 验证创建结果 assert len(ma_list) == ma_count # 车端请求云端:同步请求 res1 = vehicle_req_synchronization({"rvdc.eventId": rvdc_get_datetime(0, "TIMESTAMP"), "rvdc.requestId": 0, "rvdc.appId": 1, "rvdc.vin": self.vin, "rvdc.dataType": "maSynchronizationReq", "rvdc.payload.storageQuota": 25, "rvdc.payload.maFormatVersion": 0 }) assert res1.get("code") == 100000 # 车端请求云端:移除异常ma任务 res1 = vehicle_req_remove({"rvdc.eventId": rvdc_get_datetime(0, "TIMESTAMP"), "rvdc.requestId": 0, "rvdc.appId": 1, "rvdc.vin": self.vin, "rvdc.dataType": "rvdcRemoveMasResp", "rvdc.payload.masremoved": "true", }) assert res1.get("code") == 100000 # 因为是异步推送，验证推送结果 res = query_vehicle_cloud_log(**{"vin": self.vin, "type": "maSynchronization", "createTimeStart": "", "createTimeEnd": "", "maId": ma_list[0]}) res_log = res.get("data")["records"][0]["content"] res_content_json = json.loads(res_log) res2 = res_content_json["request"]["jsonObject"]["rvdc"]["payload"]["newMas"] assert ma_list[0] in res2 assert ma_list[1] in res2优化一下这段代码

考虑以下几点优化：

1. 将一些重复的代码抽象成函数，如创建MA任务、车端请求云端等，可以提高代码的复用性和可维护性。

2. 对于同步请求和移除异常MA任务的验证，可以使用pytest中的assert语句，使代码更加简洁易读。

3. 对于异步推送的验证，可以使用pytest中的fixture和mock，模拟推送结果，避免依赖实际的推送结果。

修改后的代码如下：

import pytest
import json

from random import randint

from utils import (
    create_schedule_or_event_ma,
    generate_ecu_cmd_list,
    vehicle_req_synchronization,
    vehicle_req_remove,
    query_vehicle_cloud_log,
    rvdc_get_datetime
)


@pytest.fixture(scope="module")
def ma_list():
    ma_count = 2
    vin_list = ["L6T79T2E1NP004452"]
    ecu_cmd_list = generate_ecu_cmd_list([
        {"model": 0, "ecu": "VGM", "service": "22", "interval": 1, "did": "F1AE"},
        {"model": 1, "ecu": "VGM", "service": "1902", "interval": 1, "input_parameter": "FF"}
    ])
    return [create_schedule_or_event_ma("schedule", "2.0多任务周期同步" + str(randint(0, 100)), vin_list, ecu_cmd_list)
            for _ in range(ma_count)]


@pytest.fixture(scope="module")
def sync_request():
    return {
        "rvdc.eventId": rvdc_get_datetime(0, "TIMESTAMP"),
        "rvdc.requestId": 0,
        "rvdc.appId": 1,
        "rvdc.vin": "L6T79T2E1NP004452",
        "rvdc.dataType": "maSynchronizationReq",
        "rvdc.payload.storageQuota": 25,
        "rvdc.payload.maFormatVersion": 0
    }


@pytest.fixture(scope="module")
def remove_request():
    return {
        "rvdc.eventId": rvdc_get_datetime(0, "TIMESTAMP"),
        "rvdc.requestId": 0,
        "rvdc.appId": 1,
        "rvdc.vin": "L6T79T2E1NP004452",
        "rvdc.dataType": "rvdcRemoveMasResp",
        "rvdc.payload.masremoved": "true"
    }


def test_create_ma(ma_list):
    assert len(ma_list) == 2


def test_sync_request(sync_request):
    res = vehicle_req_synchronization(sync_request)
    assert res.get("code") == 100000


def test_remove_request(remove_request):
    res = vehicle_req_remove(remove_request)
    assert res.get("code") == 100000


def test_query_vehicle_cloud_log(ma_list):
    # 模拟异步推送结果
    res_content_json = {
        "request": {
            "jsonObject": {
                "rvdc": {
                    "payload": {
                        "newMas": ma_list
                    }
                }
            }
        }
    }
    res = {
        "data": {
            "records": [
                {
                    "content": json.dumps(res_content_json)
                }
            ]
        }
    }
    # 验证推送结果
    res = query_vehicle_cloud_log(vin="L6T79T2E1NP004452", type="maSynchronization")
    res_log = res.get("data")["records"][0]["content"]
    res_content_json = json.loads(res_log)
    res2 = res_content_json["request"]["jsonObject"]["rvdc"]["payload"]["newMas"]
    assert set(ma_list) == set(res2)