84 lines
2.2 KiB
Python
84 lines
2.2 KiB
Python
|
"""The tests for hls streams."""
|
||
|
|
from datetime import timedelta
|
||
|
|
from io import BytesIO
|
||
|
|
from unittest.mock import patch
|
||
|
|
|
||
|
|
from homeassistant.setup import async_setup_component
|
||
|
|
from homeassistant.components.stream.core import Segment
|
||
|
|
from homeassistant.components.stream.recorder import recorder_save_worker
|
||
|
|
import homeassistant.util.dt as dt_util
|
||
|
|
|
||
|
|
from tests.common import async_fire_time_changed
|
||
|
|
from tests.components.stream.common import (
|
||
|
|
generate_h264_video, preload_stream)
|
||
|
|
|
||
|
|
|
||
|
|
async def test_record_stream(hass, hass_client):
|
||
|
|
"""
|
||
|
|
Test record stream.
|
||
|
|
|
||
|
|
Purposefully not mocking anything here to test full
|
||
|
|
integration with the stream component.
|
||
|
|
"""
|
||
|
|
await async_setup_component(hass, 'stream', {
|
||
|
|
'stream': {}
|
||
|
|
})
|
||
|
|
|
||
|
|
with patch(
|
||
|
|
'homeassistant.components.stream.recorder.recorder_save_worker'):
|
||
|
|
# Setup demo track
|
||
|
|
source = generate_h264_video()
|
||
|
|
stream = preload_stream(hass, source)
|
||
|
|
recorder = stream.add_provider('recorder')
|
||
|
|
stream.start()
|
||
|
|
|
||
|
|
segments = 0
|
||
|
|
while True:
|
||
|
|
segment = await recorder.recv()
|
||
|
|
if not segment:
|
||
|
|
break
|
||
|
|
segments += 1
|
||
|
|
|
||
|
|
stream.stop()
|
||
|
|
|
||
|
|
assert segments == 3
|
||
|
|
|
||
|
|
|
||
|
|
async def test_recorder_timeout(hass, hass_client):
|
||
|
|
"""Test recorder timeout."""
|
||
|
|
await async_setup_component(hass, 'stream', {
|
||
|
|
'stream': {}
|
||
|
|
})
|
||
|
|
|
||
|
|
with patch(
|
||
|
|
'homeassistant.components.stream.recorder.RecorderOutput.cleanup'
|
||
|
|
) as mock_cleanup:
|
||
|
|
# Setup demo track
|
||
|
|
source = generate_h264_video()
|
||
|
|
stream = preload_stream(hass, source)
|
||
|
|
recorder = stream.add_provider('recorder')
|
||
|
|
stream.start()
|
||
|
|
|
||
|
|
await recorder.recv()
|
||
|
|
|
||
|
|
# Wait a minute
|
||
|
|
future = dt_util.utcnow() + timedelta(minutes=1)
|
||
|
|
async_fire_time_changed(hass, future)
|
||
|
|
await hass.async_block_till_done()
|
||
|
|
|
||
|
|
assert mock_cleanup.called
|
||
|
|
|
||
|
|
|
||
|
|
async def test_recorder_save():
|
||
|
|
"""Test recorder save."""
|
||
|
|
# Setup
|
||
|
|
source = generate_h264_video()
|
||
|
|
output = BytesIO()
|
||
|
|
output.name = 'test.mp4'
|
||
|
|
|
||
|
|
# Run
|
||
|
|
recorder_save_worker(output, [Segment(1, source, 4)])
|
||
|
|
|
||
|
|
# Assert
|
||
|
|
assert output.getvalue()
|