132 lines
4.5 KiB
Python
132 lines
4.5 KiB
Python
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()
|