# Path: virtool/dispatcher/events.py
# class DispatcherSQLEvents:
#     """
#     Listens on SQLAlchemy session events to know what models have changes on transaction commits.
# 
#     Calls enqueue change with the correct dispatcher interface and list of modified IDs when a
#     transaction is committed. Gracefully handles rollbacks.
# 
#     Inspired by signalling in [Flask-SQLAlchemy](https://github.com/pallets/flask-sqlalchemy).
# 
#     """
# 
#     def __init__(
#         self, enqueue_change: Callable[[str, Operation, List[Union[str, int]]], None]
#     ):
#         self._enqueue_change = enqueue_change
#         self._changes = defaultdict(dict)
# 
#         event.listen(Session, "before_flush", self._record_ops)
#         event.listen(Session, "before_commit", self._record_ops)
#         event.listen(Session, "after_commit", self._after_commit)
#         event.listen(Session, "after_rollback", self._after_rollback)
# 
#     def _after_commit(self, session: Session):
#         """
#         Executed after a commit.
# 
#         Processes recorded changes and uses them to call :meth:`enqueue_change` as needed. Clears
#         the record changes when complete.
# 
#         :param session: the session that was committed
# 
#         """
#         changes = list(self._changes[id(session)].values())
# 
#         for data, change_type in changes:
#             interface = get_interface_from_model(data)
#             self._enqueue_change(interface, change_type, [data.id])
# 
#         del self._changes[id(session)]
# 
#     def _after_rollback(self, session: Session):
#         """
#         Executed after session rollback.
# 
#         Clears the records changes for the passed session.
# 
#         :param session: the session that was rolled back
# 
#         """
#         del self._changes[id(session)]
# 
#     def _record_ops(self, session: Session, flush_context=None, instances=None):
#         """
#         Called when a session fires the `after_flush` or `before_commit` events.
# 
#         Records the operations that have occurred in the session and stores them so they can be
#         sent via `enqueue_change` when the transaction is committed.
# 
#         The `flush_context` and `instances` arguments are included to match the callback function
#         signature expected by SQLAlchemy for the `after_flush` or `before_commit` events. They are
#         not used.
# 
#         :param session: the session the event was fired on
#         :param flush_context: unused
#         :param instances: unused
# 
#         """
#         session_changes = self._changes[id(session)]
# 
#         for targets, operation in (
#             (session.new, "insert"),
#             (session.dirty, "update"),
#             (session.deleted, "delete"),
#         ):
#             for target in targets:
#                 state = inspect(target)
#                 key = state.identity_key if state.has_identity else id(target)
#                 session_changes[key] = (target, operation)


import pytest
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
from virtool.dispatcher.events import DispatcherSQLEvents
from virtool.labels.models import Label



@pytest.mark.parametrize("rollback", [False, True])
async def test_events(rollback, mocker, pg: AsyncEngine):
    """
    Test that changes are recorded as needed and not recorded when a transaction is rolled back.
    Specifically, make sure changes aren't recorded for both flushed and un-flushed changes.

    """
    changes = list()

    def enqueue_change(*args):
        changes.append(args)


# Path: virtool/dispatcher/events.py
# class DispatcherSQLEvents:
#     """
#     Listens on SQLAlchemy session events to know what models have changes on transaction commits.
# 
#     Calls enqueue change with the correct dispatcher interface and list of modified IDs when a
#     transaction is committed. Gracefully handles rollbacks.
# 
#     Inspired by signalling in [Flask-SQLAlchemy](https://github.com/pallets/flask-sqlalchemy).
# 
#     """
# 
#     def __init__(
#         self, enqueue_change: Callable[[str, Operation, List[Union[str, int]]], None]
#     ):
#         self._enqueue_change = enqueue_change
#         self._changes = defaultdict(dict)
# 
#         event.listen(Session, "before_flush", self._record_ops)
#         event.listen(Session, "before_commit", self._record_ops)
#         event.listen(Session, "after_commit", self._after_commit)
#         event.listen(Session, "after_rollback", self._after_rollback)
# 
#     def _after_commit(self, session: Session):
#         """
#         Executed after a commit.
# 
#         Processes recorded changes and uses them to call :meth:`enqueue_change` as needed. Clears
#         the record changes when complete.
# 
#         :param session: the session that was committed
# 
#         """
#         changes = list(self._changes[id(session)].values())
# 
#         for data, change_type in changes:
#             interface = get_interface_from_model(data)
#             self._enqueue_change(interface, change_type, [data.id])
# 
#         del self._changes[id(session)]
# 
#     def _after_rollback(self, session: Session):
#         """
#         Executed after session rollback.
# 
#         Clears the records changes for the passed session.
# 
#         :param session: the session that was rolled back
# 
#         """
#         del self._changes[id(session)]
# 
#     def _record_ops(self, session: Session, flush_context=None, instances=None):
#         """
#         Called when a session fires the `after_flush` or `before_commit` events.
# 
#         Records the operations that have occurred in the session and stores them so they can be
#         sent via `enqueue_change` when the transaction is committed.
# 
#         The `flush_context` and `instances` arguments are included to match the callback function
#         signature expected by SQLAlchemy for the `after_flush` or `before_commit` events. They are
#         not used.
# 
#         :param session: the session the event was fired on
#         :param flush_context: unused
#         :param instances: unused
# 
#         """
#         session_changes = self._changes[id(session)]
# 
#         for targets, operation in (
#             (session.new, "insert"),
#             (session.dirty, "update"),
#             (session.deleted, "delete"),
#         ):
#             for target in targets:
#                 state = inspect(target)
#                 key = state.identity_key if state.has_identity else id(target)
#                 session_changes[key] = (target, operation)


# Path: tests/dispatcher/test_events.py
import pytest
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
from virtool.dispatcher.events import DispatcherSQLEvents
from virtool.labels.models import Label




@pytest.mark.parametrize("rollback", [False, True])
async def test_events(rollback, mocker, pg: AsyncEngine):
    """
    Test that changes are recorded as needed and not recorded when a transaction is rolled back.
    Specifically, make sure changes aren't recorded for both flushed and un-flushed changes.

    """
    changes = list()

    def enqueue_change(*args):
        changes.append(args)


    DispatcherSQLEvents(enqueue_change)