Stage 9 · L45

Assert protected effects, database rollback and coverage limits

Core · original Session 9

The stage-end capstone combines a protected state transition with an audit fact. A successful active→checked_in change must create exactly one AuditEvent in the same transaction. Repeat check-in returns the completed state without another audit. A forbidden customer action must leave the rental active and create no audit.

The service first flushes the rental state change, then adds/flushes AuditEvent before transaction exit. The audit action has a real database CHECK. The failure fixture changes the second write's action to invalid, producing an actual PostgreSQL constraint error after the first UPDATE was sent. The context rolls back both; a fresh session confirms active state and zero audit rows.

This is stronger than mocking session.rollback or looking at the same session's object. The first write actually reached PostgreSQL within a transaction, and another session sees its rollback. The public response is a safe500 internal_error rather than exposing the constraint SQL. Expected uniqueness/reference conflicts remain409 through the narrowed error classification.

Coverage measures executed lines/branches under a defined run; it does not prove assertions are meaningful, roles are complete, every race is covered or production resources work. The source supplement's80% example is a gate proposal, not a universal threshold or a passing result here. Review the permission/state/side-effect matrix and the actual fresh-session assertions.

Audit history is a synthetic teaching entity. Its foreign keys restrict deletion; production audit retention, account deletion/anonymization and erasure replay are not decided by this exercise. Do not publish this schema as if those legal/product decisions were approved. The public website's certificates and existing progress IDs remain untouched.

Savepoint isolation is a direct-service test technique. Begin one outer transaction on an explicitly owned connection, bind Session to it with join_transaction_mode="create_savepoint", and let the service finish its inner transaction. Its successful commit releases the savepoint, while outer.rollback still removes the test effect. Close the session before rolling back and verify from a fresh connection afterward.

This fixture isolates only operations using that bound connection. A service opening a new engine/session/connection escapes the outer transaction, and sharing this connection across unrelated concurrent tasks is unsafe. Actual API/failure/concurrency checks here keep disposable PostgreSQL fixtures and fresh observers; they do not claim savepoints enclose an arbitrary whole application.

with engine.connect() as connection:
    outer = connection.begin()
    session = Session(bind=connection, join_transaction_mode='create_savepoint')
    try:
        service_result = services.create_equipment(session, payload)
        # Service completion releases its savepoint, not the outer transaction.
    finally:
        session.close()
        outer.rollback()
# A fresh connection verifies the test write did not persist.

The bound session joins through savepoints, so a service commit releases its inner savepoint while the outer test transaction still owns rollback. A fresh connection after outer rollback sees no test row. A service opening a different connection escapes that fixture, so actual API rollback tests retain disposable databases/fresh observers. Coverage counts execution, not assertion quality. The capstone separately proves state/audit rollback through a real second-write failure.

python run_checks.py --stage 09 --role starter --prepare
python run_checks.py --stage 09 --role starter
python run_checks.py --stage 09 --role starter --transfer

python run_checks.py --stage 09 --role solution --transfer

@requires_permission("rental", "check_in")
def check_in(session, rental_id, actor):
    with session.begin():
        # Resolve immutable parent identity, then use the same parent lock order as reserve.
        equipment_id = session.scalar(select(Rental.equipment_id).where(Rental.id == rental_id))
        if equipment_id is None:
            raise DomainError(404, "rental_missing", "Rental not found")
        session.scalar(select(Equipment).where(Equipment.id == equipment_id).with_for_update())
        rental = session.scalar(select(Rental).where(Rental.id == rental_id).with_for_update())
        if rental.status == "active":
            rental.status = "checked_in"
            session.flush()
            session.add(AuditEvent(rental_id=rental.id,actor_id=actor.id,action="checked_in"))
            session.flush()
        output = RentalOut.model_validate(rental)
    return output