tests/test_world_normal_migration.py

"""The 20261003 move to normal terrain must not destroy anything.

Locks down: a fresh world directory (old maps preserved, auth untouched, no
automatic playerdata import on flat->normal), a fresh EMPTY Naldo checkpoint
(no seeding: upstream asserts an empty vectordb matches the skills file), and
normal-terrain server properties without flat layers.
"""
from __future__ import annotations

import json
from pathlib import Path

import minecraft_stack as mc
import voyager_stack as vs


def test_new_world_is_fresh_and_old_maps_are_kept():
    assert mc.WORLD_DIR.name == "world-20261003-normal-v1"
    assert mc.WORLD_DIR.parent == mc.PERSIST_ROOT
    assert mc.PREVIOUS_WORLD_DIR.name == "world-20260928-ores-y64-v1"
    assert mc.WORLD_DIR != mc.PREVIOUS_WORLD_DIR
    # The flat constants stay for the old maps and the map-contract tests;
    # the new map simply does not use them.
    assert mc.WORLD_PROFILE == "minecraft:normal"


def test_normal_server_properties_have_no_flat_layers():
    props = mc._server_properties()
    assert props["level-type"] == "minecraft:normal"
    assert props["generate-structures"] == "true"
    assert "generator-settings" not in props
    assert props["server-ip"] == "127.0.0.1"
    assert props["max-players"] == str(mc.MAX_PLAYERS)


def test_flat_profile_still_selects_the_flat_generator(monkeypatch):
    # Profile selection works both ways; the historic flat contract itself is
    # left untouched (its own test keeps the HEAD baseline).
    monkeypatch.setattr(mc, "WORLD_PROFILE", "minecraft:flat")
    props = mc._server_properties()
    assert props["level-type"] == "minecraft:flat"
    assert props["generator-settings"] == mc.FLAT_GENERATOR_SETTINGS
    assert props["generate-structures"] == "false"


def test_flat_to_normal_imports_no_player_saves(tmp_path, monkeypatch):
    previous = tmp_path / "world-20260928-ores-y64-v1"
    current = tmp_path / "world-20261003-normal-v1"
    (previous / "playerdata").mkdir(parents=True)
    human = previous / "playerdata" / "human.dat"
    human.write_bytes(b"human-save")
    (current).mkdir(parents=True)

    # Flat->normal: everyone spawns fresh at the natural spawn; importing an
    # old position buries the player inside a mountain, far from Naldo.
    assert mc.WORLD_PROFILE == "minecraft:normal"
    assert mc._should_carry_player_saves(current, previous) is False
    # The legacy helper still exists for flat->flat replacements.
    monkeypatch.setattr(mc, "WORLD_PROFILE", "minecraft:flat")
    assert mc._should_carry_player_saves(current, previous) is True
    assert mc._should_carry_player_saves(current, tmp_path / "missing") is False
    (current / "level.dat").write_bytes(b"generated")
    assert mc._should_carry_player_saves(current, previous) is False
    # Nothing is ever deleted from the old world by asking the question.
    assert human.read_bytes() == b"human-save"


def test_relay_auth_is_untouched_by_the_map_move():
    # The Xbox OAuth cache lives here; a map change must never cost the login.
    assert mc.RELAY_DIR == mc.PERSIST_ROOT / "relay"
    assert "token.json" not in str(mc.WORLD_DIR)


def test_new_checkpoint_starts_empty_and_seeding_is_gone():
    # Upstream SkillManager asserts vectordb count == len(skills): seeding a
    # non-empty skills.json without its vectordb fails the boot, and the
    # repaired defaults then flip resume=True. The new map must start empty.
    assert vs.CKPT_DIR.name == "ckpt-20261003-normal-v1"
    assert "gemini-voyager" not in str(vs.CKPT_DIR)
    assert not hasattr(vs, "_seed_skill_library"), "automatic skill seeding must be removed"
    assert not hasattr(vs, "PREV_CKPT_DIR"), "no previous-checkpoint wiring may remain"


def test_fresh_checkpoint_inherits_nothing_from_the_old_map(tmp_path):
    old = tmp_path / "ckpt-old"
    (old / "skill").mkdir(parents=True)
    (old / "skill" / "skills.json").write_text(
        json.dumps({"mineWood": {"code": "x"}}), encoding="utf-8"
    )
    (old / "curriculum").mkdir(parents=True)
    (old / "curriculum" / "completed_tasks.json").write_text('["Mine 1 wood"]', encoding="utf-8")
    (old / "curriculum" / "failed_tasks.json").write_text("[]", encoding="utf-8")

    new = tmp_path / "ckpt-20261003-normal-v1"
    new.mkdir(parents=True)
    # The boot creates the directory and nothing else: no skills, no tasks,
    # no positions are ever copied across maps.
    assert vs._is_fresh_checkpoint(new) is True
    assert vs._is_fresh_checkpoint(old) is False
    assert not (new / "skill" / "skills.json").exists()
    assert not (new / "curriculum" / "completed_tasks.json").exists()
    # The old checkpoint is preserved untouched for history.
    assert json.loads((old / "skill" / "skills.json").read_text(encoding="utf-8")) == {
        "mineWood": {"code": "x"}
    }


def test_supervisor_stops_before_shutdown_without_endpoints():
    # app needs the Space runtime (spaces/gradio) to import, so verify the
    # shutdown ordering statically: the supervisor loop must wait on an Event
    # and _shutdown must set it before any save/stop. No command endpoint.
    import ast

    tree = ast.parse((Path(__file__).resolve().parents[1] / "app.py").read_text(encoding="utf-8"))
    module = {node.name: node for node in tree.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))}
    supervisor = module["_service_supervisor"]
    shutdown = module["_shutdown"]
    supervisor_src = ast.get_source_segment(
        (Path(__file__).resolve().parents[1] / "app.py").read_text(encoding="utf-8"), supervisor
    )
    assert "_supervisor_stop.wait(30)" in supervisor_src
    shutdown_src = ast.get_source_segment(
        (Path(__file__).resolve().parents[1] / "app.py").read_text(encoding="utf-8"), shutdown
    )
    assert shutdown_src.index("_supervisor_stop.set()") < shutdown_src.index("STACK.stop()")
    assert "_supervisor_stop.set()" in shutdown_src
    routes_src = (Path(__file__).resolve().parents[1] / "app.py").read_text(encoding="utf-8")
    assert "/minecraft/save" not in routes_src and "/minecraft/stop" not in routes_src


def test_presence_diagnostic_counts_without_commands_or_names():
    presence = mc._recent_players([
        "[2026-10-03 10:00:00] minecraft> Player1 joined the game",
        "[2026-10-03 10:01:00] minecraft> Player1 left the game",
        "[2026-10-03 10:02:00] minecraft> Player2 joined the game",
    ])
    assert presence["estimated_online"] == 1
    assert presence["joins_this_boot"] == 2