Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 12 additions & 1 deletion tests/test_channel_of.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,14 @@ def test_channel_of_rear_variants():

def test_channel_of_interior():
assert naming.channel_of("I") == "interior"
assert naming.channel_of("PI") == "interior" # parking interior


def test_channel_of_tele():
assert naming.channel_of("T") == "tele"
assert naming.channel_of("PT") == "tele" # parking tele
assert naming.channel_of("ET") == "tele" # event tele
assert naming.channel_of("t") == "tele" # case-insensitive


def test_channel_of_unknown_and_empty():
Expand All @@ -27,8 +35,11 @@ def test_channel_of_unknown_and_empty():


def test_channel_order_and_labels():
assert naming.CHANNEL_ORDER == ["front", "rear", "interior", "other"]
assert naming.CHANNEL_ORDER == [
"front", "rear", "tele", "interior", "other",
]
assert naming.CHANNEL_LABELS["front"] == "Front"
assert naming.CHANNEL_LABELS["rear"] == "Rear"
assert naming.CHANNEL_LABELS["tele"] == "Tele"
assert naming.CHANNEL_LABELS["interior"] == "Interior"
assert naming.CHANNEL_LABELS["other"] == "Other"
57 changes: 57 additions & 0 deletions tests/test_export_types.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,57 @@
"""Validation gates for the third-camera export job types.

Two layers must both accept a type before a job can run: the
route's pydantic pattern (web/routers/exports.py) and the
worker's enqueue() allowlist (web/services/exporter.py). A type
listed in one but not the other 422s or errors at runtime — pin
them together.
"""
from __future__ import annotations

from pathlib import Path
from unittest.mock import MagicMock

import pytest
from pydantic import ValidationError

from web.db import Database
from web.routers.exports import CreateExport, Segment
from web.services.exporter import ExportWorker

NEW_TYPES = ("join_tele", "join_interior", "pip_tele", "pip_interior")
OLD_TYPES = ("join_front", "join_rear", "pip", "pip_rear")


def test_route_model_accepts_all_types():
for t in OLD_TYPES + NEW_TYPES + ("timeline",):
assert CreateExport(type=t).type == t


def test_route_model_rejects_unknown_type():
with pytest.raises(ValidationError):
CreateExport(type="join_bogus")


def test_segment_accepts_tele_and_interior_channels():
for ch in ("front", "rear", "tele", "interior", "other"):
seg = Segment(channel=ch, start_ts=0.0, end_ts=1.0)
assert seg.channel == ch
with pytest.raises(ValidationError):
Segment(channel="bogus", start_ts=0.0, end_ts=1.0)


def test_enqueue_allowlist_accepts_new_types(
tmp_path: Path, monkeypatch,
):
monkeypatch.setattr(
"web.services.exporter.ffmpeg_available", lambda: True,
)
db = Database(str(tmp_path / "test.db"))
worker = ExportWorker(
db=db, provider=MagicMock(), broadcast=MagicMock(),
)
for t in OLD_TYPES + NEW_TYPES:
job_id = worker.enqueue(t, [1, 2])
assert isinstance(job_id, int)
with pytest.raises(ValueError):
worker.enqueue("join_bogus", [1])
82 changes: 82 additions & 0 deletions tests/test_exporter_pair_clips.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
"""Tests for ExportWorker._pair_clips slot dispatch.

The pairer groups same-capture clips by (timestamp, event_type)
and assigns each clip a slot from its trailing camera letter:
F → front, T → tele, I → interior, everything else → rear.
``required`` names the slots a group must have to count — PiP
front+rear needs both classic slots; pip_tele / pip_interior
need front plus the third camera.
"""
from __future__ import annotations

from web.services.exporter import ExportWorker


def _clip(camera: str, ts: int = 1000, event: str = "normal") -> dict:
return {
"camera": camera,
"timestamp": ts,
"event_type": event,
"path": f"/x/{ts}_{camera}.MP4",
}


def test_front_rear_pairing_unchanged():
pairs = ExportWorker._pair_clips(
[_clip("F"), _clip("R")],
)
assert len(pairs) == 1
(_, p), = pairs
assert p["front"]["camera"] == "F"
assert p["rear"]["camera"] == "R"


def test_triplet_keeps_all_three_slots():
pairs = ExportWorker._pair_clips(
[_clip("F"), _clip("R"), _clip("T")],
)
(_, p), = pairs
assert set(p) == {"front", "rear", "tele"}


def test_front_tele_pair_for_pip_tele():
# A front+tele selection has no rear — the default
# front+rear requirement drops it, the pip_tele one keeps it.
clips = [_clip("F"), _clip("T")]
assert ExportWorker._pair_clips(clips) == []
pairs = ExportWorker._pair_clips(
clips, required=("front", "tele"),
)
assert len(pairs) == 1


def test_front_interior_pair_for_pip_interior():
clips = [_clip("F"), _clip("I")]
pairs = ExportWorker._pair_clips(
clips, required=("front", "interior"),
)
assert len(pairs) == 1
(_, p), = pairs
assert p["interior"]["camera"] == "I"


