bambuddy/backend/tests/unit/test_vp_wire_dump.py

"""Tests for the env-flagged VP wire-payload dump helper used to triage
shape-of-payload bugs like #1622."""

import json
import os
from unittest.mock import patch

import pytest

from backend.app.core.config import settings as app_settings
from backend.app.services.virtual_printer import _debug


@pytest.fixture
def _isolated_log_dir(tmp_path, monkeypatch):
    with patch.object(app_settings, "log_dir", tmp_path):
        yield tmp_path


def test_disabled_by_default_writes_nothing(_isolated_log_dir, monkeypatch):
    monkeypatch.delenv("BAMBUDDY_VP_DUMP_WIRE", raising=False)
    _debug.dump_wire("VP1", "out", {"hello": "world"})
    assert not (_isolated_log_dir / "vp_wire").exists()


def test_enabled_writes_dict_as_pretty_json(_isolated_log_dir, monkeypatch):
    monkeypatch.setenv("BAMBUDDY_VP_DUMP_WIRE", "1")
    payload = {"print": {"ams": {"ams": [{"id": 0, "tray": [{"id": 0, "tray_type": "PLA"}]}]}}}
    _debug.dump_wire("Bambuddy P1S", "out", payload)
    out = _isolated_log_dir / "vp_wire" / "Bambuddy_P1S_out.json"
    assert out.is_file()
    assert json.loads(out.read_text()) == payload


def test_overwrites_on_repeat_call(_isolated_log_dir, monkeypatch):
    monkeypatch.setenv("BAMBUDDY_VP_DUMP_WIRE", "1")
    _debug.dump_wire("VP1", "out", {"v": 1})
    _debug.dump_wire("VP1", "out", {"v": 2})
    out = _isolated_log_dir / "vp_wire" / "VP1_out.json"
    assert json.loads(out.read_text()) == {"v": 2}


def test_separate_files_per_direction(_isolated_log_dir, monkeypatch):
    monkeypatch.setenv("BAMBUDDY_VP_DUMP_WIRE", "1")
    _debug.dump_wire("VP1", "in", {"src": "printer"})
    _debug.dump_wire("VP1", "out", {"src": "slicer"})
    assert (_isolated_log_dir / "vp_wire" / "VP1_in.json").is_file()
    assert (_isolated_log_dir / "vp_wire" / "VP1_out.json").is_file()


def test_sanitizes_path_traversal_in_vp_name(_isolated_log_dir, monkeypatch):
    monkeypatch.setenv("BAMBUDDY_VP_DUMP_WIRE", "1")
    _debug.dump_wire("../../etc/passwd", "out", {"x": 1})
    vp_wire = _isolated_log_dir / "vp_wire"
    files = list(vp_wire.glob("*"))
    assert len(files) == 1
    # The actual safety property: the written file is inside vp_wire/.
    # `..` as a substring of a single filename component is harmless because
    # the path separator (/) is collapsed to _ before construction.
    assert files[0].resolve().parent == vp_wire.resolve()
    assert "/" not in files[0].name


def test_empty_vp_name_falls_back_to_default(_isolated_log_dir, monkeypatch):
    monkeypatch.setenv("BAMBUDDY_VP_DUMP_WIRE", "1")
    _debug.dump_wire("", "out", {"x": 1})
    assert (_isolated_log_dir / "vp_wire" / "vp_out.json").is_file()


def test_bytes_payload_decoded(_isolated_log_dir, monkeypatch):
    monkeypatch.setenv("BAMBUDDY_VP_DUMP_WIRE", "1")
    _debug.dump_wire("VP1", "in", b'{"raw": true}')
    out = _isolated_log_dir / "vp_wire" / "VP1_in.json"
    assert out.read_text() == '{"raw": true}'


def test_unwritable_dir_is_swallowed(_isolated_log_dir, monkeypatch):
    """A debug-instrumentation failure must not crash the bridge or 1Hz loop."""
    monkeypatch.setenv("BAMBUDDY_VP_DUMP_WIRE", "1")
    # Point log_dir at a location that mkdir refuses (a regular file occupying
    # the path). Failure must be swallowed.
    blocker = _isolated_log_dir / "blocker"
    blocker.write_text("not a dir")
    with patch.object(app_settings, "log_dir", blocker):
        _debug.dump_wire("VP1", "out", {"x": 1})  # must not raise


@pytest.mark.parametrize("flag_value", ["0", "false", "off", "", "no"])
def test_falsy_flag_values_disable(_isolated_log_dir, monkeypatch, flag_value):
    monkeypatch.setenv("BAMBUDDY_VP_DUMP_WIRE", flag_value)
    _debug.dump_wire("VP1", "out", {"x": 1})
    assert not (_isolated_log_dir / "vp_wire").exists()


@pytest.mark.parametrize("flag_value", ["1", "true", "TRUE", "yes", "on", "On"])
def test_truthy_flag_values_enable(_isolated_log_dir, monkeypatch, flag_value):
    monkeypatch.setenv("BAMBUDDY_VP_DUMP_WIRE", flag_value)
    _debug.dump_wire("VP1", "out", {"x": 1})
    assert (_isolated_log_dir / "vp_wire" / "VP1_out.json").is_file()


def test_idempotent_atomic_no_partial_file_visible(_isolated_log_dir, monkeypatch):
    """tmp+rename pattern means a reader never sees a half-written .json file."""
    monkeypatch.setenv("BAMBUDDY_VP_DUMP_WIRE", "1")
    _debug.dump_wire("VP1", "out", {"x": 1})
    files = sorted(p.name for p in (_isolated_log_dir / "vp_wire").iterdir())
    # No leftover .tmp file after a successful write.
    assert files == ["VP1_out.json"]


def test_env_check_is_per_call_not_module_load(_isolated_log_dir, monkeypatch):
    """Flag toggle must take effect on the next call without restarting; we
    re-read the env var inside ``dump_wire`` rather than caching at import."""
    monkeypatch.delenv("BAMBUDDY_VP_DUMP_WIRE", raising=False)
    _debug.dump_wire("VP1", "out", {"v": 1})
    assert not (_isolated_log_dir / "vp_wire").exists()

    os.environ["BAMBUDDY_VP_DUMP_WIRE"] = "1"
    try:
        _debug.dump_wire("VP1", "out", {"v": 2})
        assert (_isolated_log_dir / "vp_wire" / "VP1_out.json").is_file()
    finally:
        os.environ.pop("BAMBUDDY_VP_DUMP_WIRE", None)