8cc04ea409
- dashcam-transfer.py: log echte OSError (shutil.copy2 i.p.v. cp), pre-flight leesbaarheidscheck voor ruimte-wissen, retry (3x) + early-abort (5x), guards op bestandssysteem-calls, WhatsApp GESTOPT bij defecte kaart - dashcam-transfer-mount.sh: skip als service al draait, read-only fsck.exfat health check, %k device-naam doorgeven - dashcam-transfer-cleanup.sh: nieuw - ruimt stale mount op bij udev REMOVE - udev/99-dashcam-trigger.rules: ADD + REMOVE regels met %k - tests: 20 python + 7 mount + 12 cleanup checks - README: architectuur, robuustheid, deploy-instructies
251 lines
9.6 KiB
Python
251 lines
9.6 KiB
Python
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_copy2 = dashcam.shutil.copy2
|
|
calls = {"n": 0}
|
|
|
|
def flaky(s, d):
|
|
calls["n"] += 1
|
|
if calls["n"] < 3:
|
|
raise OSError(errno.EIO, "I/O error")
|
|
real_copy2(s, d)
|
|
|
|
with mock.patch.object(dashcam.shutil, "copy2", 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, "copy2", 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_copy2(s, d):
|
|
with open(d, "wb") as f:
|
|
f.write(b"0" * 50)
|
|
|
|
with mock.patch.object(dashcam.shutil, "copy2", side_effect=bad_copy2), 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, "copy2", 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 SendWhatsappTest(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)
|
|
|
|
def _body(self):
|
|
req = self.mock_urlopen.call_args[0][0]
|
|
return json.loads(req.data)["content"]
|
|
|
|
def test_success_message(self):
|
|
dashcam.send_whatsapp_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_whatsapp_notification(3, 2, 100_000_000)
|
|
self.assertIn("2 fout(en)", self._body())
|
|
|
|
def test_aborted_message(self):
|
|
dashcam.send_whatsapp_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_api_failure_is_swallowed(self):
|
|
self.mock_urlopen.side_effect = Exception("boom")
|
|
dashcam.send_whatsapp_notification(0, 0, 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|