fix: enforce account data boundaries and real-data integrity
This commit is contained in:
leefer
@@ -0,0 +1,193 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import sqlite3
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
from database import ReviewDatabase
|
||||
from server import DashboardService, RequestHandler
|
||||
|
||||
|
||||
FORMULA = {"all": [{"field": "change", "operator": ">", "value": 0}]}
|
||||
|
||||
|
||||
class AccountDataBoundaryTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.temp = tempfile.TemporaryDirectory()
|
||||
self.database = ReviewDatabase(Path(self.temp.name) / "review.db")
|
||||
self.first = self.database.create_user("first_user", "salt", "hash")
|
||||
self.second = self.database.create_user("second_user", "salt", "hash")
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.temp.cleanup()
|
||||
|
||||
def test_custom_strategies_are_private_and_builtins_are_shared(self):
|
||||
builtin_id = self.database.save_screener_strategy(
|
||||
None, "共享策略", "", ["repair"], FORMULA, builtin=True
|
||||
)
|
||||
private_id = self.database.save_screener_strategy(
|
||||
self.first["id"], "甲的策略", "", ["repair"], FORMULA
|
||||
)
|
||||
|
||||
first_names = {item["name"] for item in self.database.list_screener_strategies(self.first["id"])}
|
||||
second_names = {item["name"] for item in self.database.list_screener_strategies(self.second["id"])}
|
||||
self.assertEqual(first_names, {"共享策略", "甲的策略"})
|
||||
self.assertEqual(second_names, {"共享策略"})
|
||||
with self.assertRaises(ValueError):
|
||||
self.database.delete_screener_strategy(self.second["id"], private_id)
|
||||
with self.assertRaises(ValueError):
|
||||
self.database.delete_screener_strategy(self.first["id"], builtin_id)
|
||||
self.assertTrue(self.database.delete_screener_strategy(self.first["id"], private_id))
|
||||
|
||||
def test_screener_runs_are_private(self):
|
||||
self.database.save_screener_run(
|
||||
self.first["id"], "20260721", "repair", "甲的策略", FORMULA,
|
||||
{"candidates": [{"code": "002141"}], "meta": {}},
|
||||
)
|
||||
self.assertEqual(
|
||||
self.database.latest_screener_run(self.first["id"], "20260722")["candidates"][0]["code"],
|
||||
"002141",
|
||||
)
|
||||
self.assertIsNone(self.database.latest_screener_run(self.second["id"], "20260722"))
|
||||
|
||||
def test_mentor_messages_are_scoped_by_user_mentor_and_date(self):
|
||||
self.database.save_mentor_exchange(
|
||||
self.first["id"], "mentor-a", "20260721", "怎么看？", "先看承接。", "20260721"
|
||||
)
|
||||
self.assertEqual(
|
||||
[item["role"] for item in self.database.list_mentor_messages(
|
||||
self.first["id"], "mentor-a", "20260721"
|
||||
)],
|
||||
["user", "assistant"],
|
||||
)
|
||||
self.assertEqual(
|
||||
self.database.list_mentor_messages(self.second["id"], "mentor-a", "20260721"), []
|
||||
)
|
||||
self.assertEqual(
|
||||
self.database.list_mentor_messages(self.first["id"], "mentor-b", "20260721"), []
|
||||
)
|
||||
self.assertEqual(
|
||||
self.database.list_mentor_messages(self.first["id"], "mentor-a", "20260722"), []
|
||||
)
|
||||
self.assertEqual(
|
||||
self.database.delete_mentor_messages(self.second["id"], "mentor-a", "20260721"), 0
|
||||
)
|
||||
self.assertEqual(
|
||||
self.database.delete_mentor_messages(self.first["id"], "mentor-a", "20260721"), 2
|
||||
)
|
||||
|
||||
def test_latest_data_snapshot_skips_demo_and_future_records(self):
|
||||
self.database.save_data_snapshot(
|
||||
"stock_detail", "002141:20260718", "tushare", {"marker": "real"}
|
||||
)
|
||||
self.database.save_data_snapshot(
|
||||
"stock_detail", "002141:20260719", "demo", {"marker": "demo"}
|
||||
)
|
||||
self.database.save_data_snapshot(
|
||||
"stock_detail", "002141:20260723", "tushare", {"marker": "future"}
|
||||
)
|
||||
payload = self.database.get_latest_data_snapshot(
|
||||
"stock_detail", "002141:", "002141:20260722", exclude_source="demo"
|
||||
)
|
||||
self.assertEqual(payload["marker"], "real")
|
||||
|
||||
def test_stock_detail_never_falls_back_to_demo_data(self):
|
||||
self.database.save_data_snapshot(
|
||||
"stock_detail",
|
||||
"002141:20260721",
|
||||
"demo",
|
||||
{"meta": {"source": "demo"}, "stock": {"code": "002141"}, "prices": [{}]},
|
||||
)
|
||||
service = DashboardService.__new__(DashboardService)
|
||||
service.database = self.database
|
||||
service._system_credentials = {"tushare_token": ""}
|
||||
service._request_context = SimpleNamespace(user_id=self.first["id"])
|
||||
|
||||
with self.assertRaisesRegex(ValueError, "暂无 002141 的真实行情数据"):
|
||||
service.get_stock_detail("002141", "2026-07-22")
|
||||
|
||||
|
||||
class LegacyStrategyMigrationTests(unittest.TestCase):
|
||||
def test_legacy_custom_strategy_and_run_move_to_first_admin(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
database_path = Path(directory) / "legacy.db"
|
||||
connection = sqlite3.connect(database_path)
|
||||
try:
|
||||
connection.executescript(
|
||||
"""
|
||||
CREATE TABLE users (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
username TEXT NOT NULL UNIQUE,
|
||||
password_salt TEXT NOT NULL,
|
||||
password_hash TEXT NOT NULL,
|
||||
created_at TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL
|
||||
);
|
||||
INSERT INTO users VALUES (1, 'legacy_admin', 'salt', 'hash', '2026-01-01', '2026-01-01');
|
||||
CREATE TABLE screener_strategies (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
name TEXT NOT NULL,
|
||||
description TEXT NOT NULL DEFAULT '',
|
||||
regimes TEXT NOT NULL,
|
||||
formula TEXT NOT NULL,
|
||||
builtin INTEGER NOT NULL DEFAULT 0,
|
||||
created_at TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL
|
||||
);
|
||||
INSERT INTO screener_strategies VALUES
|
||||
(1, '旧策略', '', '["repair"]', '{"all":[]}', 0, '2026-01-01', '2026-01-01'),
|
||||
(2, '旧内置', '', '["repair"]', '{"all":[]}', 1, '2026-01-01', '2026-01-01');
|
||||
CREATE TABLE screener_runs (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
trade_date TEXT NOT NULL,
|
||||
regime TEXT NOT NULL,
|
||||
strategy_name TEXT NOT NULL,
|
||||
formula TEXT NOT NULL,
|
||||
result TEXT NOT NULL,
|
||||
created_at TEXT NOT NULL
|
||||
);
|
||||
INSERT INTO screener_runs VALUES
|
||||
(1, '20260721', 'repair', '旧策略', '{}', '{"meta":{}}', '2026-07-21');
|
||||
"""
|
||||
)
|
||||
connection.commit()
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
migrated = ReviewDatabase(database_path)
|
||||
strategies = migrated.list_screener_strategies(1)
|
||||
owners = {item["name"]: item["user_id"] for item in strategies}
|
||||
self.assertEqual(owners["旧策略"], 1)
|
||||
self.assertIsNone(owners["旧内置"])
|
||||
self.assertIsNotNone(migrated.latest_screener_run(1, "20260722"))
|
||||
|
||||
|
||||
class PublicKnowledgePermissionTests(unittest.TestCase):
|
||||
@staticmethod
|
||||
def handler(path: str, method_name: str):
|
||||
handler = RequestHandler.__new__(RequestHandler)
|
||||
handler.path = path
|
||||
handler.require_auth = lambda: True
|
||||
handler.require_csrf = lambda: True
|
||||
handler.require_admin = lambda: False
|
||||
handler.require_member = lambda: True
|
||||
setattr(handler, method_name, lambda: (_ for _ in ()).throw(AssertionError("mutation ran")))
|
||||
return handler
|
||||
|
||||
def test_regular_user_cannot_change_reason_or_seat_alias(self):
|
||||
RequestHandler.do_POST(self.handler("/api/reasons", "save_reason"))
|
||||
RequestHandler.do_POST(self.handler("/api/seat-aliases", "save_seat_alias"))
|
||||
|
||||
def test_regular_user_cannot_change_sector_phase(self):
|
||||
RequestHandler.do_POST(
|
||||
self.handler("/api/heaven/sector-phases", "save_sector_phase_override")
|
||||
)
|
||||
RequestHandler.do_DELETE(
|
||||
self.handler("/api/heaven/sector-phases/%E6%B2%B9%E6%B0%94", "send_json")
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user