houseplan-card/tests_backend/test_ha_import_export.py

"""Portable backup and authenticated endpoint contract tests (#50).

The file intentionally carries the ``test_ha_`` prefix: importing the custom
component executes its Home Assistant package initializer even when an
individual test exercises an otherwise pure helper.
"""
from __future__ import annotations

import asyncio
import json
from pathlib import Path
from types import SimpleNamespace
from typing import Any

import pytest
import voluptuous as vol
from homeassistant.core import HomeAssistant
from pytest_homeassistant_custom_component.common import MockConfigEntry

from custom_components.houseplan.import_export import (
    ImportFailure,
    build_space_merge,
    create_export,
    create_preview,
    get_candidate,
    parse_document,
    prepare_apply,
    revalidate_candidate,
)
from custom_components.houseplan import import_export as import_export_api
from custom_components.houseplan import websocket_api as wsapi
from custom_components.houseplan.http_api import HouseplanImportPreviewView, KEY_HASS
from custom_components.houseplan.const import (
    DEFAULT_CONFIG,
    DOMAIN,
    MAX_IMPORT_PREVIEWS_PER_USER,
    MAX_IMPORT_PREVIEWS_TOTAL,
    PLAN_MODEL_VERSION,
    FILES_DIR, PLANS_DIR,
)
from custom_components.houseplan.store import (
    async_save_layout_state,
    get_data,
    layout_store_payload,
    migrate_config_background_mode,
)
from custom_components.houseplan.validation import MAX_LAYOUT


@pytest.fixture(autouse=True)
def _enable_custom_integrations(enable_custom_integrations):
    """Allow loading custom_components in the HA harness."""
    yield


def _config() -> dict:
    return {
        "model_version": 6,
        "spaces": [{
            "id": "ground", "title": "Ground", "view_box": [0, 0, 1, 1],
            "rooms": [{"id": "living", "name": "Living", "poly": [[0, 0], [1, 0], [1, 1]]}],
            "plan_url": None,
        }],
        "markers": [{
            "id": "lamp", "binding": "entity:light.living", "space": "ground",
            "room_id": "living", "controls": ["switch.wall"],
        }],
        "settings": {
            "bg_mode": "static",
            "known_devices": ["old-device"],
            "new_device_ids": ["new-device"],
        },
    }


def _document(tmp_path: Path, kind: str = "full") -> dict:
    runtime = SimpleNamespace(instance_id="instance-a")
    document, _filename = create_export(
        runtime,
        {"config": _config(), "rev": 2},
        {"layout": {"lamp": {"x": 0.4, "y": 0.5, "s": "ground"}}, "rev": 3},
        kind=kind,
        space_id="ground" if kind == "space" else None,
        card_version="1.61.0",
        config_root=tmp_path,
    )
    return document


def test_background_defaults_and_store_migration_preserve_legacy_view() -> None:
    assert DEFAULT_CONFIG["settings"]["bg_mode"] == "daynight"

    legacy = {
        "config": {
            "spaces": [{
                "id": "ground",
                "settings": {"bg_mode": "daynight", "future_space": 1},
            }],
            "markers": [],
            "settings": {"future_global": {"kept": True}},
            "future_root": "kept",
        },
        "rev": 17,
        "future_store": [1, 2],
    }
    migrated = migrate_config_background_mode(legacy)
    assert migrated["config"]["settings"] == {
        "future_global": {"kept": True}, "bg_mode": "static",
    }
    assert migrated["config"]["spaces"][0]["settings"] == {
        "bg_mode": "daynight", "future_space": 1,
    }
    assert migrated["rev"] == 17 and migrated["future_store"] == [1, 2]
    assert "bg_mode" not in legacy["config"]["settings"]
    assert migrate_config_background_mode(migrated) is migrated

    layout = {"layout": {}, "rev": 4}
    assert migrate_config_background_mode(layout) is layout


def test_background_mode_is_materialized_across_export_and_legacy_import(tmp_path: Path) -> None:
    runtime = SimpleNamespace(instance_id="instance-a")
    inherited = _config()
    inherited["settings"]["bg_mode"] = "daynight"
    inherited["spaces"][0].setdefault("settings", {}).pop("bg_mode", None)

    space_document, _ = create_export(
        runtime, {"config": inherited}, {"layout": {}},
        kind="space", space_id="ground", card_version="1.64.0", config_root=tmp_path,
    )
    exported_space = space_document["payload"]["config"]["spaces"][0]
    assert exported_space["settings"]["bg_mode"] == "daynight"

    full_document, _ = create_export(
        runtime, {"config": {**_config(), "settings": {}}}, {"layout": {}},
        kind="full", space_id=None, card_version="1.64.0", config_root=tmp_path,
    )
    assert full_document["payload"]["config"]["settings"]["bg_mode"] == "static"

    full_document["payload"]["config"]["settings"].pop("bg_mode")
    parsed_full = parse_document(json.dumps(full_document).encode())
    assert parsed_full["payload"]["config"]["settings"]["bg_mode"] == "static"

    space_document["payload"]["config"]["spaces"][0]["settings"].pop("bg_mode")
    parsed_space = parse_document(json.dumps(space_document).encode())
    assert parsed_space["payload"]["config"]["spaces"][0]["settings"]["bg_mode"] == "static"
    target = _config()
    target["settings"]["bg_mode"] = "daynight"
    merged, _layout, _details = build_space_merge(parsed_space, target, {}, "skip")
    assert merged["spaces"][-1]["settings"]["bg_mode"] == "static"


def test_strict_parser_rejects_duplicate_proto_and_future_model(tmp_path: Path) -> None:
    with pytest.raises(ImportFailure, match="Duplicate") as duplicate:
        parse_document(b'{"format":"houseplan-export","format":"houseplan-export"}')
    assert duplicate.value.code == "invalid_json"
    with pytest.raises(ImportFailure) as proto:
        parse_document(b'{"__proto__":{},"format":"houseplan-export"}')
    assert proto.value.code == "invalid_json"
    document = _document(tmp_path)
    document["model_version"] = 999
    with pytest.raises(ImportFailure) as future:
        parse_document(json.dumps(document).encode())
    assert future.value.code == "future_model"
    document = _document(tmp_path)
    document["model_version"] = True
    with pytest.raises(ImportFailure) as boolean:
        parse_document(json.dumps(document).encode())
    assert boolean.value.code == "invalid_format"


@pytest.mark.parametrize("value", ["abc", {"bad": 1}, -1, 10**20, True])
def test_parser_rejects_invalid_dropped_marker_link_counter(tmp_path: Path, value: Any) -> None:
    document = _document(tmp_path)
    document["transfer"]["dropped_marker_links"] = value
    with pytest.raises(ImportFailure) as invalid:
        parse_document(json.dumps(document).encode())
    assert invalid.value.code == "invalid_format"


def test_full_preview_drops_dormant_broken_and_duplicate_links(tmp_path: Path) -> None:
    document = _document(tmp_path)
    markers = document["payload"]["config"]["markers"]
    markers.extend([
        {"id": "plain", "binding": "virtual", "is_light": False},
        {"id": "dumb", "binding": "virtual", "is_light": True},
        {
            "id": "controller", "binding": "virtual",
            "controls": [
                "marker:missing", "marker:plain", "marker:dumb", "marker:dumb",
                "switch.keep",
            ],
        },
    ])
    runtime = SimpleNamespace(instance_id="instance-a", import_previews={})
    response = create_preview(
        runtime, json.dumps(document).encode(), owner_id="alice", duplicate_policy="skip",
        current_config_data={"config": _config(), "rev": 1},
        current_layout_data={"layout": {}, "rev": 1}, config_root=tmp_path,
    )
    assert response["preview"]["dropped_marker_links"] == 3
    candidate = get_candidate(runtime, response["token"], "alice")
    imported = candidate["document"]["payload"]["config"]
    controller = next(marker for marker in imported["markers"] if marker["id"] == "controller")
    assert controller["controls"] == ["marker:dumb", "switch.keep"]


