103 lines
3.6 KiB
Python
103 lines
3.6 KiB
Python
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from app.bettergi_current_party import (
|
|
CURRENT_PARTY_SCRIPT_NAME,
|
|
CurrentPartyReadError,
|
|
clear_current_party_status,
|
|
prepare_current_party_read,
|
|
read_current_party_status,
|
|
)
|
|
|
|
|
|
class BetterGICurrentPartyTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp_dir = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temp_dir.name)
|
|
self.group_dir = self.root / "User" / "ScriptGroup"
|
|
self._write_json(
|
|
self.group_dir / "切换队伍.json",
|
|
{
|
|
"index": 2,
|
|
"name": "切换队伍",
|
|
"config": {"marker": "preserved"},
|
|
"projects": [{"name": "旧项目"}],
|
|
},
|
|
)
|
|
self._write_json(
|
|
self.group_dir / "其他.json",
|
|
{"index": 5, "name": "其他", "config": {}, "projects": []},
|
|
)
|
|
|
|
def tearDown(self):
|
|
self.temp_dir.cleanup()
|
|
|
|
@staticmethod
|
|
def _write_json(path: Path, data):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
def test_prepare_reader_group_clones_config_and_uses_request_id(self):
|
|
prepared = prepare_current_party_read(self.root, "request-123")
|
|
|
|
update = prepared.updates[0]
|
|
self.assertEqual(update.data["index"], 6)
|
|
self.assertEqual(update.data["config"]["marker"], "preserved")
|
|
self.assertEqual(update.data["name"], "直播系统读取当前队伍")
|
|
project = update.data["projects"][0]
|
|
self.assertEqual(project["folderName"], CURRENT_PARTY_SCRIPT_NAME)
|
|
self.assertEqual(project["jsScriptSettingsObject"]["requestId"], "request-123")
|
|
self.assertEqual(
|
|
prepared.status_path,
|
|
self.root / "User" / "JsScript" / CURRENT_PARTY_SCRIPT_NAME / "status.json",
|
|
)
|
|
|
|
def test_status_reader_ignores_stale_and_running_results(self):
|
|
prepared = prepare_current_party_read(self.root, "request-123")
|
|
self._write_json(
|
|
prepared.status_path,
|
|
{"state": "success", "request_id": "old", "party_name": "旧队伍"},
|
|
)
|
|
self.assertIsNone(read_current_party_status(prepared.status_path, "request-123"))
|
|
|
|
self._write_json(
|
|
prepared.status_path,
|
|
{"state": "running", "request_id": "request-123"},
|
|
)
|
|
self.assertIsNone(read_current_party_status(prepared.status_path, "request-123"))
|
|
|
|
self._write_json(
|
|
prepared.status_path,
|
|
{
|
|
"state": "success",
|
|
"request_id": "request-123",
|
|
"party_name": "好感队",
|
|
"candidates": ["好感队"],
|
|
},
|
|
)
|
|
result = read_current_party_status(prepared.status_path, "request-123")
|
|
self.assertEqual(result.party_name, "好感队")
|
|
self.assertEqual(result.candidates, ("好感队",))
|
|
|
|
def test_status_reader_reports_script_error_and_can_be_cleared(self):
|
|
prepared = prepare_current_party_read(self.root, "request-123")
|
|
self._write_json(
|
|
prepared.status_path,
|
|
{
|
|
"state": "error",
|
|
"request_id": "request-123",
|
|
"message": "识别到多个候选",
|
|
},
|
|
)
|
|
with self.assertRaisesRegex(CurrentPartyReadError, "多个候选"):
|
|
read_current_party_status(prepared.status_path, "request-123")
|
|
|
|
clear_current_party_status(prepared.status_path)
|
|
self.assertFalse(prepared.status_path.exists())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|