pixel/server/test_world_policy.py

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,
)


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