Test SQLite file locking

!lamin login testuser2

import pytest
import lamindb_setup as ln_setup
from lamindb_setup.core.cloud_sqlite_locker import Locker, InstanceLockedException
from lamindb_setup.core.upath import UPath

root = UPath("s3://lamindb-ci/test-load-lock", cache_regions=True)

exclusion = root / ".lamindb/_exclusion/"
# if cache is present, then .exists() for intermediate keys return False
# bug in s3fs
exclusion.fs.invalidate_cache()
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.connect("test-load-lock"), would like to test this, but need another python session

locker.unlock()

# ln_setup.connect("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.connect("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.connect("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.connect("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._connect_instance._TEST_FAILED_LOAD = True

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

# assert ln_setup.core.cloud_sqlite_locker._locker._has_lock is None

# ln_setup._connect_instance._TEST_FAILED_LOAD = False

# ln_setup.connect("test-load-lock", _test=True)

ln_setup.delete("test-load-lock", force=True)