dl-organizer/tests/test_path_safety.py

"""Security tests for path generation and execution boundaries."""

from datetime import datetime, timezone
from uuid import uuid4

from vlm.config import Config
from vlm.executor import ExecutionEngine
from vlm.models import ExecutionPlan, FileOperation, MovieIdentity, VideoFile
from vlm.planner import generate_plan


def test_generate_plan_rejects_destination_outside_library_root(tmp_path):
    config = Config(
        library_root=tmp_path / "library",
        movie_template="../../escape/{title}/",
    )
    config.library_root.mkdir(parents=True, exist_ok=True)

    source = config.library_root / "movie" / "Test.mkv"
    source.parent.mkdir(parents=True, exist_ok=True)
    source.write_text("x")

    vf = VideoFile(
        path=source,
        filename=source.name,
        size_bytes=1,
        modified_timestamp=datetime.now(timezone.utc),
        category="movie",
    )
    identity = MovieIdentity(
        title="Test",
        year=2020,
        confidence=1.0,
        needs_review=False,
        original_filename=source.name,
    )

    plan = generate_plan([(vf, identity)], config)
    op = plan.operations[0]
    assert op.operation_type == "no-op"
    assert "Unsafe destination outside library root" in op.reason


def test_generate_plan_sanitizes_title_components(tmp_path):
    config = Config(library_root=tmp_path / "library")
    config.library_root.mkdir(parents=True, exist_ok=True)

    source = config.library_root / "movie" / "Raw.mkv"
    source.parent.mkdir(parents=True, exist_ok=True)
    source.write_text("x")

    vf = VideoFile(
        path=source,
        filename=source.name,
        size_bytes=1,
        modified_timestamp=datetime.now(timezone.utc),
        category="movie",
    )
    identity = MovieIdentity(
        title="A/../../B\\C",
        year=2020,
        confidence=1.0,
        needs_review=False,
        original_filename=source.name,
    )

    plan = generate_plan([(vf, identity)], config)
    op = plan.operations[0]
    assert op.destination_path is not None
    assert ".." not in str(op.destination_path)
    assert op.destination_path.is_absolute()
    assert str(config.library_root.resolve()) in str(op.destination_path.resolve())


def test_executor_blocks_unsafe_destination_even_with_manual_plan(tmp_path):
    library_root = tmp_path / "library"
    library_root.mkdir(parents=True, exist_ok=True)
    source = library_root / "movie" / "Sample.mkv"
    source.parent.mkdir(parents=True, exist_ok=True)
    source.write_text("sample")

    operation = FileOperation(
        operation_type="move",
        source_path=source,
        destination_path=tmp_path / "outside" / "Sample.mkv",
        reason="unsafe test",
        has_conflict=False,
        conflict_reason=None,
    )
    plan = ExecutionPlan(
        plan_id=str(uuid4()),
        created_at=datetime.now(timezone.utc),
        operations=[operation],
        summary={"total": 1, "move": 1, "rename": 0, "quarantine": 0, "no-op": 0},
    )

    engine = ExecutionEngine(config=Config(library_root=library_root))
    results, summary, _ = engine.execute_plan(plan, mode="execute", confirmed=True)

    assert not results[0].success
    assert "outside library root" in (results[0].error_message or "")
    assert summary["failed"] == 1


def test_executor_blocks_unsafe_source_even_with_manual_plan(tmp_path):
    library_root = tmp_path / "library"
    library_root.mkdir(parents=True, exist_ok=True)
    outside_root = tmp_path / "outside"
    outside_root.mkdir(parents=True, exist_ok=True)
    source = outside_root / "Sample.mkv"
    source.write_text("sample")

    destination = library_root / "movie" / "Sample.mkv"
    operation = FileOperation(
        operation_type="move",
        source_path=source,
        destination_path=destination,
        reason="unsafe test",
        has_conflict=False,
        conflict_reason=None,
    )
    plan = ExecutionPlan(
        plan_id=str(uuid4()),
        created_at=datetime.now(timezone.utc),
        operations=[operation],
        summary={"total": 1, "move": 1, "rename": 0, "quarantine": 0, "no-op": 0},
    )

    engine = ExecutionEngine(config=Config(library_root=library_root))
    results, summary, _ = engine.execute_plan(plan, mode="execute", confirmed=True)

    assert not results[0].success
    assert "Unsafe source outside library root" in (results[0].error_message or "")
    assert summary["failed"] == 1
    assert source.exists()
    assert not destination.exists()