@pytest.mark.parametrize("kind", ["self", "cycle"])
def test_full_preview_still_rejects_self_and_cycle(kind: str, tmp_path: Path) -> None:
    document = _document(tmp_path)
    markers = document["payload"]["config"]["markers"]
    if kind == "self":
        markers.append({
            "id": "self", "binding": "virtual", "is_light": True,
            "controls": ["marker:self"],
        })
        expected = "marker_control_self"
    else:
        markers.extend([
            {"id": "a", "binding": "virtual", "is_light": True, "controls": ["marker:b"]},
            {"id": "b", "binding": "virtual", "is_light": True, "controls": ["marker:a"]},
        ])
        expected = "marker_control_cycle"
    with pytest.raises(ImportFailure) as invalid:
        create_preview(
            SimpleNamespace(instance_id="instance-a", import_previews={}),
            json.dumps(document).encode(), owner_id="alice", duplicate_policy="skip",
            current_config_data={"config": _config(), "rev": 1},
            current_layout_data={"layout": {}, "rev": 1}, config_root=tmp_path,
        )
    assert invalid.value.code == expected


def test_layout_writer_preserves_unrelated_metadata_and_removes_only_named_keys() -> None:
    stored = {
        "layout": {"old": {"x": 0, "y": 0}}, "rev": 4,
        "repair_backup": {"space": "ground"}, "optimize_pending": {"target": True},
        "future_metadata": {"kept": True},
    }
    result = layout_store_payload(
        stored, {"new": {"x": 1, "y": 1}}, 5, remove=("optimize_pending",)
    )
    assert result["repair_backup"] == {"space": "ground"}
    assert result["future_metadata"] == {"kept": True}
    assert "optimize_pending" not in result
    assert result["layout"] == {"new": {"x": 1, "y": 1}}
    assert result["rev"] == 5


def test_full_export_has_versioned_envelope_and_live_layout(tmp_path: Path) -> None:
    config = _config()
    config["markers"].append({"id": "gone", "binding": "virtual", "removed": True})
    runtime = SimpleNamespace(instance_id="instance-a")
    document, filename = create_export(
        runtime,
        {"config": config, "rev": 1},
        {"layout": {
            "lamp": {"x": 0.4, "y": 0.5, "s": "ground"},
            "gone": {"x": 0.1, "y": 0.2, "s": "ground"},
        }},
        kind="full", space_id=None, card_version="1.61.0", config_root=tmp_path,
    )
    assert document["format"] == "houseplan-export"
    assert document["export_version"] == 1
    assert document["model_version"] == 6
    assert "model_version" not in document["payload"]["config"]
    assert document["source_fingerprint"].startswith("sha256:")
    assert set(document["payload"]["layout"]) == {"lamp"}
    assert document["placement_manifest"][0]["layout_id"] == "lamp"
    assert filename.startswith("houseplan-full-") and filename.endswith(".json")


def _plan_only_source() -> tuple[dict[str, Any], dict[str, Any]]:
    config = _config()
    space = config["spaces"][0]
    space.update({
        "cell_cm": 5,
        "plan_url": "https://example.invalid/floor.svg",
        "plan_x": 0.1,
        "plan_y": -0.2,
        "plan_scale_x": 1.2,
        "plan_scale_y": 0.8,
        "plan_angle": 15,
        "settings": {
            "show_names": True, "north_deg": 90, "sun_rays": True,
            "future_binding": "sensor.secret",
        },
        "future_space": {"entity": "sensor.secret"},
        "openings": [{
            "id": "window", "type": "window", "x": 0.5, "y": 0,
            "angle": 0, "length": 0.2, "contact": "binary_sensor.window",
            "lock": "lock.window", "invert": True, "flip_h": True,
            "future_opening": "sensor.secret",
        }],
        "walls": [{"key": "wall-1", "cm": 20, "future_wall": "sensor.secret"}],
        "room_drafts": [{
            "id": "draft", "points": [[0, 0], [0.2, 0]],
            "segments": [{"cm": 10, "future_segment": "sensor.secret"}],
        }],
        "partitions": [{"id": "partition", "a": [0, 0.5], "b": [1, 0.5], "cm": 12}],
        "wall_columns": [{"id": "column", "shape": "circle", "center": [0.2, 0.2], "cm": 30}],
        "open_spans": [{"a": [0.4, 0], "b": [0.6, 0], "future_span": "sensor.secret"}],
        "decor": [
            {
                "id": "modern", "kind": "text", "x": 0.2, "y": 0.2,
                "text": "Temp {sensor.kitchen} / {climate.hall.current_temperature} °C",
                "future_decor": "sensor.secret",
            },
            {
                "id": "legacy", "kind": "text", "x": 0.3, "y": 0.3,
                "text": "Tank {} / {}", "entity": "sensor.tank",
                "attr": "level", "unit": "%",
            },
            {
                "id": "static", "kind": "text", "x": 0.4, "y": 0.4,
                "text": "Literal {not a reference} and sensor.user_text",
            },
            {
                "id": "furniture", "kind": "furniture", "symbol": "sofa",
                "x": 0.5, "y": 0.5, "w": 0.2, "h": 0.1, "angle": 10,
            },
        ],
    })
    room = space["rooms"][0]
    room.update({
        "area": "living-area",
        "settings": {
            "fill_mode": "custom", "custom_fill": {"c": "#123456", "a": 0.4},
            "temp_source": "sensor.room_temp", "hum_source": "sensor.room_humidity",
            "future_room_binding": "sensor.secret",
        },
        "future_room": "sensor.secret",
    })
    config["markers"].append({
        "id": "note", "binding": "virtual", "space": "ground",
        "room_id": "living", "name": "Boiler", "icon": "mdi:fire",
    })
    layout = {
        "lamp": {"x": 0.4, "y": 0.5, "s": "ground"},
        "note": {"x": 0.2, "y": 0.3, "s": "ground"},
        "lg_light.group": {"x": 0.1, "y": 0.1, "s": "ground"},
        "auto-device": {"x": 0.8, "y": 0.8, "s": "ground"},
        "rl_living": {"x": 0.45, "y": 0.55, "s": "ground", "future": "drop"},
        "rl_other": {"x": 0.1, "y": 0.1, "s": "other"},
    }
    return config, layout


