166 lines
7.2 KiB
Python
166 lines
7.2 KiB
Python
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()
|