53 lines
1.8 KiB
Python
53 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
class ServiceTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.tmp = tempfile.TemporaryDirectory()
|
|
os.environ["ASR_DATA_DIR"] = str(Path(cls.tmp.name) / "data")
|
|
os.environ["ASR_STATE_DIR"] = str(Path(cls.tmp.name) / "state")
|
|
from asrservice import config, db, worker
|
|
cls.config, cls.db, cls.worker = config, db, worker
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.tmp.cleanup()
|
|
|
|
def test_job_events_follow_updates(self):
|
|
row = self.db.create_job("a" * 24, "x.m4a", "x.m4a", source="ssh")
|
|
self.assertEqual(row["status"], "queued")
|
|
self.db.update_job(row["id"], "transcribing", "started", attempts=1)
|
|
current = self.db.get_job(row["id"])
|
|
self.assertEqual(current["attempts"], 1)
|
|
self.assertEqual([e["status"] for e in self.db.events(row["id"])],
|
|
["queued", "transcribing"])
|
|
|
|
def test_transcript_contract(self):
|
|
text = self.worker.transcript_from_response({
|
|
"segments": [
|
|
{"speaker": "speaker_0", "text": "你好。"},
|
|
{"speaker": "speaker_0", "text": "继续。"},
|
|
{"speaker": "speaker_1", "text": "收到。"},
|
|
],
|
|
"voiceprint": {"session_id": "s1", "matches": []},
|
|
})
|
|
self.assertIn("voiceprint_session: s1", text)
|
|
self.assertIn("【speaker_0】你好。继续。", text)
|
|
self.assertEqual(text.count("【speaker_0】"), 1)
|
|
|
|
def test_transcript_rejects_empty_segments(self):
|
|
with self.assertRaises(RuntimeError):
|
|
self.worker.transcript_from_response({"segments": []})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|