def test_plan_only_export_projects_geometry_and_round_trips_room_labels(tmp_path: Path) -> None:
    config, layout = _plan_only_source()
    runtime = SimpleNamespace(instance_id="instance-a", import_previews={})
    document, _ = create_export(
        runtime, {"config": config}, {"layout": layout}, kind="space",
        space_id="ground", plan_only=True, card_version="review", config_root=tmp_path,
    )
    payload = document["payload"]
    exported = payload["config"]["spaces"][0]
    assert document["transfer"] == {"dropped_marker_links": 0, "plan_only": True}
    assert payload["config"]["markers"] == []
    assert payload["layout"] == {
        "rl_living": {"x": 0.45, "y": 0.55, "s": "ground"},
    }
    assert document["placement_manifest"] == [{
        "layout_id": "rl_living", "space_id": "ground", "owner": "room_label",
        "owner_id": "living", "binding": None, "label": None, "icon": None,
    }]
    assert document["content_manifest"][0]["url"] == "https://example.invalid/floor.svg"
    assert all(item.get("owner") != "marker" for item in document["content_manifest"])

    assert "future_space" not in exported
    assert exported["settings"] == {
        "show_names": True, "north_deg": 90, "sun_rays": True, "bg_mode": "static",
    }
    room = exported["rooms"][0]
    assert "area" not in room and "future_room" not in room
    assert room["settings"] == {
        "fill_mode": "custom", "custom_fill": {"c": "#123456", "a": 0.4},
    }
    opening = exported["openings"][0]
    assert opening["flip_h"] is True
    assert not {"contact", "lock", "invert", "future_opening"} & set(opening)
    by_id = {item["id"]: item for item in exported["decor"]}
    assert by_id["modern"]["text"] == "Temp — / — °C"
    assert by_id["legacy"]["text"] == "Tank — / —"
    assert not {"entity", "attr", "unit"} & set(by_id["legacy"])
    assert by_id["static"]["text"] == "Literal {not a reference} and sensor.user_text"
    assert by_id["furniture"]["symbol"] == "sofa"

    parsed = parse_document(json.dumps(document).encode())
    preview = create_preview(
        runtime, json.dumps(document).encode(), owner_id="alice", duplicate_policy="skip",
        current_config_data={"config": {"spaces": [], "markers": []}, "rev": 0},
        current_layout_data={"layout": {}, "rev": 0}, config_root=tmp_path,
        registry_snapshot={"areas": set()},
    )
    assert preview["preview"]["plan_only"] is True
    assert preview["preview"]["counts"]["markers"] == 0
    assert preview["preview"]["counts"]["layout"] == 1
    assert preview["preview"]["duplicates"] == 0
    assert preview["preview"]["missing_areas"] == []
    assert preview["preview"]["bindings"] == {
        "device": 0, "entity": 0, "virtual": 0,
        "active": 0, "disabled": 0, "missing": 0,
    }
    candidate = get_candidate(runtime, preview["token"], "alice")
    refreshed = revalidate_candidate(
        candidate, {"config": {"spaces": [], "markers": []}, "rev": 2},
        {"layout": {}, "rev": 3}, duplicate_policy="skip", config_root=tmp_path,
    )
    assert refreshed["preview"]["plan_only"] is True
    merged, merged_layout, details = prepare_apply(
        candidate, {"spaces": [], "markers": []}, {}, confirm_missing_content=False,
    )
    assert merged["markers"] == []
    imported_room = merged["spaces"][0]["rooms"][0]
    assert imported_room.get("area") is None
    assert merged_layout == {
        "rl_" + imported_room["id"]: {
            "x": 0.45, "y": 0.55, "s": details["space_id"],
        },
    }
    assert parsed["transfer"]["plan_only"] is True


def test_ordinary_space_export_is_unchanged_when_plan_only_is_false(tmp_path: Path) -> None:
    config, layout = _plan_only_source()
    runtime = SimpleNamespace(instance_id="instance-a")
    implicit, _implicit_name = create_export(
        runtime, {"config": config}, {"layout": layout}, kind="space",
        space_id="ground", card_version="review", config_root=tmp_path,
    )
    explicit, _explicit_name = create_export(
        runtime, {"config": config}, {"layout": layout}, kind="space",
        space_id="ground", plan_only=False, card_version="review", config_root=tmp_path,
    )
    for value in (implicit, explicit):
        value.pop("created_at")
    assert implicit == explicit
    assert "plan_only" not in implicit["transfer"]
    assert implicit["payload"]["config"]["markers"]
    assert set(implicit["payload"]["layout"]) == {
        key for key, pos in layout.items() if pos.get("s") == "ground"
    }


@pytest.mark.parametrize(
    "mutation",
    [
        "marker", "layout", "area", "temperature", "opening", "legacy_decor",
        "live_token", "unknown", "placement", "marker_content", "invalid_content",
    ],
)
def test_parser_rejects_forged_plan_only_privacy_claim(
    tmp_path: Path, mutation: str,
) -> None:
    config, layout = _plan_only_source()
    document, _ = create_export(
        SimpleNamespace(instance_id="instance-a"), {"config": config}, {"layout": layout},
        kind="space", space_id="ground", plan_only=True,
        card_version="review", config_root=tmp_path,
    )
    space = document["payload"]["config"]["spaces"][0]
    if mutation == "marker":
        document["payload"]["config"]["markers"].append({
            "id": "forged", "binding": "virtual", "name": "Forged",
        })
    elif mutation == "layout":
        document["payload"]["layout"]["sensor.secret"] = {
            "x": 0.1, "y": 0.2, "s": "ground",
        }
        document["placement_manifest"].append({
            "layout_id": "sensor.secret", "space_id": "ground",
            "owner": "auto_device", "owner_id": "sensor.secret",
            "binding": "device:sensor.secret", "label": None, "icon": None,
        })
    elif mutation == "area":
        space["rooms"][0]["area"] = "secret-area"
    elif mutation == "temperature":
        space["rooms"][0].setdefault("settings", {})["temp_source"] = "sensor.secret"
    elif mutation == "opening":
        space["openings"][0]["contact"] = "binary_sensor.secret"
    elif mutation == "legacy_decor":
        space["decor"][0]["entity"] = "sensor.secret"
    elif mutation == "live_token":
        space["decor"][0]["text"] = "Leaked {sensor.secret}"
    elif mutation == "unknown":
        space["future_binding"] = "sensor.secret"
    elif mutation == "placement":
        document["placement_manifest"][0]["owner"] = "auto_device"
    elif mutation == "marker_content":
        document["content_manifest"].append({"owner": "marker"})
    elif mutation == "invalid_content":
        document["content_manifest"].append("sensor.secret")
    with pytest.raises(ImportFailure) as invalid:
        parse_document(json.dumps(document).encode())
    assert invalid.value.code == "invalid_format"


@pytest.mark.parametrize("value", [1, "true", None, {}, []])
def test_parser_requires_strict_plan_only_boolean(tmp_path: Path, value: Any) -> None:
    document = _document(tmp_path, "space")
    document["transfer"]["plan_only"] = value
    with pytest.raises(ImportFailure) as invalid:
        parse_document(json.dumps(document).encode())
    assert invalid.value.code == "invalid_format"


def test_plan_only_cannot_be_requested_for_full_export(tmp_path: Path) -> None:
    with pytest.raises(ImportFailure) as invalid:
        create_export(
            SimpleNamespace(instance_id="instance-a"), {"config": _config()}, {"layout": {}},
            kind="full", space_id=None, plan_only=True,
            card_version="review", config_root=tmp_path,
        )
    assert invalid.value.code == "invalid_format"


def test_full_export_import_round_trip_restores_model_version(tmp_path: Path) -> None:
    """The portable envelope version must return to the persisted config."""
    document = _document(tmp_path)
    runtime = SimpleNamespace(instance_id="instance-a", import_previews={})
    response = create_preview(
        runtime, json.dumps(document).encode(), owner_id="alice", duplicate_policy="skip",
        current_config_data={"config": {"spaces": [], "markers": []}, "rev": 0},
        current_layout_data={"layout": {}, "rev": 0}, config_root=tmp_path,
    )
    candidate = get_candidate(runtime, response["token"], "alice")
    config, _layout, _details = prepare_apply(
        candidate, {"spaces": [], "markers": []}, {}, confirm_missing_content=False,
    )
    assert config["model_version"] == document["model_version"] == PLAN_MODEL_VERSION
    assert config == _config()


