102 lines
3.6 KiB
Python
102 lines
3.6 KiB
Python
import threading
|
|
import os
|
|
|
|
from src.available_issue_snapshot_store import AvailableIssueSnapshotStore
|
|
|
|
|
|
def test_independent_workers_allow_only_one_catalog_refresh(tmp_path):
|
|
path = tmp_path / "available.sqlite3"
|
|
first = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
|
|
second = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
|
|
barrier = threading.Barrier(2)
|
|
results = []
|
|
|
|
def acquire(store):
|
|
barrier.wait()
|
|
results.append(store.try_acquire_refresh(lease_seconds=5))
|
|
|
|
threads = [threading.Thread(target=acquire, args=(store,)) for store in (first, second)]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join()
|
|
|
|
assert sum(owner is not None for owner in results) == 1
|
|
|
|
|
|
def test_published_catalog_and_metadata_are_visible_to_another_worker(tmp_path):
|
|
path = tmp_path / "available.sqlite3"
|
|
writer = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
|
|
reader = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
|
|
owner = writer.try_acquire_refresh(lease_seconds=5)
|
|
|
|
writer.publish(owner, items=[{"repository": "stackchain/api", "number": 7}])
|
|
|
|
state = reader.load()
|
|
assert state.items == [{"repository": "stackchain/api", "number": 7}]
|
|
assert state.created_at == 100.0
|
|
assert state.retry_at is None
|
|
assert state.refreshing is False
|
|
|
|
|
|
def test_failed_refresh_backoff_is_shared_and_lease_is_released(tmp_path):
|
|
path = tmp_path / "available.sqlite3"
|
|
writer = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
|
|
reader = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
|
|
owner = writer.try_acquire_refresh(lease_seconds=5)
|
|
|
|
writer.fail_refresh(owner, retry_at=105.0)
|
|
|
|
state = reader.load()
|
|
assert state.retry_at == 105.0
|
|
assert state.refreshing is False
|
|
assert reader.try_acquire_refresh(lease_seconds=5) is not None
|
|
|
|
|
|
def test_confirmed_claim_is_removed_for_every_worker(tmp_path):
|
|
path = tmp_path / "available.sqlite3"
|
|
writer = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
|
|
reader = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
|
|
owner = writer.try_acquire_refresh(lease_seconds=5)
|
|
writer.publish(owner, items=[
|
|
{"repository": "stackchain/api", "number": 7},
|
|
{"repository": "stackchain/web", "number": 8},
|
|
])
|
|
|
|
writer.remove_claimed("stackchain/api", 7)
|
|
|
|
assert reader.load().items == [{"repository": "stackchain/web", "number": 8}]
|
|
|
|
|
|
def test_invalidation_clears_catalog_and_claim_filters(tmp_path):
|
|
store = AvailableIssueSnapshotStore(tmp_path / "available.sqlite3", clock=lambda: 100.0)
|
|
owner = store.try_acquire_refresh(lease_seconds=5)
|
|
store.publish(owner, items=[{"repository": "stackchain/api", "number": 7}])
|
|
store.remove_claimed("stackchain/api", 7)
|
|
|
|
store.invalidate()
|
|
owner = store.try_acquire_refresh(lease_seconds=5)
|
|
store.publish(owner, items=[{"repository": "stackchain/api", "number": 7}])
|
|
|
|
assert store.load().items == [{"repository": "stackchain/api", "number": 7}]
|
|
|
|
|
|
def test_catalog_store_is_private(tmp_path):
|
|
path = tmp_path / "state" / "available.sqlite3"
|
|
|
|
AvailableIssueSnapshotStore(path)
|
|
|
|
assert os.stat(path.parent).st_mode & 0o777 == 0o700
|
|
assert os.stat(path).st_mode & 0o777 == 0o600
|
|
|
|
|
|
def test_cancelled_refresh_releases_its_lease_for_immediate_takeover(tmp_path):
|
|
path = tmp_path / "available.sqlite3"
|
|
first = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
|
|
second = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
|
|
owner = first.try_acquire_refresh(lease_seconds=30)
|
|
|
|
first.release_refresh(owner)
|
|
|
|
assert second.try_acquire_refresh(lease_seconds=30) is not None
|