horizon-airdef/tools/lan_e2e.py

#!/usr/bin/env python3
"""LAN multiplayer end to end: a host opens its world to LAN and a guest joins, both real games.

Two Horizon clients run side by side under horizon-ctl, each in its own scratch folder. The host
creates a survival world and opens it to LAN with commands allowed; the guest connects to it by
address. The script then checks, from both sides, what a guest's game used to get wrong (#366):

- tab:       the guest's player list holds the host and the guest, under their own names;
- self:      the guest's own model exists (third person, inventory preview);
- armour:    each side draws the other player in that player's own armour;
- nametags:  each side draws the other player's name above them;
- chest:     a chest the guest opens opens on the guest's screen, its lid opens on both sides and
             closes on both when the guest closes it; the host keeps playing with its mouse look (#687);
- crafting:  the guest crafts planks in the inventory's 2x2 grid and opens a crafting table, which
             leaves the host's screen and mouse look alone;
- sign:      a sign the guest places doesn't open a sign editor on the host;
- damage:    a guest in survival takes fall damage;
- mobs:      with the host in creative, a zombie and a skeleton go for the guest and hurt them;
- hazards:   a guest in survival gets hungry sprint-jumping and is hurt by lava, a block at the eyes,
             drowning and the void (then respawns);
- pvp:       the guest and the host hit each other with a sword; with the pvp game rule off, nothing;
- respawn:   a player killed is drawn dead by the other side, and alive again once it respawns (the
             respawned player is a new entity to the others); the host dies, and on the server thread
             the guest too;
- containers: the guest's ender chest holds the guest's items, not the host's, and a double chest
             opens as 9x6 with the right half first;
- machines:  a hopper, a dispenser and a dropper open on the guest's screen with their 5 and 9
             slots, and keep what the guest shift-clicks into them (#848);
- feedback:  the guest's command feedback names the guest and shows the numbers;
- chat:      a line the host sends reaches the guest's event log (`wait-for chat`), and on the
             server thread the host window's too;
- effects:   a beacon's power reaches every player in range, the host and the guest;
- distance:  (--server-thread) the server follows the host's render distance, past vanilla's 32 too
             (IntegratedServer.tickServer), and the guest's chunks follow the server's (#804)
- portals:   (--server-thread) the server holds a level per dimension: the guest goes to the Nether
             through a portal while the host stays, then the host goes to the End while the guest is in
             the Nether, and both come back.
- commands:  the guest's chat completes commands and their arguments over the server's command
             tree (Tab), and runs what it completed;
- recipes:   the guest's recipe book lists the planks recipe the logs unlock, and clicking it
             places a log in the inventory's crafting grid (the server places it).
- feedback:  the guest's command feedback names the guest and shows the numbers, and the host (an
             operator) is told of it as `[Guest: ...]`.
- clicks:    with real mouse presses (not click-slot, which skips the screen's press and drag
             handling), the guest puts a stack into a chest and a furnace, drags planks across
             its crafting grid and gathers a stack with a double click; the server has each.
- sleep:     at night the guest lies down in its own bed (not the host); alone it stays night, it
             gets up with Escape, and once the host sleeps too the night ends and both wake.
- push:      a cow standing in the guest pushes it just clear of the cow (the guest's own client
             pushes it, as in vanilla), not ever faster across the ground.
- eat:       holding right click with bread, the guest's own screen plays the eating (the item in use
             counts up) and the server eats one loaf and feeds the guest.
- xp:        experience orbs 4 blocks away fly to the guest, show on the guest's screen and give the
             guest their points (not the host).
- objects:   boats, minecarts, armor stands, item frames and paintings the host summons reach the
             guest as the host has them (kind, place, rotation, what they hold or wear, the variant).
- projectiles: primed TNT (one falling), an end crystal and an arrow that lands reach the guest where
             the host has them; a falling block lands on both sides.
- light:     the guest's light matches the host's: around a torch placed near it, under a roof and
             once a block of the roof is removed under the sky, and in the last chunk it tracks next to
             glowstone in a chunk it doesn't (only the server's light updates bring that one).

Each check is reported on its own (a failing one doesn't stop the others) and the screenshots of
both sides go to --out (host-*.png, guest-*.png).

    python3 tools/lan_e2e.py                          # build with the fast profile, run every check
    python3 tools/lan_e2e.py --bin-dir target/fast    # prebuilt binaries
    python3 tools/lan_e2e.py --check chest --check tab
    python3 tools/lan_e2e.py --server-thread --check portals   # the host's world on its server thread

Needs the prepared resources in .local/horizon-client and, on Linux, vscreen (the desktop); on
macOS both games render offscreen. Games started here have no audio (HORIZON_NO_AUDIO=1).
"""
import argparse
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import threading
import time
import zlib
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from agent_control_e2e import Cli, Failed, game_env  # noqa: E402

ROOT = Path(__file__).resolve().parents[1]
GUEST_NAME = 'Guest'
# The host plays single player on its server thread (--server-thread).
SERVER_THREAD = False
CHECKS = ['tab', 'self', 'armour', 'nametags', 'walls', 'chest', 'crafting', 'sign', 'damage', 'mobs', 'hazards', 'pvp', 'respawn', 'containers', 'feedback', 'titles', 'chat', 'commands', 'recipes', 'machines', 'trials', 'clicks', 'sleep', 'push', 'eat', 'xp', 'objects', 'projectiles', 'light', 'effects', 'portals', 'distance']


class Game:
    """One client: its scratch folder, its socket and a CLI driver."""

    def __init__(self, name, ctl, resources):
        self.name = name
        self.work = Path(tempfile.mkdtemp(prefix=f'hzlan-{name}-'))
        (self.work / '.local/horizon-client').mkdir(parents=True)
        os.symlink(resources, self.work / '.local/horizon-client/26.3')
        self.socket = self.work / 'c.sock'
        env = game_env(os.environ.get('VSCREEN', 'lan-e2e'))
        env['HORIZON_HOME'] = str(self.work / 'home')
        self.d = Cli(ctl, self.socket, self.work, env)
        self.d.name = name

    def __call__(self, tool, **args):
        return self.d(tool, **args)

    def raw(self, *words):
        """A game command horizon-ctl has no table entry for (lan, connect): the words as typed."""
        result = subprocess.run(self.d.base + [str(w) for w in words], cwd=self.work, env=self.d.env,
                                capture_output=True, text=True, timeout=600)
        lines = result.stdout.strip().splitlines()
        reply = json.loads(lines[-1]) if lines else {'error': result.stderr.strip()}
        print(f'[{self.name}] {list(words)} -> {json.dumps(reply)[:400]}')
        if result.returncode or 'error' in reply:
            raise Failed(f'{" ".join(map(str, words))}: {reply}')
        return reply

    def shot(self, out, label):
        path = out / f'{self.name}-{label}.png'
        self.d('screenshot', path=path)
        return path

    def wait(self, ticks):
        self.d('wait', ticks=ticks)

    def close(self):
        try:
            self.d('quit')
        except Exception:
            pass
        time.sleep(2)
        log = self.socket.with_suffix('.log')
        return log.read_text(errors='replace') if log.exists() else ''


def remote(guest):
    """The guest's view of the server (`status`'s server part)."""
    return guest('status')['server']


def back_to_game(game):
    """Close whatever screen a failed check left open, without opening the pause menu."""
    try:
        if game('state').get('screen') not in (None, 'playing'):
            game.raw('key', 'esc', 'tap')
            game.wait(2)
    except Failed:
        pass


def pos_of(game):
    return game('state')['player']['position']


def face_each_other(host, guest):
    """Put the guest 3 blocks east of the host, both looking at the other's head."""
    hx, hy, hz = pos_of(host)
    guest('command', command=f'tp @s {hx + 3:.2f} {hy:.2f} {hz:.2f}')
    guest.wait(30)
    host.raw('look-at', f'{hx + 3:.2f}', f'{hy + 1.6:.2f}', f'{hz:.2f}')
    guest.raw('look-at', f'{hx:.2f}', f'{hy + 1.6:.2f}', f'{hz:.2f}')
    host.wait(10)


def check_tab(host, guest, out):
    players = remote(guest)['players']
    names = sorted(p['name'] if isinstance(p, dict) else str(p) for p in players)
    assert names == sorted([host.player, GUEST_NAME]), f'guest tab list: {players}, host {host.player}'
    guest.raw('key', 'tab', 'down')
    guest.wait(5)
    guest.shot(out, 'tab')
    guest.raw('key', 'tab', 'up')
    return names


def check_self(host, guest, out):
    guest.raw('key', 'f5', 'tap')
    guest.wait(10)
    guest.shot(out, 'self-third-person')
    status = guest('state')
    guest.raw('key', 'f5', 'tap')
    guest.raw('key', 'f5', 'tap')
    guest.raw('key', 'e', 'tap')
    guest.raw('wait-for', 'screen', 'inventory')
    guest.shot(out, 'self-inventory')
    model = remote(guest).get('player_model')
    guest.raw('key', 'esc', 'tap')
    assert model, f'the guest has no player model: {status.get("camera")}'
    return 'model drawn'