def test_preview_tokens_are_bounded_owned_expiring_and_single_use(
    tmp_path: Path, monkeypatch,
) -> None:
    runtime = SimpleNamespace(instance_id="instance-a", import_previews={})
    raw = json.dumps(_document(tmp_path)).encode()
    # Separate the two equal production limits so this assertion still kills a
    # mutant that accidentally removes only the per-user eviction branch.
    monkeypatch.setattr(
        import_export_api, "MAX_IMPORT_PREVIEWS_TOTAL", MAX_IMPORT_PREVIEWS_PER_USER + 5,
    )
    tokens = []
    for _index in range(MAX_IMPORT_PREVIEWS_PER_USER + 1):
        response = create_preview(
            runtime, raw, owner_id="alice", duplicate_policy="skip",
            current_config_data={"config": _config(), "rev": 1},
            current_layout_data={"layout": {}, "rev": 1}, config_root=tmp_path,
        )
        tokens.append(response["token"])
    assert len(runtime.import_previews) == MAX_IMPORT_PREVIEWS_PER_USER
    assert tokens[0] not in runtime.import_previews

    live = tokens[-1]
    with pytest.raises(ImportFailure) as wrong_owner:
        get_candidate(runtime, live, "bob")
    assert wrong_owner.value.code == "preview_owner_mismatch"
    get_candidate(runtime, live, "alice", consume=True)
    with pytest.raises(ImportFailure) as consumed:
        get_candidate(runtime, live, "alice")
    assert consumed.value.code == "preview_expired"

    expiring = create_preview(
        runtime, raw, owner_id="alice", duplicate_policy="skip",
        current_config_data={"config": _config(), "rev": 1},
        current_layout_data={"layout": {}, "rev": 1}, config_root=tmp_path,
    )["token"]
    runtime.import_previews[expiring]["expires"] = 0
    with pytest.raises(ImportFailure) as expired:
        get_candidate(runtime, expiring, "alice")
    assert expired.value.code == "preview_expired"
    assert expiring not in runtime.import_previews

    monkeypatch.setattr(import_export_api, "MAX_IMPORT_PREVIEWS_TOTAL", MAX_IMPORT_PREVIEWS_TOTAL)
    global_runtime = SimpleNamespace(instance_id="instance-a", import_previews={})
    global_tokens = [
        create_preview(
            global_runtime, raw, owner_id=f"owner-{index}", duplicate_policy="skip",
            current_config_data={"config": _config(), "rev": 1},
            current_layout_data={"layout": {}, "rev": 1}, config_root=tmp_path,
        )["token"]
        for index in range(MAX_IMPORT_PREVIEWS_TOTAL + 1)
    ]
    assert len(global_runtime.import_previews) == MAX_IMPORT_PREVIEWS_TOTAL
    assert global_tokens[0] not in global_runtime.import_previews


@pytest.mark.parametrize(("stored", "envelope"), [(None, 0), (0, 0), (3, 3), (9, 9)])
def test_export_preserves_actual_model_only_in_the_envelope(
    tmp_path: Path, stored: int | None, envelope: int,
) -> None:
    config = _config()
    if stored is None:
        config.pop("model_version", None)
    else:
        config["model_version"] = stored
    document, _ = create_export(
        SimpleNamespace(instance_id="instance-a"),
        {"config": config}, {"layout": {}}, kind="full", space_id=None,
        card_version="1.61.0", config_root=tmp_path,
    )
    assert document["model_version"] == envelope
    assert "model_version" not in document["payload"]["config"]
    if envelope > PLAN_MODEL_VERSION:
        with pytest.raises(ImportFailure) as future:
            parse_document(json.dumps(document).encode())
        assert future.value.code == "future_model"


def test_export_rejects_boolean_stored_model_before_schema_coercion(tmp_path: Path) -> None:
    config = _config()
    config["model_version"] = True
    with pytest.raises(ImportFailure) as invalid:
        create_export(
            SimpleNamespace(instance_id="instance-a"), {"config": config}, {"layout": {}},
            kind="full", space_id=None, card_version="1.61.0", config_root=tmp_path,
        )
    assert invalid.value.code == "invalid_config"


def test_space_import_remaps_owned_ids_and_duplicate_policy(tmp_path: Path) -> None:
    document = _document(tmp_path, "space")
    current = _config()
    current["spaces"][0]["id"] = "existing"
    current["spaces"][0]["title"] = "Ground"
    current["markers"][0]["space"] = "existing"
    merged, layout, details = build_space_merge(
        document, current, {"lamp": {"x": 0.1, "y": 0.1, "s": "existing"}}, "skip"
    )
    assert len(merged["spaces"]) == 2
    assert merged["spaces"][-1]["id"] != "ground"
    assert merged["spaces"][-1]["title"] == "Ground (2)"
    assert details["duplicates"] >= 1 and details["skipped"] == 1
    assert all(position["s"] == details["space_id"] for key, position in layout.items() if key != "lamp")

    virtual, _layout, vdetails = build_space_merge(document, current, {}, "virtual")
    copied = virtual["markers"][-1]
    assert copied["binding"] == "virtual"
    assert copied["display"] == "static_icon"
    assert "controls" not in copied and vdetails["virtualized"] == 1


def test_preview_is_owner_bound_and_foreign_full_drops_discovery_lists(tmp_path: Path) -> None:
    document = _document(tmp_path)
    document["source_fingerprint"] = "sha256:foreign"
    raw = json.dumps(document).encode()
    runtime = SimpleNamespace(instance_id="instance-a", import_previews={})
    response = create_preview(
        runtime, raw, owner_id="alice", duplicate_policy="skip",
        current_config_data={"config": _config(), "rev": 4},
        current_layout_data={"layout": {}, "rev": 5}, config_root=tmp_path,
    )
    with pytest.raises(ImportFailure) as owner:
        get_candidate(runtime, response["token"], "bob")
    assert owner.value.code == "preview_owner_mismatch"
    candidate = get_candidate(runtime, response["token"], "alice")
    config, _layout, _details = prepare_apply(
        candidate, _config(), {}, confirm_missing_content=True,
    )
    assert "known_devices" not in config["settings"]
    assert "new_device_ids" not in config["settings"]


def test_preview_candidate_digest_and_global_memory_cap_are_enforced(tmp_path: Path) -> None:
    document = _document(tmp_path)
    runtime = SimpleNamespace(instance_id="instance-a", import_previews={})
    tokens = []
    for owner in range(MAX_IMPORT_PREVIEWS_TOTAL + 1):
        response = create_preview(
            runtime, json.dumps(document).encode(), owner_id=f"owner-{owner}",
            duplicate_policy="skip", current_config_data={"config": _config(), "rev": 1},
            current_layout_data={"layout": {}, "rev": 1}, config_root=tmp_path,
        )
        tokens.append(response["token"])
    assert len(runtime.import_previews) == MAX_IMPORT_PREVIEWS_TOTAL
    assert tokens[0] not in runtime.import_previews

    candidate = runtime.import_previews[tokens[-1]]
    candidate["document"]["payload"]["config"]["settings"]["mutated"] = True
    with pytest.raises(ImportFailure) as changed:
        get_candidate(runtime, tokens[-1], f"owner-{MAX_IMPORT_PREVIEWS_TOTAL}")
    assert changed.value.code == "invalid_format"


def test_foreign_content_never_binds_to_same_named_local_file(tmp_path: Path) -> None:
    plans = tmp_path / PLANS_DIR
    plans.mkdir(parents=True, exist_ok=True)
    (plans / "portable.svg").write_text("<svg/>", encoding="utf-8")
    config = _config()
    config["spaces"][0]["plan_url"] = "/api/houseplan/content/plans/_/portable.svg"
    document, _ = create_export(
        SimpleNamespace(instance_id="source-instance"), {"config": config}, {"layout": {}},
        kind="full", space_id=None, card_version="review", config_root=tmp_path,
    )
    runtime = SimpleNamespace(instance_id="target-instance", import_previews={})
    response = create_preview(
        runtime, json.dumps(document).encode(), owner_id="alice", duplicate_policy="skip",
        current_config_data={"config": _config(), "rev": 1},
        current_layout_data={"layout": {}, "rev": 1}, config_root=tmp_path,
    )
    row = response["preview"]["content"][0]
    assert row["exists_on_target"] is True
    assert row["state"] == "detach_required"
    assert response["preview"]["confirmation_required"] is True
    candidate = get_candidate(runtime, response["token"], "alice")
    with pytest.raises(ImportFailure) as unconfirmed:
        prepare_apply(candidate, _config(), {}, confirm_missing_content=False)
    assert unconfirmed.value.code == "content_confirmation_required"


def test_preview_counts_only_stored_wall_entries(tmp_path: Path) -> None:
    document = _document(tmp_path)
    document["payload"]["config"].pop("model_version", None)
    response = create_preview(
        SimpleNamespace(instance_id="instance-a", import_previews={}),
        json.dumps(document).encode(), owner_id="alice", duplicate_policy="skip",
        current_config_data={"config": _config(), "rev": 1},
        current_layout_data={"layout": {}, "rev": 1}, config_root=tmp_path,
    )
    assert response["preview"]["counts"]["walls"] == 0


