1
0
Fork 0
mirror of https://github.com/khoj-ai/khoj.git synced 2024-12-24 05:08:09 +00:00
khoj/tests/test_db_lock.py

59 lines
1.5 KiB
Python
Raw Normal View History

2024-04-17 07:52:41 +00:00
import time
import pytest
from khoj.database.adapters import ProcessLockAdapters
from khoj.database.models import ProcessLock
from tests.helpers import ProcessLockFactory
@pytest.mark.django_db(transaction=True)
def test_process_lock(default_process_lock):
# Arrange
lock: ProcessLock = default_process_lock
# Assert
assert True == ProcessLockAdapters.is_process_locked_by_name(lock.name)
2024-04-17 07:52:41 +00:00
@pytest.mark.django_db(transaction=True)
def test_expired_process_lock():
# Arrange
lock: ProcessLock = ProcessLockFactory(name="test_expired_lock", max_duration_in_seconds=2)
# Act
time.sleep(3)
# Assert
assert False == ProcessLockAdapters.is_process_locked_by_name(lock.name)
2024-04-17 07:52:41 +00:00
@pytest.mark.django_db(transaction=True)
def test_in_progress_lock(default_process_lock):
# Arrange
lock: ProcessLock = default_process_lock
# Act
ProcessLockAdapters.run_with_lock(lock.name, lambda: time.sleep(2))
# Assert
assert True == ProcessLockAdapters.is_process_locked_by_name(lock.name)
2024-04-17 07:52:41 +00:00
@pytest.mark.django_db(transaction=True)
def test_run_with_completed():
# Arrange
ProcessLockAdapters.run_with_lock("test_run_with", lambda: time.sleep(2))
# Act
time.sleep(4)
# Assert
assert False == ProcessLockAdapters.is_process_locked_by_name("test_run_with")
2024-04-17 07:52:41 +00:00
@pytest.mark.django_db(transaction=True)
def test_nonexistent_lock():
# Assert
assert False == ProcessLockAdapters.is_process_locked_by_name("nonexistent_lock")