5355fcaba8
* Add H.265 support to stream component * Change find_box to generator * Move fmp4 utilities to fmp4utils.py * Add minimum segments and segment durations * Remove MIN_SEGMENTS * Fix when container_options is None * Fix missing num_segments and update tests * Remove unnecessary mock attribute * Fix Segment construction in test_recorder_save * fix recorder with lookback Co-authored-by: Jason Hunter <hunterjm@gmail.com>
124 lines
3.6 KiB
Python
124 lines
3.6 KiB
Python
"""The tests for hls streams."""
|
|
from datetime import timedelta
|
|
from urllib.parse import urlparse
|
|
|
|
import pytest
|
|
|
|
from homeassistant.components.stream import request_stream
|
|
from homeassistant.const import HTTP_NOT_FOUND
|
|
from homeassistant.setup import async_setup_component
|
|
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
|
|
|
|
|
|
@pytest.mark.skip("Flaky in CI")
|
|
async def test_hls_stream(hass, hass_client):
|
|
"""
|
|
Test hls stream.
|
|
|
|
Purposefully not mocking anything here to test full
|
|
integration with the stream component.
|
|
"""
|
|
await async_setup_component(hass, "stream", {"stream": {}})
|
|
|
|
# Setup demo HLS track
|
|
source = generate_h264_video()
|
|
stream = preload_stream(hass, source)
|
|
stream.add_provider("hls")
|
|
|
|
# Request stream
|
|
url = request_stream(hass, source)
|
|
|
|
http_client = await hass_client()
|
|
|
|
# Fetch playlist
|
|
parsed_url = urlparse(url)
|
|
playlist_response = await http_client.get(parsed_url.path)
|
|
assert playlist_response.status == 200
|
|
|
|
# Fetch init
|
|
playlist = await playlist_response.text()
|
|
playlist_url = "/".join(parsed_url.path.split("/")[:-1])
|
|
init_url = playlist_url + "/init.mp4"
|
|
init_response = await http_client.get(init_url)
|
|
assert init_response.status == 200
|
|
|
|
# Fetch segment
|
|
playlist = await playlist_response.text()
|
|
playlist_url = "/".join(parsed_url.path.split("/")[:-1])
|
|
segment_url = playlist_url + playlist.splitlines()[-1][1:]
|
|
segment_response = await http_client.get(segment_url)
|
|
assert segment_response.status == 200
|
|
|
|
# Stop stream, if it hasn't quit already
|
|
stream.stop()
|
|
|
|
# Ensure playlist not accessible after stream ends
|
|
fail_response = await http_client.get(parsed_url.path)
|
|
assert fail_response.status == HTTP_NOT_FOUND
|
|
|
|
|
|
@pytest.mark.skip("Flaky in CI")
|
|
async def test_stream_timeout(hass, hass_client):
|
|
"""Test hls stream timeout."""
|
|
await async_setup_component(hass, "stream", {"stream": {}})
|
|
|
|
# Setup demo HLS track
|
|
source = generate_h264_video()
|
|
stream = preload_stream(hass, source)
|
|
stream.add_provider("hls")
|
|
|
|
# Request stream
|
|
url = request_stream(hass, source)
|
|
|
|
http_client = await hass_client()
|
|
|
|
# Fetch playlist
|
|
parsed_url = urlparse(url)
|
|
playlist_response = await http_client.get(parsed_url.path)
|
|
assert playlist_response.status == 200
|
|
|
|
# Wait a minute
|
|
future = dt_util.utcnow() + timedelta(minutes=1)
|
|
async_fire_time_changed(hass, future)
|
|
|
|
# Fetch again to reset timer
|
|
playlist_response = await http_client.get(parsed_url.path)
|
|
assert playlist_response.status == 200
|
|
|
|
# Wait 5 minutes
|
|
future = dt_util.utcnow() + timedelta(minutes=5)
|
|
async_fire_time_changed(hass, future)
|
|
|
|
# Ensure playlist not accessible
|
|
fail_response = await http_client.get(parsed_url.path)
|
|
assert fail_response.status == HTTP_NOT_FOUND
|
|
|
|
|
|
@pytest.mark.skip("Flaky in CI")
|
|
async def test_stream_ended(hass):
|
|
"""Test hls stream packets ended."""
|
|
await async_setup_component(hass, "stream", {"stream": {}})
|
|
|
|
# Setup demo HLS track
|
|
source = generate_h264_video()
|
|
stream = preload_stream(hass, source)
|
|
track = stream.add_provider("hls")
|
|
|
|
# Request stream
|
|
request_stream(hass, source)
|
|
|
|
# Run it dead
|
|
while True:
|
|
segment = await track.recv()
|
|
if segment is None:
|
|
break
|
|
segments = segment.sequence
|
|
|
|
assert segments > 1
|
|
assert not track.get_segment()
|
|
|
|
# Stop stream, if it hasn't quit already
|
|
stream.stop()
|