diff --git a/reolink_aio/baichuan/baichuan.py b/reolink_aio/baichuan/baichuan.py index 9efce85a..0c57d9ff 100644 --- a/reolink_aio/baichuan/baichuan.py +++ b/reolink_aio/baichuan/baichuan.py @@ -5,7 +5,7 @@ import asyncio import logging from collections.abc import Callable -from datetime import datetime, timedelta +from datetime import date, datetime, timedelta from inspect import getmembers from time import time as time_now from typing import TYPE_CHECKING, Any, Coroutine, Literal, overload @@ -121,6 +121,25 @@ } WHITELED_MODE_HTTP_TO_BC = {v: k for k, v in WHITELED_MODE_BC_TO_HTTP.items()} +# Maps a VOD_trigger to the recordType/alarmType substrings that set it. Shared +# by the cmd 272 and the FILE_FIND VOD search paths so both classify identically. +VOD_TRIGGER_TOKENS: tuple[tuple[VOD_trigger, tuple[str, ...]], ...] = ( + (VOD_trigger.MOTION, ("md", "pir", "other")), + (VOD_trigger.IO, ("io",)), + (VOD_trigger.PERSON, ("people",)), + (VOD_trigger.FACE, ("face",)), + (VOD_trigger.VEHICLE, ("vehicle",)), + (VOD_trigger.ANIMAL, ("dog_cat",)), + (VOD_trigger.DOORBELL, ("visitor",)), + (VOD_trigger.PACKAGE, ("package",)), + (VOD_trigger.CRYING, ("cry",)), + (VOD_trigger.CROSSLINE, ("crossline",)), + (VOD_trigger.INTRUSION, ("intrusion",)), + (VOD_trigger.LINGER, ("loitering",)), + (VOD_trigger.FORGOTTEN_ITEM, ("legacy",)), + (VOD_trigger.TAKEN_ITEM, ("loss",)), +) + class Baichuan: """Reolink Baichuan API class.""" @@ -4084,7 +4103,20 @@ async def search_vod_type( end_minute=end.minute, end_second=end.second, ) - mess = await self.send(cmd_id=272, channel=channel, body=xml) + try: + mess = await self.send(cmd_id=272, channel=channel, body=xml) + except ApiError as err: + if err.rspCode != 405 or request_i > 1: + raise + # Some firmwares (Reolink E1 Pro, Argus battery cameras) do not + # implement the cmd 272 VOD search and reject it with HTTP 405. + # The generic FILE_FIND search (cmd 14/15/16) returns the same SD + # recordings, so fall back to it. See issue #150. + _LOGGER.debug( + "Baichuan host %s: cmd 272 VOD search not supported (status 405), falling back to FILE_FIND", + self._host, + ) + return await self._search_vod_filefind(channel, uid, start, end, stream) fileHandle = get_value_from_xml(mess, "fileHandle") xml = xmls.FindRecVideo.format(channel=channel, fileHandle=fileHandle) @@ -4127,63 +4159,9 @@ async def search_vod_type( if time_event is not None: start_time_file = to_reolink_time_id(time_file + int((time_event - time_file) / split_time) * split_time) - vod_type_dict.setdefault(start_time_file, VOD_trigger.NONE) - if "md" in trigger or "pir" in trigger or "other" in trigger: - vod_type_dict[start_time_file] |= VOD_trigger.MOTION - vod_file.bc_triggers |= VOD_trigger.MOTION - vod_dict[VOD_trigger.MOTION].append(vod_file) - if "io" in trigger: - vod_type_dict[start_time_file] |= VOD_trigger.IO - vod_file.bc_triggers |= VOD_trigger.IO - vod_dict[VOD_trigger.IO].append(vod_file) - if "people" in trigger: - vod_type_dict[start_time_file] |= VOD_trigger.PERSON - vod_file.bc_triggers |= VOD_trigger.PERSON - vod_dict[VOD_trigger.PERSON].append(vod_file) - if "face" in trigger: - vod_type_dict[start_time_file] |= VOD_trigger.FACE - vod_file.bc_triggers |= VOD_trigger.FACE - vod_dict[VOD_trigger.FACE].append(vod_file) - if "vehicle" in trigger: - vod_type_dict[start_time_file] |= VOD_trigger.VEHICLE - vod_file.bc_triggers |= VOD_trigger.VEHICLE - vod_dict[VOD_trigger.VEHICLE].append(vod_file) - if "dog_cat" in trigger: - vod_type_dict[start_time_file] |= VOD_trigger.ANIMAL - vod_file.bc_triggers |= VOD_trigger.ANIMAL - vod_dict[VOD_trigger.ANIMAL].append(vod_file) - if "visitor" in trigger: - vod_type_dict[start_time_file] |= VOD_trigger.DOORBELL - vod_file.bc_triggers |= VOD_trigger.DOORBELL - vod_dict[VOD_trigger.DOORBELL].append(vod_file) - if "package" in trigger: - vod_type_dict[start_time_file] |= VOD_trigger.PACKAGE - vod_file.bc_triggers |= VOD_trigger.PACKAGE - vod_dict[VOD_trigger.PACKAGE].append(vod_file) - if "cry" in trigger: - vod_type_dict[start_time_file] |= VOD_trigger.CRYING - vod_file.bc_triggers |= VOD_trigger.CRYING - vod_dict[VOD_trigger.CRYING].append(vod_file) - if "crossline" in trigger: - vod_type_dict[start_time_file] |= VOD_trigger.CROSSLINE - vod_file.bc_triggers |= VOD_trigger.CROSSLINE - vod_dict[VOD_trigger.CROSSLINE].append(vod_file) - if "intrusion" in trigger: - vod_type_dict[start_time_file] |= VOD_trigger.INTRUSION - vod_file.bc_triggers |= VOD_trigger.INTRUSION - vod_dict[VOD_trigger.INTRUSION].append(vod_file) - if "loitering" in trigger: - vod_type_dict[start_time_file] |= VOD_trigger.LINGER - vod_file.bc_triggers |= VOD_trigger.LINGER - vod_dict[VOD_trigger.LINGER].append(vod_file) - if "legacy" in trigger: - vod_type_dict[start_time_file] |= VOD_trigger.FORGOTTEN_ITEM - vod_file.bc_triggers |= VOD_trigger.FORGOTTEN_ITEM - vod_dict[VOD_trigger.FORGOTTEN_ITEM].append(vod_file) - if "loss" in trigger: - vod_type_dict[start_time_file] |= VOD_trigger.TAKEN_ITEM - vod_file.bc_triggers |= VOD_trigger.TAKEN_ITEM - vod_dict[VOD_trigger.TAKEN_ITEM].append(vod_file) + self._classify_vod_triggers( + trigger, start_time_file, vod_file, vod_type_dict, vod_dict + ) if finished == 0: if time_event is None: @@ -4193,32 +4171,127 @@ async def search_vod_type( await self.send(cmd_id=274, channel=channel, body=xml) - # xml = xmls.FileInfoListOpen.format( - # channel=channel, - # uid=uid, - # stream_type=stream_type, - # start_year=start.year, - # start_month=start.month, - # start_day=start.day, - # start_hour=start.hour, - # start_minute=start.minute, - # start_second=start.second, - # end_year=end.year, - # end_month=end.month, - # end_day=end.day, - # end_hour=end.hour, - # end_minute=end.minute, - # end_second=end.second, - # ) - # mess = await self.send(cmd_id=14, body=xml) - # handle = get_value_from_xml(mess, "handle") - - # xml_file_info = xmls.FileInfoList.format(channel=channel, handle=handle, uid=uid) - # await self.send(cmd_id=15, body=xml_file_info) - # await self.send(cmd_id=16, body=xml_file_info) + return vod_type_dict, vod_dict + + @staticmethod + def _classify_vod_triggers( + trigger: str, + name: str, + vod_file: VOD_file, + vod_type_dict: dict[str, VOD_trigger], + vod_dict: dict[VOD_trigger, list[VOD_file]], + ) -> None: + """Set VOD_trigger flags on a clip from its recordType/alarmType string. + + Shared by the cmd 272 and FILE_FIND VOD search paths (see + VOD_TRIGGER_TOKENS) so both classify clips identically. + """ + vod_type_dict.setdefault(name, VOD_trigger.NONE) + for trig, tokens in VOD_TRIGGER_TOKENS: + if any(token in trigger for token in tokens): + vod_type_dict[name] |= trig + vod_file.bc_triggers = (vod_file.bc_triggers or VOD_trigger.NONE) | trig + vod_dict[trig].append(vod_file) + + async def _search_vod_filefind( + self, channel: int, uid: str, start: datetime, end: datetime, stream: str | None + ) -> tuple[dict[str, VOD_trigger], dict[VOD_trigger, list[VOD_file]]]: + """List SD recordings via the Baichuan FILE_FIND protocol (cmd 14/15/16). + + Fallback for firmwares that reject the cmd 272 VOD search with HTTP 405 + (Reolink E1 Pro, Argus battery cameras). Returns the same structure as + search_vod_type. The firmware searches a single day per FILE_FIND open + (a multi-day window returns only the start day), so iterate day-by-day + and merge the results. + """ + vod_type_dict: dict[str, VOD_trigger] = {} + vod_dict: dict[VOD_trigger, list[VOD_file]] = {trig: [] for trig in VOD_trigger} + seen: set[str] = set() + + day = start.date() + while day <= end.date(): + await self._filefind_day( + channel, uid, day, stream, vod_type_dict, vod_dict, seen + ) + day += timedelta(days=1) return vod_type_dict, vod_dict + async def _filefind_day( + self, + channel: int, + uid: str, + day: date, + stream: str | None, + vod_type_dict: dict[str, VOD_trigger], + vod_dict: dict[VOD_trigger, list[VOD_file]], + seen: set[str], + ) -> None: + """FILE_FIND one whole calendar day (00:00:00–23:59:59), merging results.""" + open_xml = xmls.FileInfoListOpen.format( + channel=channel, + uid=uid, + start_year=day.year, + start_month=day.month, + start_day=day.day, + start_hour=0, + start_minute=0, + start_second=0, + end_year=day.year, + end_month=day.month, + end_day=day.day, + end_hour=23, + end_minute=59, + end_second=59, + ) + mess = await self.send(cmd_id=14, channel=channel, body=open_xml) + handle = get_value_from_xml(mess, "handle") + if handle is None: + return + + list_xml = xmls.FileInfoList.format(channel=channel, uid=uid, handle=handle) + try: + for _ in range(50): + mess = await self.send(cmd_id=15, channel=channel, body=list_xml) + new_files = False + for item in XML.fromstring(mess).findall(".//FileInfo"): + name = get_value_from_xml(item, "name") + if name is None or name in seen: + continue + start_event = self._xml_time_to_datetime(item.find("startTime")) + end_event = self._xml_time_to_datetime(item.find("endTime")) + if start_event is None or end_event is None: + continue + seen.add(name) + new_files = True + + size_l = get_value_from_xml(item, "sizeL", int) or 0 + size_h = get_value_from_xml(item, "sizeH", int) or 0 + trigger = get_value_from_xml(item, "recordType") or "" + data = { + "type": stream, + "StartTime": datetime_to_reolink_time(start_event), + "EndTime": datetime_to_reolink_time(end_event), + "PlaybackTime": datetime_to_reolink_time(start_event), + "name": name, + "size": str(size_l + (size_h << 32)), + } + vod_file = VOD_file(data) + vod_file.bc_triggers = VOD_trigger.NONE + self._classify_vod_triggers( + trigger, name, vod_file, vod_type_dict, vod_dict + ) + + if not new_files: + break + else: + _LOGGER.warning("Baichuan host %s: FILE_FIND search exceeded 50 pages, quitting", self._host) + finally: + try: + await self.send(cmd_id=16, channel=channel, body=list_xml) + except ApiError: + pass + @property def events_active(self) -> bool: return self._events_active and time_now() - self._time_connection_lost > 120 diff --git a/tests/test_baichuan_filefind_fallback.py b/tests/test_baichuan_filefind_fallback.py new file mode 100644 index 00000000..c19305a7 --- /dev/null +++ b/tests/test_baichuan_filefind_fallback.py @@ -0,0 +1,435 @@ +from __future__ import annotations + +import unittest +from datetime import datetime +from types import SimpleNamespace +from unittest.mock import AsyncMock + +from reolink_aio.baichuan.baichuan import Baichuan +from reolink_aio.exceptions import ApiError +from reolink_aio.typings import VOD_trigger + +# A single FILE_FIND (cmd 15) page holding one clip triggered by motion + person. +_FILE_FIND_PAGE = """ + + +Mp4Record/2024-01-15/RecM01_20240115_120000_120030.mp4 +20241151200 +202411512030 +1048576 +0 +md, people + + +""" + +_CLIP_NAME = "Mp4Record/2024-01-15/RecM01_20240115_120000_120030.mp4" + + +def _page_for_day(day: int) -> str: + """One FILE_FIND page holding a single motion clip on the given day.""" + return f""" + + +Mp4Record/2024-01-{day:02d}/RecM01_202401{day:02d}_120000_120030.mp4 +20241{day}1200 +20241{day}12030 +1048576 +0 +md + + +""" + + +def _info( + name: str | None, + record_type: str = "md", + *, + day: int = 15, + size_l: int = 1048576, + size_h: int = 0, + with_times: bool = True, +) -> str: + """Build one block. name=None or with_times=False make it malformed.""" + name_tag = f"{name}" if name is not None else "" + if with_times: + times = ( + f"20241{day}" + "1200" + f"20241{day}" + "12030" + ) + else: + times = "" + return ( + f"{name_tag}{times}" + f"{size_l}{size_h}" + f"{record_type}" + ) + + +def _wrap(*infos: str) -> str: + """Wrap FileInfo blocks in a cmd-15 FileInfoList page.""" + return f'\n\n{"".join(infos)}\n\n' + + +_EMPTY_PAGE = "" + + +class TestSearchVodTypeFilefindFallback(unittest.IsolatedAsyncioTestCase): + def _make_host(self) -> Baichuan: + return Baichuan( + host="127.0.0.1", + username="user", + password="password", + http_api=SimpleNamespace( + camera_uid=lambda channel: "UID123_0", + nvr_name="test", + _updating=False, + ), + ) + + async def test_falls_back_to_filefind_on_cmd_272_status_405(self) -> None: + baichuan = self._make_host() + calls: list[int] = [] + + async def fake_send( + cmd_id: int, channel: int = 0, body: str = "", **_kwargs: object + ) -> str: + calls.append(cmd_id) + if cmd_id == 272: + raise ApiError("received status code 405 from cmd_id 272", rspCode=405) + if cmd_id == 14: + return "7" + if cmd_id == 15: + # First page returns the clip, subsequent pages are empty so the + # paging loop terminates. + return _FILE_FIND_PAGE if calls.count(15) == 1 else "" + return "" + + baichuan.send = AsyncMock(side_effect=fake_send) # type: ignore[method-assign] + + start = datetime(2024, 1, 15, 8, 0, 0) + end = datetime(2024, 1, 15, 20, 0, 0) + vod_type_dict, vod_dict = await baichuan.search_vod_type( + 0, start, end, stream="main" + ) + + # cmd 272 was attempted, then the FILE_FIND open/list/close chain ran. + self.assertIn(272, calls) + self.assertIn(14, calls) + self.assertIn(15, calls) + self.assertIn(16, calls) # handle closed + + # The clip is classified under both of its triggers. + self.assertEqual(set(vod_type_dict), {_CLIP_NAME}) + self.assertTrue(vod_type_dict[_CLIP_NAME] & VOD_trigger.MOTION) + self.assertTrue(vod_type_dict[_CLIP_NAME] & VOD_trigger.PERSON) + + motion = vod_dict[VOD_trigger.MOTION] + person = vod_dict[VOD_trigger.PERSON] + self.assertEqual(len(motion), 1) + self.assertEqual(len(person), 1) + self.assertEqual(motion[0].file_name, _CLIP_NAME) + self.assertEqual(motion[0].start_time, datetime(2024, 1, 15, 12, 0, 0)) + self.assertEqual(motion[0].size, 1048576) + # Triggers not present in recordType stay empty. + self.assertEqual(vod_dict[VOD_trigger.VEHICLE], []) + + async def test_iterates_each_day_of_a_multi_day_window(self) -> None: + # The firmware searches one day per FILE_FIND open, so a multi-day + # window must trigger one open/list/close chain per calendar day. + baichuan = self._make_host() + opens = 0 + served: set[int] = set() + + async def fake_send( + cmd_id: int, channel: int = 0, body: str = "", **_kwargs: object + ) -> str: + nonlocal opens + if cmd_id == 272: + raise ApiError("received status code 405 from cmd_id 272", rspCode=405) + if cmd_id == 14: + opens += 1 + return "7" + if cmd_id == 15: + # First page of each day's open yields that day's clip; the + # firmware maps the open window's start day onto the results. + if opens not in served: + served.add(opens) + return _page_for_day(14 + opens) # 15, 16, 17 + return "" + return "" + + baichuan.send = AsyncMock(side_effect=fake_send) # type: ignore[method-assign] + + start = datetime(2024, 1, 15, 8, 0, 0) + end = datetime(2024, 1, 17, 20, 0, 0) + vod_type_dict, vod_dict = await baichuan.search_vod_type( + 0, start, end, stream="main" + ) + + # One open per day in the [15, 17] window, and one clip harvested each. + self.assertEqual(opens, 3) + self.assertEqual(len(vod_type_dict), 3) + motion = vod_dict[VOD_trigger.MOTION] + self.assertEqual(len(motion), 3) + self.assertEqual( + sorted(f.start_time.day for f in motion), [15, 16, 17] + ) + + async def test_accumulates_clips_across_multiple_pages_in_one_day(self) -> None: + # A single day's search can span several cmd-15 pages; every page's new + # clips must be collected until an empty page ends the paging loop. + baichuan = self._make_host() + pages = iter([_wrap(_info("clipA")), _wrap(_info("clipB")), _EMPTY_PAGE]) + + async def fake_send( + cmd_id: int, channel: int = 0, body: str = "", **_kwargs: object + ) -> str: + if cmd_id == 272: + raise ApiError("received status code 405 from cmd_id 272", rspCode=405) + if cmd_id == 14: + return "7" + if cmd_id == 15: + return next(pages) + return "" + + baichuan.send = AsyncMock(side_effect=fake_send) # type: ignore[method-assign] + + vod_type_dict, vod_dict = await baichuan.search_vod_type( + 0, datetime(2024, 1, 15, 0, 0, 0), datetime(2024, 1, 15, 23, 59, 59), stream="main" + ) + self.assertEqual(set(vod_type_dict), {"clipA", "clipB"}) + self.assertEqual(len(vod_dict[VOD_trigger.MOTION]), 2) + + async def test_deduplicates_a_clip_returned_for_two_adjacent_days(self) -> None: + # A clip straddling midnight is returned by both days' searches; the + # shared `seen` set must keep it from being counted twice. + baichuan = self._make_host() + opens = 0 + served: set[int] = set() + overlap = _wrap(_info("overlap_clip")) + + async def fake_send( + cmd_id: int, channel: int = 0, body: str = "", **_kwargs: object + ) -> str: + nonlocal opens + if cmd_id == 272: + raise ApiError("received status code 405 from cmd_id 272", rspCode=405) + if cmd_id == 14: + opens += 1 + return "7" + if cmd_id == 15: + if opens not in served: + served.add(opens) + return overlap + return _EMPTY_PAGE + return "" + + baichuan.send = AsyncMock(side_effect=fake_send) # type: ignore[method-assign] + + vod_type_dict, vod_dict = await baichuan.search_vod_type( + 0, datetime(2024, 1, 15, 0, 0, 0), datetime(2024, 1, 16, 23, 59, 59), stream="main" + ) + self.assertEqual(opens, 2) # both days searched + self.assertEqual(set(vod_type_dict), {"overlap_clip"}) + self.assertEqual(len(vod_dict[VOD_trigger.MOTION]), 1) # not double-counted + + async def test_day_without_a_handle_is_skipped_and_search_continues(self) -> None: + # A day with no recordings returns no handle; that day is skipped (no + # cmd 15/16) and the loop still searches the remaining days. + baichuan = self._make_host() + opens = 0 + served = False + calls: list[int] = [] + + async def fake_send( + cmd_id: int, channel: int = 0, body: str = "", **_kwargs: object + ) -> str: + nonlocal opens, served + calls.append(cmd_id) + if cmd_id == 272: + raise ApiError("received status code 405 from cmd_id 272", rspCode=405) + if cmd_id == 14: + opens += 1 + return _EMPTY_PAGE if opens == 1 else "7" + if cmd_id == 15: + if not served: + served = True + return _wrap(_info("day2_clip", day=16)) + return _EMPTY_PAGE + return "" + + baichuan.send = AsyncMock(side_effect=fake_send) # type: ignore[method-assign] + + vod_type_dict, _ = await baichuan.search_vod_type( + 0, datetime(2024, 1, 15, 0, 0, 0), datetime(2024, 1, 16, 23, 59, 59), stream="main" + ) + self.assertEqual(opens, 2) # both days opened + self.assertEqual(set(vod_type_dict), {"day2_clip"}) + self.assertEqual(calls.count(16), 1) # only the day with a handle is closed + + async def test_classifies_every_record_type_token(self) -> None: + # Guards the full token->trigger table (and the substring matching) so a + # renamed enum or a token collision is caught. + cases = { + "md": VOD_trigger.MOTION, + "pir": VOD_trigger.MOTION, + "other": VOD_trigger.MOTION, + "io": VOD_trigger.IO, + "people": VOD_trigger.PERSON, + "face": VOD_trigger.FACE, + "vehicle": VOD_trigger.VEHICLE, + "dog_cat": VOD_trigger.ANIMAL, + "visitor": VOD_trigger.DOORBELL, + "package": VOD_trigger.PACKAGE, + "cry": VOD_trigger.CRYING, + "crossline": VOD_trigger.CROSSLINE, + "intrusion": VOD_trigger.INTRUSION, + "loitering": VOD_trigger.LINGER, + "legacy": VOD_trigger.FORGOTTEN_ITEM, + "loss": VOD_trigger.TAKEN_ITEM, + } + for token, trig in cases.items(): + with self.subTest(token=token): + baichuan = self._make_host() + served = False + + async def fake_send( + cmd_id: int, channel: int = 0, body: str = "", + _token: str = token, **_kwargs: object, + ) -> str: + nonlocal served + if cmd_id == 272: + raise ApiError("received status code 405 from cmd_id 272", rspCode=405) + if cmd_id == 14: + return "7" + if cmd_id == 15: + if not served: + served = True + return _wrap(_info("clip", record_type=_token)) + return _EMPTY_PAGE + return "" + + baichuan.send = AsyncMock(side_effect=fake_send) # type: ignore[method-assign] + + vod_type_dict, vod_dict = await baichuan.search_vod_type( + 0, datetime(2024, 1, 15, 0, 0, 0), datetime(2024, 1, 15, 23, 59, 59), stream="main" + ) + self.assertTrue(vod_type_dict["clip"] & trig) + self.assertEqual(len(vod_dict[trig]), 1) + + async def test_malformed_file_info_entries_are_skipped(self) -> None: + # Entries missing a name or an unparseable time are skipped, while valid + # siblings on the same page are still harvested. + baichuan = self._make_host() + served = False + page = _wrap( + _info(None), # no name + _info("no_times", with_times=False), # unparseable time + _info("good_clip"), + ) + + async def fake_send( + cmd_id: int, channel: int = 0, body: str = "", **_kwargs: object + ) -> str: + nonlocal served + if cmd_id == 272: + raise ApiError("received status code 405 from cmd_id 272", rspCode=405) + if cmd_id == 14: + return "7" + if cmd_id == 15: + if not served: + served = True + return page + return _EMPTY_PAGE + return "" + + baichuan.send = AsyncMock(side_effect=fake_send) # type: ignore[method-assign] + + vod_type_dict, vod_dict = await baichuan.search_vod_type( + 0, datetime(2024, 1, 15, 0, 0, 0), datetime(2024, 1, 15, 23, 59, 59), stream="main" + ) + self.assertEqual(set(vod_type_dict), {"good_clip"}) + self.assertEqual(len(vod_dict[VOD_trigger.MOTION]), 1) + + async def test_size_combines_high_and_low_dwords(self) -> None: + # Clips larger than 4 GiB carry a non-zero sizeH; size must be the 64-bit + # combination sizeL + (sizeH << 32). + baichuan = self._make_host() + served = False + + async def fake_send( + cmd_id: int, channel: int = 0, body: str = "", **_kwargs: object + ) -> str: + nonlocal served + if cmd_id == 272: + raise ApiError("received status code 405 from cmd_id 272", rspCode=405) + if cmd_id == 14: + return "7" + if cmd_id == 15: + if not served: + served = True + return _wrap(_info("big_clip", size_l=100, size_h=1)) + return _EMPTY_PAGE + return "" + + baichuan.send = AsyncMock(side_effect=fake_send) # type: ignore[method-assign] + + _, vod_dict = await baichuan.search_vod_type( + 0, datetime(2024, 1, 15, 0, 0, 0), datetime(2024, 1, 15, 23, 59, 59), stream="main" + ) + self.assertEqual(vod_dict[VOD_trigger.MOTION][0].size, (1 << 32) + 100) + + async def test_close_failure_does_not_break_the_search(self) -> None: + # A cmd 16 (close handle) failure is swallowed so results are still + # returned. + baichuan = self._make_host() + served = False + + async def fake_send( + cmd_id: int, channel: int = 0, body: str = "", **_kwargs: object + ) -> str: + nonlocal served + if cmd_id == 272: + raise ApiError("received status code 405 from cmd_id 272", rspCode=405) + if cmd_id == 14: + return "7" + if cmd_id == 15: + if not served: + served = True + return _wrap(_info("clip")) + return _EMPTY_PAGE + if cmd_id == 16: + raise ApiError("close failed", rspCode=400) + return "" + + baichuan.send = AsyncMock(side_effect=fake_send) # type: ignore[method-assign] + + vod_type_dict, _ = await baichuan.search_vod_type( + 0, datetime(2024, 1, 15, 0, 0, 0), datetime(2024, 1, 15, 23, 59, 59), stream="main" + ) + self.assertEqual(set(vod_type_dict), {"clip"}) + + async def test_non_405_api_error_is_not_swallowed(self) -> None: + baichuan = self._make_host() + + async def fake_send( + cmd_id: int, channel: int = 0, body: str = "", **_kwargs: object + ) -> str: + if cmd_id == 272: + raise ApiError("received status code 400 from cmd_id 272", rspCode=400) + return "" + + baichuan.send = AsyncMock(side_effect=fake_send) # type: ignore[method-assign] + + with self.assertRaises(ApiError): + await baichuan.search_vod_type( + 0, datetime(2024, 1, 15), datetime(2024, 1, 15), stream="main" + ) + + +if __name__ == "__main__": + unittest.main()