6.4Testing Triggers

Triggers run automatically when data is inserted, updated, or deleted. They are invisible to the application — which makes them especially important to test.

updated_at Trigger

def test_updated_at_trigger(db):
    """Verify that updated_at changes when a row is modified."""
    cursor = db.cursor()

    # Create a user
    cursor.execute("""
        INSERT INTO users (id, name, email)
        VALUES ('trigger-test', 'Original Name', 'trigger@test.com')
        RETURNING created_at, updated_at
    """)
    created_at, updated_at = cursor.fetchone()
    assert created_at == updated_at  # On creation, both should be the same

    # Wait briefly and update
    import time
    time.sleep(1)
    cursor.execute("""
        UPDATE users SET name = 'Updated Name'
        WHERE id = 'trigger-test'
    """)

    # Verify updated_at changed
    cursor.execute("SELECT updated_at FROM users WHERE id = 'trigger-test'")
    new_updated_at = cursor.fetchone()[0]
    assert new_updated_at > updated_at, "updated_at should be newer after UPDATE"

def test_updated_at_unchanged_on_same_value(db):
    """Some implementations only update timestamp if data actually changed."""
    cursor = db.cursor()
    cursor.execute("""
        INSERT INTO users (id, name, email)
        VALUES ('same-val-test', 'Same Name', 'sameval@test.com')
    """)

    import time
    time.sleep(1)

    # Update with the same value
    cursor.execute("""
        UPDATE users SET name = 'Same Name'
        WHERE id = 'same-val-test'
    """)

    cursor.execute("""
        SELECT created_at, updated_at FROM users WHERE id = 'same-val-test'
    """)
    created_at, updated_at = cursor.fetchone()
    # Behavior depends on trigger implementation:
    # Some triggers update timestamp regardless, others only on actual change

Audit Trail Trigger

def test_audit_trail_on_update(db):
    """Verify that changes to users are logged in audit table."""
    cursor = db.cursor()

    # Create user
    cursor.execute("""
        INSERT INTO users (id, name, email, role)
        VALUES ('audit-test', 'Audit User', 'audit@test.com', 'viewer')
    """)

    # Update role
    cursor.execute("""
        UPDATE users SET role = 'admin' WHERE id = 'audit-test'
    """)

    # Check audit table
    cursor.execute("""
        SELECT old_value, new_value, changed_field, changed_at
        FROM audit_log
        WHERE table_name = 'users' AND record_id = 'audit-test'
        ORDER BY changed_at DESC LIMIT 1
    """)
    row = cursor.fetchone()
    assert row is not None, "Audit log entry should exist"
    assert row[0] == "viewer"   # old_value
    assert row[1] == "admin"    # new_value
    assert row[2] == "role"     # changed_field

def test_audit_trail_on_delete(db):
    """Verify that deletions are logged."""
    cursor = db.cursor()
    cursor.execute("""
        INSERT INTO users (id, name, email)
        VALUES ('delete-audit', 'Delete Me', 'delete@test.com')
    """)
    cursor.execute("DELETE FROM users WHERE id = 'delete-audit'")

    cursor.execute("""
        SELECT action FROM audit_log
        WHERE table_name = 'users' AND record_id = 'delete-audit'
    """)
    row = cursor.fetchone()
    assert row is not None
    assert row[0] == "DELETE"