diff --git a/README.md b/README.md index f831505..34da857 100644 --- a/README.md +++ b/README.md @@ -1,16 +1,114 @@ # pybinlock -`binlock` is a python module for programmatically reading and writing Avid Media Composer bin locks (`.lck` files), which are primarily used in multiuser Avid environments. +`binlock` is a python package for programmatically reading and writing Avid Media Composer bin locks (`.lck` files). -The `binlock.BinLock` class encapsulates the name used in a bin lock and provides functionality for reading and writing a bin lock `.lck` file. It is essentially a python +Lock files are primarily used in multi-user Avid environments to indicate that a particular machine on the network (and of course, the user behind it!) has temporary ownership over an Avid bin (`.avb` file) to +potentially write changes to the bin. While one machine holds the lock, others are still able to open the bin, albeit in read-only mode, until the lock is released. In this way, two operators cannot +inadvertently make changes to a bin that would step over each other. + +The ability to lock, unlock, and otherwise parse a lock file for a bin would be similarly useful for pipeline and automation purposes, and that's where `binlock` comes in. + +--- + +The `binlock.BinLock` class encapsulates the information used in a bin lock and provides functionality for reading and writing a bin lock `.lck` file. It is essentially a python [`dataclass`](https://docs.python.org/3/library/dataclasses.html) with additional validation and convenience methods. With `binlock.Binlock`, lock files can be programmatically -created, [read from](#reading), [written to](#writing), or ["held" with a context manager](#as-a-context-manager). +created, [read from](#reading), [written to](#writing), or ["held" with a context manager](#as-a-context-manager) to indicate ownership and render bins read-only to other users in +the shared project. >[!WARNING] >While the `.lck` lock file format is a very simple one, it is officially undocumented. Use this library at your own risk -- I assume no responsibility for any damage to your >project, loss of data, or underwhelming box office performance. -## Reading +Working with bin locks can be done either by referring directly to the `.lck` files themselves, or to the Avid bins `.avb` they affect. Ultimately this depends on how you like to +work and what exacly you're doing, but often the best way is to refer to the Avid bins, which we'll cover first. + +## Bin-Referred Operations + +Many times, it's best to think of bin locking in terms of the bins you're affecting. The `binlock.BinLock` class provides methods to read, lock, unlock, or temporarily "hold the lock" for an Avid bin. + +### Reading + +Reading the lock info for an Avid bin is possible with the `BinLock.from_bin(avb_path)` class method, passing an existing `.avb` file path as a string. + +```python +from binlock import BinLock +lock = BinLock.from_bin("01_EDITS/Reel 1.avb") # Returns a `BinLock` object +if lock: + print(lock.name) +``` +Here, `BinLock.from_bin(avb_path)` returns a`BinLock` object representing the `.lck` lockfile info for a locked bin, or `None` if the bin is not locked. We then print the name of the lock, for example: +``` +zMichael +``` + +### Locking A Bin + +To lock an Avid bin (by writing a `.lck` lock file), we'll create a new `BinLock` object and use it to lock a bin. + +```python +from binlock import BinLock +new_lock = BinLock("zMichael") +try: + new_lock.lock_bin("01_EDITS/Reel 2.avb") +except BinLockExistsError as e: + print(e) +else: + print("Bin is now locked!") +``` + +Once executed, `Reel 2.avb` will appear locked by `zMichael` to all users on the project. Custom names can be used for other purposes, such as `Locked Picture` or `Delivered` to indicate +no further changes are to be made to the bins. If the bin is already locked, a `BinLockExistsError` will be raised instead. + +If a name string is not provided, the machine's host name will be used by default, just as Avid would do. Therefore, a good one-liner to lock a bin for the current machine might be: + +```python +from binlock import BinLock +BinLock(avb_path).lock_bin("01_EDITS/Reel 3.avb") +``` + +### Unlocking A Bin + +A bin can be unlocked, but only by a `BinLock` of the same name. + +```python +from binlock import BinLock +try: + BinLock("zMichael").unlock_bin("01_EDITS/Reel 2.avb") +except BinLockOwnershipError as e: + print(e) +else: + print("Bin has been unlocked.") +``` + +Because a bin should **only** be unlocked by the process that locked it in the first place, if the lock names do not match, a `BinLockOwnershipError` will be raised as a safety precaution, and the bin will not be unlocked. + +>[!CAUTION] +>Unlocking bins can be extremely risky and result in data loss if done carelessly. A bin should only be unlocked if you were the one who locked it, and you are certain that any changes you have made have been properly committed. +>Instead of manually locking and unlocking bins, you should instead [hold a lock](#holding-a-lock-on-a-bin) whenever possible, as described below. + +### Holding A Lock On A Bin + +It is often much safer to utilize a context manager to lock a bin only while you perform actions on it, then release the lock immediately after. This is possible with the `BinLock.hold_bin(avb_path)` context manager: + +```python +from binlock import BinLock + +path_bin = "01_EDITS/Reel 3.avb" + +with BinLock("zAdmin").hold_bin(path_bin) as lock: + print(f"Bin locked as {lock.name}. Now doing stuff to the bin...") + do_stuff_to_bin(path_bin) +print("Lock released.") +``` + +Here, a bin will be safely locked, then unlocked on completion. + +## Lock-Referred Operations + +Operations similar to [bin-referred operations](#bin-referred-operations) can be done by referencing the `.lck` lock files directly. This may be useful for +more specialized workflows, but should be used with caution as the are less safe. + +### Reading A Lock File Reading from an existing `.lck` file is possible using the `BinLock.from_path(lck_path)` class method, passing an existing `.lck` file path as a string. @@ -24,9 +122,9 @@ This would output the name on the lock, for example: zMichael ``` -## Writing +### Writing A Lock File -Directly writing a `.lck` file works similarly with the `BinLock.to_path(lck_path)` class method, passing a path to the `.lck` file you would like to create. +Directly writing a `.lck` lock file works similarly to the `BinLock.to_path(lck_path)` class method, passing a path to the `.lck` file you would like to create. ```python from binlock import BinLock @@ -38,50 +136,43 @@ see the result. >[!CAUTION] >Directly writing a `.lck` file in this way will allow you to overwrite any existing `.lck` file, which is almost certainly a bad idea. Take care to first ->check for an existing `.lck` file, or even better, use [the context manager approach](#as-a-context-manager) instead. +>check for an existing `.lck` file, or even better, use the context manager approach by [holding a lock file](#holding-a-lock-file) instead. -## As A Context Manager +### Holding A Lock File -The strongly recommended way to programmatically lock an Avid bin using `pybinlock` is to use `BinLock.hold(lck_path)` as a context manager. This allows you to "hold" the +The strongly recommended way to programmatically lock an Avid bin using `binlock` is to use `BinLock.hold_lock(lck_path)` as a context manager. This allows you to "hold" the lock on a bin while you do stuff to it. It includes safety checks to ensure a lock does not already exist (i.e. the bin is locked by someone else), and automatically removes the lock on exit or on fatal error. -This approach should be used whenever possible (in favor of [directly writing](#writing) a `.lck`, which can be more risky). +This approach should be used whenever possible (in favor of [directly writing](#writing-a-lock-file) a `.lck`, which can be more risky). ```python import time from binlock import BinLock -with BinLock("zMichael").hold("01_EDITS/Reel 1.lck"): +with BinLock("zMichael").hold_lock("01_EDITS/Reel 1.lck"): time.sleep(60) # Look busy ``` Here, the context manager will throw a `BinLockExistsError` if the lock already exists, and will not continue. Otherwise, it will lock the bin with `zMichael` for 60 seconds, then release the lock. -### Being A "Good Citizen" +## Being A "Good Citizen" -I don't mean to toot my own little horn here, but I have also released [`pybinlog`](https://github.com/mjiggidy/pybinlog), which is a python module for writing bin log files. It is highly +I don't mean to toot my own little horn here, but I have also released [`pybinhistory`](https://github.com/mjiggidy/pybinhistory), which is a python package for writing bin log files. It is highly recommended that any time you lock a bin, you also add an entry in the Avid bin log, just as Avid would do. Here they are together: ```python from binlock import BinLock, BinLockExistsError -from binlog import BinLog +from binhistory import BinLog path_bin = "01_EDITS/Reel 1.avb" -path_lock = BinLock.lock_path_from_bin_path(path_bin) path_log = BinLog.log_path_from_bin_path(path_bin) computer_name = "zMichael" user_name = "MJ 2024.12.2" try: - with BinLock(computer_name).hold(path_lock): + with BinLock(computer_name).hold_bin(path_bin): BinLog.touch(path_log, computer=computer_name, user=user_name) do_cool_stuff_to_bin(path_bin) - except BinLockExistsError: - try: - print("Bin is already locked by", BinLock.from_path(path_lock).name) - except Exception: - print("Bin is already locked") + print(e) ``` ->[!TIP] ->Oh! Did you see that? `BinLock.lock_path_from_bin_path(path_bin)` is a static method that returns the expected path to a `.lck` lock file for a given `.avb` bin. diff --git a/binlock/__init__.py b/binlock/__init__.py index 2324b45..c4fea40 100644 --- a/binlock/__init__.py +++ b/binlock/__init__.py @@ -4,17 +4,5 @@ https://github.com/mjiggidy/pybinlock """ -MAX_NAME_LENGTH:int = 24 -"""Maximum allowed lock name""" -# TODO: Observed max 21 in real-life locks... need to investigate - -DEFAULT_FILE_EXTENSION = ".lck" -""" -The default file extension for a lock file - -Not used directly by `BinLock`, but perhaps useful to reference -in your own scripts so that we're all on the same page -""" - -from .exceptions import BinLockFileDecodeError, BinLockLengthError, BinLockExistsError +from .exceptions import BinLockFileDecodeError, BinLockNameError, BinLockExistsError, BinLockNotFoundError, BinLockOwnershipError from .binlock import BinLock \ No newline at end of file diff --git a/binlock/binlock.py b/binlock/binlock.py index e37f7b4..33d1dd9 100644 --- a/binlock/binlock.py +++ b/binlock/binlock.py @@ -1,24 +1,40 @@ """Utilites for working with bin locks (.lck files)""" -import dataclasses, pathlib, typing, contextlib -from . import BinLockLengthError, BinLockFileDecodeError, BinLockExistsError -from . import MAX_NAME_LENGTH, DEFAULT_FILE_EXTENSION +import dataclasses, pathlib, typing, contextlib, socket +from . import BinLockNameError, BinLockFileDecodeError, BinLockExistsError, BinLockNotFoundError, BinLockOwnershipError + +DEFAULT_FILE_EXTENSION:str = ".lck" +"""The default file extension for a lock file""" + +DEFAULT_LOCK_NAME:str = socket.gethostname() +"""Default name to use on the lock, if none is provided""" + +MAX_NAME_LENGTH:int = 24 +"""Maximum allowed lock name""" +# TODO: Observed max 21 in real-life locks... need to investigate + +TOTAL_FILE_SIZE:int = 255 +"""Total size of a .lck file""" @dataclasses.dataclass(frozen=True) class BinLock: """Represents a bin lock file (.lck)""" - name:str + name:str = DEFAULT_LOCK_NAME """Name of the Avid the lock belongs to""" def __post_init__(self): """Validate lock name""" - if not self.name.strip(): - raise BinLockLengthError("Username for the lock must not be empty") + if not isinstance(self.name, str): + raise BinLockNameError(f"Lock name must be a string (got {type(self.name)})") + elif not self.name.strip(): + raise BinLockNameError("Username for the lock must not be empty") + elif not self.name.isprintable(): + raise BinLockNameError("Username for the lock must not contain non-printable characters") elif len(self.name) > MAX_NAME_LENGTH: - raise BinLockLengthError(f"Username for the lock must not exceed {MAX_NAME_LENGTH} characters (attempted {len(self.name)} characters)") - + raise BinLockNameError(f"Username for the lock must not exceed {MAX_NAME_LENGTH} characters (attempted {len(self.name)} characters)") + @staticmethod def _read_utf16le(buffer:typing.BinaryIO) -> str: """Decode as UTF-16le until we hit NULL""" @@ -30,6 +46,62 @@ def _read_utf16le(buffer:typing.BinaryIO) -> str: break b_name += b_chars return b_name.decode("utf-16le") + + def lock_bin(self, bin_path:str, missing_bin_ok:bool=True): + """Lock a given bin (.avb) with this lock""" + + if not missing_bin_ok and not pathlib.Path(bin_path).is_file(): + raise FileNotFoundError(f"Bin does not exist at {bin_path}") + + lock_path = self.get_lock_path_from_bin_path(bin_path) + + # Prevent locking an already-locked bin + if pathlib.Path(lock_path).is_file(): + try: + lock = self.from_path(lock_path) + raise BinLockExistsError(f"Bin is already locked by {lock.name}") + except Exception as e: # Flew too close to the sun + raise BinLockExistsError("Bin is already locked") + + self.to_path(lock_path) + + def unlock_bin(self, bin_path:str, missing_bin_ok:bool=True): + """ + Unlock a given bin (.avb) + + For safety, the name on the bin lock MUST match the name on this `BinLock` instance + """ + + if not missing_bin_ok and not pathlib.Path(bin_path).is_file(): + raise FileNotFoundError(f"Bin does not exist at {bin_path}") + + bin_lock = self.from_bin(bin_path) + + if not bin_lock: + raise BinLockNotFoundError("This bin is not currently locked") + + if bin_lock != self: + raise BinLockOwnershipError("Bin locks do not match") + + pathlib.Path(self.get_lock_path_from_bin_path(bin_path)).unlink(missing_ok=True) + + @classmethod + def from_bin(cls, bin_path:str, missing_bin_okay:bool=True) -> "BinLock": + """ + Get the existing lock for a given bin (.avb) path + + Returns `None` if the bin is not locked + """ + + if not missing_bin_okay and not pathlib.Path(bin_path).is_file(): + raise FileNotFoundError(f"Bin does not exist at {bin_path}") + + lock_path = cls.get_lock_path_from_bin_path(bin_path) + + if not pathlib.Path(lock_path).is_file(): + return None + + return cls.from_path(lock_path) @classmethod def from_path(cls, lock_path:str) -> "BinLock": @@ -46,15 +118,24 @@ def to_path(self, lock_path:str): """Write to .lck lockfile""" with open(lock_path, "wb") as lock_file: - lock_file.write(self.name[:MAX_NAME_LENGTH].ljust(255, '\x00').encode("utf-16le")) + lock_file.write(self.name[:MAX_NAME_LENGTH].ljust(TOTAL_FILE_SIZE, '\x00').encode("utf-16le")) - def hold(self, lock_path:str) -> "_BinLockContextManager": - """Hold the lock""" + def hold_lock(self, lock_path:str) -> "_BinLockContextManager": + """Context manager to hold a lock at a given path""" + + return _BinLockContextManager(self, lock_path) + + def hold_bin(self, bin_path:str, missing_bin_ok:bool=True) -> "_BinLockContextManager": + """Context manager to hold a lock for a given bin (.avb) path""" + + if not missing_bin_ok and not pathlib.Path(bin_path).is_file(): + raise FileNotFoundError(f"Bin does not exist at {bin_path}") + lock_path = self.get_lock_path_from_bin_path(bin_path) return _BinLockContextManager(self, lock_path) @staticmethod - def lock_path_from_bin_path(bin_path:str) -> str: + def get_lock_path_from_bin_path(bin_path:str) -> str: """Determine the lock path from a given bin path""" return str(pathlib.Path(bin_path).with_suffix(DEFAULT_FILE_EXTENSION)) @@ -68,7 +149,7 @@ def __init__(self, lock:BinLock, lock_path:str): self._lock_info = lock self._lock_path = lock_path - def __enter__(self) -> "_BinLockContextManager": + def __enter__(self) -> BinLock: """Write the lock on enter""" if pathlib.Path(self._lock_path).is_file(): @@ -80,7 +161,7 @@ def __enter__(self) -> "_BinLockContextManager": pathlib.Path(self._lock_path).unlink(missing_ok=True) raise e - return self + return self._lock_info def __exit__(self, exc_type, exc_value, traceback) -> bool: """Remove the lock on exit and call 'er a day""" diff --git a/binlock/exceptions.py b/binlock/exceptions.py index e61a8e1..c227dfa 100644 --- a/binlock/exceptions.py +++ b/binlock/exceptions.py @@ -1,8 +1,14 @@ -class BinLockLengthError(ValueError): - """User name is not a valid length (between 1 and MAX_NAME_LENGTH chars)""" +class BinLockNameError(ValueError): + """The given lock name is not valid for use""" class BinLockFileDecodeError(ValueError): """File could not be decoded; likely not a valid lock file""" class BinLockExistsError(FileExistsError): - """A lock file already exists for this bin, perhaps from another user""" \ No newline at end of file + """A lock file already exists for this bin, perhaps from another machine""" + +class BinLockNotFoundError(FileNotFoundError): + """An expected bin lock is not found""" + +class BinLockOwnershipError(PermissionError): + """The existing bin lock belongs to another entity (lock names do not match)""" \ No newline at end of file diff --git a/examples/hold_bin.py b/examples/hold_bin.py new file mode 100644 index 0000000..9f16d18 --- /dev/null +++ b/examples/hold_bin.py @@ -0,0 +1,27 @@ +""" +Given a path to a bin (.avb), "hold" a lock on it while you do stuff. +I don't know -- I think it's neat. +""" + +import sys, pathlib +from binlock import BinLock + +if __name__ == "__main__": + + if not len(sys.argv) > 1: + print(f"Usage: {pathlib.Path(__file__).name} path_to_bin.avb", file=sys.stderr) + sys.exit(1) + + path_bin = pathlib.Path(sys.argv[1]) + if not path_bin.suffix.lower() == ".avb": + print(f"Expecting a `.avb` file here, got {path_bin} instead", file=sys.stderr) + sys.exit(2) + + with BinLock().hold_bin(path_bin) as lock: + input(f"Holding lock on {path_bin} as {lock.name}... (press any key)") + + if BinLock().from_bin(path_bin): + print("Somehow unable to release bin") + sys.exit(3) + + print("Lock released") \ No newline at end of file diff --git a/examples/hold_lock.py b/examples/hold_lock.py index 7ceac43..9331499 100644 --- a/examples/hold_lock.py +++ b/examples/hold_lock.py @@ -1,38 +1,22 @@ -""" -Given a path to a bin (.avb), "hold" a lock on it while you do stuff. -I don't know -- I think it's neat. -""" - -import sys, pathlib, enum -from binlock import BinLock - -class OverEngineeredExitCodes(enum.IntEnum): - - EXIT_OK = 0 - BAD_USAGE = 1 - NOT_AVB = 2 - NO_AVB = 3 - -if __name__ == "__main__": - - if not len(sys.argv) > 1: - print(f"Usage: {pathlib.Path(__file__).name} path_to_bin.avb", file=sys.stderr) - sys.exit(OverEngineeredExitCodes.BAD_USAGE) - - path_bin = pathlib.Path(sys.argv[1]) - if not path_bin.suffix.lower() == ".avb": - print(f"Expecting a `.avb` file here, got {path_bin} instead", file=sys.stderr) - sys.exit(OverEngineeredExitCodes.NOT_AVB) - - elif not path_bin.is_file(): - print(f"Bin does not exist: {path_bin}", file=sys.stderr) - sys.exit(OverEngineeredExitCodes.NO_AVB) - - # Determine lock path from a given bin path - path_lock = BinLock.lock_path_from_bin_path(path_bin) - - # Lock bin while we do stuff, then release it - with BinLock("anon").hold(path_lock): - input("Deleting important things...") - - print("Done") \ No newline at end of file +""" +Create and hold a lock, independent of a bin +""" + +import sys, pathlib +from binlock import BinLock + +if __name__ == "__main__": + + if not len(sys.argv) > 1: + print(f"Usage: {pathlib.Path(__file__).name} path_to_lock.lck", file=sys.stderr) + sys.exit(1) + + path_lock = pathlib.Path(sys.argv[1]) + if not path_lock.suffix.lower() == ".lck": + print(f"Expecting a `.lck` file here, got {path_lock} instead", file=sys.stderr) + sys.exit(2) + + with BinLock("zMichael").hold_lock(path_lock) as lock: + input(f"Holding lock at {path_lock} as {lock.name}... (press any key)") + + print("Lock released") \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index c4dc681..0b5f4aa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,4 +4,4 @@ build-backend = "setuptools.build_meta" [project] name = "pybinlock" -version = "0.1.0" \ No newline at end of file +version = "0.2.0" \ No newline at end of file diff --git a/tests/test_binlock.py b/tests/test_binlock.py new file mode 100644 index 0000000..8e41648 --- /dev/null +++ b/tests/test_binlock.py @@ -0,0 +1,132 @@ +import pytest +import pathlib +from binlock import ( + BinLock, + BinLockNameError, + BinLockFileDecodeError, + BinLockExistsError, + BinLockOwnershipError, +) + +from binlock.binlock import MAX_NAME_LENGTH + +# Helper to create a dummy bin file (with a .avb extension) +def create_dummy_bin(tmp_path): + dummy_bin = tmp_path / "dummy.avb" + dummy_bin.write_text("dummy content") + return dummy_bin + +def test_binlock_creation_valid(): + # Creating a BinLock with a valid name should succeed. + lock = BinLock(name="valid_user") + assert lock.name == "valid_user" + +def test_binlock_invalid_empty_name(): + # An empty (or whitespace-only) name should raise an error. + with pytest.raises(BinLockNameError): + BinLock(name=" ") + +def test_binlock_invalid_non_string(): + # A non-string name should raise an error. + with pytest.raises(BinLockNameError): + BinLock(name=123) + +def test_binlock_invalid_too_long_name(): + # A name longer than MAX_NAME_LENGTH should raise an error. + long_name = "a" * (MAX_NAME_LENGTH + 1) + with pytest.raises(BinLockNameError): + BinLock(name=long_name) + +def test_to_from_path_roundtrip(tmp_path): + # Write a lock to file then read it back; the objects should be equal. + lock = BinLock(name="tester") + lock_file = tmp_path / "test.lck" + lock.to_path(str(lock_file)) + loaded_lock = BinLock.from_path(str(lock_file)) + assert loaded_lock == lock + +def test_lock_bin_creates_lock_file(tmp_path): + # Locking a bin should create the corresponding lock file. + dummy_bin = create_dummy_bin(tmp_path) + lock = BinLock(name="user1") + lock_file_path = pathlib.Path(lock.get_lock_path_from_bin_path(str(dummy_bin))) + # Ensure the lock file doesn't exist before locking. + if lock_file_path.exists(): + lock_file_path.unlink() + lock.lock_bin(str(dummy_bin)) + assert lock_file_path.exists() + +def test_lock_bin_already_locked(tmp_path): + # Trying to lock an already locked bin should raise a BinLockExistsError. + dummy_bin = create_dummy_bin(tmp_path) + lock1 = BinLock(name="user1") + lock2 = BinLock(name="user2") + # First lock the bin. + lock1.lock_bin(str(dummy_bin)) + with pytest.raises(BinLockExistsError): + lock2.lock_bin(str(dummy_bin)) + # Cleanup: remove the lock file. + lock_file = pathlib.Path(lock1.get_lock_path_from_bin_path(str(dummy_bin))) + if lock_file.exists(): + lock_file.unlink() + +def test_unlock_bin_removes_lock_file(tmp_path): + # Unlocking a bin should remove its lock file. + dummy_bin = create_dummy_bin(tmp_path) + lock = BinLock(name="user1") + lock.lock_bin(str(dummy_bin)) + lock_file = pathlib.Path(lock.get_lock_path_from_bin_path(str(dummy_bin))) + assert lock_file.exists() + lock.unlock_bin(str(dummy_bin)) + assert not lock_file.exists() + +def test_unlock_bin_wrong_owner(tmp_path): + # Unlocking with a BinLock instance that doesn't match the one that locked it + # should raise a BinLockOwnershipError. + dummy_bin = create_dummy_bin(tmp_path) + lock1 = BinLock(name="user1") + lock2 = BinLock(name="user2") + lock1.lock_bin(str(dummy_bin)) + with pytest.raises(BinLockOwnershipError): + lock2.unlock_bin(str(dummy_bin)) + # Cleanup: + lock_file = pathlib.Path(lock1.get_lock_path_from_bin_path(str(dummy_bin))) + if lock_file.exists(): + lock_file.unlink() + +def test_get_lock_from_bin_returns_none(tmp_path): + # When no lock file exists, get_lock_from_bin should return None. + dummy_bin = create_dummy_bin(tmp_path) + assert BinLock.from_bin(str(dummy_bin)) is None + +def test_hold_lock_context_manager(tmp_path): + # The hold_lock context manager should create the lock file on entry and remove it on exit, + # while returning the BinLock instance. + lock = BinLock(name="user1") + lock_file_path = tmp_path / "test.lck" + if lock_file_path.exists(): + lock_file_path.unlink() + with lock.hold_lock(str(lock_file_path)) as held_lock: + assert held_lock == lock + assert lock_file_path.exists() + assert not lock_file_path.exists() + +def test_hold_bin_context_manager(tmp_path): + # The hold_bin context manager should similarly manage the lock for a given bin. + dummy_bin = create_dummy_bin(tmp_path) + lock = BinLock(name="user1") + lock_file_path = pathlib.Path(lock.get_lock_path_from_bin_path(str(dummy_bin))) + if lock_file_path.exists(): + lock_file_path.unlink() + with lock.hold_bin(str(dummy_bin)) as held_lock: + assert held_lock == lock + assert lock_file_path.exists() + assert not lock_file_path.exists() + +def test_corrupt_lock_file(tmp_path): + # A lock file that cannot be decoded as UTF-16le should raise BinLockFileDecodeError. + lock_file = tmp_path / "corrupt.lck" + # Write an odd number of bytes to provoke a decode error. + lock_file.write_bytes(b'\xff') + with pytest.raises(BinLockFileDecodeError): + BinLock.from_path(str(lock_file))