import errno import importlib.util import json import logging import os import sys import tempfile import unittest from unittest import mock REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) MODULE_PATH = os.path.join(REPO_ROOT, "dashcam-transfer.py") spec = importlib.util.spec_from_file_location("dashcam_transfer", MODULE_PATH) dashcam = importlib.util.module_from_spec(spec) spec.loader.exec_module(dashcam) sys.modules["dashcam_transfer"] = dashcam logging.getLogger().addHandler(logging.NullHandler()) def make_mp4(path, size): os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "wb") as f: f.write(b"0" * size) class GetMp4FilesTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() def tearDown(self): self.tmp.cleanup() def test_matches_front_and_back_files(self): src = self.tmp.name make_mp4(os.path.join(src, "20260622050346_0000001F.MP4"), 10) make_mp4(os.path.join(src, "20260622050346_0000002B.MP4"), 10) make_mp4(os.path.join(src, "negeer.txt"), 10) result = dashcam.get_mp4_files(src) self.assertEqual(len(result), 2) self.assertIn(("20260622050346_0000001F.MP4", "20260622", "F"), result) self.assertIn(("20260622050346_0000002B.MP4", "20260622", "B"), result) def test_missing_dir_returns_empty_list(self): self.assertEqual(dashcam.get_mp4_files(os.path.join(self.tmp.name, "nope")), []) def test_listdir_error_returns_none(self): with mock.patch.object(dashcam.os, "listdir", side_effect=OSError(errno.EIO, "I/O error")): self.assertIsNone(dashcam.get_mp4_files(self.tmp.name)) class CheckSourceReadableTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.src = self.tmp.name def tearDown(self): self.tmp.cleanup() def test_readable_dir_returns_true(self): fname = "20260622050346_0000001F.MP4" make_mp4(os.path.join(self.src, fname), 10) self.assertTrue(dashcam.check_source_readable(self.src, [(fname, "20260622", "F")])) def test_io_error_on_read_returns_false(self): fname = "20260622050346_0000001F.MP4" make_mp4(os.path.join(self.src, fname), 10) with mock.patch("builtins.open", side_effect=OSError(errno.EIO, "I/O error")): self.assertFalse(dashcam.check_source_readable(self.src, [(fname, "20260622", "F")])) class CopyFileTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.src = os.path.join(self.tmp.name, "src.mp4") self.dst = os.path.join(self.tmp.name, "dst.mp4") make_mp4(self.src, 100) def tearDown(self): self.tmp.cleanup() def test_copies_file(self): dashcam.copy_file(self.src, self.dst, attempts=2, delay=0) self.assertEqual(os.path.getsize(self.dst), 100) def test_retries_then_succeeds(self): real_copy = dashcam.shutil.copy calls = {"n": 0} def flaky(s, d): calls["n"] += 1 if calls["n"] < 3: raise OSError(errno.EIO, "I/O error") real_copy(s, d) with mock.patch.object(dashcam.shutil, "copy", side_effect=flaky): dashcam.copy_file(self.src, self.dst, attempts=3, delay=0) self.assertEqual(calls["n"], 3) self.assertEqual(os.path.getsize(self.dst), 100) def test_raises_after_exhausting_attempts(self): with mock.patch.object(dashcam.shutil, "copy", side_effect=OSError(errno.EIO, "I/O error")): with self.assertRaises(OSError): dashcam.copy_file(self.src, self.dst, attempts=2, delay=0) class TransferFilesTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.src = os.path.join(self.tmp.name, "src") self.base = os.path.join(self.tmp.name, "base") os.makedirs(self.src) os.makedirs(self.base) def tearDown(self): self.tmp.cleanup() def _make(self, fname, size): make_mp4(os.path.join(self.src, fname), size) return (fname, fname[:8], fname[-5]) def test_transfers_files_to_camera_subdirs(self): files = [ self._make("20260622050346_0000001F.MP4", 100), self._make("20260622050346_0000002B.MP4", 200), ] ok, errors = dashcam.transfer_files(files, self.src, self.base) self.assertEqual((ok, errors), (2, 0)) self.assertFalse(os.path.exists(os.path.join(self.src, files[0][0]))) self.assertEqual( os.path.getsize(os.path.join(self.base, "20260622", "front", files[0][0])), 100 ) self.assertEqual( os.path.getsize(os.path.join(self.base, "20260622", "back", files[1][0])), 200 ) def test_skips_existing_matching_destination(self): files = [self._make("20260622050346_0000001F.MP4", 100)] dst = os.path.join(self.base, "20260622", "front", files[0][0]) make_mp4(dst, 100) ok, errors = dashcam.transfer_files(files, self.src, self.base) self.assertEqual((ok, errors), (1, 0)) self.assertFalse(os.path.exists(os.path.join(self.src, files[0][0]))) self.assertEqual(os.path.getsize(dst), 100) def test_overwrites_existing_different_destination(self): files = [self._make("20260622050346_0000001F.MP4", 100)] dst = os.path.join(self.base, "20260622", "front", files[0][0]) make_mp4(dst, 10) ok, errors = dashcam.transfer_files(files, self.src, self.base) self.assertEqual((ok, errors), (1, 0)) self.assertEqual(os.path.getsize(dst), 100) def test_verification_mismatch_counts_as_error_and_keeps_source(self): files = [self._make("20260622050346_0000001F.MP4", 100)] def bad_copy(s, d): with open(d, "wb") as f: f.write(b"0" * 50) with mock.patch.object(dashcam.shutil, "copy", side_effect=bad_copy), mock.patch.object( dashcam.time, "sleep" ): ok, errors = dashcam.transfer_files(files, self.src, self.base) self.assertEqual((ok, errors), (0, 1)) self.assertTrue(os.path.exists(os.path.join(self.src, files[0][0]))) def test_aborts_after_max_consecutive_errors(self): files = [ self._make("2026062205034%d_000000%dF.MP4" % (i, i + 1), 10) for i in range(dashcam.MAX_CONSECUTIVE_ERRORS + 3) ] with mock.patch.object( dashcam.shutil, "copy", side_effect=OSError(errno.EIO, "I/O error") ), mock.patch.object(dashcam.time, "sleep"): with self.assertRaises(dashcam.TransferAborted) as ctx: dashcam.transfer_files(files, self.src, self.base) self.assertEqual(ctx.exception.consecutive, dashcam.MAX_CONSECUTIVE_ERRORS) self.assertEqual(ctx.exception.ok, 0) for fname, _, _ in files: self.assertTrue(os.path.exists(os.path.join(self.src, fname))) class EnsureFreeSpaceTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.base = self.tmp.name def tearDown(self): self.tmp.cleanup() @staticmethod def _statvfs(free_bytes): frsize = 4096 return mock.Mock(f_frsize=frsize, f_bavail=free_bytes // frsize) def test_no_deletion_when_enough_space(self): with mock.patch.object(dashcam.os, "statvfs", return_value=self._statvfs(10_000_000)): dashcam.ensure_free_space(1000, self.base) def test_deletes_oldest_date_dirs_until_enough(self): for name, fname in (("20260301", "a.mp4"), ("20260310", "b.mp4"), ("20260811", "c.mp4")): make_mp4(os.path.join(self.base, name, fname), 5000) with mock.patch.object(dashcam.os, "statvfs", return_value=self._statvfs(4096)): dashcam.ensure_free_space(5000, self.base) self.assertFalse(os.path.exists(os.path.join(self.base, "20260301"))) self.assertTrue(os.path.exists(os.path.join(self.base, "20260310"))) self.assertTrue(os.path.exists(os.path.join(self.base, "20260811"))) def test_exits_when_still_insufficient(self): make_mp4(os.path.join(self.base, "20260301", "a.mp4"), 100) with mock.patch.object(dashcam.os, "statvfs", return_value=self._statvfs(0)): with self.assertRaises(SystemExit): dashcam.ensure_free_space(10_000, self.base) class SendMatrixTest(unittest.TestCase): def setUp(self): patcher = mock.patch.object(dashcam.urllib.request, "urlopen") self.mock_urlopen = patcher.start() self.mock_urlopen.return_value.__enter__.return_value = mock.Mock() self.addCleanup(patcher.stop) self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) token_file = os.path.join(self.tmp.name, "sherpa-token") room_file = os.path.join(self.tmp.name, "eddy-report-room") with open(token_file, "w") as f: f.write("mct_testtoken\n") with open(room_file, "w") as f: f.write("!testroom:whatsed.eddydevink.nl\n") p1 = mock.patch.object(dashcam, "MATRIX_TOKEN_FILE", token_file) p2 = mock.patch.object(dashcam, "MATRIX_ROOM_FILE", room_file) p1.start() p2.start() self.addCleanup(p1.stop) self.addCleanup(p2.stop) def _body(self): req = self.mock_urlopen.call_args[0][0] return json.loads(req.data)["body"] def _request(self): return self.mock_urlopen.call_args[0][0] def test_success_message(self): dashcam.send_matrix_notification(5, 0, 381_000_000) self.assertIn("5 bestanden", self._body()) self.assertIn("geen fouten", self._body()) def test_errors_message(self): dashcam.send_matrix_notification(3, 2, 100_000_000) self.assertIn("2 fout(en)", self._body()) def test_aborted_message(self): dashcam.send_matrix_notification( 1, 4, 100_000_000, aborted=True, reason="SD-kaart onleesbaar" ) self.assertIn("GESTOPT", self._body()) self.assertIn("SD-kaart onleesbaar", self._body()) def test_sends_room_message_with_bearer_token(self): dashcam.send_matrix_notification(1, 0, 10_000) req = self._request() self.assertEqual(req.get_method(), "PUT") self.assertEqual(req.get_header("Authorization"), "Bearer mct_testtoken") self.assertIn("/rooms/!testroom:whatsed.eddydevink.nl/send/m.room.message/", req.full_url) self.assertEqual(json.loads(req.data)["msgtype"], "m.text") def test_missing_secrets_is_swallowed(self): with mock.patch.object(dashcam, "MATRIX_TOKEN_FILE", "/nonexistent/token"): dashcam.send_matrix_notification(0, 0, 0) self.mock_urlopen.assert_not_called() def test_api_failure_is_swallowed(self): self.mock_urlopen.side_effect = Exception("boom") dashcam.send_matrix_notification(0, 0, 0) if __name__ == "__main__": unittest.main()