From baddcb414b096dcee2fb51783255f895c946edd9 Mon Sep 17 00:00:00 2001 From: Florian Bailly Date: Sun, 26 Jul 2026 16:49:31 +0200 Subject: [PATCH] Fall back to FILE_FIND when cmd 272 VOD search returns 405 Some firmwares (Reolink E1 Pro, Argus battery cameras) do not implement the cmd 272 VOD search and answer it with HTTP 405. Fall back to the Baichuan FILE_FIND protocol (cmd 14 open / 15 list / 16 close) to enumerate the SD-card recordings on those devices. Closes #150. The firmware searches a single calendar day per FILE_FIND open (a multi-day window returns only the start day), so the search iterates day-by-day and merges the results, de-duplicating clips by name across days. The recordType/alarmType -> VOD_trigger classification is factored into a shared VOD_TRIGGER_TOKENS table and _classify_vod_triggers helper, used by both the cmd 272 and FILE_FIND paths so they classify identically (this also gives FILE_FIND the full trigger set, incl. cry/crossline/intrusion/...). Live-verified on an Argus PT Ultra (fw v3.0.0.3911): a 12-day window now returns 123 clips across 11 days. Tests cover the 405 fallback, per-day iteration, multi-page paging, cross-day de-dup, days without recordings, the full trigger table, malformed entries, 64-bit sizes, and close-failure tolerance. --- reolink_aio/baichuan/baichuan.py | 237 +++++++----- tests/test_baichuan_filefind_fallback.py | 435 +++++++++++++++++++++++ 2 files changed, 590 insertions(+), 82 deletions(-) create mode 100644 tests/test_baichuan_filefind_fallback.py 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()