def test_parser_caps_placement_manifest_rows(tmp_path: Path) -> None:
    document = _document(tmp_path)
    document["placement_manifest"] = [
        {"layout_id": f"row-{index}"} for index in range(MAX_LAYOUT + 1)
    ]
    with pytest.raises(ImportFailure) as oversized:
        parse_document(json.dumps(document).encode())
    assert oversized.value.code == "too_large"


def test_preview_rejects_manifest_that_omits_a_payload_reference(tmp_path: Path) -> None:
    config = _config()
    config["spaces"][0]["plan_url"] = "/api/houseplan/content/plans/_/plan.svg"
    runtime = SimpleNamespace(instance_id="instance-a")
    document, _ = create_export(
        runtime, {"config": config}, {"layout": {}}, kind="full", space_id=None,
        card_version="1.61.0", config_root=tmp_path,
    )
    document["content_manifest"] = []
    with pytest.raises(ImportFailure) as invalid:
        create_preview(
            SimpleNamespace(instance_id="instance-a", import_previews={}),
            json.dumps(document).encode(), owner_id="alice", duplicate_policy="skip",
            current_config_data={"config": _config(), "rev": 1},
            current_layout_data={"layout": {}, "rev": 1}, config_root=tmp_path,
        )
    assert invalid.value.code == "invalid_content"


def test_revalidate_full_refreshes_revisions_and_registry_summary(tmp_path: Path) -> None:
    document = _document(tmp_path)
    runtime = SimpleNamespace(instance_id="instance-a", import_previews={})
    response = create_preview(
        runtime, json.dumps(document).encode(), owner_id="alice", duplicate_policy="skip",
        current_config_data={"config": _config(), "rev": 1},
        current_layout_data={"layout": {}, "rev": 2}, config_root=tmp_path,
    )
    candidate = get_candidate(runtime, response["token"], "alice")
    refreshed = revalidate_candidate(
        candidate, {"config": _config(), "rev": 7}, {"layout": {}, "rev": 8},
        duplicate_policy="skip", config_root=tmp_path,
        registry_snapshot={
            "active_device": set(), "disabled_device": set(),
            "active_entity": {"light.living"}, "disabled_entity": set(), "areas": set(),
        },
    )
    assert refreshed["expected_config_rev"] == 7
    assert refreshed["expected_layout_rev"] == 8
    assert refreshed["preview"]["bindings"]["active"] == 1


def test_space_remap_covers_marker_id_layout_and_vacuum_segment_map(tmp_path: Path) -> None:
    document = _document(tmp_path, "space")
    document["payload"]["config"]["markers"][0]["vacuum"] = {
        "segment_map": {"12": "living"},
    }
    merged, layout, details = build_space_merge(
        document, {"spaces": [], "markers": [], "settings": {}}, {}, "skip",
    )
    imported = next(space for space in merged["spaces"] if space["id"] == details["space_id"])
    imported_room = imported["rooms"][0]["id"]
    marker = next(marker for marker in merged["markers"] if marker.get("space") == details["space_id"])
    assert marker["id"] != "lamp"
    assert marker["vacuum"]["segment_map"] == {"12": imported_room}
    assert marker["id"] in layout and "lamp" not in layout


def test_space_merge_remaps_every_space_owned_id_and_room_link(tmp_path: Path) -> None:
    document = _document(tmp_path, "space")
    space = document["payload"]["config"]["spaces"][0]
    space["rooms"] = [
        {"id": "living", "name": "Living", "poly": [[0, 0], [1, 0], [1, 0.5]],
         "open_to": ["kitchen"]},
        {"id": "kitchen", "name": "Kitchen", "poly": [[0, 0.5], [1, 0.5], [1, 1]]},
    ]
    space["openings"] = [
        {"id": "op1", "type": "door", "x": 0.5, "y": 0.5, "angle": 0, "length": 0.1},
    ]
    space["decor"] = [
        {"id": "dec1", "kind": "line", "x1": 0, "y1": 0, "x2": 1, "y2": 1,
         "color": "#ffffff", "opacity": 1, "width_cm": 1},
    ]
    space["partitions"] = [{"id": "part1", "a": [0, 0.25], "b": [1, 0.25], "cm": 15}]
    space["wall_columns"] = [
        {"id": "column1", "shape": "square", "center": [0.2, 0.2], "cm": 30},
    ]
    space["room_drafts"] = [{
        "id": "draft1", "points": [[0, 0], [0.2, 0], [0.2, 0.2]],
        "segments": [{"cm": 10}, {"cm": 10}],
    }]
    merged, _layout, details = build_space_merge(
        document, {"spaces": [], "markers": [], "settings": {}}, {}, "skip",
    )
    imported = next(item for item in merged["spaces"] if item["id"] == details["space_id"])
    old_ids = {"ground", "living", "kitchen", "op1", "dec1", "part1", "column1", "draft1"}
    imported_ids = {imported["id"]}
    for collection in ("rooms", "openings", "decor", "partitions", "wall_columns", "room_drafts"):
        imported_ids.update(str(item["id"]) for item in imported.get(collection) or [])
    assert not old_ids & imported_ids
    room_ids = {room["id"] for room in imported["rooms"]}
    assert set(imported["rooms"][0]["open_to"]) <= room_ids


def test_space_merge_remaps_internal_marker_light_links(tmp_path: Path) -> None:
    document = _document(tmp_path, "space")
    markers = document["payload"]["config"]["markers"]
    markers[0]["name"] = "Living lamp"
    markers[0]["controls"] = ["marker:dumb", "light.external"]
    markers.append({
        "id": "dumb", "binding": "virtual", "space": "ground",
        "room_id": "living", "name": "Dumb lamp", "is_light": True,
    })
    merged, _layout, details = build_space_merge(
        document, {"spaces": [], "markers": [], "settings": {}}, {}, "skip",
    )
    imported = [m for m in merged["markers"] if m.get("space") == details["space_id"]]
    by_name = {m.get("name"): m for m in imported}
    assert by_name["Living lamp"]["controls"] == [
        "marker:" + by_name["Dumb lamp"]["id"], "light.external",
    ]
    assert details["dropped_marker_links"] == 0


def test_issue_90_space_merge_remaps_internal_badge_marker_link(tmp_path: Path) -> None:
    document = _document(tmp_path, "space")
    markers = document["payload"]["config"]["markers"]
    markers[0]["name"] = "Controller"
    markers[0]["value_badge"] = {
        "enabled": True,
        "source": {"kind": "derived_marker_state", "ref": "marker:dumb"},
        "position": "bottom",
    }
    markers.append({
        "id": "dumb", "binding": "virtual", "space": "ground",
        "room_id": "living", "name": "Dumb lamp", "is_light": True,
    })
    merged, _layout, details = build_space_merge(
        document, {"spaces": [], "markers": [], "settings": {}}, {}, "skip",
    )
    imported = [m for m in merged["markers"] if m.get("space") == details["space_id"]]
    by_name = {m.get("name"): m for m in imported}
    assert by_name["Controller"]["value_badge"]["source"]["ref"] == \
        "marker:" + by_name["Dumb lamp"]["id"]
    assert details["dropped_marker_links"] == 0


def test_space_export_drops_marker_links_outside_selection(tmp_path: Path) -> None:
    config = _config()
    config["markers"][0]["controls"] = ["marker:outside", "light.keep"]
    config["markers"].append({
        "id": "outside", "binding": "virtual", "space": "other",
        "name": "Other lamp", "is_light": True,
    })
    document, _ = create_export(
        SimpleNamespace(instance_id="instance-a"), {"config": config},
        {"layout": {"lamp": {"s": "ground", "x": 0.5, "y": 0.5}}},
        kind="space", space_id="ground", card_version="1.61.0", config_root=tmp_path,
    )
    marker = document["payload"]["config"]["markers"][0]
    assert marker["controls"] == ["light.keep"]
    assert document["transfer"]["dropped_marker_links"] == 1


