Nora
38 lines
1.3 KiB
Python
38 lines
1.3 KiB
Python
"""Claim-id dedup at the SFTP pre-flight stage."""
|
|
from datetime import datetime
|
|
from cyclone.store.submission_dedup import (
|
|
DuplicateClaimError,
|
|
record_submission,
|
|
check_duplicate,
|
|
DEFAULT_WINDOW_DAYS,
|
|
)
|
|
|
|
|
|
def test_check_duplicate_returns_none_for_first_submission(tmp_path):
|
|
db = tmp_path / "test.db"
|
|
assert check_duplicate("ORIG-001", db_url=f"sqlite:///{db}") is None
|
|
|
|
|
|
def test_check_duplicate_raises_within_window(tmp_path):
|
|
db = tmp_path / "test.db"
|
|
record_submission("ORIG-002", submitted_at=datetime(2026, 6, 1), db_url=f"sqlite:///{db}")
|
|
try:
|
|
check_duplicate("ORIG-002", db_url=f"sqlite:///{db}",
|
|
now=datetime(2026, 6, 15))
|
|
except DuplicateClaimError as e:
|
|
assert e.claim_id == "ORIG-002"
|
|
assert e.original_submission_at == datetime(2026, 6, 1)
|
|
else:
|
|
raise AssertionError("expected DuplicateClaimError")
|
|
|
|
|
|
def test_check_duplicate_passes_after_window(tmp_path):
|
|
db = tmp_path / "test.db"
|
|
record_submission("ORIG-003", submitted_at=datetime(2026, 1, 1), db_url=f"sqlite:///{db}")
|
|
# 100 days later — past the 30-day default window
|
|
assert check_duplicate("ORIG-003", db_url=f"sqlite:///{db}",
|
|
now=datetime(2026, 4, 11)) is None
|
|
|
|
|
|
def test_default_window_is_30_days():
|
|
assert DEFAULT_WINDOW_DAYS == 30 |