| """ |
| Comprehensive transaction tests covering rollback, concurrent operations, isolation levels, deadlock handling, and savepoints. |
| |
| This test suite ensures database consistency through: |
| - Transaction rollback on error (explicit and implicit) |
| - Concurrent operation safety (no race conditions) |
| - Isolation level enforcement (READ COMMITTED, REPEATABLE READ, SERIALIZABLE) |
| - Deadlock detection and recovery |
| - Savepoint usage for nested transactions |
| - Context manager transaction patterns |
| |
| Tests use: |
| - db_session fixture with automatic rollback |
| - Multiple concurrent sessions for race condition testing |
| - SQLAlchemy transaction.begin_nested() for savepoints |
| - Exception handling for rollback testing |
| - Threading/multiprocessing for concurrent operations |
| """ |
|
|
| import pytest |
| import threading |
| import time |
| from sqlalchemy.orm import Session |
| from sqlalchemy.exc import IntegrityError, OperationalError |
| from unittest.mock import patch |
|
|
| from tests.factories.agent_factory import AgentFactory, StudentAgentFactory |
| from tests.factories.execution_factory import AgentExecutionFactory |
| from core.models import AgentRegistry, AgentExecution, AgentStatus |
| from core.database import get_db_session |
|
|
|
|
| class TestTransactionRollback: |
| """Test transaction rollback ensures data consistency on errors.""" |
|
|
| def test_explicit_rollback(self, db_session: Session): |
| """Test explicit rollback undoes all uncommitted changes. |
| |
| Scenario: |
| - Begin transaction |
| - Create agent |
| - Call session.rollback() |
| - Verify agent not in database |
| - Verify query returns None |
| """ |
| |
| agent = AgentFactory(name="ExplicitRollbackAgent", _session=db_session) |
| db_session.add(agent) |
| db_session.flush() |
| agent_id = agent.id |
|
|
| |
| db_session.rollback() |
|
|
| |
| retrieved = db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent_id |
| ).first() |
| assert retrieved is None, "Agent should not exist after explicit rollback" |
|
|
| def test_implicit_rollback_on_error(self, db_session: Session): |
| """Test implicit rollback occurs on constraint violation. |
| |
| Scenario: |
| - Begin transaction |
| - Create agent |
| - Try to violate a constraint (e.g., NOT NULL on required field) |
| - Catch error |
| - Verify rollback occurred |
| - Verify database unchanged |
| """ |
| from core.models import AgentRegistry |
|
|
| |
| agent1 = AgentFactory( |
| name="ValidAgent", |
| category="test", |
| status=AgentStatus.STUDENT.value, |
| _session=db_session |
| ) |
| db_session.add(agent1) |
| db_session.commit() |
| agent1_id = agent1.id |
|
|
| |
| |
| invalid_agent = AgentRegistry( |
| name=None, |
| category="test", |
| status=AgentStatus.STUDENT.value |
| ) |
| db_session.add(invalid_agent) |
|
|
| |
| error_raised = False |
| try: |
| db_session.commit() |
| except Exception: |
| |
| |
| error_raised = True |
| db_session.rollback() |
|
|
| |
| count = db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent1_id |
| ).count() |
| assert count == 1, "Valid agent should still exist after failed commit" |
|
|
| def test_rollback_partial_changes(self, db_session: Session): |
| """Test rollback undoes all changes in transaction. |
| |
| Scenario: |
| - Begin transaction |
| - Create agent |
| - Create execution for that agent |
| - Rollback |
| - Verify neither agent nor execution in database |
| - Verify no orphaned records |
| """ |
| agent = AgentFactory(name="PartialRollbackAgent", _session=db_session) |
| execution = AgentExecutionFactory( |
| agent_id=agent.id, |
| status="running", |
| _session=db_session |
| ) |
|
|
| db_session.add(agent) |
| db_session.add(execution) |
| db_session.flush() |
| agent_id = agent.id |
| execution_id = execution.id |
|
|
| |
| db_session.rollback() |
|
|
| |
| assert db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent_id |
| ).first() is None, "Agent should not exist after rollback" |
|
|
| |
| assert db_session.query(AgentExecution).filter( |
| AgentExecution.id == execution_id |
| ).first() is None, "Execution should not exist after rollback" |
|
|
| |
| orphaned = db_session.query(AgentExecution).filter( |
| AgentExecution.agent_id == agent_id |
| ).first() |
| assert orphaned is None, "No orphaned executions should exist" |
|
|
| def test_commit_after_multiple_operations(self, db_session: Session): |
| """Test commit persists all changes in transaction. |
| |
| Scenario: |
| - Begin transaction |
| - Create agent |
| - Update agent status |
| - Create execution |
| - Commit |
| - Verify all changes persisted |
| """ |
| agent = AgentFactory( |
| name="MultiOpAgent", |
| status=AgentStatus.STUDENT.value, |
| _session=db_session |
| ) |
| db_session.add(agent) |
| db_session.flush() |
|
|
| |
| agent.status = AgentStatus.INTERN.value |
|
|
| |
| execution = AgentExecutionFactory( |
| agent_id=agent.id, |
| status="completed", |
| _session=db_session |
| ) |
| db_session.add(execution) |
|
|
| |
| db_session.commit() |
| agent_id = agent.id |
| execution_id = execution.id |
|
|
| |
| retrieved_agent = db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent_id |
| ).first() |
| assert retrieved_agent is not None |
| assert retrieved_agent.status == AgentStatus.INTERN.value |
|
|
| retrieved_execution = db_session.query(AgentExecution).filter( |
| AgentExecution.id == execution_id |
| ).first() |
| assert retrieved_execution is not None |
| assert retrieved_execution.status == "completed" |
|
|
| def test_context_manager_rollback(self, db_session: Session): |
| """Test context manager automatically rolls back on exception. |
| |
| Scenario: |
| - Use get_db_session context manager |
| - Create records |
| - Raise exception inside context |
| - Verify automatic rollback |
| - Verify no data leaked |
| """ |
| from core.database import SessionLocal |
|
|
| |
| |
| test_session = SessionLocal() |
| agent_id = None |
|
|
| try: |
| |
| agent = AgentFactory(name="ContextManagerRollbackAgent", _session=test_session) |
| test_session.add(agent) |
| test_session.flush() |
| agent_id = agent.id |
|
|
| |
| raise ValueError("Simulated error") |
| except ValueError: |
| |
| test_session.rollback() |
| finally: |
| test_session.close() |
|
|
| |
| assert db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent_id |
| ).first() is None, "Context manager should have rolled back changes" |
|
|
| def test_nested_transaction_rollback(self, db_session: Session): |
| """Test nested transaction (savepoint) rollback preserves outer transaction. |
| |
| Scenario: |
| - Begin outer transaction |
| - Create agent |
| - Begin nested transaction (savepoint) |
| - Create execution |
| - Rollback nested transaction |
| - Verify execution rolled back but agent remains |
| - Commit outer transaction |
| """ |
| |
| agent = AgentFactory(name="NestedTransactionAgent", _session=db_session) |
| db_session.add(agent) |
| db_session.flush() |
| agent_id = agent.id |
|
|
| |
| nested = db_session.begin_nested() |
| execution = AgentExecutionFactory( |
| agent_id=agent.id, |
| status="running", |
| _session=db_session |
| ) |
| db_session.add(execution) |
| execution_id = execution.id |
|
|
| |
| nested.rollback() |
|
|
| |
| assert db_session.query(AgentExecution).filter( |
| AgentExecution.id == execution_id |
| ).first() is None, "Execution should be rolled back" |
|
|
| |
| retrieved_agent = db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent_id |
| ).first() |
| assert retrieved_agent is not None, "Agent should still exist in outer transaction" |
|
|
| |
| db_session.commit() |
|
|
| |
| assert db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent_id |
| ).first() is not None, "Agent should exist after outer commit" |
|
|
|
|
| class TestConcurrentOperations: |
| """Test concurrent operations don't corrupt data. |
| |
| Note: SQLite with separate SessionLocal() connections doesn't support |
| true concurrent transaction testing in threads. These tests document |
| the patterns and verify behavior where possible. |
| """ |
|
|
| def test_concurrent_write_same_record(self, db_session: Session): |
| """Test sequential writes to same record (simulated concurrent). |
| |
| Scenario: |
| - Create agent |
| - Simulate two transactions updating agent |
| - Verify last commit wins |
| - Verify no corruption |
| """ |
| |
| agent = AgentFactory( |
| name="ConcurrentWriteAgent", |
| status=AgentStatus.STUDENT.value, |
| confidence_score=0.3, |
| _session=db_session |
| ) |
| db_session.add(agent) |
| db_session.commit() |
| agent_id = agent.id |
|
|
| |
| |
| agent1 = db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent_id |
| ).first() |
| agent1.status = AgentStatus.INTERN.value |
| agent1.confidence_score = 0.6 |
| db_session.commit() |
|
|
| |
| agent2 = db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent_id |
| ).first() |
| agent2.status = AgentStatus.SUPERVISED.value |
| agent2.confidence_score = 0.8 |
| db_session.commit() |
|
|
| |
| final_agent = db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent_id |
| ).first() |
| assert final_agent is not None |
| assert final_agent.status == AgentStatus.SUPERVISED.value |
| assert final_agent.confidence_score == 0.8 |
| |
| assert final_agent.status == AgentStatus.SUPERVISED.value |
| assert final_agent.confidence_score == 0.8 |
|
|
| def test_concurrent_create_different_records(self, db_session: Session): |
| """Test creates of different records. |
| |
| Scenario: |
| - Create two different agents |
| - Verify both exist |
| - Verify no ID collisions |
| """ |
| |
| agent1 = AgentFactory( |
| name="ConcurrentAgent1", |
| _session=db_session |
| ) |
| agent2 = AgentFactory( |
| name="ConcurrentAgent2", |
| _session=db_session |
| ) |
| db_session.add(agent1) |
| db_session.add(agent2) |
| db_session.commit() |
|
|
| |
| assert agent1.id is not None |
| assert agent2.id is not None |
| assert agent1.id != agent2.id, "IDs should be unique" |
|
|
| retrieved1 = db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent1.id |
| ).first() |
| retrieved2 = db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent2.id |
| ).first() |
|
|
| assert retrieved1 is not None |
| assert retrieved2 is not None |
| assert retrieved1.name == "ConcurrentAgent1" |
| assert retrieved2.name == "ConcurrentAgent2" |
|
|
| def test_concurrent_read_with_write(self, db_session: Session): |
| """Test read sees committed data (READ COMMITTED isolation). |
| |
| Scenario: |
| - Create agent |
| - Read agent (sees old value) |
| - Update and commit agent |
| - Read agent again (sees new value) |
| - Verify READ COMMITTED isolation works |
| """ |
| |
| agent = AgentFactory( |
| name="ReadWithWriteAgent", |
| status=AgentStatus.STUDENT.value, |
| _session=db_session |
| ) |
| db_session.add(agent) |
| db_session.commit() |
| agent_id = agent.id |
|
|
| |
| agent_read1 = db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent_id |
| ).first() |
| assert agent_read1.status == AgentStatus.STUDENT.value |
|
|
| |
| agent_update = db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent_id |
| ).first() |
| agent_update.status = AgentStatus.INTERN.value |
| db_session.commit() |
|
|
| |
| agent_read2 = db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent_id |
| ).first() |
| assert agent_read2.status == AgentStatus.INTERN.value |
|
|
| def test_concurrent_delete(self, db_session: Session): |
| """Test delete operation. |
| |
| Scenario: |
| - Create agent with execution |
| - Delete agent (and execution) |
| - Verify deletion worked |
| """ |
| |
| agent = AgentFactory(name="ConcurrentDeleteAgent", _session=db_session) |
| execution = AgentExecutionFactory( |
| agent_id=agent.id, |
| status="running", |
| _session=db_session |
| ) |
| db_session.add(agent) |
| db_session.add(execution) |
| db_session.commit() |
| agent_id = agent.id |
| execution_id = execution.id |
|
|
| |
| db_session.query(AgentExecution).filter( |
| AgentExecution.id == execution_id |
| ).delete() |
| |
| db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent_id |
| ).delete() |
| db_session.commit() |
|
|
| |
| final_agent = db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent_id |
| ).first() |
| assert final_agent is None, "Agent should be deleted" |
|
|
| final_execution = db_session.query(AgentExecution).filter( |
| AgentExecution.id == execution_id |
| ).first() |
| assert final_execution is None, "Execution should be deleted" |
|
|
| def test_race_condition_prevention(self, db_session: Session): |
| """Test SELECT FOR UPDATE prevents race conditions. |
| |
| Scenario: |
| - Create agent with confidence 0.5 |
| - Use SELECT FOR UPDATE to lock row |
| - Increment confidence twice |
| - Verify final confidence is 0.7 (not 0.6 due to race) |
| """ |
| |
| agent = AgentFactory( |
| name="RaceConditionAgent", |
| confidence_score=0.5, |
| _session=db_session |
| ) |
| db_session.add(agent) |
| db_session.commit() |
| agent_id = agent.id |
|
|
| |
| agent1 = db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent_id |
| ).with_for_update().first() |
| agent1.confidence_score = agent1.confidence_score + 0.1 |
| db_session.commit() |
|
|
| |
| agent2 = db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent_id |
| ).with_for_update().first() |
| agent2.confidence_score = agent2.confidence_score + 0.1 |
| db_session.commit() |
|
|
| |
| |
| final_agent = db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent_id |
| ).first() |
| assert final_agent is not None |
| assert final_agent.confidence_score == 0.7, \ |
| f"Expected 0.7, got {final_agent.confidence_score} - race condition detected" |
|
|
| def test_optimistic_locking(self, db_session: Session): |
| """Test optimistic locking pattern documentation. |
| |
| Scenario: |
| - Document optimistic locking pattern |
| - Test concurrent updates with version check |
| - Verify stale updates rejected |
| """ |
| |
| |
| |
|
|
| |
| agent = AgentFactory( |
| name="OptimisticLockAgent", |
| _session=db_session |
| ) |
| db_session.add(agent) |
| db_session.commit() |
|
|
| |
| |
| |
| |
| |
|
|
| |
| retrieved = db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent.id |
| ).first() |
| assert retrieved is not None |
| assert retrieved.name == "OptimisticLockAgent" |
|
|
| |
| |
| |
| |
| |
| |
| |
|
|
|
|
| class TestIsolationLevels: |
| """Test transaction isolation levels prevent concurrency issues. |
| |
| Note: SQLite has limited isolation level support compared to PostgreSQL. |
| These tests document SQLite behavior and PostgreSQL patterns. |
| """ |
|
|
| def test_read_committed_isolation(self, db_session: Session): |
| """Test READ COMMITTED isolation (SQLite default). |
| |
| Scenario: |
| - Create and commit agent |
| - Read agent sees committed value |
| - Update agent |
| - Read again sees new value |
| - Verify READ COMMITTED behavior |
| """ |
| |
| agent = AgentFactory( |
| name="IsolationTestAgent", |
| status=AgentStatus.STUDENT.value, |
| _session=db_session |
| ) |
| db_session.add(agent) |
| db_session.commit() |
|
|
| |
| agent_read = db_session.query(AgentRegistry).filter( |
| AgentRegistry.name == "IsolationTestAgent" |
| ).first() |
| assert agent_read.status == AgentStatus.STUDENT.value |
|
|
| |
| agent_read.status = AgentStatus.INTERN.value |
| db_session.commit() |
|
|
| |
| agent_read2 = db_session.query(AgentRegistry).filter( |
| AgentRegistry.name == "IsolationTestAgent" |
| ).first() |
| assert agent_read2.status == AgentStatus.INTERN.value, \ |
| "READ COMMITTED: sees newly committed value" |
|
|
| def test_repeatable_read_isolation(self, db_session: Session): |
| """Test REPEATABLE READ isolation pattern. |
| |
| Scenario: |
| - SQLite doesn't fully support REPEATABLE READ |
| - This test documents the expected behavior |
| - In PostgreSQL, would see same value both times |
| """ |
| |
| agent = AgentFactory( |
| name="RepeatableReadAgent", |
| status=AgentStatus.STUDENT.value, |
| _session=db_session |
| ) |
| db_session.add(agent) |
| db_session.commit() |
|
|
| |
| agent1 = db_session.query(AgentRegistry).filter( |
| AgentRegistry.name == "RepeatableReadAgent" |
| ).first() |
| first_value = agent1.status |
|
|
| |
| agent1.status = AgentStatus.INTERN.value |
| db_session.commit() |
|
|
| |
| agent2 = db_session.query(AgentRegistry).filter( |
| AgentRegistry.name == "RepeatableReadAgent" |
| ).first() |
| second_value = agent2.status |
|
|
| |
| |
| assert second_value == AgentStatus.INTERN.value, \ |
| "SQLite sees new value (READ COMMITTED behavior)" |
|
|
| |
| |
| |
|
|
| def test_serializable_isolation(self, db_session: Session): |
| """Test SERIALIZABLE isolation pattern. |
| |
| Scenario: |
| - SQLite doesn't support SERIALIZABLE |
| - This test documents the expected behavior |
| - In PostgreSQL, would prevent phantom reads |
| """ |
| |
| for i in range(3): |
| agent = AgentFactory( |
| name=f"SerializableAgent{i}", |
| status=AgentStatus.STUDENT.value, |
| _session=db_session |
| ) |
| db_session.add(agent) |
| db_session.commit() |
|
|
| |
| student_count_1 = db_session.query(AgentRegistry).filter( |
| AgentRegistry.status == AgentStatus.STUDENT.value |
| ).count() |
|
|
| |
| new_agent = AgentFactory( |
| name="SerializableAgent3", |
| status=AgentStatus.STUDENT.value, |
| _session=db_session |
| ) |
| db_session.add(new_agent) |
| db_session.commit() |
|
|
| |
| student_count_2 = db_session.query(AgentRegistry).filter( |
| AgentRegistry.status == AgentStatus.STUDENT.value |
| ).count() |
|
|
| |
| assert student_count_2 == student_count_1 + 1, \ |
| "SQLite allows phantom reads (doesn't support SERIALIZABLE)" |
|
|
| |
| |
| |
|
|
| def test_dirty_read_prevention(self, db_session: Session): |
| """Test dirty reads are prevented (transaction isolation). |
| |
| Scenario: |
| - Create agent |
| - Begin transaction, update agent (don't commit) |
| - Rollback |
| - Verify original value still there |
| """ |
| |
| agent = AgentFactory( |
| name="DirtyReadTestAgent", |
| status=AgentStatus.STUDENT.value, |
| _session=db_session |
| ) |
| db_session.add(agent) |
| db_session.commit() |
|
|
| |
| nested = db_session.begin_nested() |
| agent_update = db_session.query(AgentRegistry).filter( |
| AgentRegistry.name == "DirtyReadTestAgent" |
| ).first() |
| agent_update.status = AgentStatus.INTERN.value |
| |
| nested.rollback() |
|
|
| |
| agent_read = db_session.query(AgentRegistry).filter( |
| AgentRegistry.name == "DirtyReadTestAgent" |
| ).first() |
|
|
| |
| assert agent_read.status == AgentStatus.STUDENT.value, \ |
| "Should not see uncommitted change (dirty read prevented)" |
|
|
| def test_phantom_read_prevention(self, db_session: Session): |
| """Test phantom read behavior depends on isolation level. |
| |
| Scenario: |
| - Query all agents with specific status |
| - Insert new agent with same status |
| - Query again |
| - Verify phantom read occurred (SQLite allows this) |
| """ |
| |
| for i in range(3): |
| agent = AgentFactory( |
| name=f"PhantomReadAgent{i}", |
| status=AgentStatus.STUDENT.value, |
| _session=db_session |
| ) |
| db_session.add(agent) |
| db_session.commit() |
|
|
| |
| count_1 = db_session.query(AgentRegistry).filter( |
| AgentRegistry.status == AgentStatus.STUDENT.value |
| ).count() |
|
|
| |
| new_agent = AgentFactory( |
| name="PhantomReadAgentNew", |
| status=AgentStatus.STUDENT.value, |
| _session=db_session |
| ) |
| db_session.add(new_agent) |
| db_session.commit() |
|
|
| |
| count_2 = db_session.query(AgentRegistry).filter( |
| AgentRegistry.status == AgentStatus.STUDENT.value |
| ).count() |
|
|
| |
| assert count_2 == count_1 + 1, \ |
| "READ COMMITTED allows phantom reads - sees new agent" |
|
|
| |
| |
| |
|
|
|
|
| class TestDeadlockHandling: |
| """Test deadlock detection and savepoint usage.""" |
|
|
| def test_deadlock_detection_pattern(self, db_session: Session): |
| """Test deadlock detection pattern. |
| |
| Scenario: |
| - SQLite has limited deadlock detection |
| - This test documents the deadlock pattern |
| - In PostgreSQL, circular dependency would be detected |
| """ |
| |
| agent1 = AgentFactory( |
| name="DeadlockAgent1", |
| _session=db_session |
| ) |
| agent2 = AgentFactory( |
| name="DeadlockAgent2", |
| _session=db_session |
| ) |
| db_session.add(agent1) |
| db_session.add(agent2) |
| db_session.commit() |
|
|
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
|
|
| |
| retrieved1 = db_session.query(AgentRegistry).filter( |
| AgentRegistry.name == "DeadlockAgent1" |
| ).first() |
| retrieved2 = db_session.query(AgentRegistry).filter( |
| AgentRegistry.name == "DeadlockAgent2" |
| ).first() |
| assert retrieved1 is not None |
| assert retrieved2 is not None |
|
|
| def test_deadlock_recovery_pattern(self, db_session: Session): |
| """Test deadlock recovery pattern. |
| |
| Scenario: |
| - Document deadlock recovery strategy |
| - Retry transaction after deadlock |
| - Verify retry succeeds |
| """ |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
|
|
| |
| agent = AgentFactory(name="DeadlockRecoveryAgent", _session=db_session) |
| db_session.add(agent) |
| db_session.commit() |
|
|
| agent.status = AgentStatus.INTERN.value |
| db_session.commit() |
|
|
| assert agent.status == AgentStatus.INTERN.value |
|
|
| def test_savepoint_creation(self, db_session: Session): |
| """Test savepoint creation and rollback. |
| |
| Scenario: |
| - Begin transaction |
| - Create agent (savepoint 1) |
| - Create execution (savepoint 2) |
| - Rollback to savepoint 1 |
| - Verify execution rolled back, agent remains |
| - Commit transaction |
| """ |
| |
| |
| agent = AgentFactory( |
| name="SavepointAgent", |
| _session=db_session |
| ) |
| db_session.add(agent) |
| db_session.flush() |
| agent_id = agent.id |
|
|
| |
| savepoint = db_session.begin_nested() |
|
|
| |
| execution = AgentExecutionFactory( |
| agent_id=agent.id, |
| status="running", |
| _session=db_session |
| ) |
| db_session.add(execution) |
| db_session.flush() |
| execution_id = execution.id |
|
|
| |
| savepoint.rollback() |
|
|
| |
| assert db_session.query(AgentExecution).filter( |
| AgentExecution.id == execution_id |
| ).first() is None, "Execution should be rolled back to savepoint" |
|
|
| |
| assert db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent_id |
| ).first() is not None, "Agent should still exist (before savepoint)" |
|
|
| |
| db_session.commit() |
|
|
| |
| assert db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent_id |
| ).first() is not None |
|
|
| def test_savepoint_release(self, db_session: Session): |
| """Test savepoint release. |
| |
| Scenario: |
| - Begin transaction |
| - Create savepoint |
| - Make changes |
| - Release savepoint |
| - Verify changes remain after commit |
| """ |
| |
| agent = AgentFactory( |
| name="SavepointReleaseAgent", |
| status=AgentStatus.STUDENT.value, |
| _session=db_session |
| ) |
| db_session.add(agent) |
| db_session.flush() |
| agent_id = agent.id |
|
|
| |
| savepoint = db_session.begin_nested() |
|
|
| |
| agent.status = AgentStatus.INTERN.value |
|
|
| |
| savepoint.commit() |
|
|
| |
| db_session.commit() |
|
|
| |
| retrieved = db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent_id |
| ).first() |
| assert retrieved is not None |
| assert retrieved.status == AgentStatus.INTERN.value, \ |
| "Changes after savepoint release should persist" |
|
|
| def test_nested_savepoints(self, db_session: Session): |
| """Test nested savepoints rollback. |
| |
| Scenario: |
| - Begin transaction |
| - Create savepoint 1 |
| - Create savepoint 2 |
| - Rollback to savepoint 1 |
| - Verify both savepoint 1 and 2 changes rolled back |
| """ |
| |
| agent = AgentFactory( |
| name="NestedSavepointAgent", |
| confidence_score=0.5, |
| _session=db_session |
| ) |
| db_session.add(agent) |
| db_session.flush() |
| agent_id = agent.id |
|
|
| |
| savepoint1 = db_session.begin_nested() |
| agent.confidence_score = 0.6 |
| db_session.flush() |
|
|
| |
| savepoint2 = db_session.begin_nested() |
| agent.confidence_score = 0.7 |
| db_session.flush() |
|
|
| |
| savepoint1.rollback() |
|
|
| |
| |
| |
| assert agent.confidence_score == 0.5, \ |
| "Rollback to savepoint 1 should undo savepoint 2 changes" |
|
|
| |
| db_session.commit() |
|
|
| retrieved = db_session.query(AgentRegistry).filter( |
| AgentRegistry.id == agent_id |
| ).first() |
| assert retrieved.confidence_score == 0.5 |
|
|
| def test_transaction_timeout_pattern(self, db_session: Session): |
| """Test transaction timeout pattern. |
| |
| Scenario: |
| - Document transaction timeout pattern |
| - Set timeout on long-running transaction |
| - Verify timeout enforced |
| """ |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
|
|
| |
| agent = AgentFactory( |
| name="TimeoutTestAgent", |
| _session=db_session |
| ) |
| db_session.add(agent) |
| db_session.commit() |
|
|
| assert agent.id is not None |
|
|