51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from bakrest.registry import ChangeEvent, ChangeRegistry
|
|
|
|
|
|
def test_pending_changes_persist_until_marked(tmp_path: Path) -> None:
|
|
db_path = tmp_path / "changes.sqlite"
|
|
first = ChangeRegistry(db_path)
|
|
first.record(ChangeEvent(path=str(tmp_path / "image.jpg"), event_type="modified"))
|
|
|
|
second = ChangeRegistry(db_path)
|
|
|
|
assert second.pending_count() == 1
|
|
|
|
|
|
def test_mark_ignored_removes_from_pending_queue(tmp_path: Path) -> None:
|
|
db_path = tmp_path / "changes.sqlite"
|
|
registry = ChangeRegistry(db_path)
|
|
path = str(tmp_path / "image.jpg")
|
|
registry.record(ChangeEvent(path=path, event_type="modified"))
|
|
|
|
registry.mark_ignored(path)
|
|
|
|
assert registry.pending_count() == 0
|
|
|
|
|
|
def test_snapshot_change_detection(tmp_path: Path) -> None:
|
|
registry = ChangeRegistry(tmp_path / "changes.sqlite")
|
|
path = str(tmp_path / "image.jpg")
|
|
|
|
assert registry.has_snapshot_changed(path, 100, 123)
|
|
|
|
registry.update_snapshot(path, 100, 123)
|
|
|
|
assert not registry.has_snapshot_changed(path, 100, 123)
|
|
assert registry.has_snapshot_changed(path, 101, 123)
|
|
assert registry.has_snapshot_changed(path, 100, 124)
|
|
|
|
|
|
def test_record_stores_snapshot_and_mtime(tmp_path: Path) -> None:
|
|
registry = ChangeRegistry(tmp_path / "changes.sqlite")
|
|
path = str(tmp_path / "image.jpg")
|
|
|
|
registry.record(ChangeEvent(path=path, event_type="modified", size=100, mtime_ns=123))
|
|
|
|
pending = registry.list_pending()
|
|
assert pending == [ChangeEvent(path=path, event_type="modified", size=100, mtime_ns=123)]
|
|
assert not registry.has_snapshot_changed(path, 100, 123)
|