def check_armour(host, guest, out):
    host('command', command='item replace entity @s armor.head with minecraft:golden_helmet')
    host('command', command='item replace entity @s armor.chest with minecraft:golden_chestplate')
    guest('command', command='item replace entity @s armor.chest with minecraft:diamond_chestplate')
    guest('command', command='item replace entity @s armor.legs with minecraft:diamond_leggings')
    face_each_other(host, guest)
    host.shot(out, 'armour')
    guest.shot(out, 'armour')
    seen = lan_guest(host).get('model_armor')
    assert seen == [None, 'minecraft:diamond_chestplate', 'minecraft:diamond_leggings', None], \
        f'the host draws the guest in {seen}'
    host_gear = [e for e in remote(guest).get('player_equipment', [])]
    assert any('golden_chestplate' in json.dumps(e) for e in host_gear), f'the guest sees the host in {host_gear}'
    return 'each in their own'


def check_nametags(host, guest, out):
    face_each_other(host, guest)
    h = host('status').get('name_tags') or []
    g = guest('status').get('name_tags') or []
    host.shot(out, 'nametags')
    guest.shot(out, 'nametags')
    assert GUEST_NAME in h, f'host draws name tags {h}'
    assert host.player in g, f'guest draws name tags {g}'
    return {'host': h, 'guest': g}


def grab(game):
    """Mouse look on, as a player's grabbed pointer: an agent's mouse-move stands in for it."""
    game.raw('mouse-move', 1, 0)
    game.raw('mouse-move', -1, 0)


def host_unaffected(host, what):
    """What a guest opens is the guest's: the host keeps its world on screen and its mouse look (#687)."""
    state = host('state')
    assert state.get('screen') == 'playing', f'{what} opened the {state.get("screen")} screen on the host'
    assert state.get('pointer_grabbed'), f"{what} released the host's pointer"


def lid(game, pos):
    return game.raw('block', *pos).get('lid', 0.0)


