122 lines
5.0 KiB
Python
122 lines
5.0 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from govoplan_core.core.change_sequence import (
|
|
ChangeSequenceEntry,
|
|
ChangeSequenceRetentionFloor,
|
|
decode_sequence_watermark,
|
|
encode_sequence_watermark,
|
|
max_sequence_id,
|
|
min_sequence_id,
|
|
prune_sequence_entries,
|
|
record_change,
|
|
retained_sequence_floor,
|
|
sequence_entries_since,
|
|
sequence_watermark_is_expired,
|
|
)
|
|
from govoplan_core.db.base import Base
|
|
|
|
|
|
class ChangeSequenceTests(unittest.TestCase):
|
|
def test_records_changes_and_returns_entries_after_watermark(self) -> None:
|
|
engine = create_engine("sqlite:///:memory:")
|
|
SessionLocal = sessionmaker(bind=engine)
|
|
try:
|
|
Base.metadata.create_all(bind=engine, tables=[ChangeSequenceEntry.__table__, ChangeSequenceRetentionFloor.__table__])
|
|
with SessionLocal() as session:
|
|
first = record_change(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
module_id="files",
|
|
collection="files.assets",
|
|
resource_type="file",
|
|
resource_id="file-1",
|
|
operation="created",
|
|
)
|
|
second = record_change(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
module_id="files",
|
|
collection="files.assets",
|
|
resource_type="file",
|
|
resource_id="file-1",
|
|
operation="updated",
|
|
)
|
|
session.commit()
|
|
|
|
self.assertEqual(encode_sequence_watermark(first.id), "seq:1")
|
|
self.assertEqual(decode_sequence_watermark("seq:1"), 1)
|
|
self.assertEqual(max_sequence_id(session, tenant_id="tenant-1", module_id="files"), second.id)
|
|
self.assertEqual(min_sequence_id(session, tenant_id="tenant-1", module_id="files"), first.id)
|
|
self.assertFalse(sequence_watermark_is_expired(session, tenant_id="tenant-1", module_id="files", since=first.id - 1))
|
|
self.assertEqual(
|
|
[entry.operation for entry in sequence_entries_since(session, tenant_id="tenant-1", module_id="files", since=first.id)],
|
|
["updated"],
|
|
)
|
|
|
|
session.delete(first)
|
|
session.commit()
|
|
self.assertFalse(sequence_watermark_is_expired(session, tenant_id="tenant-1", module_id="files", since=0))
|
|
finally:
|
|
engine.dispose()
|
|
|
|
def test_pruning_records_retention_floor_for_stale_watermarks(self) -> None:
|
|
engine = create_engine("sqlite:///:memory:")
|
|
SessionLocal = sessionmaker(bind=engine)
|
|
try:
|
|
Base.metadata.create_all(bind=engine, tables=[ChangeSequenceEntry.__table__, ChangeSequenceRetentionFloor.__table__])
|
|
with SessionLocal() as session:
|
|
first = record_change(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
module_id="files",
|
|
collection="files.assets",
|
|
resource_type="file",
|
|
resource_id="file-1",
|
|
operation="created",
|
|
)
|
|
second = record_change(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
module_id="files",
|
|
collection="files.assets",
|
|
resource_type="file",
|
|
resource_id="file-2",
|
|
operation="created",
|
|
)
|
|
session.commit()
|
|
first_id = first.id
|
|
second_resource_id = second.resource_id
|
|
|
|
deleted = prune_sequence_entries(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
module_id="files",
|
|
collections=("files.assets",),
|
|
before_or_equal_id=first_id,
|
|
)
|
|
session.commit()
|
|
|
|
self.assertEqual(deleted, 1)
|
|
self.assertEqual(retained_sequence_floor(session, tenant_id="tenant-1", module_id="files", collections=("files.assets",)), first_id)
|
|
self.assertTrue(sequence_watermark_is_expired(session, tenant_id="tenant-1", module_id="files", collections=("files.assets",), since=0))
|
|
self.assertFalse(sequence_watermark_is_expired(session, tenant_id="tenant-1", module_id="files", collections=("files.assets",), since=first_id))
|
|
self.assertEqual(
|
|
[entry.resource_id for entry in sequence_entries_since(session, tenant_id="tenant-1", module_id="files", collections=("files.assets",), since=first_id)],
|
|
[second_resource_id],
|
|
)
|
|
finally:
|
|
engine.dispose()
|
|
|
|
def test_rejects_invalid_watermark(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
decode_sequence_watermark("seq:not-a-number")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|