wehub-resource-sync
57 lines
2.0 KiB
Python
57 lines
2.0 KiB
Python
import ast
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
EXAMPLE_DIR = ROOT / "examples" / "industrial_data_pretraining" / "qwen3_asr"
|
|
SERVER_PATH = EXAMPLE_DIR / "serve_qwen3_asr_ws.py"
|
|
NOTE_PATHS = [
|
|
EXAMPLE_DIR / "serve_qwen3_asr_ws_notes.md",
|
|
EXAMPLE_DIR / "serve_qwen3_asr_ws_notes_en.md",
|
|
]
|
|
|
|
|
|
class Qwen3AsrWebsocketExampleTest(unittest.TestCase):
|
|
def test_notes_do_not_contain_review_placeholders(self):
|
|
forbidden_phrases = ["以下是我的个人理解", "The following is my personal understanding"]
|
|
for note_path in NOTE_PATHS:
|
|
text = note_path.read_text(encoding="utf-8")
|
|
for phrase in forbidden_phrases:
|
|
with self.subTest(path=note_path.name, phrase=phrase):
|
|
self.assertNotIn(phrase, text)
|
|
|
|
def test_handler_logs_unexpected_exceptions(self):
|
|
tree = ast.parse(SERVER_PATH.read_text(encoding="utf-8"), filename=str(SERVER_PATH))
|
|
handle_client = next(
|
|
node
|
|
for node in tree.body
|
|
if isinstance(node, ast.AsyncFunctionDef) and node.name == "handle_client"
|
|
)
|
|
|
|
catches_generic_exception = False
|
|
logs_exception = False
|
|
for node in ast.walk(handle_client):
|
|
if not isinstance(node, ast.ExceptHandler):
|
|
continue
|
|
if isinstance(node.type, ast.Name) and node.type.id == "Exception":
|
|
catches_generic_exception = True
|
|
for child in ast.walk(node):
|
|
if not isinstance(child, ast.Call):
|
|
continue
|
|
func = child.func
|
|
if (
|
|
isinstance(func, ast.Attribute)
|
|
and isinstance(func.value, ast.Name)
|
|
and func.value.id == "logging"
|
|
and func.attr == "exception"
|
|
):
|
|
logs_exception = True
|
|
|
|
self.assertTrue(catches_generic_exception)
|
|
self.assertTrue(logs_exception)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|