69 lines
2.5 KiB
Python
69 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from yovision_brain.app import run_pipeline
|
|
from yovision_brain.events import JsonLinesSink
|
|
from yovision_brain.input import CancellationToken
|
|
|
|
|
|
FIXTURE = Path(__file__).parents[1] / "fixtures" / "events" / "area.json"
|
|
|
|
|
|
def test_pipeline_generates_stable_internal_event() -> None:
|
|
raw = json.loads(FIXTURE.read_text(encoding="utf-8"))
|
|
first_stream, second_stream = io.StringIO(), io.StringIO()
|
|
first = run_pipeline(raw, JsonLinesSink(first_stream), base_dir=FIXTURE.parent)
|
|
second = run_pipeline(raw, JsonLinesSink(second_stream), base_dir=FIXTURE.parent)
|
|
assert first.events == second.events == 1
|
|
assert first_stream.getvalue() == second_stream.getvalue()
|
|
event = json.loads(first_stream.getvalue())
|
|
assert event["event_type"] == "danger_area_entered"
|
|
assert event["schema"] == "brain.internal.event-candidate/v1"
|
|
|
|
|
|
def test_no_hit_has_explicit_zero_event_summary() -> None:
|
|
raw = json.loads(FIXTURE.read_text(encoding="utf-8"))
|
|
raw["detector"]["threshold"] = 255
|
|
raw["detector"]["minimum_area"] = 1000
|
|
stream = io.StringIO()
|
|
summary = run_pipeline(raw, JsonLinesSink(stream))
|
|
assert (summary.status, summary.events, stream.getvalue()) == ("completed", 0, "")
|
|
|
|
|
|
def test_cli_runs_without_sense_or_bell() -> None:
|
|
result = subprocess.run(
|
|
[sys.executable, "-m", "yovision_brain.app", "--config", str(FIXTURE), "--output", "-"],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert result.returncode == 0
|
|
assert json.loads(result.stdout)["schema"] == "brain.internal.event-candidate/v1"
|
|
assert json.loads(result.stderr)["events"] == 1
|
|
|
|
|
|
def test_pre_cancelled_run_is_explicit() -> None:
|
|
raw = json.loads(FIXTURE.read_text(encoding="utf-8"))
|
|
token = CancellationToken()
|
|
token.cancel()
|
|
summary = run_pipeline(raw, JsonLinesSink(io.StringIO()), cancellation=token)
|
|
assert (summary.status, summary.frames, summary.events) == ("cancelled", 0, 0)
|
|
|
|
|
|
def test_cli_config_failure_is_nonzero_and_does_not_echo_path(tmp_path: Path) -> None:
|
|
missing = tmp_path / "private-machine-path.json"
|
|
result = subprocess.run(
|
|
[sys.executable, "-m", "yovision_brain.app", "--config", str(missing)],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert result.returncode == 2
|
|
assert json.loads(result.stderr)["status"] == "error"
|
|
assert str(tmp_path) not in result.stderr
|