| import threading |
| from concurrent.futures import Future |
|
|
| import pytest |
| from support import Clock, snapshot |
|
|
| from pompom.api import ApiError |
| from pompom.config import Settings, SettingsStore |
| from pompom.runtime import Runtime |
|
|
|
|
| class Hardware: |
| def __init__(self): |
| self.captures = self.notifications = self.cancels = 0 |
|
|
| def capabilities(self): |
| return {"camera": True, "actions": ["reminder", "antenna_wiggle"]} |
|
|
| def capture(self): |
| self.captures += 1 |
| return b"jpeg" |
|
|
| def notify(self, settings): |
| self.notifications += 1 |
| return [{"action": "reminder", "status": "started"}] |
|
|
| def celebrate(self, settings): |
| return [{"action": "cheerful1", "status": "started"}] |
|
|
| def cancel(self): |
| self.cancels += 1 |
|
|
| def step(self): |
| pass |
|
|
|
|
| class Api: |
| def __init__(self): |
| self.controls = [] |
| self.counter = 0 |
| self.frames = [] |
|
|
| def start(self, settings): |
| self.counter += 1 |
| return snapshot("session-" + str(self.counter), next_capture_after_ms=0) |
|
|
| def control(self, settings, sid, action): |
| self.controls.append((sid, action)) |
| return snapshot( |
| sid, |
| state={"pause": "paused", "resume": "observing", "end": "ended"}[action], |
| ) |
|
|
| def frame(self, settings, sid, frame_id, jpeg): |
| self.frames.append(frame_id) |
| return snapshot(sid) |
|
|
|
|
| class DancingHardware(Hardware): |
| def __init__(self): |
| super().__init__() |
| self.dancing = True |
| self.dance_targets = 0 |
|
|
| def is_dancing(self): |
| return self.dancing |
|
|
| def notify(self, settings): |
| self.dancing = False |
| return super().notify(settings) |
|
|
| def cancel(self): |
| self.dancing = False |
| super().cancel() |
|
|
| def set_focus_progress(self, seconds): |
| if seconds < 10: |
| self.dancing = False |
|
|
| def step(self): |
| self.dance_targets += int(self.dancing) |
|
|
|
|
| @pytest.fixture |
| def runtime(tmp_path): |
| store = SettingsStore(tmp_path / "settings.json") |
| store.save(Settings(api_key="test-secret")) |
| r = Runtime(Hardware(), Api(), store, Clock()) |
| yield r |
| r.close() |
|
|
|
|
| def test_controls_require_consent_and_token(runtime): |
| with pytest.raises(ValueError): |
| runtime.control("start") |
| runtime.configure({"api_key": ""}) |
| with pytest.raises(ValueError): |
| runtime.control("start", consent=True) |
|
|
|
|
| def test_phone_release_executes_happy_once(runtime): |
| r = runtime |
| r.control("start", consent=True) |
| for phone in (1, 0, 0): |
| f = Future() |
| f.set_result( |
| snapshot( |
| phone_probability=phone, |
| observable_probability=1, |
| reason="phone_use_suspected" if phone else "phone_use_not_detected", |
| ) |
| ) |
| r.pending, r.pending_generation = f, r.session.generation |
| r.step() |
| events = [e for e in r.session.events if e["event"] == "phone_put_down"] |
| assert len(events) == 1 |
| assert events[0]["actions"] == [{"action": "cheerful1", "status": "started"}] |
|
|
|
|
| @pytest.mark.parametrize( |
| "interruption", ["phone", "unknown", "failure", "paused", "ended"] |
| ) |
| def test_first_phone_result_interrupts_dance_before_next_motion_target( |
| runtime, interruption |
| ): |
| r = runtime |
| r.control("start", consent=True) |
| r.hardware = DancingHardware() |
| f = Future() |
| r.pending, r.pending_generation = f, r.session.generation |
| result = snapshot( |
| state="suspected", |
| reason="phone_use_suspected", |
| phone_probability=0.9, |
| observable_probability=1, |
| notify=False, |
| continuous_hold_ms=0, |
| ) |
| if interruption in ("paused", "ended"): |
| r.control("pause" if interruption == "paused" else "end") |
| if interruption == "failure": |
| f.set_exception(ApiError("timeout")) |
| else: |
| if interruption == "unknown": |
| result |= { |
| "state": "unknown", |
| "reason": "insufficient_visibility", |
| "observable_probability": 0.1, |
| } |
| f.set_result(result) |
| r.step() |
| assert not r.hardware.dancing |
| assert r.hardware.dance_targets == 0 |
| assert r.hardware.notifications == (1 if interruption == "phone" else 0) |
| assert r.session.notification_count == 0 |
|
|
|
|
| @pytest.mark.parametrize( |
| "change", ["pause", "end", "settings", "new_session", "deadline"] |
| ) |
| def test_old_frame_never_notifies(runtime, change): |
| r = runtime |
| r.control("start", consent=True) |
| sid = r.session.session_id |
| f = Future() |
| r.pending, r.pending_generation = f, r.session.generation |
| if change == "settings": |
| r.configure({"motion_enabled": True}) |
| elif change == "new_session": |
| r.control("end") |
| r.control("start", consent=True) |
| elif change == "deadline": |
| r.session.clock.now += 1500 |
| else: |
| r.control(change) |
| f.set_result(snapshot(sid, notify=True, notification_id="notice-1")) |
| r.step() |
| assert r.hardware.notifications == 0 |
|
|
|
|
| def test_server_delay_is_used_without_extra_local_interval(runtime): |
| r = runtime |
| r.control("start", consent=True) |
| f = Future() |
| f.set_result(snapshot(next_capture_after_ms=250)) |
| r.pending, r.pending_generation = f, r.session.generation |
| r.step() |
| assert r.next_capture == r.session.clock() + 0.25 |
| assert r.hardware.captures == 0 |
|
|
|
|
| def test_only_one_inflight_and_stop_stays_responsive(runtime): |
| r = runtime |
| entered, release = threading.Event(), threading.Event() |
|
|
| def frame(settings, sid, fid, jpeg): |
| entered.set() |
| assert release.wait(3) |
| return snapshot(sid, notify=True, notification_id="late") |
|
|
| r.api.frame = frame |
| r.control("start", consent=True) |
| r.step() |
| assert entered.wait(2) |
| for _ in range(10): |
| r.session.clock.now += 1 |
| r.step() |
| assert r.hardware.captures == 1 |
| r.control("pause") |
| release.set() |
| r.pending.result(timeout=2) |
| r.step() |
| assert r.hardware.notifications == 0 and r.hardware.captures == 1 |
|
|
|
|
| def test_failed_pause_still_stops_capture(runtime): |
| r = runtime |
| r.control("start", consent=True) |
|
|
| def fail(*args): |
| raise ApiError("connection") |
|
|
| r.api.control = fail |
| with pytest.raises(ValueError): |
| r.control("pause") |
| r.step() |
| assert r.session.state == "paused" and r.hardware.captures == 0 |
|
|
|
|
| def test_end_during_slow_start_closes_remote_session(runtime): |
| r = runtime |
| entered, release = threading.Event(), threading.Event() |
|
|
| def slow(settings): |
| entered.set() |
| assert release.wait(3) |
| return snapshot() |
|
|
| r.api.start = slow |
| thread = threading.Thread(target=lambda: r.control("start", consent=True)) |
| thread.start() |
| assert entered.wait(2) |
| |
| |
| ended = threading.Event() |
| locally_stopped = threading.Event() |
| original_cancel = r.hardware.cancel |
|
|
| def cancelled(): |
| original_cancel() |
| locally_stopped.set() |
|
|
| r.hardware.cancel = cancelled |
|
|
| def end(): |
| r.control("end") |
| ended.set() |
|
|
| t = threading.Thread(target=end) |
| t.start() |
| assert locally_stopped.wait(2) |
| with r.lock: |
| assert r.session.state == "ended" |
| release.set() |
| thread.join(3) |
| t.join(3) |
| r.cleanup_pool.submit(lambda: None).result(timeout=3) |
| assert ended.is_set() and ("session-1", "end") in r.api.controls |
| assert r.hardware.captures == 0 |
|
|
|
|
| def test_settings_owner_only_no_token_echo_and_preview_local(runtime): |
| assert runtime.preview() == b"jpeg" |
| runtime.configure({"motion_enabled": True}) |
| assert runtime.store.path.stat().st_mode & 0o777 == 0o600 |
| assert "test-secret" not in str(runtime.snapshot()) |
| assert "test-secret" not in repr(runtime.session.settings) |
| runtime.control("start", consent=True) |
| with pytest.raises(ValueError): |
| runtime.preview() |
|
|
|
|
| def test_queued_capture_cancelled_and_local_timer_runs_without_frames(runtime): |
| r = runtime |
| r.control("start", consent=True) |
| g = r.session.generation |
| r.session.clock.now += 1500 |
| r.step() |
| assert r.session.state == "completed" |
| assert r._observe(g, r.session.settings, "session-1") is None |
| assert r.hardware.captures == r.hardware.notifications == 0 |
|
|
