test_client.pyを改善: 起動時のconfig.json削除を追加、初期化待機ロジックを拡張(VRCT_INIT_TIMEOUT対応、ソフトタイムアウト警告、30秒ごとの進捗ログ、プロセス終了検出とログ化、JSONログ表示の安定化)およびAutomatedEndpointTesterクラスを追加して run_all / run_random / run_specific を実装・メニューに統合
This commit is contained in:
@@ -7,7 +7,7 @@ stdin/stdoutを介してバックエンドと通信し、エンドポイント
|
||||
2. 別のターミナルでこのスクリプトを実行: python test_client.py
|
||||
3. エンドポイントとデータを指定してテストを実行
|
||||
"""
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import json
|
||||
import base64
|
||||
@@ -15,6 +15,9 @@ import sys
|
||||
import time
|
||||
from typing import Optional, Dict, Any
|
||||
|
||||
if os.path.exists("config.json"):
|
||||
os.remove("config.json")
|
||||
|
||||
class Color:
|
||||
GREEN = '\033[32m'
|
||||
RED = '\033[31m'
|
||||
@@ -43,44 +46,70 @@ class TestClient:
|
||||
self._wait_for_initialization()
|
||||
print(f"{Color.GREEN}バックエンド起動完了{Color.RESET}\n")
|
||||
|
||||
def _wait_for_initialization(self, timeout: float = 60.0):
|
||||
"""
|
||||
バックエンドの初期化完了を待機
|
||||
/run/initialization_complete のレスポンスを受信するまで待つ
|
||||
def _wait_for_initialization(self, timeout: Optional[float] = None):
|
||||
"""バックエンド初期化完了 (/run/initialization_complete) を待機する。
|
||||
|
||||
旧仕様: 60秒で TimeoutError を発生させていた。
|
||||
新仕様:
|
||||
- timeout が None の場合は無期限待機。
|
||||
- 'VRCT_INIT_TIMEOUT' 環境変数が設定されていれば soft timeout 値として使用。
|
||||
- soft timeout 経過時は ERROR ではなく WARN を表示し継続待機。
|
||||
- 進捗: 30秒ごとに経過時間と最後に受信した endpoint を表示。
|
||||
- バックエンドプロセスが終了した場合のみ例外を投げる。
|
||||
"""
|
||||
import os
|
||||
env_timeout = os.getenv("VRCT_INIT_TIMEOUT")
|
||||
if timeout is None:
|
||||
try:
|
||||
timeout = float(env_timeout) if env_timeout else None
|
||||
except ValueError:
|
||||
timeout = None
|
||||
|
||||
start_time = time.time()
|
||||
last_progress_endpoint = None
|
||||
last_progress_time_log = 0.0
|
||||
|
||||
while True:
|
||||
if time.time() - start_time > timeout:
|
||||
print(f"{Color.RED}[ERROR]{Color.RESET} 初期化タイムアウト ({timeout}秒)")
|
||||
raise TimeoutError(f"Backend initialization timeout after {timeout} seconds")
|
||||
|
||||
# stdoutから1行読み込み
|
||||
line = self.process.stdout.readline()
|
||||
|
||||
if not line:
|
||||
# プロセスが終了した場合
|
||||
# プロセス終了検知
|
||||
if self.process.poll() is not None:
|
||||
raise RuntimeError("Backend process terminated during initialization")
|
||||
|
||||
# soft timeout 警告表示
|
||||
if timeout is not None and (time.time() - start_time) > timeout:
|
||||
# 一度だけ警告を出し timeout を解除(以降は継続待機)
|
||||
print(f"{Color.YELLOW}[WARN]{Color.RESET} 初期化が {timeout:.1f} 秒を超過しました。ダウンロード等で長時間かかっています。引き続き待機します。環境変数 VRCT_INIT_TIMEOUT を調整できます。")
|
||||
timeout = None # 解除
|
||||
|
||||
# 30秒ごとの進捗ログ
|
||||
now = time.time()
|
||||
if now - last_progress_time_log >= 30:
|
||||
elapsed = now - start_time
|
||||
ep_info = last_progress_endpoint or "(受信なし)"
|
||||
print(f"{Color.CYAN}[進捗]{Color.RESET} 初期化経過 {elapsed:.1f} 秒 / 最終 endpoint: {ep_info}")
|
||||
last_progress_time_log = now
|
||||
|
||||
line = self.process.stdout.readline()
|
||||
if not line:
|
||||
# 何も来ていないがプロセスは動作中 -> 継続
|
||||
continue
|
||||
|
||||
# JSON解析を試みる
|
||||
stripped = line.strip()
|
||||
if not stripped:
|
||||
continue
|
||||
|
||||
# JSON解析試行
|
||||
try:
|
||||
response = json.loads(line.strip())
|
||||
response = json.loads(stripped)
|
||||
endpoint = response.get("endpoint", "")
|
||||
status = response.get("status", 0)
|
||||
|
||||
# 初期化メッセージを表示
|
||||
last_progress_endpoint = endpoint or last_progress_endpoint
|
||||
print(f"{Color.CYAN} [初期化中]{Color.RESET} {endpoint} (Status: {status})")
|
||||
|
||||
# 初期化完了を検出
|
||||
if endpoint == "/run/initialization_complete" and status == 200:
|
||||
print(f"{Color.GREEN} [初期化完了]{Color.RESET}")
|
||||
total_elapsed = time.time() - start_time
|
||||
print(f"{Color.GREEN} [初期化完了]{Color.RESET} 経過 {total_elapsed:.1f} 秒")
|
||||
return
|
||||
|
||||
except json.JSONDecodeError:
|
||||
# JSON以外の行(ログなど)を表示
|
||||
stripped = line.strip()
|
||||
if stripped:
|
||||
# ログ行として扱う
|
||||
print(f"{Color.CYAN} [Backend]{Color.RESET} {stripped}")
|
||||
continue
|
||||
|
||||
@@ -263,6 +292,492 @@ def run_example_tests(client: TestClient):
|
||||
|
||||
print(f"{Color.BOLD}=== サンプルテスト終了 ==={Color.RESET}\n")
|
||||
|
||||
class AutomatedEndpointTester:
|
||||
"""backend_test.py のロジックを stdin/stdout 通信向けに移植した自動テストクラス"""
|
||||
def __init__(self, client: TestClient):
|
||||
self.client = client
|
||||
# config 的な値をキャッシュ
|
||||
self.cache: Dict[str, Any] = {}
|
||||
# エンドポイント分類 (動的取得手段が無いため暫定的にハードコード)
|
||||
self.validity_endpoints = [
|
||||
"/set/enable/translation",
|
||||
"/set/disable/translation",
|
||||
"/set/enable/transcription_send",
|
||||
"/set/disable/transcription_send",
|
||||
"/set/enable/transcription_receive",
|
||||
"/set/disable/transcription_receive",
|
||||
"/set/enable/websocket_server",
|
||||
"/set/disable/websocket_server",
|
||||
"/set/enable/convert_message_to_romaji",
|
||||
"/set/disable/convert_message_to_romaji",
|
||||
"/set/enable/convert_message_to_hiragana",
|
||||
"/set/disable/convert_message_to_hiragana",
|
||||
]
|
||||
self.set_data_endpoints = [
|
||||
"/set/data/selected_tab_no",
|
||||
"/set/data/selected_translation_engines",
|
||||
"/set/data/selected_your_languages",
|
||||
"/set/data/selected_target_languages",
|
||||
"/set/data/selected_transcription_engine",
|
||||
"/set/data/transparency",
|
||||
"/set/data/ui_scaling",
|
||||
"/set/data/textbox_ui_scaling",
|
||||
"/set/data/message_box_ratio",
|
||||
"/set/data/send_message_button_type",
|
||||
"/set/data/font_family",
|
||||
"/set/data/ui_language",
|
||||
"/set/data/main_window_geometry",
|
||||
"/set/data/selected_translation_compute_device",
|
||||
"/set/data/selected_transcription_compute_device",
|
||||
"/set/data/ctranslate2_weight_type",
|
||||
"/set/data/plamo_model",
|
||||
"/set/data/plamo_auth_key",
|
||||
"/set/data/gemini_model",
|
||||
"/set/data/gemini_auth_key",
|
||||
"/set/data/openai_model",
|
||||
"/set/data/openai_auth_key",
|
||||
"/set/data/lmstudio_model",
|
||||
"/set/data/lmstudio_url",
|
||||
"/set/data/ollama_model",
|
||||
"/set/data/deepl_auth_key",
|
||||
"/set/data/selected_mic_host",
|
||||
"/set/data/selected_mic_device",
|
||||
"/set/data/mic_threshold",
|
||||
"/set/data/mic_record_timeout",
|
||||
"/set/data/mic_phrase_timeout",
|
||||
"/set/data/mic_max_phrases",
|
||||
"/set/data/hotkeys",
|
||||
"/set/data/plugins_status",
|
||||
"/set/data/mic_avg_logprob",
|
||||
"/set/data/mic_no_speech_prob",
|
||||
"/set/data/mic_word_filter",
|
||||
"/set/data/selected_speaker_device",
|
||||
"/set/data/speaker_threshold",
|
||||
"/set/data/speaker_record_timeout",
|
||||
"/set/data/speaker_phrase_timeout",
|
||||
"/set/data/speaker_max_phrases",
|
||||
"/set/data/speaker_avg_logprob",
|
||||
"/set/data/speaker_no_speech_prob",
|
||||
"/set/data/whisper_weight_type",
|
||||
"/set/data/overlay_small_log_settings",
|
||||
"/set/data/overlay_large_log_settings",
|
||||
"/set/data/send_message_format_parts",
|
||||
"/set/data/received_message_format_parts",
|
||||
"/set/data/websocket_host",
|
||||
"/set/data/websocket_port",
|
||||
"/set/data/osc_ip_address",
|
||||
"/set/data/osc_port",
|
||||
"/set/data/selected_translation_compute_type",
|
||||
"/set/data/selected_transcription_compute_type",
|
||||
]
|
||||
self.run_endpoints = [
|
||||
"/run/send_message_box",
|
||||
"/run/typing_message_box",
|
||||
"/run/stop_typing_message_box",
|
||||
"/run/send_text_overlay",
|
||||
"/run/swap_your_language_and_target_language",
|
||||
"/run/update_software",
|
||||
"/run/update_cuda_software",
|
||||
"/run/download_ctranslate2_weight",
|
||||
"/run/download_whisper_weight",
|
||||
"/run/open_filepath_logs",
|
||||
"/run/open_filepath_config_file",
|
||||
"/run/feed_watchdog",
|
||||
"/run/lmstudio_connection",
|
||||
"/run/ollama_connection",
|
||||
]
|
||||
self.delete_data_endpoints = [
|
||||
"/delete/data/deepl_auth_key",
|
||||
]
|
||||
self.results: Dict[str, Dict[str, Any]] = {}
|
||||
|
||||
# ---------------------------------- Utility ----------------------------------
|
||||
def _record(self, endpoint: str, status: Optional[int], result: Any, expected_status: list[int]):
|
||||
self.results[endpoint] = {
|
||||
"status": status,
|
||||
"result": result,
|
||||
"expected_status": expected_status,
|
||||
"success": status in expected_status if status is not None else False
|
||||
}
|
||||
|
||||
def _get(self, endpoint: str) -> Any:
|
||||
"""/get/data/* の結果をキャッシュしつつ取得"""
|
||||
resp = self.client.send_request(endpoint)
|
||||
if resp.get("status") == 200:
|
||||
self.cache[endpoint.split("/")[-1]] = resp.get("result")
|
||||
return resp.get("result")
|
||||
return None
|
||||
|
||||
# ---------------------------------- Generators ----------------------------------
|
||||
def _gen_set_data(self, endpoint: str):
|
||||
expected = [200]
|
||||
data = None
|
||||
# ほぼ backend_test.py のロジックを踏襲
|
||||
if endpoint == "/set/data/selected_tab_no":
|
||||
data = sys.modules.get('__random_tab_choices', None) or None # placeholder for future dynamic
|
||||
data = data or "1"
|
||||
elif endpoint == "/set/data/selected_translation_engines":
|
||||
engines = self._get("/get/data/translation_engines") or []
|
||||
data = {i: (engines and (engines[0] if len(engines) else None)) for i in ["1","2","3"]}
|
||||
elif endpoint == "/set/data/selected_your_languages":
|
||||
lang_list = self._get("/get/data/selectable_language_list") or []
|
||||
if lang_list:
|
||||
choice = lang_list[0]
|
||||
data = {i: {"1": {**choice, "enable": True}} for i in ["1","2","3"]}
|
||||
elif endpoint == "/set/data/selected_target_languages":
|
||||
lang_list = self._get("/get/data/selectable_language_list") or []
|
||||
if lang_list:
|
||||
base = lang_list[0]
|
||||
data = {i: {j: {**base, "enable": (j=="1")} for j in ["1","2","3"]} for i in ["1","2","3"]}
|
||||
elif endpoint == "/set/data/selected_transcription_engine":
|
||||
engines = self._get("/get/data/transcription_engines") or []
|
||||
data = engines[0] if engines else None
|
||||
elif endpoint == "/set/data/transparency":
|
||||
import random
|
||||
data = random.randint(0,100)
|
||||
elif endpoint == "/set/data/ui_scaling" or endpoint == "/set/data/textbox_ui_scaling":
|
||||
import random
|
||||
data = random.randint(50,200)
|
||||
elif endpoint == "/set/data/message_box_ratio":
|
||||
import random
|
||||
data = round(random.uniform(0.1,0.9),2)
|
||||
elif endpoint == "/set/data/send_message_button_type":
|
||||
import random
|
||||
data = random.choice(["show","hide","show_and_disable_enter_key"])
|
||||
elif endpoint == "/set/data/font_family":
|
||||
import random
|
||||
data = random.choice(["Arial","Verdana","Times New Roman"])
|
||||
elif endpoint == "/set/data/ui_language":
|
||||
import random
|
||||
data = random.choice(["en","ja","ko","zh-Hant","zh-Hans"])
|
||||
elif endpoint == "/set/data/main_window_geometry":
|
||||
import random
|
||||
data = {
|
||||
"x_pos": random.randint(0,1920),
|
||||
"y_pos": random.randint(0,1080),
|
||||
"width": random.randint(800,1920),
|
||||
"height": random.randint(600,1080)
|
||||
}
|
||||
elif endpoint == "/set/data/selected_translation_compute_device":
|
||||
lst = self._get("/get/data/translation_compute_device_list") or []
|
||||
import random
|
||||
data = random.choice(lst) if lst else None
|
||||
elif endpoint == "/set/data/selected_transcription_compute_device":
|
||||
lst = self._get("/get/data/transcription_compute_device_list") or []
|
||||
import random
|
||||
data = random.choice(lst) if lst else None
|
||||
elif endpoint == "/set/data/ctranslate2_weight_type":
|
||||
dct = self._get("/get/data/selectable_ctranslate2_weight_type_dict") or {}
|
||||
keys = list(dct.keys())
|
||||
import random
|
||||
data = random.choice(keys) if keys else None
|
||||
elif endpoint == "/set/data/plamo_model":
|
||||
lst = self._get("/get/data/plamo_model_list") or []
|
||||
import random
|
||||
data = random.choice(lst) if lst else None
|
||||
elif endpoint == "/set/data/plamo_auth_key":
|
||||
data = "PLAMO_DUMMY_KEY"
|
||||
expected = [200,400]
|
||||
elif endpoint == "/set/data/gemini_model":
|
||||
lst = self._get("/get/data/gemini_model_list") or []
|
||||
import random
|
||||
data = random.choice(lst) if lst else None
|
||||
elif endpoint == "/set/data/gemini_auth_key":
|
||||
data = "GEMINI_DUMMY_KEY"
|
||||
expected = [200,400]
|
||||
elif endpoint == "/set/data/openai_model":
|
||||
lst = self._get("/get/data/openai_model_list") or []
|
||||
import random
|
||||
data = random.choice(lst) if lst else None
|
||||
elif endpoint == "/set/data/openai_auth_key":
|
||||
data = "OPENAI_DUMMY_KEY"
|
||||
expected = [200,400]
|
||||
elif endpoint == "/set/data/lmstudio_model":
|
||||
lst = self._get("/get/data/lmstudio_model_list") or []
|
||||
import random
|
||||
data = random.choice(lst) if lst else None
|
||||
elif endpoint == "/set/data/lmstudio_url":
|
||||
import random
|
||||
data = random.choice(["http://localhost:1234/v1","http://127.0.0.1:1234/v1","http://invalid_host:9999/v1"])
|
||||
expected=[200,400]
|
||||
elif endpoint == "/set/data/ollama_model":
|
||||
lst = self._get("/get/data/ollama_model_list") or []
|
||||
import random
|
||||
data = random.choice(lst) if lst else None
|
||||
elif endpoint == "/set/data/deepl_auth_key":
|
||||
data = "DEEPL_DUMMY_KEY"
|
||||
expected=[200,400]
|
||||
elif endpoint == "/set/data/selected_mic_host":
|
||||
lst = self._get("/get/data/mic_host_list") or []
|
||||
import random
|
||||
data = random.choice(lst) if lst else None
|
||||
elif endpoint == "/set/data/selected_mic_device":
|
||||
lst = self._get("/get/data/mic_device_list") or []
|
||||
import random
|
||||
data = random.choice(lst) if lst else None
|
||||
elif endpoint == "/set/data/mic_threshold":
|
||||
import random
|
||||
val = random.randint(-1000,3000)
|
||||
data = val
|
||||
expected=[200] if 0 <= val <= 2000 else [400]
|
||||
elif endpoint == "/set/data/mic_record_timeout":
|
||||
import random
|
||||
val = random.randint(-1,10)
|
||||
phrase = self._get("/get/data/mic_phrase_timeout")
|
||||
data = val
|
||||
expected=[200] if (phrase is not None and 0 <= val <= phrase) else [400]
|
||||
elif endpoint == "/set/data/mic_phrase_timeout":
|
||||
import random
|
||||
val = random.randint(-1,10)
|
||||
record = self._get("/get/data/mic_record_timeout")
|
||||
data = val
|
||||
expected=[200] if (record is not None and record <= val) else [400]
|
||||
elif endpoint == "/set/data/mic_max_phrases":
|
||||
import random
|
||||
val = random.randint(-1,10)
|
||||
data = val
|
||||
expected=[200] if val >= 0 else [400]
|
||||
elif endpoint == "/set/data/hotkeys":
|
||||
data = {'toggle_vrct_visibility': None,'toggle_translation': None,'toggle_transcription_send': None,'toggle_transcription_receive': None}
|
||||
elif endpoint == "/set/data/plugins_status":
|
||||
plugins = self._get("/get/data/plugins") or []
|
||||
import random
|
||||
data = {p: random.choice([True,False]) for p in plugins}
|
||||
elif endpoint == "/set/data/mic_avg_logprob":
|
||||
import random
|
||||
data = random.uniform(-5,0)
|
||||
elif endpoint == "/set/data/mic_no_speech_prob":
|
||||
import random
|
||||
data = random.uniform(0,1)
|
||||
elif endpoint == "/set/data/mic_word_filter":
|
||||
import random
|
||||
data = random.choice([["test_0_0","test_0_1","test_0_2",None],["test_1_0","test_1_1",None],["test_2_0",None],[None]])
|
||||
elif endpoint == "/set/data/selected_speaker_device":
|
||||
lst = self._get("/get/data/speaker_device_list") or []
|
||||
import random
|
||||
data = random.choice(lst) if lst else None
|
||||
elif endpoint == "/set/data/speaker_threshold":
|
||||
import random
|
||||
val = random.randint(-1000,5000)
|
||||
data = val
|
||||
expected=[200] if 0 <= val <= 4000 else [400]
|
||||
elif endpoint == "/set/data/speaker_record_timeout":
|
||||
import random
|
||||
val = random.randint(-1,10)
|
||||
phrase = self._get("/get/data/speaker_phrase_timeout")
|
||||
data = val
|
||||
expected=[200] if (phrase is not None and 0 <= val <= phrase) else [400]
|
||||
elif endpoint == "/set/data/speaker_phrase_timeout":
|
||||
import random
|
||||
val = random.randint(-1,10)
|
||||
record = self._get("/get/data/speaker_record_timeout")
|
||||
data = val
|
||||
expected=[200] if (record is not None and record <= val) else [400]
|
||||
elif endpoint == "/set/data/speaker_max_phrases":
|
||||
import random
|
||||
val = random.randint(-1,10)
|
||||
data = val
|
||||
expected=[200] if val >= 0 else [400]
|
||||
elif endpoint == "/set/data/speaker_avg_logprob":
|
||||
import random
|
||||
data = random.uniform(-5,0)
|
||||
elif endpoint == "/set/data/speaker_no_speech_prob":
|
||||
import random
|
||||
data = random.uniform(0,1)
|
||||
elif endpoint == "/set/data/whisper_weight_type":
|
||||
dct = self._get("/get/data/selectable_whisper_weight_type_dict") or {}
|
||||
import random
|
||||
keys=[k for k,v in dct.items() if v]
|
||||
data = random.choice(keys) if keys else None
|
||||
elif endpoint == "/set/data/overlay_small_log_settings" or endpoint == "/set/data/overlay_large_log_settings":
|
||||
import random
|
||||
data = {
|
||||
"x_pos": random.random(),
|
||||
"y_pos": random.random(),
|
||||
"z_pos": random.random(),
|
||||
"x_rotation": random.random(),
|
||||
"y_rotation": random.random(),
|
||||
"z_rotation": random.random(),
|
||||
"display_duration": random.randint(0,100),
|
||||
"fadeout_duration": random.randint(0,100),
|
||||
"opacity": random.random(),
|
||||
"ui_scaling": random.random(),
|
||||
"tracker": random.choice(["HMD","LeftHand","RightHand"])
|
||||
}
|
||||
elif endpoint == "/set/data/send_message_format_parts":
|
||||
fmt = self._get("/get/data/send_message_format_parts")
|
||||
data = fmt
|
||||
elif endpoint == "/set/data/received_message_format_parts":
|
||||
fmt = self._get("/get/data/received_message_format_parts")
|
||||
data = fmt
|
||||
elif endpoint == "/set/data/websocket_host":
|
||||
import random
|
||||
val = random.choice(["127.0.0.1","aaaaadwafasdsd","0210.1564.845.0"])
|
||||
data = val
|
||||
expected = [200,400] if val=="127.0.0.1" else [400]
|
||||
elif endpoint == "/set/data/websocket_port":
|
||||
import random
|
||||
data = random.randint(1024,65535)
|
||||
expected=[200,400]
|
||||
elif endpoint == "/set/data/osc_ip_address":
|
||||
import random
|
||||
val = random.choice(["127.0.0.1","aaaaadwafasdsd","0210.1564.845.0"])
|
||||
data = val
|
||||
expected = [200] if val=="127.0.0.1" else [400]
|
||||
elif endpoint == "/set/data/osc_port":
|
||||
import random
|
||||
data = random.randint(1024,65535)
|
||||
elif endpoint == "/set/data/selected_translation_compute_type":
|
||||
device = self.cache.get("selected_translation_compute_device") or self._get("/get/data/selected_translation_compute_device")
|
||||
if device and isinstance(device, dict):
|
||||
import random
|
||||
data = random.choice(device.get("compute_types", [])) if device.get("compute_types") else None
|
||||
elif endpoint == "/set/data/selected_transcription_compute_type":
|
||||
device = self.cache.get("selected_transcription_compute_device") or self._get("/get/data/selected_transcription_compute_device")
|
||||
if device and isinstance(device, dict):
|
||||
import random
|
||||
data = random.choice(device.get("compute_types", [])) if device.get("compute_types") else None
|
||||
return data, expected
|
||||
|
||||
def _gen_run_data(self, endpoint: str):
|
||||
expected = [200]
|
||||
data = None
|
||||
import random
|
||||
if endpoint == "/run/send_message_box":
|
||||
choices=[{"data":{"id":"000001","message":"test"},"status":[200]},
|
||||
{"data":{"id":"000002","message":"Hello World!"},"status":[200]},
|
||||
{"data":{"id":"000003","message":"こんにちわ 世界!"},"status":[200]},
|
||||
{"data":{"id":"000004","message":"안녕하세요 세계!"},"status":[200]},
|
||||
{"data":{"id":"000005","message":"你好,世界!"},"status":[200]}]
|
||||
choice = random.choice(choices)
|
||||
data = choice["data"]
|
||||
expected = choice["status"]
|
||||
elif endpoint in ["/run/typing_message_box","/run/stop_typing_message_box","/run/send_text_overlay","/run/swap_your_language_and_target_language"]:
|
||||
data = "test_overlay" if endpoint == "/run/send_text_overlay" else None
|
||||
elif endpoint in ["/run/update_software","/run/update_cuda_software","/run/download_ctranslate2_weight","/run/download_whisper_weight","/run/open_filepath_logs","/run/open_filepath_config_file","/run/feed_watchdog"]:
|
||||
expected=[401]
|
||||
elif endpoint in ["/run/lmstudio_connection","/run/ollama_connection"]:
|
||||
expected=[200,400]
|
||||
return data, expected
|
||||
|
||||
# ---------------------------------- Tests ----------------------------------
|
||||
def test_validity_single(self, endpoint: str):
|
||||
expected=[200]
|
||||
if endpoint == "/set/enable/websocket_server":
|
||||
expected=[200,400]
|
||||
resp = self.client.send_request(endpoint)
|
||||
status = resp.get("status")
|
||||
result = resp.get("result")
|
||||
self._record(endpoint, status, result, expected)
|
||||
ok = status in expected
|
||||
tag = f"{Color.GREEN}PASS{Color.RESET}" if ok else f"{Color.RED}FAIL{Color.RESET}"
|
||||
print(f"[Validity] {endpoint} -> {tag} ({status})")
|
||||
return ok
|
||||
|
||||
def test_set_data_single(self, endpoint: str):
|
||||
data, expected = self._gen_set_data(endpoint)
|
||||
if expected == [404]:
|
||||
self._record(endpoint, None, None, expected)
|
||||
print(f"[SetData] {endpoint} -> {Color.RED}UNKNOWN{Color.RESET}")
|
||||
return False
|
||||
if data is None:
|
||||
self._record(endpoint, None, None, expected)
|
||||
print(f"[SetData] {endpoint} -> {Color.YELLOW}SKIP(no data){Color.RESET}")
|
||||
return True
|
||||
resp = self.client.send_request(endpoint, data)
|
||||
status = resp.get("status")
|
||||
result = resp.get("result")
|
||||
self._record(endpoint, status, result, expected)
|
||||
ok = status in expected
|
||||
tag = f"{Color.GREEN}PASS{Color.RESET}" if ok else f"{Color.RED}FAIL{Color.RESET}"
|
||||
print(f"[SetData] {endpoint} -> {tag} ({status}) data={data}")
|
||||
return ok
|
||||
|
||||
def test_run_single(self, endpoint: str):
|
||||
data, expected = self._gen_run_data(endpoint)
|
||||
if expected == [401]:
|
||||
self._record(endpoint, None, None, expected)
|
||||
print(f"[Run] {endpoint} -> {Color.YELLOW}SKIP(401){Color.RESET}")
|
||||
return True
|
||||
resp = self.client.send_request(endpoint, data)
|
||||
status = resp.get("status")
|
||||
result = resp.get("result")
|
||||
self._record(endpoint, status, result, expected)
|
||||
ok = status in expected
|
||||
tag = f"{Color.GREEN}PASS{Color.RESET}" if ok else f"{Color.RED}FAIL{Color.RESET}"
|
||||
print(f"[Run] {endpoint} -> {tag} ({status})")
|
||||
return ok
|
||||
|
||||
def test_delete_single(self, endpoint: str):
|
||||
expected=[200]
|
||||
resp = self.client.send_request(endpoint)
|
||||
status = resp.get("status")
|
||||
result = resp.get("result")
|
||||
self._record(endpoint, status, result, expected)
|
||||
ok = status in expected
|
||||
tag = f"{Color.GREEN}PASS{Color.RESET}" if ok else f"{Color.RED}FAIL{Color.RESET}"
|
||||
print(f"[Delete] {endpoint} -> {tag} ({status})")
|
||||
return ok
|
||||
|
||||
def run_all(self):
|
||||
print(f"{Color.BOLD}=== 有効/無効エンドポイントテスト ==={Color.RESET}")
|
||||
for ep in self.validity_endpoints:
|
||||
self.test_validity_single(ep)
|
||||
print(f"{Color.BOLD}=== データ設定エンドポイントテスト ==={Color.RESET}")
|
||||
for ep in self.set_data_endpoints:
|
||||
self.test_set_data_single(ep)
|
||||
print(f"{Color.BOLD}=== 実行エンドポイントテスト ==={Color.RESET}")
|
||||
for ep in self.run_endpoints:
|
||||
self.test_run_single(ep)
|
||||
print(f"{Color.BOLD}=== 削除エンドポイントテスト ==={Color.RESET}")
|
||||
for ep in self.delete_data_endpoints:
|
||||
self.test_delete_single(ep)
|
||||
|
||||
def run_random(self, iterations: int = 500):
|
||||
import random
|
||||
print(f"{Color.BOLD}=== ランダムアクセステスト(iter={iterations}) ==={Color.RESET}")
|
||||
groups = ["validity","set","run","delete"]
|
||||
for i in range(iterations):
|
||||
g = random.choice(groups)
|
||||
if g == "validity":
|
||||
ep = random.choice(self.validity_endpoints)
|
||||
self.test_validity_single(ep)
|
||||
elif g == "set":
|
||||
ep = random.choice(self.set_data_endpoints)
|
||||
self.test_set_data_single(ep)
|
||||
elif g == "run":
|
||||
ep = random.choice(self.run_endpoints)
|
||||
self.test_run_single(ep)
|
||||
else:
|
||||
ep = random.choice(self.delete_data_endpoints)
|
||||
self.test_delete_single(ep)
|
||||
# 最後に disable 系を OFF
|
||||
for ep in self.validity_endpoints:
|
||||
if ep.startswith("/set/disable/"):
|
||||
self.client.send_request(ep)
|
||||
|
||||
def run_specific_random(self, iterations: int = 200):
|
||||
import random
|
||||
print(f"{Color.BOLD}=== 特定(osc/websocket)ランダムテスト(iter={iterations}) ==={Color.RESET}")
|
||||
set_specific = ["/set/data/osc_ip_address","/set/data/osc_port","/set/data/websocket_host","/set/data/websocket_port"]
|
||||
for i in range(iterations):
|
||||
ep = random.choice(set_specific)
|
||||
self.test_set_data_single(ep)
|
||||
|
||||
def summary(self):
|
||||
total = len(self.results)
|
||||
passed = sum(1 for r in self.results.values() if r["success"])
|
||||
skipped = sum(1 for r in self.results.values() if r["expected_status"] == [401])
|
||||
failed = total - passed - skipped
|
||||
print(f"\n{Color.BOLD}==== テストサマリー ==== {Color.RESET}")
|
||||
print(f"総数: {total} / 成功: {passed} / 失敗: {failed} / スキップ(401): {skipped}")
|
||||
if failed:
|
||||
print(f"{Color.RED}失敗詳細:{Color.RESET}")
|
||||
for ep, r in self.results.items():
|
||||
if not r["success"] and r["expected_status"] != [401]:
|
||||
print(f"- {ep} status={r['status']} expected={r['expected_status']} result={r['result']}")
|
||||
print(f"{Color.BOLD}======================={Color.RESET}\n")
|
||||
|
||||
def run_interactive_mode(client: TestClient):
|
||||
"""対話モードでテストを実行"""
|
||||
print(f"{Color.BOLD}=== 対話モード ==={Color.RESET}")
|
||||
@@ -316,7 +831,10 @@ def main():
|
||||
print("モードを選択してください:")
|
||||
print("1. サンプルテストを実行")
|
||||
print("2. 対話モードで実行")
|
||||
mode = input(f"{Color.CYAN}選択 (1 or 2): {Color.RESET}").strip()
|
||||
print("3. 自動テスト(全)を実行")
|
||||
print("4. ランダムアクセステスト")
|
||||
print("5. 特定(osc/websocket)ランダムテスト")
|
||||
mode = input(f"{Color.CYAN}選択 (1-5): {Color.RESET}").strip()
|
||||
|
||||
print()
|
||||
|
||||
@@ -324,6 +842,18 @@ def main():
|
||||
run_example_tests(client)
|
||||
elif mode == "2":
|
||||
run_interactive_mode(client)
|
||||
elif mode == "3":
|
||||
tester = AutomatedEndpointTester(client)
|
||||
tester.run_all()
|
||||
tester.summary()
|
||||
elif mode == "4":
|
||||
tester = AutomatedEndpointTester(client)
|
||||
tester.run_random()
|
||||
tester.summary()
|
||||
elif mode == "5":
|
||||
tester = AutomatedEndpointTester(client)
|
||||
tester.run_specific_random()
|
||||
tester.summary()
|
||||
else:
|
||||
print(f"{Color.YELLOW}無効な選択です。対話モードを開始します。{Color.RESET}\n")
|
||||
run_interactive_mode(client)
|
||||
|
||||
Reference in New Issue
Block a user