WIP: test/mig.shared.localfile

tests/support.py

@@ -9,6 +9,7 @@
 from unittest import TestCase, main as testmain
 
 TEST_BASE = os.path.dirname(__file__)
+TEST_DATA_DIR = os.path.join(TEST_BASE, "data")
 TEST_OUTPUT_DIR = os.path.join(TEST_BASE, "output")
 MIG_BASE = os.path.realpath(os.path.join(TEST_BASE, ".."))
 PY2 = sys.version_info[0] == 2
@@ -157,6 +158,17 @@ def cleanpath(relative_path, test_case):
     tmp_path = os.path.join(TEST_OUTPUT_DIR, relative_path)
     test_case._cleanup_paths.add(tmp_path)
 
+
+def fixturepath(relative_path):
+    tmp_path = os.path.join(TEST_DATA_DIR, relative_path)
+    assert (not stat.ISDIR(os.stat(tmp_path)))
+    return tmp_path
+
+
+def projectrelative(tmp_path):
+    return os.path.relpath(tmp_path, start=MIG_BASE)
+
+
 def temppath(relative_path, test_case, skip_clean=False):
     assert(isinstance(test_case, MigTestCase))
     tmp_path = os.path.join(TEST_OUTPUT_DIR, relative_path)

tests/test_mig_shared_localfile.py

@@ -0,0 +1,60 @@
+# -*- coding: utf-8 -*-
+
+import binascii
+from contextlib import contextmanager
+import fcntl
+import os
+import sys
+import zipfile
+
+sys.path.append(os.path.realpath(os.path.join(os.path.dirname(__file__), ".")))
+
+from support import MigTestCase, fixturepath, projectrelative, temppath, testmain
+from mig.shared.serverfile import LOCK_EX
+from mig.shared.localfile import LocalFile
+
+DUMMY_FILE = 'some_file'
+
+
+@contextmanager
+def managed_localfile(f):
+    assert isinstance(f, LocalFile)
+    try:
+        yield f
+    finally:
+        if f.get_lock_mode() != fcntl.LOCK_UN:
+            pass
+        if not f.closed:
+            f.close()
+
+
+class MigSharedLocalfile(MigTestCase):
+    def assertPathLocked(self, file_path):
+        with open(file_path) as conflicting_f:
+            reraise = False
+            try:
+                self.assertRaises(OSError, lambda: fcntl.flock(
+                    conflicting_f, fcntl.LOCK_NB | LOCK_EX))
+            except AssertionError:
+                # if it did raise, clean up the lock we have but shouldn't
+                fcntl.flock(conflicting_f, fcntl.LOCK_NB | fcntl.LOCK_UN)
+                # delay throwing a user-friendly error to aovid nested raise
+                reraise = True
+            if reraise:
+                # now raise something hospitable
+                raise AssertionError(
+                    "expected locked file: %s" % projectrelative(file_path))
+
+    def test_localfile_locking(self):
+        some_file = temppath(DUMMY_FILE, self)
+
+        with managed_localfile(LocalFile(some_file, 'w')) as f:
+            f.lock(LOCK_EX)
+
+            self.assertEqual(f.get_lock_mode(), LOCK_EX)
+
+            self.assertPathLocked(some_file)
+
+
+if __name__ == '__main__':
+    testmain()