Source code for autoease.core.reporter

"""
@author: meili
@contact: meili02@corp.netease.com
@file: reporter.py
@date: 2023/9/5 14:50
@desc: 
"""
import time
import os
import json

import requests

from autoease.core.stage import (
    Stage
)


[docs]def send_report(case_name, backend_url, record_id, game_log_path, start_time, end_time, game_name, file_manager, case_info_log, case_error_log): from autoease.core.autoease_global import AutoEaseG as AE_G files = [] files_hwnd = [] # 1. copy game log file copy_log_file_name = f"{case_name}-{game_name}.log" copy_log_file_path = file_manager.copy_file(game_log_path, copy_log_file_name) AE_G.FILE_MANAGER.FILE_LIST.append(copy_log_file_path) # 2. get all dependency imgs path, copy to root path dependencies_dict = AE_G.STAGE_MANAGER.get_dependency_file_dict_from_args() dependencies_set = set() for img_path, file_full_name in dependencies_dict.items(): file_r_path = file_manager.copy_file(img_path, file_full_name) dependencies_set.add(file_r_path) # 3. stop performance monitor, get all performance data AE_G.PERFORMANCE_TOOL.stop_profile() time.sleep(2) performance_data_list = AE_G.PERFORMANCE_TOOL.get_all_tags_performance_data() p_idx = 0 for stage_or_step in AE_G.STAGE_MANAGER.STAGES_LIST: if isinstance(stage_or_step, Stage): for step_dict in stage_or_step.step_list: step_dict['performance'] = performance_data_list[p_idx] p_idx += 1 else: stage_or_step['performance'] = performance_data_list[p_idx] p_idx += 1 # 4. get start time and end time from airperf time_list = AE_G.PERFORMANCE_TOOL.get_start_time_and_end_time() if time_list is not None: start_time = time_list[0] end_time = time_list[1] # 5. save full performance data to a file r_path = AE_G.FILE_MANAGER.get_file_relative_path(f"{case_name}-{AE_G.GAME_NAME}-performance.json") AE_G.PERFORMANCE_TOOL.save_performance_data(r_path) if AE_G.PERFORMANCE_TOOL.PERFORMANCE_DATA_FILE_PATH is not None: AE_G.FILE_MANAGER.FILE_LIST.append(AE_G.PERFORMANCE_TOOL.PERFORMANCE_DATA_FILE_PATH) if AE_G.IS_RECORD: AE_G.FILE_MANAGER.FILE_LIST = AE_G.FILE_MANAGER.FILE_LIST + AE_G.DEVICE.get_screenshot_file_list if AE_G.IS_UPLOAD_VIDEO: AE_G.FILE_MANAGER.FILE_LIST = AE_G.FILE_MANAGER.FILE_LIST + AE_G.DEVICE.get_video_file_list else: video_excluded_whole_video = [video_file for video_file in AE_G.DEVICE.get_video_file_list if video_file != AE_G.DEVICE.whole_video] AE_G.FILE_MANAGER.FILE_LIST = AE_G.FILE_MANAGER.FILE_LIST + video_excluded_whole_video for depen_file_path in dependencies_set: AE_G.FILE_MANAGER.FILE_LIST.append(depen_file_path) for _ in AE_G.FILE_MANAGER.FILE_LIST: try: f = open(_, 'rb') print(_) files_hwnd.append(f) files.append(('files', f)) except FileNotFoundError: print(f"file not found: {_}") whole_result = AE_G.STAGE_MANAGER.get_whole_result() with open(case_info_log, "r") as file: case_info_log_content = file.read() with open(case_error_log, "r") as file: case_error_log_content = file.read() data = { "package_name": game_name, "record_id": int(record_id), "result": whole_result, "start_time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(start_time)), "end_time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(end_time)), "duration": int(end_time - start_time), "log": os.path.basename(copy_log_file_path), "content": json.dumps(AE_G.STAGE_MANAGER.get_stages_list()), "case_info_log": case_info_log_content, "case_error_log":case_error_log_content, } if AE_G.IS_RECORD and AE_G.IS_UPLOAD_VIDEO: data.update({"video": os.path.basename(AE_G.DEVICE.whole_video)}) if AE_G.PERFORMANCE_TOOL.PERFORMANCE_DATA_FILE_PATH is not None: data['performance'] = os.path.basename(AE_G.PERFORMANCE_TOOL.PERFORMANCE_DATA_FILE_PATH) header = { "X-ADMIN-TOKEN": "cgX0D0eOynTQ4OZa" } print(json.dumps(data, indent=4)) resp = requests.post( url=backend_url, data=data, files=files, headers=header ) if resp.status_code == 200: print("post finished!") else: print("backend_url:", backend_url) print(resp.text) raise Exception for f in files_hwnd: f.close() time.sleep(2) for f in file_manager.FILE_LIST: if f != game_log_path: try: pass # os.remove(f) except PermissionError: pass pass
if __name__ == '__main__': pass