evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Use async with to fetch HTTP streams in tests (#82788)

Browse source
This commit is contained in:
uvjustin 2022-11-28 03:35:03 +08:00 • committed by GitHub
parent 444ad52757
commit 31ad208500
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 23 additions and 21 deletions
Download patch file Download diff file Expand all files Collapse all files

34
tests/components/api/test_init.py
View file

@ -327,35 +327,35 @@ async def test_stream(hass, mock_api_client):
"""Test the stream.""" """Test the stream."""
listen_count = _listen_count(hass) listen_count = _listen_count(hass)
resp = await mock_api_client.get(const.URL_API_STREAM) async with mock_api_client.get(const.URL_API_STREAM) as resp:
assert resp.status == HTTPStatus.OK assert resp.status == HTTPStatus.OK
assert listen_count + 1 == _listen_count(hass) assert listen_count + 1 == _listen_count(hass)
hass.bus.async_fire("test_event") hass.bus.async_fire("test_event")
data = await _stream_next_event(resp.content) data = await _stream_next_event(resp.content)
assert data["event_type"] == "test_event" assert data["event_type"] == "test_event"
async def test_stream_with_restricted(hass, mock_api_client): async def test_stream_with_restricted(hass, mock_api_client):
"""Test the stream with restrictions.""" """Test the stream with restrictions."""
listen_count = _listen_count(hass) listen_count = _listen_count(hass)
resp = await mock_api_client.get( async with mock_api_client.get(
f"{const.URL_API_STREAM}?restrict=test_event1,test_event3" f"{const.URL_API_STREAM}?restrict=test_event1,test_event3"
) ) as resp:
assert resp.status == HTTPStatus.OK assert resp.status == HTTPStatus.OK
assert listen_count + 1 == _listen_count(hass) assert listen_count + 1 == _listen_count(hass)
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
data = await _stream_next_event(resp.content) data = await _stream_next_event(resp.content)
assert data["event_type"] == "test_event1" assert data["event_type"] == "test_event1"
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
hass.bus.async_fire("test_event3") hass.bus.async_fire("test_event3")
data = await _stream_next_event(resp.content) data = await _stream_next_event(resp.content)
assert data["event_type"] == "test_event3" assert data["event_type"] == "test_event3"
async def _stream_next_event(stream): async def _stream_next_event(stream):

10
tests/components/camera/test_init.py
View file

@ -503,15 +503,17 @@ async def test_camera_proxy_stream(hass, mock_camera, hass_client):
client = await hass_client() client = await hass_client()
response = await client.get("/api/camera_proxy_stream/camera.demo_camera") async with client.get("/api/camera_proxy_stream/camera.demo_camera") as response:
assert response.status == HTTPStatus.OK assert response.status == HTTPStatus.OK
with patch( with patch(
"homeassistant.components.demo.camera.DemoCamera.handle_async_mjpeg_stream", "homeassistant.components.demo.camera.DemoCamera.handle_async_mjpeg_stream",
return_value=None, return_value=None,
): ):
response = await client.get("/api/camera_proxy_stream/camera.demo_camera") async with await client.get(
assert response.status == HTTPStatus.BAD_GATEWAY "/api/camera_proxy_stream/camera.demo_camera"
) as response:
assert response.status == HTTPStatus.BAD_GATEWAY
async def test_websocket_web_rtc_offer( async def test_websocket_web_rtc_offer(