import unittest import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) from world_policy import ( ACTIVE, VIOLATION_HIDDEN, apply_command, apply_server_event, can_delete_asset, can_import_full_state, can_modify_object, make_dynamic_move_event, make_world_phase_event, ) def base_state(): return { "worldMode": "shared", "assets": [ {"id": "asset-a", "ownerAccountId": "alice", "author": "Alice", "version": 1}, {"id": "asset-b", "ownerAccountId": "bob", "author": "Bob", "version": 1}, ], "placed": [ {"id": "obj-a", "assetId": "asset-a", "ownerAccountId": "alice", "x": 1, "y": 1, "version": 1, "status": ACTIVE} ], "dynamicSummons": [ {"id": "dyn-b", "assetId": "asset-b", "ownerAccountId": "bob", "homeX": 2, "homeY": 2, "version": 1, "status": ACTIVE} ], "objectVotes": {}, "assetVotes": {}, "hiddenObjects": {}, "hiddenAssets": {}, "moderationReports": [], "tombstones": {"assets": {}, "objects": {}}, "serverSync": {}, } class WorldPolicyTest(unittest.TestCase): def test_non_owner_cannot_delete_static_object(self): state = base_state() allowed, meta = can_modify_object(state, {"id": "bob"}, "obj-a") self.assertFalse(allowed) self.assertEqual(meta["reason"], "not_object_owner") result = apply_command(state, {"id": "bob"}, {"type": "object.delete", "kind": "static", "objectId": "obj-a"}, at=100) self.assertFalse(result["accepted"]) self.assertEqual(result["reason"], "not_object_owner") self.assertEqual(len(state["placed"]), 1) def test_non_owner_cannot_move_dynamic_object(self): state = base_state() command = { "type": "object.move", "kind": "dynamic", "object": {"id": "dyn-b", "assetId": "asset-b", "homeX": 9, "homeY": 9, "version": 1}, } result = apply_command(state, {"id": "alice"}, command, at=100) self.assertFalse(result["accepted"]) self.assertEqual(result["reason"], "not_object_owner") def test_owner_can_move_and_delete_own_object(self): state = base_state() move = apply_command( state, {"id": "bob"}, {"type": "object.move", "kind": "dynamic", "object": {"id": "dyn-b", "assetId": "asset-b", "homeX": 9, "homeY": 9, "version": 1}}, at=100, ) self.assertTrue(move["accepted"]) self.assertTrue(apply_server_event(state, move["event"])) self.assertEqual(state["dynamicSummons"][0]["homeX"], 9) self.assertEqual(state["dynamicSummons"][0]["ownerAccountId"], "bob") delete = apply_command(state, {"id": "bob"}, {"type": "object.delete", "kind": "dynamic", "objectId": "dyn-b"}, at=200) self.assertTrue(delete["accepted"]) self.assertTrue(apply_server_event(state, delete["event"])) self.assertEqual(state["dynamicSummons"], []) self.assertIn("dyn-b", state["tombstones"]["objects"]) def test_admin_can_violation_hide_any_object(self): state = base_state() result = apply_command(state, {"id": "mod", "admin": True}, {"type": "admin.hide_violation", "objectId": "obj-a"}, at=100) self.assertTrue(result["accepted"]) self.assertTrue(apply_server_event(state, result["event"])) self.assertEqual(state["placed"][0]["status"], VIOLATION_HIDDEN) self.assertTrue(state["placed"][0]["permanentHidden"]) def test_deleted_object_cannot_be_resurrected_by_stale_upsert(self): state = base_state() delete = apply_command(state, {"id": "alice"}, {"type": "object.delete", "kind": "static", "objectId": "obj-a"}, at=100) self.assertTrue(apply_server_event(state, delete["event"])) stale = { "serverEventId": "sev_90_object_upsert", "serverAt": 90, "actorAccountId": "alice", "type": "object.upsert", "kind": "static", "object": {"id": "obj-a", "assetId": "asset-a", "ownerAccountId": "alice", "x": 7, "y": 7, "version": 1}, "objectVersion": 1, } self.assertFalse(apply_server_event(state, stale)) self.assertEqual(state["placed"], []) def test_asset_delete_cascades_and_tombstones_owned_objects(self): state = base_state() allowed, _ = can_delete_asset(state, {"id": "alice"}, "asset-a") self.assertTrue(allowed) result = apply_command(state, {"id": "alice"}, {"type": "asset.delete", "assetId": "asset-a"}, at=100) self.assertTrue(result["accepted"]) self.assertTrue(apply_server_event(state, result["event"])) self.assertNotIn("asset-a", [asset["id"] for asset in state["assets"]]) self.assertEqual(state["placed"], []) self.assertIn("asset-a", state["tombstones"]["assets"]) self.assertIn("obj-a", state["tombstones"]["objects"]) self.assertEqual(len(state["dynamicSummons"]), 1) def test_non_owner_can_delete_asset_and_all_placements(self): state = base_state() allowed, _ = can_delete_asset(state, {"id": "alice"}, "asset-b") self.assertTrue(allowed) result = apply_command(state, {"id": "alice"}, {"type": "asset.delete", "assetId": "asset-b"}, at=100) self.assertTrue(result["accepted"]) self.assertTrue(apply_server_event(state, result["event"])) self.assertNotIn("asset-b", [asset["id"] for asset in state["assets"]]) self.assertEqual(state["dynamicSummons"], []) self.assertIn("asset-b", state["tombstones"]["assets"]) self.assertIn("dyn-b", state["tombstones"]["objects"]) def test_world_phase_event_updates_authoritative_clock(self): state = base_state() event = make_world_phase_event(at=600_000) self.assertTrue(apply_server_event(state, event)) self.assertEqual(state["serverSync"]["clock"]["worldTimeMs"], 600_000) self.assertEqual(state["serverSync"]["clock"]["dayMs"], 600_000) self.assertEqual(state["serverSync"]["lastServerEventId"], event["serverEventId"]) def test_dynamic_move_event_is_server_authoritative(self): state = base_state() event = make_dynamic_move_event(state, "dyn-b", x=4.5, y=7.0, target_x=8.0, target_y=9.0, at=700, actor="server") self.assertIsNotNone(event) self.assertEqual(event["type"], "dynamic.move") self.assertTrue(apply_server_event(state, event)) target = state["serverSync"]["dynamicTargets"]["dyn-b"] self.assertEqual(target["targetX"], 8.0) self.assertEqual(target["targetY"], 9.0) self.assertEqual(state["dynamicSummons"][0]["serverState"]["x"], 4.5) self.assertEqual(state["dynamicSummons"][0]["version"], 2) def test_shared_full_state_import_is_rejected(self): state = base_state() allowed, meta = can_import_full_state(state, {"id": "alice"}) self.assertFalse(allowed) self.assertEqual(meta["reason"], "shared_full_import_disabled") local_state = {**state, "worldMode": "local"} allowed, _ = can_import_full_state(local_state, {"id": "alice"}) self.assertTrue(allowed) if __name__ == "__main__": unittest.main()