Travis Vasceannie
1ce24cdf7b
- Moved all hookify configuration files from `.claude/` to `.claude/hooks/` subdirectory for better organization - Added four new blocking hooks to prevent common error handling anti-patterns: - `block-broad-exception-handler`: Prevents catching generic `Exception` with only logging - `block-datetime-now-fallback`: Blocks returning `datetime.now()` as fallback on parse failures to prevent data corruption - `block-default
259 lines
10 KiB
Python
259 lines
10 KiB
Python
"""Tests for TriggerService application logic."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
import time
|
|
from dataclasses import dataclass
|
|
|
|
import pytest
|
|
|
|
from noteflow.application.services.triggers import (
|
|
TriggerService,
|
|
TriggerServiceSettings,
|
|
)
|
|
from noteflow.domain.triggers import TriggerAction, TriggerSignal, TriggerSource
|
|
|
|
|
|
@dataclass
|
|
class FakeProvider:
|
|
"""Simple signal provider for testing."""
|
|
|
|
signal: TriggerSignal | None
|
|
enabled: bool = True
|
|
calls: int = 0
|
|
|
|
@property
|
|
def source(self) -> TriggerSource:
|
|
return TriggerSource.AUDIO_ACTIVITY
|
|
|
|
@property
|
|
def max_weight(self) -> float:
|
|
return 1.0
|
|
|
|
def is_enabled(self) -> bool:
|
|
return self.enabled
|
|
|
|
def get_signal(self) -> TriggerSignal | None:
|
|
self.calls += 1
|
|
return self.signal
|
|
|
|
|
|
def _settings(
|
|
*,
|
|
enabled: bool = True,
|
|
auto_start: bool = False,
|
|
rate_limit_seconds: int = 60,
|
|
snooze_seconds: int = 30,
|
|
threshold_ignore: float = 0.2,
|
|
threshold_auto: float = 0.8,
|
|
) -> TriggerServiceSettings:
|
|
return TriggerServiceSettings(
|
|
enabled=enabled,
|
|
auto_start_enabled=auto_start,
|
|
rate_limit_seconds=rate_limit_seconds,
|
|
snooze_seconds=snooze_seconds,
|
|
threshold_ignore=threshold_ignore,
|
|
threshold_auto_start=threshold_auto,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"attr,expected",
|
|
[("action", TriggerAction.IGNORE), ("confidence", 0.0), ("signals", ())],
|
|
)
|
|
def test_trigger_service_disabled_skips_providers(attr: str, expected: object) -> None:
|
|
"""Disabled trigger service should ignore without evaluating providers."""
|
|
provider = FakeProvider(signal=TriggerSignal(TriggerSource.AUDIO_ACTIVITY, weight=0.5))
|
|
service = TriggerService([provider], settings=_settings(enabled=False))
|
|
decision = service.evaluate()
|
|
assert getattr(decision, attr) == expected
|
|
|
|
|
|
def test_trigger_service_disabled_never_calls_provider() -> None:
|
|
"""Disabled service does not call providers."""
|
|
provider = FakeProvider(signal=TriggerSignal(TriggerSource.AUDIO_ACTIVITY, weight=0.5))
|
|
service = TriggerService([provider], settings=_settings(enabled=False))
|
|
service.evaluate()
|
|
assert provider.calls == 0
|
|
|
|
|
|
def test_trigger_service_snooze_ignores_signals(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Snoozed trigger service ignores signals until snooze expires."""
|
|
provider = FakeProvider(signal=TriggerSignal(TriggerSource.AUDIO_ACTIVITY, weight=0.5))
|
|
service = TriggerService([provider], settings=_settings())
|
|
|
|
monkeypatch.setattr(time, "monotonic", lambda: 100.0)
|
|
service.snooze(seconds=20)
|
|
|
|
monkeypatch.setattr(time, "monotonic", lambda: 110.0)
|
|
decision = service.evaluate()
|
|
assert decision.action == TriggerAction.IGNORE, "Should ignore signals during active snooze"
|
|
|
|
monkeypatch.setattr(time, "monotonic", lambda: 130.0)
|
|
decision = service.evaluate()
|
|
assert decision.action == TriggerAction.NOTIFY, "Should notify after snooze expires"
|
|
|
|
|
|
def test_trigger_service_rate_limit(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""TriggerService enforces rate limit between prompts."""
|
|
provider = FakeProvider(signal=TriggerSignal(TriggerSource.AUDIO_ACTIVITY, weight=0.5))
|
|
service = TriggerService([provider], settings=_settings(rate_limit_seconds=60))
|
|
|
|
monkeypatch.setattr(time, "monotonic", lambda: 100.0)
|
|
first = service.evaluate()
|
|
assert first.action == TriggerAction.NOTIFY, "First evaluation should notify"
|
|
|
|
monkeypatch.setattr(time, "monotonic", lambda: 120.0)
|
|
second = service.evaluate()
|
|
assert second.action == TriggerAction.IGNORE, "Should ignore within rate limit window"
|
|
|
|
monkeypatch.setattr(time, "monotonic", lambda: 200.0)
|
|
third = service.evaluate()
|
|
assert third.action == TriggerAction.NOTIFY, "Should notify after rate limit expires"
|
|
|
|
|
|
def test_trigger_service_auto_start(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Auto-start fires when confidence passes threshold and auto-start is enabled."""
|
|
provider = FakeProvider(signal=TriggerSignal(TriggerSource.AUDIO_ACTIVITY, weight=0.9))
|
|
service = TriggerService([provider], settings=_settings(auto_start=True, threshold_auto=0.8))
|
|
|
|
monkeypatch.setattr(time, "monotonic", lambda: 100.0)
|
|
decision = service.evaluate()
|
|
|
|
assert decision.action == TriggerAction.AUTO_START
|
|
|
|
|
|
def test_trigger_service_auto_start_disabled_notifies(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""High confidence should still notify when auto-start is disabled."""
|
|
provider = FakeProvider(signal=TriggerSignal(TriggerSource.AUDIO_ACTIVITY, weight=0.9))
|
|
service = TriggerService([provider], settings=_settings(auto_start=False, threshold_auto=0.8))
|
|
|
|
monkeypatch.setattr(time, "monotonic", lambda: 100.0)
|
|
decision = service.evaluate()
|
|
|
|
assert decision.action == TriggerAction.NOTIFY
|
|
|
|
|
|
def test_trigger_service_below_ignore_threshold(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Signals below ignore threshold should be ignored."""
|
|
provider = FakeProvider(signal=TriggerSignal(TriggerSource.AUDIO_ACTIVITY, weight=0.1))
|
|
service = TriggerService([provider], settings=_settings(threshold_ignore=0.2))
|
|
|
|
monkeypatch.setattr(time, "monotonic", lambda: 100.0)
|
|
decision = service.evaluate()
|
|
|
|
assert decision.action == TriggerAction.IGNORE
|
|
|
|
|
|
def test_trigger_service_threshold_validation() -> None:
|
|
"""Invalid threshold ordering should raise."""
|
|
with pytest.raises(ValueError, match="threshold_auto_start"):
|
|
TriggerServiceSettings(
|
|
enabled=True,
|
|
auto_start_enabled=False,
|
|
rate_limit_seconds=10,
|
|
snooze_seconds=5,
|
|
threshold_ignore=0.9,
|
|
threshold_auto_start=0.2,
|
|
)
|
|
|
|
|
|
def test_trigger_service_skips_disabled_providers() -> None:
|
|
"""Disabled providers should be skipped when evaluating."""
|
|
enabled_signal = FakeProvider(signal=TriggerSignal(TriggerSource.AUDIO_ACTIVITY, weight=0.3))
|
|
disabled_signal = FakeProvider(
|
|
signal=TriggerSignal(TriggerSource.AUDIO_ACTIVITY, weight=0.7), enabled=False
|
|
)
|
|
service = TriggerService([enabled_signal, disabled_signal], settings=_settings())
|
|
|
|
decision = service.evaluate()
|
|
|
|
assert math.isclose(decision.confidence, 0.3, rel_tol=1e-9), "Confidence should reflect only enabled provider"
|
|
assert enabled_signal.calls == 1, "Enabled provider should be called once"
|
|
assert disabled_signal.calls == 0, "Disabled provider should not be called"
|
|
|
|
|
|
def test_trigger_service_snooze_state_active_is_snoozed(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""is_snoozed returns True during active snooze."""
|
|
service = TriggerService([], settings=_settings())
|
|
monkeypatch.setattr(time, "monotonic", lambda: 50.0)
|
|
service.snooze(seconds=10)
|
|
monkeypatch.setattr(time, "monotonic", lambda: 55.0)
|
|
assert service.is_snoozed is True
|
|
|
|
|
|
def test_trigger_service_snooze_state_active_remaining_seconds(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""snooze_remaining_seconds reflects active snooze duration."""
|
|
service = TriggerService([], settings=_settings())
|
|
monkeypatch.setattr(time, "monotonic", lambda: 50.0)
|
|
service.snooze(seconds=10)
|
|
monkeypatch.setattr(time, "monotonic", lambda: 55.0)
|
|
assert math.isclose(service.snooze_remaining_seconds, 5.0, rel_tol=1e-9)
|
|
|
|
|
|
def test_trigger_service_snooze_state_cleared_is_snoozed(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""is_snoozed returns False after snooze is cleared."""
|
|
service = TriggerService([], settings=_settings())
|
|
monkeypatch.setattr(time, "monotonic", lambda: 50.0)
|
|
service.snooze(seconds=10)
|
|
service.clear_snooze()
|
|
assert service.is_snoozed is False
|
|
|
|
|
|
def test_trigger_service_snooze_state_cleared_remaining_seconds(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""snooze_remaining_seconds returns 0.0 after snooze is cleared."""
|
|
service = TriggerService([], settings=_settings())
|
|
monkeypatch.setattr(time, "monotonic", lambda: 50.0)
|
|
service.snooze(seconds=10)
|
|
service.clear_snooze()
|
|
assert service.snooze_remaining_seconds == 0.0
|
|
|
|
|
|
def test_trigger_service_rate_limit_with_existing_prompt(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Existing prompt time inside rate limit should short-circuit to IGNORE."""
|
|
provider = FakeProvider(signal=TriggerSignal(TriggerSource.AUDIO_ACTIVITY, weight=0.9))
|
|
service = TriggerService([provider], settings=_settings(rate_limit_seconds=30))
|
|
|
|
# First call at t=90 triggers a prompt, setting _last_prompt internally
|
|
monkeypatch.setattr(time, "monotonic", lambda: 90.0)
|
|
first_decision = service.evaluate()
|
|
assert first_decision.action == TriggerAction.NOTIFY, "First evaluation should notify"
|
|
|
|
# Second call at t=100 (10s later) should be rate-limited (< 30s)
|
|
monkeypatch.setattr(time, "monotonic", lambda: 100.0)
|
|
second_decision = service.evaluate()
|
|
|
|
assert second_decision.action == TriggerAction.IGNORE, "Should ignore within rate limit"
|
|
assert service.is_enabled is True, "Service should remain enabled after rate limit"
|
|
|
|
|
|
def test_trigger_service_enable_toggles(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""set_enabled and set_auto_start should update settings."""
|
|
# Use a high-confidence provider to verify auto_start behavior
|
|
provider = FakeProvider(signal=TriggerSignal(TriggerSource.AUDIO_ACTIVITY, weight=0.9))
|
|
service = TriggerService(
|
|
[provider], settings=_settings(enabled=True, auto_start=False, threshold_auto=0.8)
|
|
)
|
|
monkeypatch.setattr(time, "monotonic", lambda: 100.0)
|
|
|
|
# Verify set_enabled(False) disables the service
|
|
service.set_enabled(False)
|
|
assert service.is_enabled is False, "Service should be disabled after set_enabled(False)"
|
|
|
|
# Re-enable and verify set_auto_start(True) enables auto-start behavior
|
|
service.set_enabled(True)
|
|
service.set_auto_start(True)
|
|
|
|
# With auto_start enabled and high confidence, should get AUTO_START action
|
|
decision = service.evaluate()
|
|
assert decision.action == TriggerAction.AUTO_START, "Should auto-start with high confidence and auto_start enabled"
|