"""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()