37 lines
1.3 KiB
Python
37 lines
1.3 KiB
Python
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from archive_clients.probes import probe_root, probe_writable_directory
|
|
from archive_clients.protocol import ProtocolError, decode, encode, new_envelope
|
|
|
|
|
|
class ProtocolAndProbeTests(unittest.TestCase):
|
|
def test_protocol_round_trip_and_duplicate_key_rejection(self):
|
|
envelope = new_envelope()
|
|
envelope.heartbeat.sequence = 1
|
|
self.assertEqual(decode(encode(envelope)).heartbeat.sequence, 1)
|
|
raw = json.loads(encode(envelope))
|
|
duplicate = json.dumps(raw).replace("{", '{"messageId":"duplicate",', 1)
|
|
with self.assertRaises(ProtocolError):
|
|
decode(duplicate)
|
|
raw["messageId"] = "not-a-uuid"
|
|
with self.assertRaisesRegex(ProtocolError, "canonical UUID"):
|
|
decode(json.dumps(raw))
|
|
|
|
def test_filesystem_probe_cleans_up(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
result = probe_root(root)
|
|
self.assertTrue(result.readable)
|
|
self.assertTrue(result.writable)
|
|
self.assertIsInstance(result.reflink, bool)
|
|
self.assertEqual(list(root.iterdir()), [])
|
|
probe_writable_directory(root)
|
|
self.assertEqual(list(root.iterdir()), [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|