0461fbc074
Root was bad practice (oude staat); aiaiaiaiagent (uid 1001) kon geen chmod/utime op de exfat-mount (uid=1000) -> elke copy2-kopie faalde met EPERM (10-15, 0/68). Correcte laag is de identity-laag, niet de kopieer- primitive: User=eddy (mount-eigenaar). copy2 behouden (timestamps op doel). Matrix-secrets sherpa-token/eddy-report-room -> mode 640 group eddy + traverse op /home/aiaiaiaiagent zodat de notificatie werkt. Revert eerdere copyfile-workaround (dd46ad6/69d08ba): die maskeerde de mismatch in plaats van de principal te aligneren.
282 lines
11 KiB
Python
282 lines
11 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 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()
|