認証成功時のAPI実装

from common.http_client import *
from common.shared_state import *

class Authentication(HttpClient):

    def authenticate_system(self, username, credential):
        # システム認証処理
        logger.info("システム認証を実行中")
        endpoint = f"https://signinuat.midea.com/login?service=http%3A%2F%2Fdevsecopsuat.midea.com%2Fagile%2Ftest%2Ffunction%2Fcase%3FsysId%3DITS000001096&locale=zh"
        self._execute_get(endpoint=endpoint)
        payload = f'username={username}&password={credential}&execution=e1s1&bVisitorId=5e96ed57febc37d079725bd8f40ea042&_eventId=submit&authType=sms123&geolocation='
        response = self._execute_form_post(endpoint=endpoint, data=payload)
        set_shared_state("SystemSession", self.session)
        return response.text

認証前処理とシステムページ取得API

from common.http_client import *
from common.shared_state import *

class SystemApi(HttpClient):

    def __init__(self):
        self.session = get_shared_state("SystemSession")
        if self.session is None:
            logger.error("システムに認証されていません")
            raise SystemExit

    def fetch_system_data(self, payload):
        endpoint = "https://devsecopsuat.midea.com/api/system-project-server/systemInfo/getListPage"
        response = self._execute_json_post(endpoint=endpoint, data=payload)
        return response.json()

テストケース作成APIの実装

from common.http_client import *
from common.shared_state import *

class TestCaseCreator(HttpClient):
    def __init__(self):
        self.session = get_shared_state("SystemSession")
        if self.session is None:
            logger.error("システムに認証されていません")
            raise SystemExit

    def create_test_case(self, parameters):
        # 新規テストケース作成ボタンクリック
        endpoint = "https://devsecopsuat.midea.com/api/agile-service/workItem/v1/getItemView/VIRTUAL_SYSTEM_TESTCASE_TEMPLATE_ID"
        response = self._execute_get(endpoint=endpoint, params=parameters)
        return response.json()

テストケース削除APIの実装（Gpath式を使用）

from common.http_client import *
from common.shared_state import *

class TestCaseRemover(HttpClient):
    def __init__(self):
        self.session = get_shared_state("SystemSession")
        if self.session is None:
            logger.error("システムに認証されていません")
            raise SystemExit

    def remove_test_case(self, parameters):
        # headersの設定
        headers = {"Teamid": "", "Orgid": "bbe42cba6adb4d268d72ce087123e4c5", "Systemid": "ITS000001096", "View": "system"}
        endpoint = f"https://devsecopsuat.midea.com/api/agile-service/test/delete/<strong>{get_shared_state('case_id')}</strong>/ITS000001096"
        print("endpoint=", endpoint)
        response = self._execute_get(endpoint, data=parameters, headers=headers)
        return response.json()

初期化用の空クラス

動的値取得用ユーティリティクラス（環境変数のカプセル化）

# 共有状態-初期化
def init_shared_state():
    global _shared_dict
    _shared_dict = {}


# 共有状態の設定
def set_shared_state(key, value):
    try:
        # フレームワークが使用する共有変数はログに出力しない
        _shared_dict[key] = value
    except Exception:
        try:
            init_shared_state()
            _shared_dict[key] = value
        except Exception as e:
            raise Exception(f"共有変数の設定中にエラーが発生しました：{key}={value}、例外：{e}")


# 共有変数の取得（存在しない場合はNoneを返す）
def get_shared_state(key):
    try:
        return _shared_dict[key]
    except Exception as e:
        # デフォルトはsit環境
        if key == "environment":
            set_shared_state("environment", "sit")
            return "sit"
        print(f"共有変数の取得中にエラーが発生しました：【{key}】例外：{e}")
        return None

HTTPリクエストをカプセル化したクラス：

import json
import requests
from loguru import logger