def test_tele_only_never_pairs():
assert ExportWorker._pair_clips(
[_clip("T")], required=("front", "tele"),
) == []


def test_parking_prefixes_assign_correct_slots():
pairs = ExportWorker._pair_clips(
[_clip("PF", event="parking"), _clip("PT", event="parking")],
required=("front", "tele"),
)
(_, p), = pairs
assert p["front"]["camera"] == "PF"
assert p["tele"]["camera"] == "PT"


def test_rear_tele_without_front_never_pairs():
assert ExportWorker._pair_clips(
[_clip("R"), _clip("T")], required=("front", "tele"),
) == []
16 changes: 13 additions & 3 deletions tests/test_importer.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,29 +13,39 @@ def test_classify_event_type():
assert importer.classify_event_type("F", "DCIM/Movie/RO/X.MP4") == "ro"
# RO wins even for a parking-named clip living under /RO/.
assert importer.classify_event_type("PF", "Movie/RO/X.MP4") == "ro"
# Third cameras: telephoto (T) and interior (I) classify the
# same way as front/rear.
assert importer.classify_event_type("T", "DCIM/Movie/X.MP4") == "normal"
assert importer.classify_event_type("PT", "DCIM/Parking/X.MP4") == "parking"
assert importer.classify_event_type("I", "DCIM/Movie/X.MP4") == "normal"
assert importer.classify_event_type("PI", "Movie/RO/X.MP4") == "ro"


def test_scan_source_recurses_and_sorts_newest_first(tmp_path: Path):
from web.services import importer
root = tmp_path / "card"
(root / "DCIM" / "Movie").mkdir(parents=True)
(root / "DCIM" / "Movie" / "RO").mkdir()
# Recognised: one older normal front, one newer locked front.
# Recognised: one older normal front, one newer locked front,
# and a telephoto clip from a 3-camera model.
(root / "DCIM" / "Movie" / "2026_0101_080000_0001F.MP4").write_bytes(b"a" * 10)
(root / "DCIM" / "Movie" / "RO" / "2026_0102_090000_0002F.MP4").write_bytes(b"b" * 20)
(root / "DCIM" / "Movie" / "2026_0101_081000_0003T.MP4").write_bytes(b"t" * 7)
# Junk: ignored + reported.
(root / "DCIM" / "Movie" / "notes.txt").write_text("hi")
# Matches the filename regex but has an impossible date -> bad_timestamp.
(root / "DCIM" / "Movie" / "2026_1399_250000_0009F.MP4").write_bytes(b"c" * 5)

manifest = importer.scan_source(str(root))
assert manifest.total_bytes == 30
assert manifest.total_bytes == 37
assert [it.basename for it in manifest.items] == [
"2026_0102_090000_0002F.MP4", # newest first
"2026_0101_081000_0003T.MP4",
"2026_0101_080000_0001F.MP4",
]
assert manifest.items[0].event_type == "ro"
assert manifest.items[1].event_type == "normal"
assert manifest.items[1].event_type == "normal" # the T clip
assert manifest.items[2].event_type == "normal"
skipped = {s["name"]: s["reason"] for s in manifest.skipped}
assert skipped["notes.txt"] == "not_recognised"
assert skipped["2026_1399_250000_0009F.MP4"] == "bad_timestamp"
Expand Down
17 changes: 16 additions & 1 deletion tests/test_naming.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,10 @@ def test_input_order_does_not_matter() -> None:

def test_each_label_passes_through() -> None:
clips = [_clip(_ts(2024, 3, 15, 14, 30))]
for label in ("front", "rear", "pip-front", "pip-rear"):
for label in (
"front", "rear", "tele", "interior",
"pip-front", "pip-rear", "pip-tele", "pip-interior",
):
assert build_basename(clips, label).endswith(f"_{label}_1clip")


Expand All @@ -75,6 +78,18 @@ def test_export_download_name_maps_type_and_adds_ext() -> None:
assert export_download_name("pip_rear", clips, 7) == (
"2024-03-15_1430-1502_pip-rear_2clips.mp4"
)
assert export_download_name("join_tele", clips, 7) == (
"2024-03-15_1430-1502_tele_2clips.mp4"
)
assert export_download_name("join_interior", clips, 7) == (
"2024-03-15_1430-1502_interior_2clips.mp4"
)
assert export_download_name("pip_tele", clips, 7) == (
"2024-03-15_1430-1502_pip-tele_2clips.mp4"
)
assert export_download_name("pip_interior", clips, 7) == (
"2024-03-15_1430-1502_pip-interior_2clips.mp4"
)


def test_export_download_name_falls_back_when_no_clips() -> None:
Expand Down
18 changes: 18 additions & 0 deletions tests/test_pip_filter.py
Original file line number Diff line number Diff line change
Expand Up @@ -72,3 +72,21 @@ def test_front_main_is_the_default() -> None:
assert _pip_filter_complex("top_right") == (
_pip_filter_complex("top_right", main="front")
)