def check_chest(host, guest, out):
    x, y, z = (int(v // 1) for v in pos_of(guest))
    chest = (x, y, z + 2)
    guest('command', command=f'setblock {chest[0]} {chest[1]} {chest[2]} minecraft:chest')
    guest.wait(10)
    grab(host)
    guest.raw('interact', *chest)
    screen = guest.raw('wait-for', 'screen', 'container', '--timeout', 10)
    guest.wait(20)
    host_unaffected(host, "the guest's chest")
    opened = {'guest': lid(guest, chest), 'host': lid(host, chest)}
    host.raw('look-at', chest[0] + 0.5, chest[1] + 0.5, chest[2] + 0.5)
    host.wait(2)
    host.shot(out, 'chest-open')
    guest.shot(out, 'chest-open')
    guest.raw('key', 'esc', 'tap')
    guest.wait(20)
    closed = {'guest': lid(guest, chest), 'host': lid(host, chest)}
    host.shot(out, 'chest-closed')
    assert screen, 'no container screen on the guest'
    assert opened['guest'] > 0.9 and opened['host'] > 0.9, f'lid while open: {opened}'
    assert closed['guest'] < 0.1 and closed['host'] < 0.1, f'lid after closing: {closed}'
    return {'open': opened, 'closed': closed}


def check_crafting(host, guest, out):
    guest('command', command='clear @s')
    guest('command', command='give @s minecraft:oak_log 1')
    guest.wait(10)
    guest.raw('key', 'e', 'tap')
    guest.raw('wait-for', 'screen', 'inventory')
    inv = guest('inventory')
    slot = next(s['slot'] for s in inv['hotbar'] if s and s.get('item') == 'minecraft:oak_log')
    # Horizon's slot ids (click-slot): the hotbar is 0-8, the grid 100-108, the result 200.
    guest.raw('click-slot', slot)
    guest.raw('click-slot', 100)
    guest.wait(5)
    result = [s for s in guest('screen').get('slots', []) if s.get('slot') == 200]
    guest.shot(out, 'crafting-grid')
    guest.raw('click-slot', 200, 'shift')
    guest.wait(5)
    guest.raw('key', 'esc', 'tap')
    guest.wait(5)
    planks = guest('inventory')['counts'].get('minecraft:oak_planks', 0)
    assert planks == 4, f'crafted {planks} planks (result slot: {result})'
    x, y, z = (int(v // 1) for v in pos_of(guest))
    table = (x, y, z - 2)
    guest('command', command=f'setblock {table[0]} {table[1]} {table[2]} minecraft:crafting_table')
    guest.wait(10)
    grab(host)
    guest.raw('interact', *table)
    guest.raw('wait-for', 'screen', 'inventory', '--timeout', 10)
    host_unaffected(host, "the guest's crafting table")
    size = guest('screen').get('craft_size')
    assert size == 3, f'the crafting table opened a {size}x{size} grid'
    # Two planks one above the other in the 3x3 grid: four sticks.
    slot = next(s['slot'] for s in guest('inventory')['hotbar'] if s and s.get('item') == 'minecraft:oak_planks')
    guest.raw('click-slot', slot)
    guest.raw('click-slot', 100, 'right')
    guest.raw('click-slot', 103, 'right')
    guest.raw('click-slot', slot)
    guest.wait(5)
    guest.shot(out, 'crafting-table')
    guest.raw('click-slot', 200, 'shift')
    guest.wait(5)
    guest.raw('key', 'esc', 'tap')
    guest.wait(5)
    counts = guest('inventory')['counts']
    assert counts.get('minecraft:stick') == 4 and counts.get('minecraft:oak_planks') == 2, f'after the table: {counts}'
    return {'planks': planks, 'table': counts}


def check_sign(host, guest, out):
    x, y, z = (int(v // 1) for v in pos_of(guest))
    spot = (x + 2, y, z)
    guest('command', command=f'setblock {spot[0]} {spot[1] - 1} {spot[2]} minecraft:stone')
    guest('command', command=f'setblock {spot[0]} {spot[1]} {spot[2]} minecraft:air')
    guest('command', command='give @s minecraft:oak_sign 1')
    guest.wait(10)
    grab(host)
    guest.raw('place', *spot, '--item', 'minecraft:oak_sign')
    guest.wait(20)
    placed = host.raw('block', *spot).get('block')
    host.shot(out, 'sign')
    assert placed and 'sign' in placed, f'the guest placed {placed}'
    host_unaffected(host, "the guest's sign")
    return placed


def effect_ids(game):
    return [e.get('id', '').removeprefix('minecraft:') for e in game('state')['player'].get('effects', [])]


def check_effects(host, guest, out):
    x, y, z = (int(v // 1) for v in pos_of(host))
    bx, by, bz = x - 3, y, z - 3
    host('command', command=f'fill {bx - 1} {by - 1} {bz - 1} {bx + 1} {by - 1} {bz + 1} minecraft:iron_block')
    host('command', command=f'setblock {bx} {by} {bz} minecraft:beacon{{primary_effect:"minecraft:speed"}}')
    # BeaconBlockEntity.tick applies its powers every 80 ticks, once the beam has reached the sky.
    seen = {}
    for _ in range(15):
        host.wait(20)
        seen = {'host': effect_ids(host), 'guest': effect_ids(guest)}
        if all('speed' in v for v in seen.values()):
            break
    host.shot(out, 'effects')
    guest.shot(out, 'effects')
    assert 'speed' in seen['host'] and 'speed' in seen['guest'], f'beacon effects: {seen}'
    return seen


def wait_dimension(game, dimension, ticks):
    """Wait up to `ticks` for the game to be in `dimension` with its world loaded around it."""
    for _ in range(ticks // 20):
        state = game('state')
        if state.get('dimension') == dimension and state.get('player'):
            return state
        game.wait(20)
    return game('state')


def nether_portal(game, x, y, z):
    """A lit portal along x, two blocks wide and three tall, its inside starting at x y z."""
    game('command', command=f'fill {x - 1} {y - 1} {z} {x + 2} {y + 3} {z} minecraft:obsidian')
    game('command', command=f'fill {x} {y} {z} {x + 1} {y + 2} {z} minecraft:nether_portal[axis=x]')


def lan_guest(host):
    """The guest's entry in the host's `lan status` (on the server thread the owner is listed too)."""
    guests = host.raw('lan', 'status')['guests']
    return next(g for g in guests if g['name'] == GUEST_NAME)

def check_distance(host, guest, out):
    """IntegratedServer.tickServer: the server takes the host's render distance as its view distance
    (past 32, which Horizon's options allow) and sends it to every player (SetChunkCacheRadius)."""
    if not SERVER_THREAD:
        return 'skipped (needs --server-thread)'
    seen = {}
    try:
        for distance in (48, 6):
            host.raw('render-distance', distance)
            # The pause screen doesn't hold the server here: it follows on its next tick.
            host.wait(20)
            guest.wait(20)
            seen[distance] = (host('status').get('render_distance'), guest('status').get('render_distance'))
    finally:
        host.raw('render-distance', 12)
    # A client streams out to min(its render distance, the server's view distance + 1).
    assert seen[48][0] == 48, f'the host at 48 streams {seen[48][0]} chunks out (all: {seen})'
    assert seen[6][0] == 6, f'the host at 6 streams {seen[6][0]} chunks out (all: {seen})'
    assert seen[6][1] == 7, f'the guest streams {seen[6][1]} with the server at 6 (all: {seen})'
    return seen


def check_portals(host, guest, out):
    if not SERVER_THREAD:
        return 'skipped (needs --server-thread)'
    x, y, z = (int(v // 1) for v in pos_of(guest))
    nether_portal(guest, x, y, z + 3)
    guest.wait(5)
    guest('command', command=f'tp @s {x + 0.5} {y} {z + 3.5}')
    # players_nether_portal_default_delay: 80 ticks in survival.
    there = wait_dimension(guest, 'minecraft:the_nether', 600)
    guest.wait(40)
    guest.shot(out, 'portals-nether')
    host_dim = host('state').get('dimension')
    assert there.get('dimension') == 'minecraft:the_nether', f'the guest is in {there.get("dimension")}'
    assert host_dim == 'minecraft:overworld', f'the host went to {host_dim} with the guest'
    arrived = [int(v // 1) for v in pos_of(guest)]
    below = guest.raw('block', arrived[0], arrived[1] - 1, arrived[2])
    assert below.get('loaded') and below.get('block') != 'minecraft:air', f'under the guest in the Nether: {below}'
    # The host takes an End portal while the guest is in the Nether: three levels at once.
    hx, hy, hz = (int(v // 1) for v in pos_of(host))
    host('command', command=f'setblock {hx} {hy} {hz} minecraft:end_portal')
    end = wait_dimension(host, 'minecraft:the_end', 600)
    host.wait(40)
    host.shot(out, 'portals-end')
    guest_dim = guest('state').get('dimension')
    assert end.get('dimension') == 'minecraft:the_end', f'the host is in {end.get("dimension")}'
    assert guest_dim == 'minecraft:the_nether', f'the guest left the Nether for {guest_dim}'
    # Both come back: the host through an exit portal, the guest out of its portal and in again.
    ex, ey, ez = (int(v // 1) for v in pos_of(host))
    host('command', command=f'setblock {ex + 2} {ey} {ez} minecraft:end_portal')
    host('command', command=f'tp @s {ex + 2.5} {ey} {ez + 0.5}')
    guest('command', command=f'tp @s {arrived[0] + 0.5} {arrived[1]} {arrived[2] - 2.5}')
    guest.wait(30)
    guest('command', command=f'tp @s {arrived[0] + 0.5} {arrived[1]} {arrived[2] + 0.5}')
    home = wait_dimension(host, 'minecraft:overworld', 600)
    back = wait_dimension(guest, 'minecraft:overworld', 600)
    guest.wait(40)
    host.shot(out, 'portals-back')
    guest.shot(out, 'portals-back')
    assert home.get('dimension') == 'minecraft:overworld', f'the host stayed in {home.get("dimension")}'
    assert back.get('dimension') == 'minecraft:overworld', f'the guest stayed in {back.get("dimension")}'
    return {'guest_nether': arrived, 'host_end': [ex, ey, ez], 'back': pos_of(guest)}


def check_damage(host, guest, out):
    guest('command', command='gamemode survival')
    guest.wait(10)
    before = remote(guest)['health']
    x, y, z = pos_of(guest)
    guest('command', command=f'tp @s {x:.2f} {y + 12:.2f} {z:.2f}')
    guest.wait(50)
    after = remote(guest)['health']
    guest.shot(out, 'damage')
    assert after < before, f'health {before} -> {after} after a 12-block fall'
    # Into water: the fall does no damage.
    x, y, z = pos_of(guest)
    guest('command', command=f'setblock {int(x // 1)} {int(y // 1)} {int(z // 1)} minecraft:water')
    guest('command', command=f'tp @s {x:.2f} {y + 12:.2f} {z:.2f}')
    guest.wait(50)
    water = remote(guest)['health']
    assert water >= after, f'health {after} -> {water} after falling into water'
    return {'before': before, 'after': after, 'water': water}


def first_hurt(guest, before, ticks):
    """The guest's health once it drops below `before`, waiting at most `ticks`."""
    health = remote(guest)['health']
    for _ in range(ticks // 10):
        if health < before:
            break
        guest.wait(10)
        health = remote(guest)['health']
    return health


def check_mobs(host, guest, out):
    host('command', command='gamemode creative')
    host('command', command='difficulty normal')
    host('command', command='time set midnight')
    guest('command', command='gamemode survival')
    guest.wait(10)
    hurt = {}
    try:
        for mob, distance in (('zombie', 3), ('skeleton', 7)):
            x, y, z = pos_of(guest)
            before = remote(guest)['health']
            host('command', command=f'summon minecraft:{mob} {x:.2f} {y:.2f} {z + distance:.2f}')
            low = first_hurt(guest, before, 300)
            guest.shot(out, f'mobs-{mob}')
            host.shot(out, f'mobs-{mob}')
            host('command', command=f'kill @e[type=minecraft:{mob}]')
            hurt[mob] = (before, low)
    finally:
        host('command', command='gamemode survival')
        host('command', command='time set day')
    for mob, (before, low) in hurt.items():
        assert low < before, f'the {mob} never hurt the guest: health {before} -> {low} (all: {hurt})'
    return hurt


def check_hazards(host, guest, out):
    guest('command', command='gamemode survival')
    # A fresh life: full health, so no regeneration eats food while the guest runs.
    guest('command', command='kill @s')
    guest.wait(20)
    # ServerPlayer.doTick sends the dead player's health of 0 too.
    died = {'screen': guest('state').get('screen'), 'health': remote(guest)['health']}
    assert died == {'screen': 'dead', 'health': 0.0}, f'the guest was killed: {died}'
    guest.raw('respawn')
    guest.wait(40)
    # Sprint-jumping down a stone lane: ServerPlayer.jumpFromGround and checkMovementStatistics.
    guest('command', command='fill ~ ~-1 ~-1 ~60 ~-1 ~1 minecraft:stone')
    guest('command', command='fill ~ ~ ~-1 ~60 ~3 ~1 minecraft:air')
    guest.wait(10)
    x, y, z = pos_of(guest)
    guest.raw('look-at', f'{x + 60:.2f}', f'{y + 1.6:.2f}', f'{z:.2f}')
    food = lan_guest(host)['food']
    for k in ('w', 'ctrl', 'space'):
        guest.raw('key', k, 'down')
    guest.wait(160)
    for k in ('space', 'ctrl', 'w'):
        guest.raw('key', k, 'up')
    guest.wait(10)
    fed = lan_guest(host)['food']
    hunger = {'before': food, 'after': fed}
    hurt = {}
    hx, hy, hz = (int(v // 1) for v in pos_of(host))
    x, y, z = hx + 3, hy, hz
    for name, fill, ticks in (('lava', 'minecraft:lava', 60), ('wall', 'minecraft:stone', 60),
                              ('drowning', 'minecraft:water', 500), ('void', None, 60)):
        if remote(guest)['health'] <= 0:
            guest.raw('respawn')
            guest.wait(40)
        # A stone box around the guest holds the lava and the water in.
        guest('command', command=f'fill {x - 1} {y - 1} {z - 1} {x + 1} {y + 2} {z + 1} minecraft:stone')
        guest('command', command=f'fill {x} {y} {z} {x} {y + 1} {z} minecraft:air')
        guest('command', command=f'tp @s {x + 0.5} {y} {z + 0.5}')
        guest.wait(100)
        before = remote(guest)['health']
        if fill:
            guest('command', command=f'fill {x} {y} {z} {x} {y + 1} {z} {fill}')
        else:
            guest('command', command=f'tp @s {x + 0.5} -140 {z + 0.5}')
        hurt[name] = (before, first_hurt(guest, before, ticks))
        guest.shot(out, f'hazards-{name}')
        # Water puts the lava's fire out.
        guest('command', command=f'fill {x} {y} {z} {x} {y + 1} {z} minecraft:water')
        guest.wait(5)
        guest('command', command=f'fill {x - 1} {y - 1} {z - 1} {x + 1} {y + 2} {z + 1} minecraft:air')
        guest('command', command=f'fill {x - 1} {y - 1} {z - 1} {x + 1} {y - 1} {z + 1} minecraft:stone')
    guest.wait(20)
    if remote(guest)['health'] <= 0:
        guest.raw('respawn')
        guest.wait(40)
    # [food, saturation, exhaustion]: 8 s of sprint-jumping is about 2 of exhaustion; every 4
    # of it takes a point of saturation, or of food once saturation is gone (FoodData.tick).
    spent = 4 * (food[0] + food[1] - fed[0] - fed[1]) + fed[2] - food[2]
    assert spent > 1.0, f'no hunger from moving: {hunger} ({spent:.2f} spent)'
    for name, (before, after) in hurt.items():
        assert after < before, f'{name} never hurt the guest: health {before} -> {after} (all: {hurt})'
    return {'died': died, 'hunger': hunger, **hurt}


def check_pvp(host, guest, out):
    for g in (host, guest):
        g('command', command='gamemode survival')
        g('command', command='item replace entity @s weapon.mainhand with minecraft:iron_sword')
    face_each_other(host, guest)
    hits = {}
    try:
        for name, rule, attacker in (('guest hits host', 'true', guest), ('host hits guest', 'true', host),
                                     ('pvp off', 'false', guest)):
            host('command', command=f'gamerule pvp {rule}')
            face_each_other(host, guest)
            attacker.wait(30)  # a full attack charge
            victim = (lambda: host('state')['player']['health']) if attacker is guest else (lambda: remote(guest)['health'])
            before = victim()
            attacker.raw('attack')
            attacker.wait(10)
            hits[name] = (before, victim())
            attacker.shot(out, 'pvp-' + name.replace(' ', '-'))
    finally:
        host('command', command='gamerule pvp true')
    assert hits['guest hits host'][1] < hits['guest hits host'][0], f'the guest did not hurt the host: {hits}'
    assert hits['host hits guest'][1] < hits['host hits guest'][0], f'the host did not hurt the guest: {hits}'
    assert hits['pvp off'][1] == hits['pvp off'][0], f'pvp off still hurts: {hits}'
    return hits


def dead_players(game, want, ticks=60):
    """The other players `game`'s client draws dead, once that is `want` (empty or not) or after `ticks`."""
    for _ in range(ticks // 5):
        dead = remote(game)['dead_players']
        if bool(dead) == want:
            return dead
        game.wait(5)
    return dead


def check_respawn(host, guest, out):
    # ServerLevel.removePlayerImmediately and addRespawnedPlayer: a respawned player is removed
    # from the other players and paired again as a new entity, so their clients drop the body.
    # The host window only has a client's view of the guest on the server thread.
    for g in (host, guest):
        back_to_game(g)
    results = {}
    sides = [('host', host, guest)] + ([('guest', guest, host)] if SERVER_THREAD else [])
    for name, dying, watching in sides:
        face_each_other(host, guest)
        dying('command', command='kill @s')
        dying.wait(30)
        dead = dead_players(watching, True)
        watching.shot(out, f'respawn-{name}-dead')
        dying.raw('respawn')
        dying.wait(40)
        back_to_game(dying)
        # Back within tracking range of the other player, wherever the respawn point is.
        x, y, z = pos_of(dying)
        watching('command', command=f'tp @s {x + 3:.2f} {y:.2f} {z:.2f}')
        watching.wait(40)
        alive = dead_players(watching, False)
        watching.shot(out, f'respawn-{name}-alive')
        results[name] = {'dead': dead, 'after_respawn': alive}
        assert dead, f'the {name} was not drawn dead: {results}'
        assert not alive, f'the {name} still lies dead after respawning: {results}'
    return results


def container_items(game):
    """The open container's stacks by slot (Horizon's container slots are 300 and up) and its size."""
    slots = [s for s in game('screen').get('slots', []) if s.get('slot', 0) >= 300]
    return {s['slot'] - 300: s['item'] for s in slots if s.get('item')}, len(slots)


def open_container(guest, pos, out, label, take=None):
    """Open the container at `pos` and read it; `take` shift-clicks that container slot out first."""
    guest.raw('interact', *pos)
    guest.raw('wait-for', 'screen', 'container', '--timeout', 10)
    # The server's menu (an ender or double chest has no stand-in block entity: key i32::MIN),
    # then its contents, can come a few ticks after the screen.
    for _ in range(20):
        guest.wait(2)
        screen = guest('screen')
        items = {s['slot']: s.get('item') for s in screen.get('slots', [])}
        if screen.get('container') == [-2**31, 0, 0] and (take is None or items.get(300 + take)):
            break
    guest.wait(5)
    if take is not None:
        guest.raw('click-slot', 300 + take, 'shift')
        guest.wait(5)
    items = container_items(guest)
    guest.shot(out, label)
    guest.raw('key', 'esc', 'tap')
    guest.wait(5)
    return items


def check_containers(host, guest, out):
    host('command', command='item replace entity @s enderchest.0 with minecraft:diamond')
    guest('command', command='item replace entity @s enderchest.4 with minecraft:emerald')
    x, y, z = (int(v // 1) for v in pos_of(guest))
    guest('command', command=f'setblock {x} {y} {z + 2} minecraft:ender_chest')
    guest.wait(10)
    ender, ender_size = open_container(guest, (x, y, z + 2), out, 'ender-chest')
    # Facing north, a left half pairs with the block to its east (ChestBlock.getConnectedDirection).
    guest('command', command=f'setblock {x} {y} {z - 2} minecraft:chest[facing=north,type=left]')
    guest('command', command=f'setblock {x + 1} {y} {z - 2} minecraft:chest[facing=north,type=right]')
    guest('command', command=f'item replace block {x + 1} {y} {z - 2} container.0 with minecraft:gold_ingot')
    guest('command', command=f'item replace block {x} {y} {z - 2} container.0 with minecraft:iron_ingot')
    guest.wait(10)
    double, double_size = open_container(guest, (x, y, z - 2), out, 'double-chest')
    # Taking from each writes back to the guest's ender inventory and to the right half.
    guest('command', command='clear @s')
    open_container(guest, (x, y, z + 2), out, 'ender-chest-take', take=4)
    open_container(guest, (x, y, z - 2), out, 'double-chest-take', take=0)
    taken = guest('inventory')['counts']
    after = (open_container(guest, (x, y, z + 2), out, 'ender-chest-after')[0],
             open_container(guest, (x, y, z - 2), out, 'double-chest-after')[0])
    host_ender = host('command', command='data get entity @s EnderItems')
    for at in ((x, y, z + 2), (x, y, z - 2), (x + 1, y, z - 2)):
        guest('command', command=f'setblock {at[0]} {at[1]} {at[2]} minecraft:air')
    assert ender == {4: 'minecraft:emerald'} and ender_size == 27, f'the guest\'s ender chest: {ender} ({ender_size})'
    assert double_size == 54, f'the double chest opened with {double_size} slots'
    assert double == {0: 'minecraft:gold_ingot', 27: 'minecraft:iron_ingot'}, f'the double chest: {double}'
    assert 'emerald' not in json.dumps(host_ender), f'the guest\'s item went to the host: {host_ender}'
    assert taken.get('minecraft:emerald') == 1 and taken.get('minecraft:gold_ingot') == 1, f'taken: {taken}'
    assert after == ({}, {27: 'minecraft:iron_ingot'}), f'after taking: {after}'
    return {'ender': ender, 'double': double, 'taken': taken}


def check_machines(host, guest, out):
    # After `recipes`: the stacks given here unlock recipes, which would move the book's first one.
    guest('command', command='gamemode survival')
    x, y, z = (int(v // 1) for v in pos_of(guest))
    at = (x, y, z + 2)
    seen = {}
    for block in ('hopper', 'dispenser', 'dropper'):
        guest('command', command='clear @s')
        guest('command', command='give @s minecraft:cobblestone 4')
        guest('command', command=f'setblock {at[0]} {at[1]} {at[2]} minecraft:{block}')
        guest.wait(10)
        guest.raw('interact', *at)
        guest.raw('wait-for', 'screen', 'container', '--timeout', 10)
        guest.wait(10)
        size = container_items(guest)[1]
        guest.shot(out, f'machines-{block}')
        guest.raw('click-slot', hotbar_slot(guest, 'minecraft:cobblestone'), 'shift')
        guest.wait(10)
        guest.raw('key', 'esc', 'tap')
        guest.wait(10)
        # Opened again, the menu holds what the server has.
        seen[block] = (size, open_container(guest, at, out, f'machines-{block}-after')[0])
        guest('command', command=f'setblock {at[0]} {at[1]} {at[2]} minecraft:air')
    want = {'hopper': (5, {0: 'minecraft:cobblestone'}), 'dispenser': (9, {0: 'minecraft:cobblestone'}),
            'dropper': (9, {0: 'minecraft:cobblestone'})}
    assert seen == want, f'the machines on the guest: {seen}'
    return seen


def slot_point(game, slot):
    """The GUI point in the middle of slot `slot` of the open screen (`ui`'s slot-N)."""
    for w in game.raw('ui')['widgets']:
        if w.get('id') == f'slot-{slot}':
            x, y, width, height = w['bounds']
            return x + width / 2, y + height / 2
    raise Failed(f'no slot {slot} on the screen')


def mouse(game, *slots, right=False):
    """Press over the first slot, move over the others and release: one click, or a drag."""
    points = [f'{v:.1f}' for s in slots for v in slot_point(game, s)]
    game.raw('drag-at', *(['right'] if right else []), *points)
    game.wait(4)


def hotbar_slot(game, item):
    return next(s['slot'] for s in game('inventory')['hotbar'] if s and s.get('item') == item)


def check_clicks(host, guest, out):
    guest('command', command='gamemode survival')
    guest('command', command='clear @s')
    guest('command', command='give @s minecraft:cobblestone 10')
    guest('command', command='give @s minecraft:coal 3')
    guest('command', command='give @s minecraft:raw_iron 2')
    guest('command', command='give @s minecraft:oak_planks 2')
    guest.wait(10)
    x, y, z = (int(v // 1) for v in pos_of(guest))
    chest, furnace = (x, y, z + 2), (x, y, z - 2)
    guest('command', command=f'setblock {chest[0]} {chest[1]} {chest[2]} minecraft:chest')
    guest('command', command=f'setblock {furnace[0]} {furnace[1]} {furnace[2]} minecraft:furnace')
    guest.wait(10)
    # Pick the cobblestone up and put it down in the chest's first slot.
    guest.raw('interact', *chest)
    guest.raw('wait-for', 'screen', 'container', '--timeout', 10)
    guest.wait(10)
    mouse(guest, hotbar_slot(guest, 'minecraft:cobblestone'))
    mouse(guest, 300)
    guest.wait(10)
    guest.shot(out, 'clicks-chest')
    guest.raw('key', 'esc', 'tap')
    guest.wait(10)
    in_chest, _ = open_container(guest, chest, out, 'clicks-chest-after')
    # Coal into the fuel slot, raw iron into the input.
    guest.raw('interact', *furnace)
    guest.raw('wait-for', 'screen', 'container', '--timeout', 10)
    guest.wait(10)
    mouse(guest, hotbar_slot(guest, 'minecraft:coal'))
    mouse(guest, 301)
    mouse(guest, hotbar_slot(guest, 'minecraft:raw_iron'))
    mouse(guest, 300)
    guest.wait(10)
    guest.shot(out, 'clicks-furnace')
    guest.raw('key', 'esc', 'tap')
    guest.wait(10)
    # Opening the furnace again resends the whole menu: the server's own view.
    in_furnace, _ = open_container(guest, furnace, out, 'clicks-furnace-after')
    # Two planks dragged over the 2x2 grid's left column: one each, four sticks.
    guest.raw('key', 'e', 'tap')
    guest.raw('wait-for', 'screen', 'inventory')
    guest.wait(5)
    mouse(guest, hotbar_slot(guest, 'minecraft:oak_planks'))
    mouse(guest, 100, 102)
    guest.wait(10)
    grid = {s['slot']: (s.get('item'), s.get('count')) for s in guest('screen').get('slots', [])
            if s.get('slot') in (100, 102, 200) and s.get('item')}
    guest.shot(out, 'clicks-drag')
    guest.raw('click-slot', 200, 'shift')
    guest.wait(10)
    guest.raw('key', 'esc', 'tap')
    guest.wait(10)
    # The chest's menu brings the whole inventory as the server has it.
    open_container(guest, chest, out, 'clicks-inventory')
    counts = guest('inventory')['counts']
    for at in (chest, furnace):
        guest('command', command=f'setblock {at[0]} {at[1]} {at[2]} minecraft:air')
    assert in_chest == {0: 'minecraft:cobblestone'}, f'the chest holds {in_chest} after the guest put cobblestone in'
    assert in_furnace == {0: 'minecraft:raw_iron', 1: 'minecraft:coal'}, f'the furnace holds {in_furnace}'
    assert grid.get(100) == ('minecraft:oak_planks', 1) and grid.get(102) == ('minecraft:oak_planks', 1), \
        f'the drag left the grid as {grid}'
    assert counts.get('minecraft:stick') == 4 and 'minecraft:cobblestone' not in counts, f'inventory after: {counts}'
    return {'chest': in_chest, 'furnace': in_furnace, 'grid': grid, 'inventory': counts}


def check_feedback(host, guest, out):
    x, y, z = (round(v, 1) for v in pos_of(guest))
    guest('command', command='item replace entity @s weapon.mainhand with minecraft:stick')
    guest('command', command=f'tp @s {x:.1f} {y:.1f} {z:.1f}')
    guest.wait(10)
    chat = guest.raw('messages', 5)['chat']
    # The host is an operator: broadcastToAdmins tells it, as `[Guest: ...]`.
    host_chat = host.raw('messages', 5)['chat']
    guest.raw('key', 't', 'tap')
    guest.wait(5)
    guest.shot(out, 'feedback')
    guest.raw('key', 'esc', 'tap')
    guest.wait(5)
    screen = guest('state').get('screen')
    replaced = f'Replaced 1 slot(s) on {GUEST_NAME} with [Stick]'
    assert any(replaced in line for line in chat), f'no "{replaced}" in {chat}'
    assert any(line.startswith(f'Teleported {GUEST_NAME} to {x:.6f}') for line in chat), f'teleport feedback: {chat}'
    assert screen in (None, 'playing'), f'the chat did not close: {screen}'
    told = f'[{GUEST_NAME}: Teleported {GUEST_NAME} to {x:.6f}'
    assert any(line.startswith(told) for line in host_chat), f'no "{told}" for the host: {host_chat}'
    return chat + host_chat


def chat_input(game):
    return game('screen').get('chat_input')


def complete(guest, text):
    """Opens the chat on `/`, types `text` and presses Tab: the input after it."""
    guest.raw('key', 'slash', 'tap')
    guest.wait(3)
    guest.raw('type', text)
    guest.wait(2)
    guest.raw('key', 'tab', 'tap')
    guest.wait(2)
    return chat_input(guest)


def check_commands(host, guest, out):
    literal = complete(guest, 'gamem')
    # The first Escape hides the suggestions, the next closes the chat.
    for _ in range(3):
        if guest('screen').get('screen') != 'chat':
            break
        guest.raw('key', 'esc', 'tap')
        guest.wait(3)
    argument = complete(guest, 'gamemode cr')
    guest.shot(out, 'commands')
    guest.raw('key', 'enter', 'tap')
    guest.wait(20)
    mode = guest('state').get('gamemode')
    screen = guest('state').get('screen')
    guest('command', command='gamemode survival')
    guest.wait(10)
    # A node the tree marks ask_server (a bossbar id): the guest asks the
    # server, which completes it from its own bossbars.
    host('command', command='bossbar add pt6:check "Check"')
    host.wait(5)
    guest.raw('key', 'slash', 'tap')
    guest.wait(3)
    guest.raw('type', 'bossbar get pt')
    guest.wait(10)
    guest.raw('key', 'tab', 'tap')
    guest.wait(2)
    asked = chat_input(guest)
    for _ in range(3):
        if guest('screen').get('screen') != 'chat':
            break
        guest.raw('key', 'esc', 'tap')
        guest.wait(3)
    host('command', command='bossbar remove pt6:check')
    assert literal == '/gamemode', f'Tab after /gamem: {literal!r}'
    assert argument == '/gamemode creative', f'Tab after /gamemode cr: {argument!r}'
    assert mode == 'creative', f'the completed command did not run: {mode}'
    assert screen in (None, 'playing'), f'the chat did not close: {screen}'
    assert asked == '/bossbar get pt6:check', f'Tab after /bossbar get pt: {asked!r}'
    return {'literal': literal, 'argument': argument, 'mode': mode, 'ask_server': asked}


def centre(rect):
    x, y, w, h = rect
    return x + w / 2, y + h / 2


def grid_items(game):
    slots = game('screen').get('slots') or []
    return {s['slot']: s.get('item') for s in slots if 100 <= s['slot'] < 109 and s.get('item')}


def check_recipes(host, guest, out):
    guest('command', command='clear @s')
    guest('command', command='give @s minecraft:oak_log 4')
    guest.wait(40)
    guest.raw('key', 'e', 'tap')
    guest.wait(5)
    try:
        book = guest('screen').get('recipe_book')
        assert book, 'the guest\'s inventory has no recipe book'
        if len(book['targets']) == 1:
            guest.raw('click-at', *centre(book['targets'][0]))
            guest.wait(5)
            book = guest('screen').get('recipe_book')
        known = book['known']
        assert known > 0, f'the guest\'s book is empty: {book}'
        guest.raw('click-at', *centre(book['targets'][1]))
        guest.wait(10)
        grid = grid_items(guest)
        guest.shot(out, 'recipes')
    finally:
        guest.raw('key', 'esc', 'tap')
        guest.wait(5)
    assert any('oak_log' in str(item) for item in grid.values()), f'nothing placed in the grid: {grid}'
    return {'known': known, 'grid': grid}


def check_titles(host, guest, out):
    # TitleCommand sends the guest its title packets; its HUD shows and clears them.
    for command in ('title @s times 10 400 10', 'title @s subtitle "Under"', 'title @s title "Over"',
                    'title @s actionbar "Bar"'):
        guest('command', command=command)
    guest.wait(5)
    shown = {k: guest.raw('messages', 1)[k] for k in ('title', 'subtitle', 'actionbar')}
    guest.shot(out, 'titles')
    guest('command', command='title @s clear')
    guest.wait(5)
    cleared = {k: guest.raw('messages', 1)[k] for k in ('title', 'subtitle')}
    assert shown == {'title': 'Over', 'subtitle': 'Under', 'actionbar': 'Bar'}, f'titles shown: {shown}'
    assert cleared == {'title': None, 'subtitle': None}, f'titles cleared: {cleared}'
    return {'shown': shown, 'cleared': cleared}


def check_chat(host, guest, out):
    # The agent's event log follows a remote client's chat: `wait-for chat` sees the line another
    # player sends (and on the server thread the host window's own line, back from the server).
    line = f'chat check {int(time.time())}'
    host.raw('chat', line)
    seen = {g.name: g.raw('wait-for', 'chat', line, '--timeout', 10, '--since', 0).get('ok')
            for g in ((guest, host) if SERVER_THREAD else (guest,))}
    guest.shot(out, 'chat')
    return seen


def png_rows(path):
    """The rows of an 8-bit RGB or RGBA PNG, unfiltered, as bytes of 4 (or 3) per pixel."""
    data = Path(path).read_bytes()
    pos, idat, head = 8, b'', None
    while pos < len(data):
        size = int.from_bytes(data[pos:pos + 4], 'big')
        kind, body = data[pos + 4:pos + 8], data[pos + 8:pos + 8 + size]
        if kind == b'IHDR':
            head = body
        elif kind == b'IDAT':
            idat += body
        pos += 12 + size
    width, height = int.from_bytes(head[:4], 'big'), int.from_bytes(head[4:8], 'big')
    assert head[8] == 8 and head[9] in (2, 6) and head[12] == 0, f'unsupported PNG {head.hex()}'
    bpp = 4 if head[9] == 6 else 3
    raw, stride, rows, prev = zlib.decompress(idat), width * bpp, [], bytearray(width * bpp)
    for y in range(height):
        kind, line = raw[y * (stride + 1)], bytearray(raw[y * (stride + 1) + 1:(y + 1) * (stride + 1)])
        for i in range(stride):
            a = line[i - bpp] if i >= bpp else 0
            b, c = prev[i], prev[i - bpp] if i >= bpp else 0
            if kind == 1:
                line[i] = (line[i] + a) & 255
            elif kind == 2:
                line[i] = (line[i] + b) & 255
            elif kind == 3:
                line[i] = (line[i] + (a + b) // 2) & 255
            elif kind == 4:
                pa, pb, pc = abs(b - c), abs(a - c), abs(a + b - 2 * c)
                line[i] = (line[i] + (a if pa <= pb and pa <= pc else b if pb <= pc else c)) & 255
        rows.append(bytes(line))
        prev = line
    return rows, bpp


def changed_pixels(a, b):
    (ra, bpp), (rb, _) = png_rows(a), png_rows(b)
    return sum(1 for la, lb in zip(ra, rb) for i in range(0, len(la), bpp)
               if max(abs(la[i + k] - lb[i + k]) for k in range(3)) > 24)


def sneaking_on_host(host, want, ticks=40):
    """Whether the host sees the guest crouching, once it matches `want` or after `ticks`."""
    for _ in range(ticks // 5):
        sneaking = lan_guest(host)['sneaking']
        if sneaking == want:
            return sneaking
        host.wait(5)
    return sneaking


def check_walls(host, guest, out):
    # A stone wall between the two: the host still sees the guest's name through it
    # (TEXT_SEE_THROUGH), but not once the guest crouches (Entity.isDiscrete).
    for g in (host, guest):
        back_to_game(g)
    # The trial spawner's zombies may have killed the guest.
    if guest('state').get('screen') == 'dead' or remote(guest)['health'] <= 0:
        guest.raw('respawn')
        guest.wait(40)
    face_each_other(host, guest)
    hx, hy, hz = pos_of(host)
    gx, gy, gz = pos_of(guest)
    # Halfway: the players stand 3 apart, so the wall's block touches neither.
    wall = int((hx + gx) / 2 // 1)
    host('command', command=f'fill {wall} {int(hy // 1)} {int(hz // 1) - 3} {wall} {int(hy // 1) + 4} '
                            f'{int(hz // 1) + 3} minecraft:stone')
    try:
        host.raw('look-at', f'{gx:.2f}', f'{gy + 2.3:.2f}', f'{gz:.2f}')
        # Chat lines fade out after 200 ticks; let them go so only the name tag can differ.
        host.wait(210)
        standing = host.shot(out, 'walls-standing')
        guest.raw('key', 'shift', 'down')
        crouched = sneaking_on_host(host, True)
        guest.wait(10)
        crouching = host.shot(out, 'walls-crouching')
        guest.raw('key', 'shift', 'up')
        stood = not sneaking_on_host(host, False)
        guest.wait(10)
        again = host.shot(out, 'walls-standing-again')
    finally:
        guest.raw('key', 'shift', 'up')
        host('command', command=f'fill {wall} {int(hy // 1)} {int(hz // 1) - 3} {wall} {int(hy // 1) + 4} '
                                f'{int(hz // 1) + 3} minecraft:air')
        face_each_other(host, guest)
    assert crouched and stood, f'the host never saw the guest crouch and stand: {crouched}, {stood}'
    seen, steady = changed_pixels(standing, crouching), changed_pixels(standing, again)
    assert seen > 50, f'no name tag through the wall: {seen} pixels differ standing vs crouching'
    assert steady < seen // 4, f'the view itself changed: {steady} pixels between two standing shots ({seen})'
    return {'standing_vs_crouching': seen, 'standing_twice': steady}


def state_of(game, pos, key):
    state = game.raw('block', *pos)['state']
    return dict(kv.split('=') for kv in state.partition('[')[2].rstrip(']').split(',') if kv).get(key)


def wait_state(game, pos, key, want, ticks):
    for _ in range(ticks // 10):
        if state_of(game, pos, key) == want:
            return want
        game.wait(10)
    return state_of(game, pos, key)


def check_trials(host, guest, out):
    # The host watches as a spectator, whom neither block detects: only the guest can wake them.
    hx, hy, hz = pos_of(host)
    host('command', command='gamemode spectator')
    host('command', command='difficulty normal')
    host('command', command='gamerule spawn_mobs true')
    guest('command', command='gamemode survival')
    x, y, z = (int(v // 1) for v in pos_of(guest))
    spawner, vault = (x, y, z + 3), (x, y, z - 2)
    # Open air from the vault to the spawner: the spawner wants to see the guest's eyes.
    guest('command', command=f'fill {x - 1} {y} {z - 2} {x + 1} {y + 2} {z + 3} minecraft:air')
    try:
        guest('command', command=f'setblock {spawner[0]} {spawner[1]} {spawner[2]} minecraft:trial_spawner'
                                 '{spawn_data:{entity:{id:"minecraft:zombie"}},required_player_range:6}')
        guest('command', command=f'setblock {vault[0]} {vault[1]} {vault[2]} minecraft:vault'
                                 '{config:{key_item:{id:"minecraft:trial_key",count:1}}}')
        trial = wait_state(host, spawner, 'trial_spawner_state', 'active', 100)
        active = wait_state(host, vault, 'vault_state', 'active', 60)
        host.shot(out, 'trials-active')
        guest('command', command='item replace entity @s weapon.mainhand with minecraft:trial_key')
        guest.wait(5)
        guest.raw('interact', *vault)
        unlocking = wait_state(host, vault, 'vault_state', 'unlocking', 20)
        guest.shot(out, 'trials-unlocking')
        rewarded = host('command', command=f'data get block {vault[0]} {vault[1]} {vault[2]} server_data.rewarded_players')
        host_uuid = host('command', command='data get entity @s UUID')
    finally:
        host('command', command='gamerule spawn_mobs false')
        host('command', command='kill @e[type=minecraft:zombie]')
        host('command', command='gamemode survival')
        # Back where it stood: a spectator drifts, and may be inside the ground when it lands.
        host('command', command=f'tp @s {hx:.2f} {hy:.2f} {hz:.2f}')
        guest('command', command=f'setblock {spawner[0]} {spawner[1]} {spawner[2]} minecraft:air')
        guest('command', command=f'setblock {vault[0]} {vault[1]} {vault[2]} minecraft:air')
    uuid = lambda reply: re.findall(r'-?\d+', json.dumps(reply).partition('[I;')[2].partition(']')[0])
    assert trial == 'active', f'the trial spawner never saw the guest: {trial}'
    assert active == 'active', f'the vault never saw the guest: {active}'
    assert unlocking == 'unlocking', f'the guest\'s key did not unlock the vault: {unlocking}'
    assert uuid(rewarded) and uuid(rewarded) != uuid(host_uuid), f'rewarded {rewarded}, host {host_uuid}'
    return {'spawner': trial, 'vault': active, 'rewarded': uuid(rewarded), 'host': uuid(host_uuid)}


def bed_at(game, foot):
    """A bed whose foot is at `foot`, its head one block east."""
    x, y, z = foot
    game('command', command=f'setblock {x} {y} {z} minecraft:red_bed[facing=east,part=foot]')
    game('command', command=f'setblock {x + 1} {y} {z} minecraft:red_bed[facing=east,part=head]')
    return [x + 1, y, z]


def check_sleep(host, guest, out):
    for g in (host, guest):
        g('command', command='gamemode survival')
    host('command', command='time set midnight')
    gx, gy, gz = (int(v // 1) for v in pos_of(guest))
    hx, hy, hz = (int(v // 1) for v in pos_of(host))
    guest_bed = bed_at(guest, (gx, gy, gz + 2))
    host_bed = bed_at(host, (hx, hy, hz + 2))
    guest.wait(10)
    guest.raw('interact', *guest_bed)
    guest.wait(10)
    asleep = {'guest': guest('state').get('sleeping'), 'host': host('state').get('sleeping'),
              'server': lan_guest(host).get('sleeping')}
    guest.shot(out, 'sleep-guest')
    # Alone in bed: still night 120 ticks later (everyone must sleep).
    guest.wait(120)
    night = host('state')['time']['time_of_day']
    # Escape gets the guest up (InBedChatScreen), then back to bed.
    guest.raw('key', 'esc', 'tap')
    guest.wait(10)
    up = guest('state').get('sleeping')
    guest.raw('interact', *guest_bed)
    host.raw('interact', *host_bed)
    host.wait(130)
    after = {'guest': guest('state').get('sleeping'), 'host': host('state').get('sleeping'),
             'time': host('state')['time']['time_of_day']}
    guest.shot(out, 'sleep-morning')
    for at in (guest_bed, host_bed):
        guest('command', command=f'fill {at[0] - 1} {at[1]} {at[2]} {at[0]} {at[1]} {at[2]} minecraft:air')
    host('command', command='time set day')
    detail = {'asleep': asleep, 'night': night, 'up': up, 'after': after}
    assert asleep['guest'] == guest_bed and asleep['server'] == guest_bed, f'the guest is not in its bed: {detail}'
    assert asleep['host'] is None, f'the guest\'s bed put the host to sleep: {detail}'
    assert 13000 < night < 23000, f'one of two players asleep ended the night: {detail}'
    assert up is None, f'Escape did not get the guest up: {detail}'
    assert after['guest'] is None and after['host'] is None and after['time'] < 1000, f'no morning: {detail}'
    return detail
def check_push(host, guest, out):
    guest('command', command='gamemode survival')
    x, y, z = (int(v // 1) + 0.5 for v in pos_of(guest))
    guest('command', command=f'tp @s {x} {y - 0.5} {z}')
    guest.wait(20)
    x, y, z = pos_of(guest)
    # A cow that doesn't walk, 0.3 east of the guest: the two boxes overlap until their centres are
    # 0.75 apart, so the push stops a fraction of a block away.
    guest('command', command=f'summon minecraft:cow {x + 0.3:.2f} {y:.2f} {z:.2f} {{NoAI:1b}}')
    guest.wait(60)
    nx, _, nz = pos_of(guest)
    guest.shot(out, 'push')
    guest('command', command='kill @e[type=minecraft:cow]')
    moved = ((nx - x) ** 2 + (nz - z) ** 2) ** 0.5
    detail = {'from': [round(x, 3), round(z, 3)], 'to': [round(nx, 3), round(nz, 3)], 'moved': round(moved, 3)}
    assert nx < x, f'the cow did not push the guest west: {detail}'
    assert moved < 1.5, f'the cow pushed the guest too far: {detail}'
    return detail
def check_eat(host, guest, out):
    guest('command', command='gamemode survival')
    guest('command', command='clear @s')
    # Hungry: the eating needs food below 20 (Player.canEat).
    guest('command', command='effect give @s minecraft:hunger 3 255')
    guest.wait(70)
    guest('command', command='effect clear @s')
    guest('command', command='give @s minecraft:bread 3')
    guest.wait(5)
    guest.raw('select', 'minecraft:bread')
    hungry = guest('state')['player']['food']
    # use-item --hold returns once released: watch the guest's screen meanwhile.
    hold = threading.Thread(target=guest.raw, args=('use-item', '--hold', 2.2))
    hold.start()
    using = []
    for _ in range(4):
        time.sleep(0.2)
        using.append(guest('state').get('using'))
    guest.shot(out, 'eat')
    hold.join()
    guest.wait(10)
    fed = guest('state')['player']['food']
    left = guest('inventory')['counts'].get('minecraft:bread', 0)
    detail = {'hungry': hungry, 'fed': fed, 'bread': left, 'using': using, 'after': guest('state').get('using')}
    assert hungry < 20, f'the guest is not hungry: {detail}'
    ticks = [u['ticks'] for u in using if u]
    assert len(ticks) >= 3 and ticks == sorted(ticks) and ticks[-1] > ticks[0], f'no eating on the guest\'s screen: {detail}'
    assert left == 2 and fed > hungry, f'the server did not eat one loaf: {detail}'
    assert detail['after'] is None, f'still eating after the release: {detail}'
    return detail


def check_xp(host, guest, out):
    for g in (host, guest):
        g('command', command='gamemode survival')
        g('command', command='xp set @s 0 levels')
        g('command', command='xp set @s 0 points')
    guest.wait(10)
    x, y, z = pos_of(guest)
    # Three orbs of 7 points 4 blocks east of the guest (out of reach, within the 8 blocks orbs
    # follow from): they fly to the guest.
    for _ in range(3):
        guest('command', command=f'summon minecraft:experience_orb {x + 4:.2f} {y + 0.5:.2f} {z:.2f} {{Value:7s}}')
    guest.wait(2)
    seen = remote(guest).get('entities', {}).get('minecraft:experience_orb', 0)
    guest.shot(out, 'xp-orbs')
    guest.wait(60)
    xp = guest('state')['player']['xp']
    host_xp = host('state')['player']['xp']
    left = remote(guest).get('entities', {}).get('minecraft:experience_orb', 0)
    detail = {'seen': seen, 'guest': xp, 'host': host_xp, 'left': left}
    # ExperienceOrb.scanForMerges: orbs of one value side by side can merge into one.
    assert 1 <= seen <= 3, f'the guest saw {seen} orbs: {detail}'
    assert xp['total'] == 21 and xp['level'] == 2, f'the guest\'s experience after three 7-point orbs: {detail}'
    assert host_xp['total'] == 0, f'the host took the orbs: {detail}'
    assert left == 0, f'orbs left on the guest\'s screen: {detail}'
    return {'seen': seen, 'guest': xp, 'host': host_xp}


def objects_near(game, centre, key='objects'):
    """`status`'s objects within 8 blocks of centre, without their ids (each side numbers its own)."""
    out = []
    for o in game('status')[key]:
        if all(abs(o['position'][i] - centre[i]) <= 8 for i in (0, 2)):
            o = dict(o)
            o.pop('id')
            o['position'] = [round(v, 2) for v in o['position']]
            if 'yaw' in o:  # as the packets carry it, in 256ths of a turn
                o['yaw'] = round(o['yaw'] * 256 / 360) % 256
            out.append(o)
    return sorted(out, key=json.dumps)


def check_objects(host, guest, out):
    x, y, z = (int(v // 1) for v in pos_of(host))
    commands = [
        f'fill {x - 6} {y} {z - 6} {x + 6} {y + 3} {z - 6} minecraft:stone',
        f'summon minecraft:minecart {x - 4.5} {y} {z + 3.5}',
        f'summon minecraft:chest_minecart {x - 2.5} {y} {z + 3.5}',
        f'summon minecraft:furnace_minecart {x + 1.5} {y} {z + 3.5}',
        f'summon minecraft:tnt_minecart {x + 4.5} {y} {z + 3.5}',
        f'summon minecraft:oak_boat {x - 4.5} {y} {z + 6.5} {{Rotation:[45f,0f]}}',
        f'summon minecraft:birch_chest_boat {x - 1.5} {y} {z + 6.5}',
        f'summon minecraft:bamboo_raft {x + 2.5} {y} {z + 6.5}',
        f'summon minecraft:armor_stand {x - 2.5} {y} {z - 2.5} {{Rotation:[90f,0f]}}',
        f'summon minecraft:armor_stand {x - 4.5} {y} {z - 2.5} {{ShowArms:1b,Small:1b,NoBasePlate:1b,'
        'equipment:{head:{id:"minecraft:diamond_helmet",count:1}}}',
        f'summon minecraft:item_frame {x - 3.5} {y + 1} {z - 4.97} {{block_pos:[I;{x - 4},{y + 1},{z - 5}],Facing:3b,'
        'Item:{id:"minecraft:diamond",count:1},ItemRotation:3b}',
        f'summon minecraft:glow_item_frame {x - 1.5} {y + 1} {z - 4.97} {{block_pos:[I;{x - 2},{y + 1},{z - 5}],Facing:3b}}',
        f'summon minecraft:painting {x + 1.5} {y + 1} {z - 4.97} {{block_pos:[I;{x + 1},{y + 1},{z - 5}],facing:0b,'
        'variant:"minecraft:kebab"}',
        f'summon minecraft:painting {x + 4} {y + 1} {z - 4.97} {{block_pos:[I;{x + 4},{y + 1},{z - 5}],facing:0b,'
        'variant:"minecraft:courbet"}',
    ]
    try:
        for command in commands:
            host('command', command=command)
        host.wait(60)
        guest.raw('look-at', x, y, z - 5)
        guest.wait(10)
        guest.shot(out, 'objects')
        host.shot(out, 'objects')
        mine, theirs = objects_near(host, (x, y, z)), objects_near(guest, (x, y, z))
    finally:
        host('command', command=f'kill @e[type=!minecraft:player,x={x - 8},y={y - 2},z={z - 8},dx=16,dy=8,dz=16]')
        host('command', command=f'fill {x - 6} {y} {z - 6} {x + 6} {y + 3} {z - 6} minecraft:air')
    assert len(mine) == len(commands) - 1, f'the host has {len(mine)} objects: {mine}'
    assert theirs == mine, f'the guest sees {theirs}\nthe host has {mine}'
    return f'{len(theirs)} objects identical'


def check_projectiles(host, guest, out):
    x, y, z = (int(v // 1) for v in pos_of(host))
    commands = [
        f'summon minecraft:tnt {x + 4.5} {y} {z + 4.5} {{fuse:2000s}}',
        f'summon minecraft:end_crystal {x - 4.5} {y} {z + 4.5} {{ShowBottom:0b}}',
        f'summon minecraft:tnt {x + 2.5} {y + 5} {z - 3.5} {{fuse:2000s}}',
        f'summon minecraft:arrow {x - 2.5} {y + 4} {z - 2.5} {{Motion:[0.2d,0d,0d]}}',
        f'summon minecraft:falling_block {x - 0.5} {y + 6} {z - 4.5} {{BlockState:{{Name:"minecraft:sand"}}}}',
    ]
    try:
        for command in commands:
            host('command', command=command)
        host.wait(80)
        guest.raw('look-at', x, y, z)
        guest.wait(5)
        guest.shot(out, 'projectiles')
        mine = objects_near(host, (x, y, z), 'projectiles')
        theirs = objects_near(guest, (x, y, z), 'projectiles')
    finally:
        host('command', command=f'kill @e[type=!minecraft:player,x={x - 8},y={y - 2},z={z - 8},dx=16,dy=10,dz=16]')
        host('command', command=f'setblock {x - 1} {y} {z - 5} minecraft:air')
    assert len(mine) == len(commands) - 1, f'the host has {len(mine)}: {mine}'
    # The fuse counts down every tick: the guest's, read a moment later, may be a tick or two behind.
    fuses = [(m.pop('fuse', 0), t.pop('fuse', 0)) for m, t in zip(mine, theirs)]
    assert theirs == mine, f'the guest sees {theirs}\nthe host has {mine}'
    assert all(abs(a - b) <= 2 for a, b in fuses), f'fuses (host, guest): {fuses}'
    return f'{len(theirs)} identical, fuses (host, guest) {fuses}'


def light_differences(host, guest, lo, hi):
    """Positions of the box loaded on both sides, and those whose sky or block light differ."""
    h, g = host.raw('light', *lo, *hi), guest.raw('light', *lo, *hi)
    both = [i for i in range(len(h['sky'])) if h['sky'][i] != '.' and g['sky'][i] != '.']
    differ = [i for i in both if (h['sky'][i], h['block'][i]) != (g['sky'][i], g['block'][i])]
    return h, g, both, differ


def same_light(host, guest, lo, hi, what, ticks=100):
    """Wait until the guest's light over the box is the host's; the number of positions compared."""
    for _ in range(ticks // 10):
        h, g, both, differ = light_differences(host, guest, lo, hi)
        if not differ:
            assert both, f'{what}: no position of the box is loaded on both sides'
            return len(both)
        host.wait(10)
    i = differ[0]
    nx, nz = hi[0] - lo[0] + 1, hi[2] - lo[2] + 1
    pos = (lo[0] + i % nx, lo[1] + i // (nx * nz), lo[2] + i // nx % nz)
    raise AssertionError(f'{what}: {len(differ)} of {len(both)} positions differ, first {pos}: host sky '
                         f'{h["sky"][i]} block {h["block"][i]}, guest sky {g["sky"][i]} block {g["block"][i]}')


def check_light(host, guest, out):
    gx, gy, gz = (int(v // 1) for v in pos_of(guest))
    results = {}
    # A torch near the guest: the guest lights it itself, as vanilla's client does.
    torch = (gx + 3, gy, gz + 3)
    host('command', command=f'setblock {torch[0]} {torch[1]} {torch[2]} minecraft:torch')
    try:
        results['torch'] = same_light(host, guest, (gx - 12, gy - 1, gz - 12), (gx + 16, gy + 3, gz + 16), 'torch')
    finally:
        host('command', command=f'setblock {torch[0]} {torch[1]} {torch[2]} minecraft:air')
    # A roof over the guest, then a hole in it: sky light falls through.
    roof = f'{gx - 3} {gy + 4} {gz - 3} {gx + 3} {gy + 4} {gz + 3}'
    box = (gx - 6, gy - 1, gz - 6), (gx + 6, gy + 5, gz + 6)
    host('command', command=f'fill {roof} minecraft:stone')
    try:
        results['roof'] = same_light(host, guest, *box, 'roof')
        host('command', command=f'setblock {gx} {gy + 4} {gz} minecraft:air')
        results['hole'] = same_light(host, guest, *box, 'hole in the roof')
    finally:
        host('command', command=f'fill {roof} minecraft:air')
    # The last chunk the guest tracks eastwards, and glowstone high up just past it in a chunk it
    # doesn't track: the guest never sees that block, only the light updates of its border chunk.
    y = 200
    row = guest.raw('light', gx, y, gz, gx + 16 * 24, y, gz)['sky']
    assert '.' in row, 'the guest has every chunk 24 chunks east'
    untracked = (gx + row.index('.')) >> 4
    near = untracked * 16 + 1
    # On its server thread the host's own game is a client too, sent the same light updates.
    if not SERVER_THREAD:
        assert host.raw('light', near, y, gz, near, y, gz)['sky'] != '.', 'the host lacks the chunk past the guest'
    host('command', command=f'setblock {near} {y} {gz} minecraft:glowstone')
    try:
        lo, hi = (untracked * 16 - 16, y - 3, gz - 8), (untracked * 16 - 1, y + 3, gz + 8)
        results['border'] = same_light(host, guest, lo, hi, 'border chunk')
        lit = guest.raw('light', near - 2, y, gz, near - 2, y, gz)['block']
        assert lit == 'd', f'the border chunk by the glowstone has block light {lit}, not 13'
    finally:
        host('command', command=f'setblock {near} {y} {gz} minecraft:air')
    results['border_cleared'] = same_light(host, guest, lo, hi, 'border chunk without the glowstone')
    return results


CHECK_FNS = {'tab': check_tab, 'self': check_self, 'armour': check_armour, 'nametags': check_nametags,
             'chest': check_chest, 'crafting': check_crafting, 'sign': check_sign, 'damage': check_damage,
             'mobs': check_mobs, 'hazards': check_hazards, 'pvp': check_pvp,
             'respawn': check_respawn,
             'containers': check_containers, 'feedback': check_feedback, 'titles': check_titles, 'chat': check_chat,
             'machines': check_machines,
             'commands': check_commands,
             'recipes': check_recipes, 'trials': check_trials, 'walls': check_walls,
             'effects': check_effects, 'portals': check_portals, 'distance': check_distance, 'clicks': check_clicks,
             'sleep': check_sleep, 'push': check_push, 'eat': check_eat,
             'xp': check_xp, 'objects': check_objects, 'projectiles': check_projectiles,
             'light': check_light}


def run(ctl, out, resources, checks):
    host, guest = Game('host', ctl, resources), Game('guest', ctl, resources)
    if SERVER_THREAD:
        host.d.env['HORIZON_SERVER_THREAD'] = '1'
    results = {}
    try:
        host('launch', wait=False, offscreen=sys.platform == 'darwin')
        # Guests' commands need the world's Allow Commands as well as the LAN's (#372): a creative
        # world has them; everyone plays in survival.
        host.raw('open', 'Lan', 5, 'creative')
        host('wait_ready', seconds=400)
        host('command', command='defaultgamemode survival')
        host('command', command='gamemode survival')
        host('command', command='time set day')
        # An open stone floor around the host, so the players see each other and nothing is in the way.
        host('command', command='fill ~-8 ~-1 ~-8 ~8 ~-1 ~8 minecraft:stone')
        host('command', command='fill ~-8 ~ ~-8 ~8 ~6 ~8 minecraft:air')
        host('command', command='gamerule spawn_mobs false')
        port = host.raw('lan', 'open', 0, 'commands')['port']
        host.player = host.raw('lan', 'status').get('host', 'Player')  # the host's profile name
        guest('launch', wait=False, offscreen=sys.platform == 'darwin')
        guest.raw('connect', f'127.0.0.1:{port}', GUEST_NAME)
        guest.raw('wait-for', 'loaded', '--timeout', 300)
        guest.wait(40)
        face_each_other(host, guest)
        host.shot(out, 'joined')
        guest.shot(out, 'joined')
        for name in checks:
            try:
                results[name] = {'pass': True, 'detail': CHECK_FNS[name](host, guest, out)}
            except (Failed, AssertionError, StopIteration, KeyError) as e:
                results[name] = {'pass': False, 'detail': repr(e)}
                for g in (host, guest):
                    back_to_game(g)
            print(f'[lan] {name}: {"PASS" if results[name]["pass"] else "FAIL"} {results[name]["detail"]}')
    finally:
        logs = {g.name: g.close() for g in (guest, host)}
        for name, text in logs.items():
            (out / f'{name}.log').write_text(text)
        for g in (host, guest):
            shutil.rmtree(g.work, ignore_errors=True)
    return results


def main():
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument('--bin-dir', help='prebuilt horizon-client and horizon-ctl (default: build with the fast profile)')
    parser.add_argument('--check', action='append', choices=CHECKS, help='run only these checks (repeatable)')
    parser.add_argument('--out', default=str(ROOT / '.local/lan-e2e'), help='screenshots and game logs go here')
    parser.add_argument('--server-thread', action='store_true', help="the host's world runs on its server thread")
    args = parser.parse_args()
    global SERVER_THREAD
    SERVER_THREAD = args.server_thread
    if args.bin_dir:
        bins = Path(args.bin_dir).resolve()
    else:
        subprocess.run(['cargo', 'build', '--locked', '--profile', 'fast', '-p', 'horizon-client', '--bins'],
                       cwd=ROOT, check=True)
        meta = json.loads(subprocess.run(['cargo', 'metadata', '--format-version', '1', '--no-deps'], cwd=ROOT,
                                         capture_output=True, text=True, check=True).stdout)
        bins = Path(meta['target_directory']) / 'fast'
    resources = (ROOT / '.local/horizon-client/26.3').resolve()
    if not (resources / 'horizon-resources.json').exists():
        sys.exit(f'no prepared resources in {resources}')
    out = Path(args.out)
    out.mkdir(parents=True, exist_ok=True)
    results = run(bins / 'horizon-ctl', out, resources, args.check or CHECKS)
    print(json.dumps(results, indent=1))
    sys.exit(0 if results and all(r['pass'] for r in results.values()) else 1)


if __name__ == '__main__':
    main()