class HttpClient:

    session = requests.session()

    def _execute_get(self, endpoint, params=None, data=None, headers=None):
        """
        GETリクエストの実行

        :param endpoint: リクエスト先URL
        :param headers: 任意、リクエストヘッダー
        :param data: 任意、リクエストボディ
        :return: レスポンス内容
        """
        logger.debug(f"リクエストデータ：{data}")
        self.headers = {"Connection": "keep-alive",
                        "Content-Type": "application/x-www-form-urlencoded;charset=utf-8",
                        "Accept": "application/json, text/javascript, */*; q=0.01"}
        if headers is not None:
            self.headers.update(headers)
        response = self.session.request(url=endpoint, method="GET", params=params, data=data, headers=self.headers,
                                    allow_redirects=True, timeout=180)
        logger.debug(f"レスポンスデータ：{response.text}")
        return response

    def _execute_form_post(self, endpoint, params=None, data=None, headers=None):
        """
        フォーム形式のPOSTリクエストの実行

        :param endpoint: リクエスト先URL
        :param headers: 任意、リクエストヘッダー
        :param data: 任意、リクエストボディ
        :return: レスポンス内容
        """
        logger.debug(f"リクエストデータ：{data}")
        self.headers = {"Connection": "keep-alive",
                        "Content-Type": "application/x-www-form-urlencoded;charset=utf-8",
                        "Accept": "application/json, text/javascript, */*; q=0.01"}
        if headers is not None:
            self.headers.update(headers)
        response = self.session.request(url=endpoint, method="POST", params=params, data=data, headers=self.headers,
                                    allow_redirects=True, timeout=180)
        logger.debug(f"レスポンスデータ：{response.text}")
        return response

    def _execute_json_post(self, endpoint, params=None, data=None, headers=None):
        """
        JSON形式のPOSTリクエストの実行

        :param endpoint: リクエスト先URL
        :param headers: 任意、リクエストヘッダー
        :param data: 任意、リクエストボディ
        :return: レスポンス内容
        """
        if isinstance(data, dict):
            data = json.dumps(data)
        logger.debug(f"リクエストデータ：{data}")
        self.headers = {"Connection": "keep-alive", "Content-Type": "application/json; charset=utf-8",
                        "Accept": "application/json, text/javascript, */*; q=0.01"}
        if headers is not None:
            self.headers.update(headers)
        response = self.session.request(url=endpoint, method="POST", params=params, data=data, headers=self.headers,
                                    allow_redirects=True, timeout=180)
        logger.debug(f"レスポンスデータ：{response.text}")
        return response

リクエストパラメータ読み込み用ユーティリティクラス（YAMLファイル読み込み）：

import os, yaml, re

BASE_DIR = os.path.dirname(os.path.realpath(__file__))
ROOT_PATH = os.path.abspath(os.path.join(BASE_DIR, "../"))

# 単一YAMLファイルからデータを読み込む
def load_yaml(yamlPath):
    BASE_DIR = os.path.dirname(os.path.realpath(__file__))
    root_path = os.path.abspath(os.path.join(BASE_DIR, "../"))
    yamlPath = root_path + yamlPath
    with open(yamlPath, encoding='utf-8') as f:
        # YAMLデータの読み込み
        data = yaml.safe_load(f)
        return data

def generate_yaml(data, yamlFile):
    """
    辞書やリストからYAMLファイルを生成する

    :param data: 入力データ（辞書、リスト）
    :param yamlFile: 生成するYAMLファイル名
    """
    try:
        yamlFile = os.path.join(ROOT_PATH + yamlFile)
        with open(yamlFile, "w", encoding="utf-8") as f:
            yaml.dump(data, f, allow_unicode=True)
    except Exception as e:
        raise Exception(f"YAMLファイル生成エラー：{e}")


if __name__ == '__main__':
    pass
    # リクエストデータ：配列、辞書、文字列が可能
    request_data = {"param":{"org_id":"bbe42cba6adb4d268d72ce087123e4c5","sys_name":""},"current":1,"size":10,"startPage":1,"pageSize":10}
    # YAML生成メソッドの呼び出し
    generate_yaml(request_data, "/test_data/sample.yaml")

テストレポート生成用ユーティリティクラス：

# -*- coding: utf-8 -*-
# @File  : shell.py
import subprocess

class CommandExecutor:
    """シェルコマンド実行メソッドをカプセル化"""
    @staticmethod
    def run(command):
        output, errors = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).communicate()
        o = output.decode("utf-8")
        return o

リクエスト実行用クラス：

import allure

from apis.StoreCase import StoreCase
from apis.ValidateCase import ValidateCase
from apis.DuplicateCase import DuplicateCase
from apis.RemoveCase import RemoveCase
from apis.ExtractCase import ExtractCase
from apis.auth import *
from apis.system import *
from apis.creator import *
from common.yaml_loader import *