def test_issue_90_space_export_disables_badge_link_outside_selection(tmp_path: Path) -> None:
    config = _config()
    config["markers"][0]["value_badge"] = {
        "enabled": True,
        "source": {"kind": "derived_marker_state", "ref": "marker:outside"},
        "position": "left",
    }
    config["markers"].append({
        "id": "outside", "binding": "virtual", "space": "other",
        "name": "Other lamp", "is_light": True,
    })
    document, _ = create_export(
        SimpleNamespace(instance_id="instance-a"), {"config": config},
        {"layout": {"lamp": {"s": "ground", "x": 0.5, "y": 0.5}}},
        kind="space", space_id="ground", card_version="1.61.0", config_root=tmp_path,
    )
    badge = document["payload"]["config"]["markers"][0]["value_badge"]
    assert badge == {"enabled": False, "source": None, "position": "left"}
    assert document["transfer"]["dropped_marker_links"] == 1


def test_space_merge_drops_link_when_target_is_virtualized_duplicate(tmp_path: Path) -> None:
    document = _document(tmp_path, "space")
    controller = document["payload"]["config"]["markers"][0]
    controller["controls"] = ["marker:smart", "light.keep"]
    document["payload"]["config"]["markers"].append({
        "id": "smart", "binding": "device:shared-target", "space": "ground",
        "room_id": "living", "name": "Smart target", "is_light": True,
    })
    current = {
        "spaces": [], "settings": {},
        "markers": [{"id": "existing", "binding": "device:shared-target"}],
    }
    merged, _layout, details = build_space_merge(document, current, {}, "virtual")
    imported_controller = next(
        marker for marker in merged["markers"]
        if marker.get("binding") == controller.get("binding")
    )
    assert imported_controller["controls"] == ["light.keep"]
    assert details["dropped_marker_links"] == 1


def test_space_merge_drops_dormant_link_and_counts_virtualized_controller_links(
    tmp_path: Path,
) -> None:
    dormant = _document(tmp_path, "space")
    controller = dormant["payload"]["config"]["markers"][0]
    controller["controls"] = ["marker:plain", "switch.keep"]
    dormant["payload"]["config"]["markers"].append({
        "id": "plain", "binding": "virtual", "space": "ground", "is_light": False,
    })
    merged, _layout, details = build_space_merge(dormant, {"spaces": [], "markers": []}, {}, "skip")
    imported_controller = next(
        marker for marker in merged["markers"] if marker.get("binding") == controller["binding"]
    )
    assert imported_controller["controls"] == ["switch.keep"]
    assert details["dropped_marker_links"] == 1

    virtualized = _document(tmp_path, "space")
    controller = virtualized["payload"]["config"]["markers"][0]
    controller["name"] = "Controller"
    controller["controls"] = ["marker:dumb"]
    virtualized["payload"]["config"]["markers"].append({
        "id": "dumb", "binding": "virtual", "space": "ground", "is_light": True,
    })
    current = {
        "spaces": [], "markers": [{"id": "existing", "binding": controller["binding"]}],
    }
    merged, _layout, details = build_space_merge(virtualized, current, {}, "virtual")
    imported_controller = next(
        marker for marker in merged["markers"]
        if marker.get("name") == controller.get("name") and marker.get("binding") == "virtual"
    )
    assert "controls" not in imported_controller
    assert details["dropped_marker_links"] == 1


@pytest.mark.asyncio
async def test_async_layout_writer_preserves_and_can_replace_metadata() -> None:
    class _Store:
        saved: dict[str, Any] | None = None

        async def async_save(self, value: dict[str, Any]) -> None:
            self.saved = value

    store = _Store()
    runtime = SimpleNamespace(store=store)
    stored = {
        "layout": {"old": {"x": 0}}, "rev": 2,
        "geom_pending": {"s": 1}, "repair_backup": {"kept": True},
    }
    await async_save_layout_state(runtime, stored, {"new": {"x": 1}}, 3)
    assert store.saved["geom_pending"] == {"s": 1}
    assert store.saved["repair_backup"] == {"kept": True}
    await async_save_layout_state(
        runtime, store.saved, {}, 4, metadata={"only": True}, replace_metadata=True,
    )
    assert store.saved == {"only": True, "layout": {}, "rev": 4}


async def _setup(hass: HomeAssistant) -> MockConfigEntry:
    entry = MockConfigEntry(domain=DOMAIN, title="House Plan", data={}, options={})
    entry.add_to_hass(hass)
    assert await hass.config_entries.async_setup(entry.entry_id)
    await hass.async_block_till_done()
    return entry


class _Connection:
    def __init__(self, *, is_admin: bool = True) -> None:
        self.user = SimpleNamespace(id="review-owner", is_admin=is_admin)
        self.result: dict[str, Any] | None = None
        self.error: tuple[str, str] | None = None

    def send_result(self, _msg_id: int, result: dict[str, Any]) -> None:
        self.result = result

    def send_error(self, _msg_id: int, code: str, message: str) -> None:
        self.error = (code, message)


async def _candidate(
    hass: HomeAssistant,
    _tmp_path: Path,
    *,
    target_config: dict[str, Any] | None = None,
    kind: str = "full",
    duplicate_policy: str = "skip",
) -> tuple[Any, dict[str, Any], dict[str, Any]]:
    rt = get_data(hass)
    assert rt is not None
    current_config = _config()
    current_layout = {"lamp": {"x": 0.1, "y": 0.2, "s": "ground"}}
    await rt.config_store.async_save({"config": current_config, "rev": 1})
    await rt.store.async_save({
        "layout": current_layout, "rev": 1,
        "repair_backup": {"must": "survive-failure"},
        # The durable geometry-migration intent is ``space_id -> old aspect``.
        # Keep a valid intent here so rollback preservation can also be proven
        # safe across the subsequent integration reload.
        "geom_pending": {"ground": 1.5},
    })
    incoming = json.loads(json.dumps(target_config or current_config))
    incoming["spaces"][0]["title"] = "Imported"
    config_root = Path(hass.config.path(""))
    document, _ = create_export(
        rt, {"config": incoming, "rev": 7},
        {"layout": {"lamp": {"x": 0.7, "y": 0.8, "s": "ground"}}, "rev": 9},
        kind=kind, space_id="ground" if kind == "space" else None,
        card_version="1.61.0", config_root=config_root,
    )
    response = create_preview(
        rt, json.dumps(document).encode(), owner_id="review-owner",
        duplicate_policy=duplicate_policy,
        current_config_data={"config": current_config, "rev": 1},
        current_layout_data={"layout": current_layout, "rev": 1}, config_root=config_root,
    )
    return rt, response, document


async def _apply(hass: HomeAssistant, response: dict[str, Any], *, confirm: bool = False) -> _Connection:
    connection = _Connection()
    await wsapi.ws_import_apply.__wrapped__(hass, connection, {
        "id": 42,
        "type": "houseplan/import/apply",
        "token": response["token"],
        "expected_config_rev": response["expected_config_rev"],
        "expected_layout_rev": response["expected_layout_rev"],
        "confirm_missing_content": confirm,
    })
    return connection


async def test_export_and_revalidate_ws_endpoints_use_server_owned_state(
    hass: HomeAssistant, tmp_path: Path,
) -> None:
    """The WS boundary must never accept an export payload or revalidate one."""
    await _setup(hass)
    _rt, response, _document_value = await _candidate(hass, tmp_path, kind="space")

    exported = _Connection()
    await wsapi.ws_export_create.__wrapped__(hass, exported, {
        "id": 40, "type": "houseplan/export/create", "kind": "full",
        "card_version": "review",
    })
    assert exported.error is None and exported.result
    document = exported.result["document"]
    assert document["model_version"] == PLAN_MODEL_VERSION
    assert "model_version" not in document["payload"]["config"]

    plan_exported = _Connection()
    await wsapi.ws_export_create.__wrapped__(hass, plan_exported, {
        "id": 44, "type": "houseplan/export/create", "kind": "space",
        "space_id": "ground", "plan_only": True, "card_version": "review",
    })
    assert plan_exported.error is None and plan_exported.result
    plan_document = plan_exported.result["document"]
    assert plan_document["transfer"]["plan_only"] is True
    assert plan_document["payload"]["config"]["markers"] == []
    assert plan_document["payload"]["layout"] == {}

    refreshed = _Connection()
    await wsapi.ws_import_revalidate.__wrapped__(hass, refreshed, {
        "id": 41, "type": "houseplan/import/revalidate",
        "token": response["token"], "duplicate_policy": "skip",
    })
    assert refreshed.error is None and refreshed.result
    assert refreshed.result["token"] == response["token"]
    assert refreshed.result["expected_config_rev"] == response["expected_config_rev"]
    assert refreshed.result["expected_layout_rev"] == response["expected_layout_rev"]


