| """Framework-free contracts: transactions, validation and authoritative sessions.""" |
|
|
| import ast |
| import copy |
| import json |
| import threading |
| import unittest |
| from pathlib import Path |
| from unittest.mock import patch |
|
|
| from data.save_system import restore_save |
| from game.action_engine import action_choices |
| from game.service import allocate_stat, new_game |
| from game.turn_engine import TurnEvent, play_turn |
| from server.http import Assets |
| from server.sessions import Conflict, Sessions |
|
|
|
|
| class WebEngineTests(unittest.TestCase): |
| def setUp(self): |
| self.store = Sessions() |
| self.addCleanup(self.store.pool.shutdown) |
| self.token, self.session = self.store.get(None) |
|
|
| def create(self): |
| self.store.replace( |
| self.session, |
| { |
| "revision": self.session.revision, |
| "character": {"canonical_id": "luna_lovegood"}, |
| }, |
| ) |
| return self.session.state |
|
|
| def test_core_has_no_ui_framework_imports(self): |
| for folder in ("game", "systems", "data"): |
| for file in Path(folder).glob("*.py"): |
| tree = ast.parse(file.read_text()) |
| for node in ast.walk(tree): |
| names = ( |
| [n.name for n in node.names] |
| if isinstance(node, ast.Import) |
| else [node.module or ""] |
| if isinstance(node, ast.ImportFrom) |
| else [] |
| ) |
| self.assertFalse( |
| any( |
| n.split(".")[0] in ("gradio", "adapters", "server", "app") |
| for n in names |
| ), |
| file, |
| ) |
|
|
| def test_state_is_authoritative_and_sessions_are_isolated(self): |
| self.create() |
| token, other = self.store.get(None) |
| self.assertNotEqual(token, self.token) |
| self.assertIsNone(other.state) |
| self.assertIs(self.session, self.store.get(self.token)[1]) |
| view = self.store.snapshot(self.session, lambda _: None) |
| view["game"]["player"]["name"] = "Tampered" |
| self.assertEqual("Luna Lovegood", self.session.state["name"]) |
|
|
| def test_custom_creation_validation_and_allocation(self): |
| config = dict( |
| name="Rowan", |
| gender="Non-binary", |
| house="Ravenclaw", |
| year=3, |
| traits=["Brave", "Loyal", "Curious"], |
| aspirations=["Best Friend", "Spell Master"], |
| stats=dict(strength=7, agility=7, intelligence=7, charisma=7, willpower=7), |
| ) |
| state = new_game(config) |
| self.assertEqual(35, sum(state[k] for k in config["stats"])) |
| with self.assertRaises(ValueError): |
| new_game({**config, "traits": ["Brave"] * 3}) |
| with self.assertRaises(ValueError): |
| new_game({**config, "stats": {**config["stats"], "strength": 15}}) |
| with self.assertRaises(ValueError): |
| allocate_stat(state, "strength") |
| state["unspent_stat_points"] = 1 |
| allocated = allocate_stat(state, "strength") |
| self.assertEqual(8, allocated["strength"]) |
| self.assertEqual(7, state["strength"]) |
|
|
| def test_save_roundtrip_and_invalid_load_is_atomic(self): |
| state = self.create() |
| self.session.history = [["hello", "welcome"]] |
| save = self.store.save(self.session) |
| restored, history = restore_save(save) |
| self.assertEqual(state["relationships"], restored["relationships"]) |
| self.assertEqual(self.session.history, history) |
| self.assertNotEqual(state["id"], restored["id"]) |
| self.assertEqual(state["id"], save["state"]["id"]) |
| before = copy.deepcopy(state) |
| save["state"]["memory"] = [] |
| with self.assertRaises(ValueError): |
| self.store.replace(self.session, {"revision": 1, "save": save}, load=True) |
| self.assertEqual(before, self.session.state) |
|
|
| def test_future_version_and_malformed_nested_save_rejected(self): |
| self.create() |
| for bad in ({"version": 99}, {"state": {"name": "X", "year": "three"}}, []): |
| with self.assertRaises(ValueError): |
| restore_save(bad) |
| save = self.store.save(self.session) |
| save["state"]["relationships"] = {"harry_potter": "broken"} |
| with self.assertRaises(ValueError): |
| self.store.replace(self.session, {"revision": 1, "save": save}, load=True) |
|
|
| def test_missing_credentials_does_not_change_game(self): |
| state = copy.deepcopy(self.create()) |
| with self.assertRaisesRegex(ValueError, "HF_TOKEN"): |
| self.store.start_turn(self.session, {"revision": 1, "opening": True}) |
| self.assertEqual(state, self.session.state) |
| self.assertFalse(self.session.busy) |
|
|
| def test_busy_double_submit_load_and_stale_revision(self): |
| self.create() |
| entered, release = threading.Event(), threading.Event() |
| self.addCleanup(release.set) |
|
|
| def runner(*args, **kwargs): |
| state = json.loads(args[2]) |
| entered.set() |
| release.wait(5) |
| yield TurnEvent("complete", [["Begin", "Hello"]], state) |
|
|
| self.store.runner = runner |
| self.store.text_client = object() |
| self.store.start_turn(self.session, {"revision": 1, "opening": True}) |
| self.assertTrue(entered.wait(2)) |
| with self.assertRaises(Conflict): |
| self.store.start_turn(self.session, {"revision": 1, "opening": True}) |
| with self.assertRaises(Conflict): |
| self.store.replace( |
| self.session, |
| {"revision": 1, "character": {"canonical_id": "harry_potter"}}, |
| ) |
| with self.assertRaises(Conflict): |
| self.store.save(self.session) |
| release.set() |
| self.store.pool.shutdown(wait=True) |
| self.assertFalse(self.session.busy) |
| self.assertFalse(self.session.state.get("_opening_pending")) |
| with self.assertRaises(Conflict): |
| self.store.allocate(self.session, {"revision": 1, "stat": "strength"}) |
|
|
| def test_partial_stream_and_exception_never_commit_rewards(self): |
| state = self.create() |
| before = json.loads(json.dumps(state)) |
| with ( |
| patch( |
| "game.turn_engine.generate_llm_stream", |
| return_value=iter([("Hello", "Hello")]), |
| ), |
| patch( |
| "game.turn_engine.generate_scene_data", |
| side_effect=RuntimeError("failure"), |
| ), |
| patch("game.turn_engine.evaluate_alignment", return_value=(0, 0, 0, 0)), |
| ): |
| frames = list( |
| play_turn("Hello", [], json.dumps(state), None, None, "test", "test") |
| ) |
| self.assertEqual("text", frames[0].kind) |
| self.assertEqual(before, frames[0].state) |
| self.assertEqual("error", frames[-1].kind) |
| self.assertEqual(before, frames[-1].state) |
|
|
| def test_complete_engine_event_is_plain_data(self): |
| state = self.create() |
| from systems.interaction_system import enter_mode |
|
|
| enter_mode(state, "exploration", "main") |
| command = next( |
| c for label, c in action_choices(state) if c.endswith("open:travel") |
| ) |
| frames = list( |
| play_turn(command, [], json.dumps(state), None, None, "test", "test") |
| ) |
| self.assertEqual("complete", frames[-1].kind) |
| self.assertEqual(state["time"], frames[-1].state["time"]) |
| json.dumps(frames[-1].__dict__) |
|
|
| def test_assets_reject_external_paths_and_lfs_pointers(self): |
| assets = Assets() |
| self.assertIsNone(assets.url("/etc/passwd")) |
| self.assertIsNone(assets.url("data/../prompt/default_system.txt")) |
| self.assertIsNone(assets.url("https://example.com/image.png")) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
