(self, lockstore)
| 87 | assert len(list(lock.store.list("locks"))) == 0 # stale lock was removed from store |
| 88 | |
| 89 | def test_migrate_lock(self, lockstore): |
| 90 | old_id, new_id = ID1, ID2 |
| 91 | assert old_id[1] != new_id[1] # different PIDs (like when doing daemonize()) |
| 92 | lock = Lock(lockstore, id=old_id).acquire() |
| 93 | old_locks = lock._find_locks(only_mine=True) |
| 94 | assert lock.id == old_id # lock is for old id / PID |
| 95 | lock.migrate_lock(old_id, new_id) # fix the lock |
| 96 | assert lock.id == new_id # lock corresponds to the new id / PID |
| 97 | new_locks = lock._find_locks(only_mine=True) |
| 98 | assert old_locks != new_locks |
| 99 | assert len(old_locks) == len(new_locks) == 1 |
| 100 | assert old_locks[0]["hostid"] == old_id[0] |
| 101 | assert new_locks[0]["hostid"] == new_id[0] |
nothing calls this directly
no test coverage detected