260 lines
9.9 KiB
Python
260 lines
9.9 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import importlib.util
|
|
import json
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
MODULE_PATH = PROJECT_ROOT / "jwebot_release_update.py"
|
|
SPEC = importlib.util.spec_from_file_location("jwebot_release_update", MODULE_PATH)
|
|
MODULE = importlib.util.module_from_spec(SPEC)
|
|
assert SPEC.loader is not None
|
|
sys.modules[SPEC.name] = MODULE
|
|
SPEC.loader.exec_module(MODULE)
|
|
|
|
|
|
class ReleaseUpdateTests(unittest.TestCase):
|
|
release = "beta 1.2.3.1019"
|
|
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temporary.name)
|
|
self.database = self.root / "edge.sqlite3"
|
|
self.backups = self.root / "backups"
|
|
self.version = self.root / "VERSION"
|
|
self.changelog = self.root / "CHANGELOG.md"
|
|
self.delivery_helper = self.root / "jwebot_edge_delivery.py"
|
|
self.version.write_text(self.release + "\n", encoding="utf-8")
|
|
self.changelog.write_text(
|
|
(
|
|
"# Changelog\n\n"
|
|
f"## {self.release} - 2026-10-16\n\n"
|
|
"- 更新摩拉账户并修复重复回复问题。\n"
|
|
"- Internal implementation detail.\n"
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
self.delivery_helper.write_text("# test helper\n", encoding="utf-8")
|
|
connection = sqlite3.connect(self.database)
|
|
connection.executescript(
|
|
"""
|
|
CREATE TABLE metadata (
|
|
key TEXT PRIMARY KEY,
|
|
value TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
);
|
|
CREATE TABLE player_accounts (
|
|
group_fingerprint TEXT NOT NULL,
|
|
person_key TEXT NOT NULL,
|
|
state_json TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL,
|
|
PRIMARY KEY (group_fingerprint, person_key)
|
|
);
|
|
"""
|
|
)
|
|
for person_key, balance in (("person-a", 1000), ("person-b", 50)):
|
|
connection.execute(
|
|
"INSERT INTO player_accounts VALUES (?, ?, ?, ?)",
|
|
(
|
|
"group-a",
|
|
person_key,
|
|
json.dumps({"balance": balance}),
|
|
"2026-10-16T00:00:00Z",
|
|
),
|
|
)
|
|
connection.commit()
|
|
connection.close()
|
|
|
|
def tearDown(self):
|
|
self.temporary.cleanup()
|
|
|
|
def account_states(self):
|
|
connection = sqlite3.connect(self.database)
|
|
try:
|
|
rows = connection.execute(
|
|
"SELECT state_json FROM player_accounts ORDER BY person_key"
|
|
).fetchall()
|
|
finally:
|
|
connection.close()
|
|
return [json.loads(row[0]) for row in rows]
|
|
|
|
def announcement_marker(self):
|
|
key = MODULE.announcement_key(self.release, "group-a")
|
|
connection = sqlite3.connect(self.database)
|
|
try:
|
|
row = connection.execute(
|
|
"SELECT value FROM metadata WHERE key = ?",
|
|
(key,),
|
|
).fetchone()
|
|
finally:
|
|
connection.close()
|
|
return json.loads(row[0]) if row else None
|
|
|
|
def test_changelog_first_bullet_is_the_announcement_summary(self):
|
|
self.changelog.write_text(
|
|
(
|
|
"# Changelog\n\n"
|
|
f"## {self.release} - 2026-10-16\n\n"
|
|
"- 第一行更新摘要，\n"
|
|
" 同一条摘要可以换行。\n"
|
|
"- 第二条不进入群公告。\n"
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
summary = MODULE.read_changelog_summary(self.changelog, self.release)
|
|
self.assertEqual(summary, "第一行更新摘要， 同一条摘要可以换行。")
|
|
announcement = MODULE.build_announcement(self.release, summary)
|
|
self.assertIn(f"{self.release} 更新现已完成", announcement)
|
|
self.assertIn("更新补偿：摩拉 x200", announcement)
|
|
self.assertNotIn("原石 x1", announcement)
|
|
self.assertNotIn("第二条", announcement)
|
|
|
|
def test_policy_release_announces_primogem(self):
|
|
for release in (
|
|
"beta 1.0.5.1016",
|
|
"beta 1.1.0.1016",
|
|
"beta 1.2.0.1019",
|
|
):
|
|
self.assertEqual(MODULE.primogem_compensation_amount(release), 1)
|
|
announcement = MODULE.build_announcement(release, "原石系统上线。")
|
|
self.assertIn("更新补偿：摩拉 x200，原石 x1", announcement)
|
|
|
|
def test_patch_release_without_policy_does_not_announce_primogem(self):
|
|
release = "beta 1.1.2.1019"
|
|
self.assertEqual(MODULE.primogem_compensation_amount(release), 0)
|
|
announcement = MODULE.build_announcement(release, "普通小版本更新。")
|
|
self.assertIn("更新补偿：摩拉 x200", announcement)
|
|
self.assertNotIn("原石 x1", announcement)
|
|
|
|
def test_lifecycle_stage_sends_guarded_notice_without_compensation(self):
|
|
delivered = []
|
|
|
|
def deliverer(path, group, message, event_id):
|
|
delivered.append((path, group, message, event_id))
|
|
|
|
result = MODULE.run_release_lifecycle(
|
|
release_file=self.version,
|
|
database_path=self.database,
|
|
delivery_helper=self.delivery_helper,
|
|
stage="shutdown",
|
|
deliverer=deliverer,
|
|
)
|
|
|
|
self.assertEqual(result["stage"], "shutdown")
|
|
self.assertEqual(result["delivered"], 1)
|
|
self.assertEqual(delivered[0][1], "group-a")
|
|
self.assertIn("准备停机更新", delivered[0][2])
|
|
self.assertIn(self.release, delivered[0][2])
|
|
self.assertIn("shutdown", delivered[0][3])
|
|
self.assertEqual([state["balance"] for state in self.account_states()], [1000, 50])
|
|
|
|
def test_release_update_grants_once_and_announces_once(self):
|
|
delivered = []
|
|
|
|
def deliverer(path, group, announcement, event_id):
|
|
delivered.append((path, group, announcement, event_id))
|
|
|
|
applied_at = datetime(2026, 10, 16, 8, 0, tzinfo=timezone.utc)
|
|
first = MODULE.run_release_update(
|
|
release_file=self.version,
|
|
changelog=self.changelog,
|
|
database_path=self.database,
|
|
backup_dir=self.backups,
|
|
delivery_helper=self.delivery_helper,
|
|
deliverer=deliverer,
|
|
applied_at=applied_at,
|
|
)
|
|
second = MODULE.run_release_update(
|
|
release_file=self.version,
|
|
changelog=self.changelog,
|
|
database_path=self.database,
|
|
backup_dir=self.backups,
|
|
delivery_helper=self.delivery_helper,
|
|
deliverer=deliverer,
|
|
applied_at=applied_at,
|
|
)
|
|
|
|
self.assertEqual(first["compensation"], "applied")
|
|
self.assertEqual(second["compensation"], "already_applied")
|
|
self.assertEqual(first["accounts"], 2)
|
|
self.assertEqual(len(delivered), 1)
|
|
self.assertEqual(delivered[0][1], "group-a")
|
|
self.assertIn("更新摩拉账户并修复重复回复问题", delivered[0][2])
|
|
self.assertEqual([state["balance"] for state in self.account_states()], [1200, 250])
|
|
for state in self.account_states():
|
|
self.assertEqual(state["primogems"], 0)
|
|
matches = [
|
|
event
|
|
for event in state["coin_events"]
|
|
if event["delta"] == 200
|
|
and event["reason"] == f"{self.release} 版本更新补偿"
|
|
]
|
|
self.assertEqual(len(matches), 1)
|
|
self.assertEqual(matches[0]["created_at"], applied_at.timestamp())
|
|
primogem_matches = [
|
|
event
|
|
for event in state["primogem_events"]
|
|
if event["reason"] == f"{self.release} 版本更新补偿"
|
|
]
|
|
self.assertEqual(primogem_matches, [])
|
|
self.assertEqual(self.announcement_marker()["state"], "delivered")
|
|
self.assertEqual(len(list(self.backups.glob("*.sqlite3"))), 1)
|
|
|
|
def test_uncertain_announcement_never_duplicates_compensation(self):
|
|
def uncertain(*_args):
|
|
raise MODULE.AnnouncementUncertainError("uncertain")
|
|
|
|
with self.assertRaises(MODULE.AnnouncementUncertainError):
|
|
MODULE.run_release_update(
|
|
release_file=self.version,
|
|
changelog=self.changelog,
|
|
database_path=self.database,
|
|
backup_dir=self.backups,
|
|
delivery_helper=self.delivery_helper,
|
|
deliverer=uncertain,
|
|
)
|
|
self.assertEqual([state["balance"] for state in self.account_states()], [1200, 250])
|
|
self.assertEqual([state["primogems"] for state in self.account_states()], [0, 0])
|
|
self.assertEqual(self.announcement_marker()["state"], "uncertain")
|
|
|
|
called = []
|
|
|
|
def delivered(*args):
|
|
called.append(args)
|
|
|
|
with self.assertRaises(MODULE.AnnouncementUncertainError):
|
|
MODULE.run_release_update(
|
|
release_file=self.version,
|
|
changelog=self.changelog,
|
|
database_path=self.database,
|
|
backup_dir=self.backups,
|
|
delivery_helper=self.delivery_helper,
|
|
deliverer=delivered,
|
|
)
|
|
self.assertFalse(called)
|
|
|
|
result = MODULE.run_release_update(
|
|
release_file=self.version,
|
|
changelog=self.changelog,
|
|
database_path=self.database,
|
|
backup_dir=self.backups,
|
|
delivery_helper=self.delivery_helper,
|
|
deliverer=delivered,
|
|
retry_uncertain=True,
|
|
)
|
|
self.assertEqual(result["compensation"], "already_applied")
|
|
self.assertEqual(len(called), 1)
|
|
self.assertEqual([state["balance"] for state in self.account_states()], [1200, 250])
|
|
self.assertEqual([state["primogems"] for state in self.account_states()], [0, 0])
|
|
self.assertEqual(self.announcement_marker()["state"], "delivered")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|