async def test_full_export_waits_for_a_concurrent_paired_write(
    hass: HomeAssistant,
) -> None:
    """An export sees either complete pair, never config from mid-commit."""
    await _setup(hass)
    rt = get_data(hass)
    assert rt is not None
    old_config = _config()
    old_layout = {"lamp": {"x": 0.1, "y": 0.2, "s": "ground"}}
    await rt.config_store.async_save({"config": old_config, "rev": 1})
    await rt.store.async_save({"layout": old_layout, "rev": 1})

    config_written = asyncio.Event()
    finish_pair = asyncio.Event()
    new_config = json.loads(json.dumps(old_config))
    new_config["spaces"][0]["title"] = "Concurrent target"
    new_layout = {"lamp": {"x": 0.8, "y": 0.9, "s": "ground"}}

    async def write_pair() -> None:
        async with rt.write_lock:
            await rt.config_store.async_save({"config": new_config, "rev": 2})
            config_written.set()
            await finish_pair.wait()
            await rt.store.async_save({"layout": new_layout, "rev": 2})

    writer = asyncio.create_task(write_pair())
    await config_written.wait()
    exported = _Connection()
    export = asyncio.create_task(wsapi.ws_export_create.__wrapped__(hass, exported, {
        "id": 43, "type": "houseplan/export/create", "kind": "full",
        "card_version": "review",
    }))
    await asyncio.sleep(0)
    assert not export.done(), "export must wait while the paired writer owns the lock"
    finish_pair.set()
    await writer
    await export

    assert exported.error is None and exported.result
    payload = exported.result["document"]["payload"]
    assert payload["config"]["spaces"][0]["title"] == "Concurrent target"
    assert payload["layout"] == new_layout


@pytest.mark.parametrize(("endpoint", "message"), [
    (wsapi.ws_export_create, {
        "id": 50, "type": "houseplan/export/create", "kind": "full", "card_version": "review",
    }),
    (wsapi.ws_import_revalidate, {
        "id": 51, "type": "houseplan/import/revalidate", "token": "opaque",
        "duplicate_policy": "skip",
    }),
    (wsapi.ws_import_apply, {
        "id": 52, "type": "houseplan/import/apply", "token": "opaque",
        "expected_config_rev": 0, "expected_layout_rev": 0,
        "confirm_missing_content": False,
    }),
])
async def test_portable_backup_ws_endpoints_reject_non_editors(
    hass: HomeAssistant, endpoint, message: dict[str, Any],
) -> None:
    connection = _Connection(is_admin=False)
    await endpoint.__wrapped__(hass, connection, message)
    assert connection.result is None
    assert connection.error and connection.error[0] == "unauthorized"


class _ImportPreviewRequest:
    def __init__(self, hass: HomeAssistant, *, is_admin: bool) -> None:
        self.app = {KEY_HASS: hass}
        self.query: dict[str, str] = {}
        self.content_length = 0
        self._user = SimpleNamespace(id="http-review", is_admin=is_admin)

    def get(self, key: str, default=None):
        return self._user if key == "hass_user" else default


async def test_import_preview_http_rejects_non_editor_and_reports_not_ready(
    hass: HomeAssistant,
) -> None:
    view = HouseplanImportPreviewView()
    forbidden = await view.post(_ImportPreviewRequest(hass, is_admin=False))
    assert forbidden.status == 403
    unavailable = await view.post(_ImportPreviewRequest(hass, is_admin=True))
    assert unavailable.status == 503


async def test_import_apply_conflict_preserves_state_and_preview_token(
    hass: HomeAssistant, tmp_path: Path,
) -> None:
    await _setup(hass)
    rt, response, _document_value = await _candidate(hass, tmp_path)
    current = await rt.config_store.async_load()
    await rt.config_store.async_save({**current, "rev": 2})
    connection = await _apply(hass, response)
    assert connection.result is None
    assert connection.error and connection.error[0] == "conflict"
    assert response["token"] in rt.import_previews
    assert (await rt.config_store.async_load())["config"]["spaces"][0]["title"] == "Ground"
    assert (await rt.store.async_load())["layout"]["lamp"]["x"] == 0.1


async def test_apply_confirmed_detach_does_not_collect_files_and_commits_pair(
    hass: HomeAssistant, tmp_path: Path, monkeypatch,
) -> None:
    await _setup(hass)
    config = _config()
    config["spaces"][0]["plan_url"] = "/api/houseplan/content/plans/_/missing.svg"
    rt, response, _ = await _candidate(hass, tmp_path, target_config=config)
    assert response["preview"]["confirmation_required"] is True
    monkeypatch.setattr(wsapi, "collect_plans", lambda *_args, **_kwargs: (_ for _ in ()).throw(
        AssertionError("import apply must not collect plans")
    ))
    monkeypatch.setattr(wsapi, "collect_attachments", lambda *_args, **_kwargs: (_ for _ in ()).throw(
        AssertionError("import apply must not collect attachments")
    ))
    connection = await _apply(hass, response, confirm=True)
    assert connection.error is None and connection.result and connection.result["ok"]
    stored_config = await rt.config_store.async_load()
    stored_layout = await rt.store.async_load()
    assert stored_config["config"]["spaces"][0].get("plan_url") is None
    assert stored_config["rev"] == stored_layout["rev"] == 2
    assert "optimize_pending" not in stored_layout
    assert "repair_backup" not in stored_layout
    assert "geom_pending" not in stored_layout
    assert stored_layout["optimize_backup"]["kind"] == "import"


async def test_full_import_undo_is_one_shot_and_reports_its_kind(
    hass: HomeAssistant, tmp_path: Path,
) -> None:
    await _setup(hass)
    rt, response, _ = await _candidate(hass, tmp_path)
    applied = await _apply(hass, response)
    assert applied.result and applied.result["can_undo"] is True

    status = _Connection()
    await wsapi.ws_config_get.__wrapped__(hass, status, {
        "id": 60, "type": "houseplan/config/get",
    })
    assert status.result and status.result["undo_kind"] == "import"

    undone = _Connection()
    await wsapi.ws_plan_optimize_undo.__wrapped__(hass, undone, {
        "id": 61, "type": "houseplan/plan/optimize_undo",
        "expected_config_rev": 2, "expected_layout_rev": 2,
    })
    assert undone.error is None and undone.result and undone.result["can_undo"] is False
    assert (await rt.config_store.async_load())["config"]["spaces"][0]["title"] == "Ground"
    assert (await rt.store.async_load())["layout"]["lamp"]["x"] == 0.1

    repeated = _Connection()
    await wsapi.ws_plan_optimize_undo.__wrapped__(hass, repeated, {
        "id": 62, "type": "houseplan/plan/optimize_undo",
        "expected_config_rev": 3, "expected_layout_rev": 3,
    })
    assert repeated.error and repeated.error[0] == "no_backup"


