Test locking on load#

!lamin login testuser2
import pytest
import lamindb_setup as ln_setup
from lamindb_setup.dev.cloud_sqlite_locker import Locker, InstanceLockedException
from lamindb_setup.dev.upath import UPath
root = UPath("s3://lamindb-ci/test-load-lock", cache_regions=True)
exclusion = root / ".lamindb/_exclusion/"
if exclusion.exists():
    exclusion.rmdir()
assert ln_setup.settings.user.handle == "testuser2"
ln_setup.init(storage="s3://lamindb-ci/test-load-lock", name="test-load-lock")
instance_id = ln_setup.settings.instance.id
ln_setup.close()
# lock with some random id
locker = Locker("randuseridtt", storage_root=root, instance_id=instance_id)
locker.lock()
with pytest.raises(InstanceLockedException):
    ln_setup.load("test-load-lock")
locker.unlock()
ln_setup.load("test-load-lock")
ln_setup.close()
# lock through testuser1, who has user id "DzTjkKse"
locker = Locker("DzTjkKse", storage_root=root, instance_id=instance_id)
locker.lock()
with pytest.raises(InstanceLockedException) as error:
    ln_setup.load("test-load-lock")

assert (
    "InstanceLockedException: Cannot load the instance, it is locked by 'testuser1'"
    " (uid: 'DzTjkKse', name: 'Test User1')."
    in error.exconly()
)
locker.unlock()
ln_setup.load("test-load-lock")
# test ignore_prev_locker=True in unlock_cloud_sqlite_upon_exception
# i.e. test that the locker doesn't unlock if the locker hasn't changed during the function execution
with pytest.raises(RuntimeError):
    ln_setup.load("test-load-lock")

assert ln_setup.settings.instance._cloud_sqlite_locker._has_lock
ln_setup.close()
# purely technical varibale to test failed load after locking
ln_setup._load_instance._TEST_FAILED_LOAD = True

with pytest.raises(RuntimeError):
    ln_setup.load("test-load-lock")

assert ln_setup.dev.cloud_sqlite_locker._locker._has_lock is None

ln_setup._load_instance._TEST_FAILED_LOAD = False
ln_setup.load("test-load-lock", _test=True)
ln_setup.delete("test-load-lock", force=True)