thomas
1f110866f5
CI / backend (pull_request) Canceled after 0s
CI / shell (pull_request) Canceled after 0s
CI / frontend (pull_request) Canceled after 0s
CI / arm64-smoke (pull_request) Canceled after 0s
CI / backend (push) Canceled after 0s
CI / shell (push) Canceled after 0s
CI / frontend (push) Canceled after 0s
CI / arm64-smoke (push) Canceled after 0s
56 lines
1.5 KiB
Python
56 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
from app.effects.service import EffectService
|
|
from app.models.schemas import EffectPayload
|
|
|
|
|
|
class FakeEngine:
|
|
def __init__(self) -> None:
|
|
self.layers: dict[str, object] = {}
|
|
|
|
def set_layer(self, layer) -> None: # type: ignore[no-untyped-def]
|
|
self.layers[layer.name] = layer
|
|
|
|
def remove_layer(self, name: str) -> None:
|
|
self.layers.pop(name, None)
|
|
|
|
|
|
class FakeBpm:
|
|
def __init__(self) -> None:
|
|
self.current_bpm = 180.0
|
|
self.audio_connected = False
|
|
self.beat_counter = 0
|
|
|
|
|
|
def test_beat_flash_effect_pulses_layer_from_bpm_clock() -> None:
|
|
async def run_test() -> None:
|
|
engine = FakeEngine()
|
|
bpm = FakeBpm()
|
|
service = EffectService(engine, bpm) # type: ignore[arg-type]
|
|
payload = EffectPayload(
|
|
name="Beat dimmer",
|
|
slug="beat-dimmer",
|
|
effect_type="beat-flash",
|
|
priority=60,
|
|
parameters={
|
|
"duration_ms": 90,
|
|
"channels": {"1": 255},
|
|
"precedence": {"1": "htp"},
|
|
},
|
|
)
|
|
|
|
service.save(payload)
|
|
service.trigger(payload.slug)
|
|
await asyncio.sleep(0.03)
|
|
assert "effect:beat-dimmer" in engine.layers
|
|
|
|
await asyncio.sleep(0.16)
|
|
assert "effect:beat-dimmer" not in engine.layers
|
|
|
|
service.stop(payload.slug)
|
|
await service.shutdown()
|
|
|
|
asyncio.run(run_test())
|