# Tele-main / interior-main: the partner clip is ffmpeg input 1
# (the input swap happens in _pip's argv construction, not here),
# so any non-front ``main`` yields the same graph as rear-main —
# partner fullscreen, front (input 0) scaled to the inset.


def test_tele_main_matches_rear_main_graph() -> None:
assert _pip_filter_complex("top_right", main="tele") == (
_pip_filter_complex("top_right", main="rear")
)


def test_interior_main_matches_rear_main_graph() -> None:
assert _pip_filter_complex("bottom_left", main="interior") == (
_pip_filter_complex("bottom_left", main="rear")
)
50 changes: 50 additions & 0 deletions tests/test_queue_filename_parsing.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
"""Tests for the queue's filename → camera / event-type parsers.

Viofo filenames end ``…NNNNN[PE]?[FRTI].MP4`` — the trailing
letter is the camera (F=front, R=rear, T=telephoto, I=interior)
and the optional P/E prefix encodes parking/event clips.
3-channel models pair F+R with either T or I.
"""
from __future__ import annotations

from web.services.queue import (
_camera_from_filename,
_event_from_filename,
)


def test_camera_front_rear():
assert _camera_from_filename("2026_0513_172152_000120F.MP4") == "F"
assert _camera_from_filename("2026_0513_172152_000121R.MP4") == "R"
assert _camera_from_filename("2026_0513_172152_000122PF.MP4") == "F"
assert _camera_from_filename("2026_0513_172152_000123ER.MP4") == "R"


def test_camera_tele():
assert _camera_from_filename("2026_0513_172152_000124T.MP4") == "T"
assert _camera_from_filename("2026_0513_172152_000125PT.MP4") == "T"
assert _camera_from_filename("2026_0513_172152_000126ET.MP4") == "T"


def test_camera_interior():
assert _camera_from_filename("2026_0109_143514_000487I.MP4") == "I"
assert _camera_from_filename("2026_0109_145126_000520PI.MP4") == "I"
assert _camera_from_filename("2026_0109_145126_000521EI.MP4") == "I"


def test_camera_case_insensitive():
assert _camera_from_filename("2026_0513_172152_000124t.mp4") == "T"


def test_camera_unknown_letter_rejected():
assert _camera_from_filename("2026_0513_172152_000124X.MP4") is None
assert _camera_from_filename("notes.txt") is None


def test_event_type_for_third_cameras():
assert _event_from_filename("2026_0513_172152_000124T.MP4") == "normal"
assert _event_from_filename("2026_0513_172152_000125PT.MP4") == "parking"
assert _event_from_filename("2026_0513_172152_000126ET.MP4") == "event"
assert _event_from_filename("2026_0109_143514_000487I.MP4") == "normal"
assert _event_from_filename("2026_0109_145126_000520PI.MP4") == "parking"
assert _event_from_filename("2026_0109_145126_000521EI.MP4") == "event"
27 changes: 27 additions & 0 deletions tests/test_timeline_endpoint.py
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,33 @@ def test_timeline_day_mode_channels_clips_bounds(logged_in_client):
assert body["gps"] is None


def test_timeline_third_camera_channels(logged_in_client):
"""Tele and interior clips each get their own channel, in
CHANNEL_ORDER after rear. A real device has only one of the
two, but the endpoint must order any mix it finds — e.g. an
archive that spans both a telephoto and an interior setup."""
app = logged_in_client.app
_insert_clip(app, 1, 1_717_312_440, "F", 60.0)
_insert_clip(app, 2, 1_717_312_440, "R", 60.0)
_insert_clip(app, 3, 1_717_312_440, "T", 60.0)
_insert_clip(app, 4, 1_717_312_440, "I", 60.0)

r = logged_in_client.get("/api/archive/timeline?date=2026-06-02")
assert r.status_code == 200
body = r.json()
assert [ch["key"] for ch in body["channels"]] == [
"front", "rear", "tele", "interior",
]
labels = {ch["key"]: ch["label"] for ch in body["channels"]}
assert labels["tele"] == "Tele"
assert labels["interior"] == "Interior"
by_channel = {}
for c in body["clips"]:
by_channel.setdefault(c["channel"], []).append(c)
assert len(by_channel["tele"]) == 1
assert len(by_channel["interior"]) == 1


def test_timeline_journey_mode_windows_clips(logged_in_client, monkeypatch):
app = logged_in_client.app
_insert_clip(app, 1, 1_717_312_440, "F", 60.0)
Expand Down
6 changes: 4 additions & 2 deletions viofosync_lib/_archive.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,11 +30,13 @@
"yearly": "[0-9][0-9][0-9][0-9]",
}

# Downloaded recording filename glob pattern.
# Downloaded recording filename glob pattern. The trailing
# letter is the camera: F=front, R=rear, T=telephoto,
# I=interior.
downloaded_filename_glob = (
"[0-9][0-9][0-9][0-9]_[0-9][0-9][0-9][0-9]"
"_[0-9][0-9][0-9][0-9][0-9][0-9]"
"_*[FR].MP4"
"_*[FRTI].MP4"
)

# Downloaded recording filename regular expression.
Expand Down
Loading
Loading