pixel/server/test_publish_api.py

import json
import os
import shutil
import socket
import subprocess
import tempfile
import time
import unittest
import urllib.error
import urllib.request
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]


def free_port():
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.bind(("127.0.0.1", 0))
        return sock.getsockname()[1]


def asset(asset_id, owner="acct-a"):
    return {
        "id": asset_id,
        "name": "Test Block",
        "category": "static",
        "subtype": "other",
        "width": 1,
        "height": 1,
        "size": 1,
        "pixels": "a",
        "ownerAccountId": owner,
    }


def publish_command(command_id, asset_id="asset-a", object_id="object-a"):
    return {
        "id": command_id,
        "type": "publish.asset_object",
        "kind": "static",
        "asset": asset(asset_id),
        "object": {
            "id": object_id,
            "assetId": asset_id,
            "x": 3,
            "y": 4,
            "publishedAt": 1,
            "version": 1,
        },
    }


@unittest.skipIf(shutil.which("php") is None, "PHP CLI is not installed")
class PublishApiTest(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.port = free_port()
        env = os.environ.copy()
        env["PIXEL_ISLAND_DB_PATH"] = str(Path(self.temp.name) / "pixel.sqlite")
        self.proc = subprocess.Popen(
            ["php", "-S", f"127.0.0.1:{self.port}", "-t", str(ROOT)],
            cwd=str(ROOT),
            env=env,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
        self.base = f"http://127.0.0.1:{self.port}/api/index.php"
        deadline = time.time() + 5
        while time.time() < deadline:
            try:
                self.get("health")
                return
            except Exception:
                time.sleep(0.05)
        self.fail("PHP test server did not start")

    def tearDown(self):
        self.proc.terminate()
        self.proc.wait(timeout=5)
        self.temp.cleanup()

    def get(self, action):
        with urllib.request.urlopen(f"{self.base}?action={action}", timeout=5) as response:
            return json.loads(response.read().decode("utf-8"))

    def post_commands(self, commands, account_id="acct-a", password="pw"):
        payload = {
            "account": {"id": account_id, "name": account_id, "password": password},
            "commands": commands,
        }
        request = urllib.request.Request(
            f"{self.base}?action=commands",
            data=json.dumps(payload).encode("utf-8"),
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        with urllib.request.urlopen(request, timeout=5) as response:
            return json.loads(response.read().decode("utf-8"))

    def test_publish_asset_object_is_atomic_and_server_timestamped(self):
        data = self.post_commands([publish_command("cmd-one")])
        self.assertEqual(data["appliedCommandIds"], ["cmd-one"])
        self.assertEqual(data["rejectedCommands"], [])
        placed = data["snapshot"]["placed"]
        self.assertEqual(len(placed), 1)
        self.assertEqual(placed[0]["assetId"], "asset-a")
        self.assertNotEqual(placed[0]["publishedAt"], 1)
        self.assertEqual(data["snapshot"]["assets"][0]["ownerAccountId"], "acct-a")

    def test_duplicate_publish_command_replays_without_new_object(self):
        self.post_commands([publish_command("cmd-dupe")])
        data = self.post_commands([publish_command("cmd-dupe")])
        self.assertEqual(data["appliedCommandIds"], ["cmd-dupe"])
        self.assertEqual(len(data["snapshot"]["placed"]), 1)

    def test_owner_rejection_is_reported_in_batch(self):
        self.post_commands([publish_command("cmd-owner", "shared-asset", "owner-object")])
        stolen = publish_command("cmd-stolen", "shared-asset", "stolen-object")
        data = self.post_commands([stolen], account_id="acct-b", password="pw")
        self.assertEqual(data["appliedCommandIds"], [])
        self.assertEqual(data["rejectedCommands"][0]["error"], "asset_id_already_owned")

    def test_quota_rejection_does_not_abort_response(self):
        commands = [publish_command(f"cmd-{i}", f"asset-{i}", f"object-{i}") for i in range(6)]
        data = self.post_commands(commands)
        self.assertEqual(len(data["appliedCommandIds"]), 5)
        self.assertEqual(data["rejectedCommands"][0]["error"], "publish_limit_reached")


if __name__ == "__main__":
    unittest.main()