Zum Inhalt

pipeline/tests/test_failure.py

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
# SPDX-License-Identifier: AGPL-3.0-or-later

"""Tests for failure classification + log-derived reason (v1.8.0)."""

from __future__ import annotations

from pathlib import Path

from autosplat.failure import classify_failure, failure_reason_from_log


def test_classify_blur_rejection() -> None:
    info = classify_failure(
        "All 250 extracted frames were rejected as blurry "
        "(Laplacian variance below blur_threshold=100.0).",
        stage="preprocess",
    )
    assert info.category == "blur"
    assert "blurry" in info.headline.lower()
    assert "blur_threshold" in info.hint


def test_classify_sfm_no_matches() -> None:
    info = classify_failure(
        "Command '['colmap', 'mapper', ...]' returned non-zero exit status 1.",
        stage="sfm",
    )
    assert info.category == "sfm"
    # Auto-rescue already ran in the pipeline — the hint must not imply a manual
    # `autosplat rescue` step; it points at the real fix (footage overlap).
    assert "already ran" in info.hint.lower()
    assert "overlap" in info.hint.lower()


def test_classify_sfm_no_images_phrase() -> None:
    info = classify_failure("No images with matches", stage="sfm")
    assert info.category == "sfm"


def test_classify_oom() -> None:
    info = classify_failure("Brush hit OOM at resolution_cap=1600", stage="train")
    assert info.category == "oom"
    assert "resolution_cap" in info.hint


def test_classify_no_video() -> None:
    info = classify_failure("No source video found in /x/y", stage=None)
    assert info.category == "no_video"


def test_classify_interrupted() -> None:
    info = classify_failure("interrupted — the run ended without producing a result")
    assert info.category == "interrupted"
    assert "resume" in info.hint.lower()


def test_classify_bisection_no_culprit() -> None:
    """Multi-video rescue where every flight registers alone but the set won't
    co-register → a cross-video problem, distinct from the generic SfM headline."""
    info = classify_failure("bisection_no_culprit", stage="bisection")
    assert info.category == "multi_video"
    assert "co-register" in info.hint.lower() or "overlap between" in info.hint.lower()


def test_classify_unknown_fallback() -> None:
    info = classify_failure("some totally novel explosion", stage="train")
    assert info.category == "unknown"
    assert info.headline
    assert info.hint  # never empty — always actionable


def test_classify_none_reason_is_fallback() -> None:
    info = classify_failure(None, stage=None)
    assert info.category == "unknown"
    assert info.hint


def test_failure_reason_from_log_extracts_last_error_event(tmp_path: Path) -> None:
    log = tmp_path / "pipeline.log"
    log.write_text(
        '{"event": "sfm.mapper.start", "level": "info", "ts": "t1"}\n'
        '{"error": "Failed to create any sparse model", '
        '"event": "sfm.subprocess_failed", "level": "error", "ts": "t2"}\n'
    )
    reason = failure_reason_from_log(tmp_path)
    assert reason is not None
    assert "sparse model" in reason


def test_failure_reason_from_log_none_when_no_log(tmp_path: Path) -> None:
    assert failure_reason_from_log(tmp_path) is None


def test_failure_reason_from_log_none_when_no_error_lines(tmp_path: Path) -> None:
    log = tmp_path / "pipeline.log"
    log.write_text('{"event": "preprocess.done", "level": "info", "ts": "t"}\n')
    # No error event → return the last line so the panel still shows something.
    reason = failure_reason_from_log(tmp_path)
    assert reason is None or "preprocess.done" in reason