28 lines
1.1 KiB
Python
28 lines
1.1 KiB
Python
import os
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.importer import ImportAlreadyRunning, _official_import_lock
|
|
|
|
|
|
@pytest.mark.skipif(not os.environ.get("DATABASE_URL", "").startswith("postgresql"), reason="requires PostgreSQL")
|
|
def test_postgresql_import_lock_blocks_only_same_source_category() -> None:
|
|
engine = create_engine(os.environ["DATABASE_URL"])
|
|
first = Session(engine)
|
|
second = Session(engine)
|
|
try:
|
|
with _official_import_lock(first, url="https://example.test/records", region="RU", category="records"):
|
|
with pytest.raises(ImportAlreadyRunning):
|
|
with _official_import_lock(second, url="https://example.test/records", region="RU", category="records"):
|
|
pass
|
|
with _official_import_lock(second, url="https://example.test/records", region="RU", category="weekly"):
|
|
pass
|
|
with _official_import_lock(second, url="https://example.test/records", region="RU", category="records"):
|
|
pass
|
|
finally:
|
|
first.close()
|
|
second.close()
|
|
engine.dispose()
|