class TestSuite:

    #認証
    @allure.feature('認証')
    @allure.story('システムログイン')
    def test_login(self):
        # BlazeMonitorでパケットキャプチャ
        Authentication().authenticate_system("ex_shijl2", "vi3fYRFnStfYLXBriYRLlg==")
        print("認証成功")

        system_api = SystemApi()
        data = load_yaml("/test_data/sample.yaml")
        system_api.fetch_system_data(data)

    #新規テストケース作成ボタンクリック
    @allure.feature('テストケース作成')
    @allure.story('新規作成')
    def test_create_case(self):
        creator = TestCaseCreator()
        data = load_yaml("/test_data/newCase01.yaml")
        creator.create_test_case(data)

    #【保存】ボタンクリック（一意性チェックAPIが呼ばれる）
    @allure.feature('一意性チェック')
    @allure.story('重複確認')
    def test_validate_case(self):
        validator = ValidateCase()
        data = load_yaml("/test_data/checkCase.yaml")
        validator.validate_case(data)

    @allure.feature('テストケース保存')
    @allure.story('保存処理')
    def test_store_case(self):
        storer = StoreCase()
        data = load_yaml("/test_data/saveCase.yaml")
        storer.store_case(data)

    #コピー
    @allure.feature('テストケースコピー')
    @allure.story('複製処理')
    def test_duplicate_case(self):
        duplicator = DuplicateCase()
        data = load_yaml("/test_data/copyCase01.yaml")
        duplicator.duplicate_case(data)

    @allure.feature('コピーしたテストケースの保存')
    @allure.story('複製保存')
    def test_store_duplicate(self):
        storer = StoreCase()
        data = load_yaml("/test_data/saveCopyCase.yaml")
        storer.store_case(data)

    #エクスポート
    @allure.feature('テストケースエクスポート')
    @allure.story('データ出力')
    def test_export_case(self):
        #クラス名の先頭は大文字、YAML内のデータはフォーマットしないこと
        #Authentication().authenticate_system("ex_shijl2", "vi3fYRFnStfYLXBriYRLlg==")
        extractor = ExtractCase()
        extractor.get_system_info({"systemId":"ITS000001096","listView":"drop_down_tree","extMap":{}})
        data = load_yaml("/test_data/exportCase.yaml")
        response = extractor.extract_case(data)
        assert response.status_code == 200
        # with open("./test.xls", 'wb') as f:
        #     f.write(response.content)

    #削除
    @allure.feature('テストケース削除')
    @allure.story('削除処理')
    # allure serve ./result/
    # allure serve ./allure-results
    def test_remove_case(self):
       remover = RemoveCase()
       data = load_yaml("/test_data/deleteCase.yaml")
       remover.remove_case(data)

リクエストパラメータ用YAMLファイルの実際の読み込み設定

起動クラスとレポート読み込みクラス：

import pytest, os, sys
from common.command_executor import CommandExecutor
from loguru import logger

# 実行システム識別子
systemMark = str(sys.platform)

if __name__ == "__main__":
    BASE_DIR = os.path.dirname(os.path.realpath(__file__))
    root_path = os.path.abspath(os.path.join(BASE_DIR))
    # コマンドビルド
    executor = CommandExecutor()
    allure_report = "--alluredir=./allure-results"
    allure_result_path = "./allure-results"
    allure_report_path = "./allure-report"

    try:
        # テストケース実行
        args = ['-v', '-s', 'test_case/', allure_report, '--clean-alluredir', "--reruns=1"]
        pytest.main(args)
        # Allureレポート生成
        cmd = 'allure generate %s -o %s --clean' % (allure_result_path, allure_report_path)
        executor.run(cmd)
        logger.debug("Allureレポートの生成に成功しました")
    except Exception as e:
        logger.error(f'テストケースの実行に失敗しました、{e}')

allure-pytest==2.9.45
allure-python-commons==2.9.45
jmespath==0.9.5
jsonpath==0.82
logger==1.4
numpy==1.24.3
openpyxl==3.0.10
orjson==3.7.7
pandas==2.0.1
pytest==7.1.2
pytest-dependency==0.6.0
pytest-forked==1.4.0
pytest-html==3.1.1
pytest-metadata==2.0.1
pytest-rerunfailures==10.2
pytest-sugar==0.9.4
pytest-xdist==2.5.0
python-dateutil==2.8.2
PyYAML==6.0
regex==2023.8.8
requests==2.27.1
setuptools==56.0.0
simple-settings==1.2.0
six==1.16.0
xlrd==2.0.1