mirror of
https://github.com/ApfelTeeSaft/restream_playout.git
synced 2026-08-26 19:33:32 +00:00
74 lines
2.2 KiB
Python
74 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from restream_playout.exceptions import StreamDisconnected
|
|
from restream_playout.session import StreamSession
|
|
|
|
|
|
class _FakeClient:
|
|
attempts = 0
|
|
|
|
def __init__(self, url: str) -> None:
|
|
self.url = url
|
|
self.connected = False
|
|
self.last_pong = 0.0
|
|
self.fail_send = False
|
|
self._closed = asyncio.Event()
|
|
|
|
async def connect(self) -> None:
|
|
type(self).attempts += 1
|
|
if type(self).attempts == 1:
|
|
raise OSError("simulated connect failure")
|
|
self.connected = True
|
|
self.last_pong = 10**12
|
|
|
|
async def monitor(self) -> None:
|
|
await self._closed.wait()
|
|
|
|
async def ping(self) -> None:
|
|
pass
|
|
|
|
async def send_message(self, *args: object, **kwargs: object) -> None:
|
|
if self.fail_send:
|
|
raise OSError("simulated send failure")
|
|
|
|
async def close(self) -> None:
|
|
self.connected = False
|
|
self._closed.set()
|
|
|
|
async def unpublish(self) -> None:
|
|
pass
|
|
|
|
|
|
class SessionTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_reconnects_and_increments_generation(self) -> None:
|
|
_FakeClient.attempts = 0
|
|
states: list[bool] = []
|
|
stop = asyncio.Event()
|
|
|
|
async def state_callback(connected: bool) -> None:
|
|
states.append(connected)
|
|
|
|
with patch("restream_playout.session.RTMPClient", _FakeClient):
|
|
session = StreamSession(
|
|
"rtmp://localhost/live/key",
|
|
stop_event=stop,
|
|
state_callback=state_callback,
|
|
reconnect_min_delay=0.001,
|
|
reconnect_max_delay=0.001,
|
|
heartbeat_interval=100,
|
|
)
|
|
self.assertEqual(await session.ensure_connected(), 1)
|
|
client = session._client
|
|
self.assertIsNotNone(client)
|
|
client.fail_send = True
|
|
with self.assertRaises(StreamDisconnected):
|
|
await session.send_message(9, 0, b"frame")
|
|
self.assertEqual(await session.ensure_connected(), 2)
|
|
await session.close()
|
|
|
|
self.assertEqual(states, [True, False, True, False])
|