32 lines
1.1 KiB
Python
32 lines
1.1 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.config import settings
|
|
from app.database import Base
|
|
from app.models import ImportStatus, OfficialRecordImport
|
|
from app.scheduler import import_is_due
|
|
|
|
|
|
def test_import_is_due_respects_minimum_interval() -> None:
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
now = datetime.now(timezone.utc)
|
|
with Session(engine) as db:
|
|
assert import_is_due(db, now=now) is True
|
|
db.add(OfficialRecordImport(
|
|
started_at=now - timedelta(seconds=settings.import_interval_seconds - 1),
|
|
finished_at=now,
|
|
status=ImportStatus.success,
|
|
source_url=settings.official_records_url,
|
|
rows_seen=0,
|
|
rows_created=0,
|
|
rows_updated=0,
|
|
))
|
|
db.commit()
|
|
assert import_is_due(db, now=now) is False
|
|
assert import_is_due(db, now=now + timedelta(seconds=settings.import_interval_seconds)) is True
|