@pytest.mark.parametrize("kind", ["import", "import_rollback"])
async def test_setup_recovers_durable_import_pair(
    hass: HomeAssistant, kind: str,
) -> None:
    entry = await _setup(hass)
    rt = get_data(hass)
    assert rt is not None
    current = _config()
    target = json.loads(json.dumps(current))
    target["spaces"][0]["title"] = f"Recovered {kind}"
    target_layout = {"lamp": {"x": 0.8, "y": 0.7, "s": "ground"}}
    await rt.config_store.async_save({"config": current, "rev": 1})
    await rt.store.async_save({
        "layout": {"lamp": {"x": 0.1, "y": 0.2, "s": "ground"}},
        "rev": 1,
        "stale_metadata": True,
        "optimize_pending": {
            "kind": kind,
            "config": target,
            "layout": target_layout,
            "config_rev": 2,
            "layout_rev": 2,
            "final_metadata": {"future_metadata": {"kept": True}},
        },
    })

    assert await hass.config_entries.async_reload(entry.entry_id)
    await hass.async_block_till_done()
    recovered = get_data(hass)
    assert recovered is not None
    config_data = await recovered.config_store.async_load()
    layout_data = await recovered.store.async_load()
    assert config_data == {"config": target, "rev": 2}
    assert layout_data["layout"] == target_layout
    assert layout_data["rev"] == 2
    assert layout_data["future_metadata"] == {"kept": True}
    assert "stale_metadata" not in layout_data
    assert "optimize_pending" not in layout_data


async def test_apply_rechecks_plan_file_under_the_write_lock(
    hass: HomeAssistant, tmp_path: Path,
) -> None:
    await _setup(hass)
    plans = Path(hass.config.path(PLANS_DIR))
    plans.mkdir(parents=True, exist_ok=True)
    plan = plans / "race.svg"
    plan.write_text("<svg/>", encoding="utf-8")
    config = _config()
    config["spaces"][0]["plan_url"] = "/api/houseplan/content/plans/_/race.svg"
    _rt, response, _ = await _candidate(hass, tmp_path, target_config=config)
    assert response["preview"]["confirmation_required"] is False
    plan.unlink()
    connection = await _apply(hass, response)
    assert connection.result is None
    assert connection.error and connection.error[0] == "missing_plan"


async def test_apply_rechecks_attachment_under_the_write_lock(
    hass: HomeAssistant, tmp_path: Path,
) -> None:
    await _setup(hass)
    attachment_dir = Path(hass.config.path(FILES_DIR)) / "lamp"
    attachment_dir.mkdir(parents=True, exist_ok=True)
    attachment = attachment_dir / "manual.pdf"
    attachment.write_bytes(b"%PDF-1.4\n")
    config = _config()
    config["markers"][0]["pdfs"] = [{
        "name": "Manual", "url": "/api/houseplan/content/files/lamp/manual.pdf",
    }]
    _rt, response, _ = await _candidate(hass, tmp_path, target_config=config)
    assert response["preview"]["confirmation_required"] is False
    attachment.unlink()
    connection = await _apply(hass, response)
    assert connection.result is None
    assert connection.error and connection.error[0] == "missing_content"


async def test_space_apply_commits_remapped_config_and_layout_as_one_pair(
    hass: HomeAssistant, tmp_path: Path,
) -> None:
    await _setup(hass)
    rt, response, _ = await _candidate(
        hass, tmp_path, kind="space", duplicate_policy="virtual",
    )
    connection = await _apply(hass, response)
    assert connection.error is None and connection.result
    assert connection.result["kind"] == "space" and connection.result["can_undo"] is False
    config_data = await rt.config_store.async_load()
    layout_data = await rt.store.async_load()
    assert len(config_data["config"]["spaces"]) == 2
    assert config_data["rev"] == layout_data["rev"] == 2
    assert len(layout_data["layout"]) == 2
    assert "optimize_pending" not in layout_data


async def test_pair_retries_a_fail_after_layout_write_before_success(
    hass: HomeAssistant, tmp_path: Path, monkeypatch,
) -> None:
    await _setup(hass)
    rt, response, _ = await _candidate(hass, tmp_path)
    real_save = rt.store.async_save
    state = {"calls": 0}

    async def fail_after_target_write(value: dict[str, Any]) -> None:
        state["calls"] += 1
        await real_save(value)
        if state["calls"] == 2:
            raise OSError("fail after durable target layout write")

    monkeypatch.setattr(rt.store, "async_save", fail_after_target_write)
    connection = await _apply(hass, response)
    assert connection.error is None and connection.result and connection.result["ok"]
    config_data = await rt.config_store.async_load()
    layout_data = await rt.store.async_load()
    assert config_data["config"]["spaces"][0]["title"] == "Imported"
    assert layout_data["layout"]["lamp"]["x"] == 0.7
    assert "optimize_pending" not in layout_data


async def test_success_events_are_emitted_only_after_both_target_writes(
    hass: HomeAssistant, tmp_path: Path, monkeypatch,
) -> None:
    await _setup(hass)
    rt, response, _ = await _candidate(hass, tmp_path)
    real_config_save = rt.config_store.async_save
    real_layout_save = rt.store.async_save
    durable = {"config": False, "layout": False}
    observed: list[tuple[str, bool, bool]] = []

    async def config_save(value: dict[str, Any]) -> None:
        await real_config_save(value)
        if value.get("rev") == 2:
            durable["config"] = True

    async def layout_save(value: dict[str, Any]) -> None:
        await real_layout_save(value)
        if value.get("rev") == 2 and "optimize_pending" not in value:
            durable["layout"] = True

    def fire(
        _bus: Any,
        event_type: str,
        _event_data: dict[str, Any] | None = None,
        **_kwargs: Any,
    ) -> None:
        if event_type in {"houseplan_config_updated", "houseplan_layout_updated"}:
            observed.append((event_type, durable["config"], durable["layout"]))

    monkeypatch.setattr(rt.config_store, "async_save", config_save)
    monkeypatch.setattr(rt.store, "async_save", layout_save)
    # EventBus uses slots, so patch the class rather than trying to attach a
    # replacement callable to this particular Home Assistant instance.
    monkeypatch.setattr(type(hass.bus), "async_fire", fire)
    connection = await _apply(hass, response)
    assert connection.error is None and connection.result
    assert [item[0] for item in observed] == [
        "houseplan_config_updated", "houseplan_layout_updated",
    ]
    assert all(config_done and layout_done for _event, config_done, layout_done in observed)


async def test_pair_rolls_back_before_reporting_a_persistent_target_failure(
    hass: HomeAssistant, tmp_path: Path, monkeypatch,
) -> None:
    entry = await _setup(hass)
    rt, response, _ = await _candidate(hass, tmp_path)
    real_save = rt.config_store.async_save

    async def refuse_target(value: dict[str, Any]) -> None:
        if value.get("rev") == 2:
            raise OSError("target config cannot be saved")
        await real_save(value)

    monkeypatch.setattr(rt.config_store, "async_save", refuse_target)
    connection = await _apply(hass, response)
    assert connection.result is None
    assert connection.error and connection.error[0] == "commit_failed"
    config_data = await rt.config_store.async_load()
    layout_data = await rt.store.async_load()
    assert config_data["rev"] == layout_data["rev"] == 1
    assert config_data["config"]["spaces"][0]["title"] == "Ground"
    assert layout_data["layout"]["lamp"]["x"] == 0.1
    assert layout_data["repair_backup"] == {"must": "survive-failure"}
    assert layout_data["geom_pending"] == {"ground": 1.5}
    assert "optimize_pending" not in layout_data

    monkeypatch.setattr(rt.config_store, "async_save", real_save)
    assert await hass.config_entries.async_reload(entry.entry_id)
    await hass.async_block_till_done()
    rt = get_data(hass)
    assert rt is not None
    assert (await rt.config_store.async_load())["config"]["spaces"][0]["title"] == "Ground"


def test_apply_schema_accepts_only_an_opaque_preview_token() -> None:
    with pytest.raises(vol.Invalid):
        wsapi.ws_import_apply._ws_schema({
            "id": 1,
            "type": "houseplan/import/apply",
            "payload": _config(),
            "expected_config_rev": 1,
            "expected_layout_rev": 1,
        })