feat: harden dashcam transfer tegen flakke SD-lezer + stale-mount cleanup
- 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
This commit is contained in:
Eddy de Vink
parent
8ef7078ba4
commit
8cc04ea409
9 changed files with 779 additions and 76 deletions
251
tests/test_dashcam_transfer.py
Normal file
251
tests/test_dashcam_transfer.py
Normal file
|
|
@ -0,0 +1,251 @@
|
|||
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()
|
||||
Loading…
Reference in a new issue