73 lines
2.4 KiB
Python
73 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import DateTime, String, create_engine
|
|
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
|
|
|
|
from govoplan_core.core.sqlalchemy_change_tracking import (
|
|
ensure_object_id,
|
|
operation_for_object,
|
|
operation_for_soft_deletable,
|
|
previous_value,
|
|
)
|
|
|
|
|
|
class Base(DeclarativeBase):
|
|
pass
|
|
|
|
|
|
class ExampleRecord(Base):
|
|
__tablename__ = "example_records"
|
|
|
|
id: Mapped[str] = mapped_column(String, primary_key=True)
|
|
name: Mapped[str] = mapped_column(String)
|
|
deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
|
|
|
|
|
|
class SqlalchemyChangeTrackingTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(self.engine)
|
|
self.session = Session(self.engine, expire_on_commit=False)
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_pending_object_is_created_and_can_receive_id(self) -> None:
|
|
record = ExampleRecord(id="", name="Draft", deleted_at=None)
|
|
self.session.add(record)
|
|
|
|
self.assertEqual(ensure_object_id(record, lambda: "record-1"), "record-1")
|
|
self.assertEqual(record.id, "record-1")
|
|
self.assertEqual(operation_for_object(record, changed_attrs=("name",)), "created")
|
|
|
|
def test_changed_object_reports_update_and_previous_value(self) -> None:
|
|
record = ExampleRecord(id="record-1", name="Before", deleted_at=None)
|
|
self.session.add(record)
|
|
self.session.flush()
|
|
|
|
record.name = "After"
|
|
|
|
self.assertEqual(operation_for_object(record, changed_attrs=("name",)), "updated")
|
|
self.assertEqual(previous_value(record, "name"), "Before")
|
|
|
|
def test_soft_delete_and_restore_are_classified(self) -> None:
|
|
deleted_at = datetime(2026, 7, 13, tzinfo=timezone.utc)
|
|
record = ExampleRecord(id="record-1", name="Record", deleted_at=None)
|
|
self.session.add(record)
|
|
self.session.flush()
|
|
|
|
record.deleted_at = deleted_at
|
|
self.assertEqual(operation_for_soft_deletable(record, changed_attrs=("deleted_at",)), "deleted")
|
|
|
|
self.session.flush()
|
|
record.deleted_at = None
|
|
self.assertEqual(operation_for_soft_deletable(record, changed_attrs=("deleted_at",)), "created")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|