YAMLデータファイル形式

設定とテストステップを定義するYAML構造:

config:
  name: 'テストケース'
  request:
    timeout: 30
    headers:
      x-test: abc123
  variables: 
    client_id: kPoFYw85FXsnojsy5bB9hu6x

tests:
  - name: 認証トークン取得
    request:
      url: https://aip.baidubce.com/oauth/2.0/token
      params:
        grant_type: client_credentials
        client_id: $client_id
    extract:
      access_token:  "response_data.json()['access_token']"
    verify:
      - "status_code == 200"
  - name: OCR処理
    request:
      url: https://aip.baidubce.com/rest/2.0/ocr/v1/general_basic?access_token=${access_token}
    verify:
      - "response_data.json()['words_result_num'] == 6"
  - name: スキップ例
    skip: True
  - name: 繰り返し実行
    times: 3
    request:
      url: https://httpbin.org/get

主要コンポーネント実装

定数定義

SETTINGS = 'config'
TEST_STEPS = 'tests'
ENV_VARS = 'variables'
BASE_URL = 'baseurl'
HTTP_REQUEST = 'request'
ASSERTIONS = 'verify'
VAR_EXTRACTION = 'extract'
SKIP_STEP = 'skip'
LOOP_COUNT = 'times'

設定初期化

session = requests.Session()
config_data = test_def.get(SETTINGS)
context = {}

if config_data:
    base_url = config_data.get(BASE_URL)
    session_attrs = config_data.get(HTTP_REQUEST, {})
    for attr, val in session_attrs.items():
        setattr(session, attr, val)
    context.update(config_data.get(ENV_VARS, {}))

ステップ実行処理

context['step_results'] = []
for step in test_def.get(TEST_STEPS, []):
    if step.get(SKIP_STEP) or not step.get(HTTP_REQUEST):
        continue
        
    for i in range(step.get(LOOP_COUNT, 1)):
        # リクエスト変数解決
        req_data = step[HTTP_REQUEST].copy()
        if '$' in str(req_data):
            template = Template(yaml.dump(req_data))
            req_data = yaml.safe_load(template.safe_substitute(context))
        
        # HTTPメソッド自動設定
        req_data.setdefault('method', 
                           'post' if any(k in req_data for k in ['data','json','files']) 
                           else 'get')
        
        # ベースURL適用
        if base_url and not req_data['url'].startswith('http'):
            req_data['url'] = f"{base_url.rstrip('/')}/{req_data['url'].lstrip('/')}"

リクエスト実行と応答処理

response = session.request(**req_data)
step_result = {
    'request': req_data,
    'response': response,
    'status_code': response.status_code,
    'response_data': response
}
context.update(step_result)
context['step_results'].append(step_result)

# 変数抽出
for var_name, expr in step.get(VAR_EXTRACTION, {}).items():
    context[var_name] = eval(expr, {'__builtins__': None}, context)

# アサーション検証
for assertion in step.get(ASSERTIONS, []):
    eval_result = eval(assertion, {'__builtins__': None}, context)
    print(f"Assertion: {assertion} -> {'PASS' if eval_result else 'FAIL'}")

完全な実装コード

import yaml
import requests
from string import Template

SETTINGS = 'config'
TEST_STEPS = 'tests'
ENV_VARS = 'variables'
BASE_URL = 'baseurl'
HTTP_REQUEST = 'request'
ASSERTIONS = 'verify'
VAR_EXTRACTION = 'extract'
SKIP_STEP = 'skip'
LOOP_COUNT = 'times'

def execute_test_flow(test_def):
    session = requests.Session()
    config_data = test_def.get(SETTINGS, {})
    base_url = config_data.get(BASE_URL, '')
    context = config_data.get(ENV_VARS, {}).copy()
    
    # セッション設定適用
    for attr, val in config_data.get(HTTP_REQUEST, {}).items():
        setattr(session, attr, val)
    
    context['step_results'] = []
    for step in test_def.get(TEST_STEPS, []):
        if step.get(SKIP_STEP) or not step.get(HTTP_REQUEST):
            continue
            
        for i in range(step.get(LOOP_COUNT, 1)):
            req_data = step[HTTP_REQUEST].copy()
            # 変数置換
            if '$' in str(req_data):
                template = Template(yaml.dump(req_data))
                req_data = yaml.safe_load(template.safe_substitute(context))
            
            # メソッド自動設定
            req_data.setdefault('method', 'post' if any(k in req_data for k in ['data','json','files']) else 'get')
            
            # URL結合
            if base_url and not req_data['url'].startswith('http'):
                req_data['url'] = f"{base_url.rstrip('/')}/{req_data['url'].lstrip('/')}"
            
            # リクエスト実行
            response = session.request(**req_data)
            step_result = {
                'request': req_data,
                'response': response,
                'status_code': response.status_code,
                'response_data': response
            }
            context.update(step_result)
            context['step_results'].append(step_result)
            
            # 変数抽出
            for var_name, expr in step.get(VAR_EXTRACTION, {}).items():
                context[var_name] = eval(expr, {'__builtins__': None}, context)
            
            # アサーション
            for assertion in step.get(ASSERTIONS, []):
                assert_result = eval(assertion, {'__builtins__': None}, context)
                print(f"Verify: {assertion} => {assert_result}")

if __name__ == "__main__":
    with open('test_scenario.yaml') as f:
        execute_test_flow(yaml.safe_load(f))