Advanced Python unit testing framework for customer support tech enablement, covering FastAPI, SQLAlchemy, PostgreSQL, async operations, mocking, fixtures, parametrization, coverage, and comprehensive testing strategies for backend support systems
Installer avec Codex ou Claude Copiez ce prompt, collez-le dans Codex, Claude ou un autre assistant, puis laissez-le vérifier la page du skill et l'installer pour vous.
Une commande directe contourne le prompt de vérification. Examinez la source avant de l'exécuter.
Advanced Python unit testing framework for customer support tech enablement, covering FastAPI, SQLAlchemy, PostgreSQL, async operations, mocking, fixtures, parametrization, coverage, and comprehensive testing strategies for backend support systems
["Testing customer support ticket APIs and workflows","Validating database operations and transactions with PostgreSQL","Testing authentication and authorization flows for support systems","Mocking external services like email, SMS, webhooks, and CRM integrations","Testing async background tasks and queue processing","Data validation and curation testing for customer records","Integration testing with PostgreSQL and SQLAlchemy ORM","API endpoint testing for ticketing and knowledge base systems","Testing error handling and edge cases in support applications","Coverage reporting for support applications with quality gates","Testing multi-tenant customer support architectures","Validating SLA calculations and escalation logic","Testing notification systems and email templates","Database migration testing with Alembic","Performance and load testing for support APIs"]
pytest - Advanced Python Unit Testing for Customer Support
Overview
pytest is the industry-standard testing framework for Python applications, offering powerful features that enable comprehensive, maintainable, and scalable test suites. This skill focuses specifically on customer support tech enablement, providing patterns and practices for testing backend support systems, ticketing platforms, knowledge bases, and customer data platforms.
Customer support systems require rigorous testing due to their mission-critical nature. Downtime or bugs directly impact customer satisfaction, agent productivity, and business operations. This skill provides comprehensive guidance on testing all aspects of support systems using pytest.
Why pytest for Customer Support Systems
Unique Requirements
Customer support applications have specific testing needs:
High Reliability: Support systems are mission-critical; failures directly affect customer experience
Complex Data Relationships: Tickets, customers, agents, comments, attachments, and knowledge articles
Real-time Features: WebSocket connections, live chat, real-time ticket updates
Compliance: GDPR, CCPA, data retention policies, audit logging
pytest Advantages
pytest addresses these needs through:
Powerful Fixture System: Manage complex database setups, API clients, and test data
Parametrization: Test multiple scenarios efficiently (various ticket types, priority levels, user roles)
Rich Plugin Ecosystem: pytest-asyncio for async testing, pytest-mock for mocking, pytest-cov for coverage
Excellent Integration: Works seamlessly with FastAPI, SQLAlchemy, PostgreSQL, Pydantic
Clear Output: Readable test results and comprehensive error reporting
Scalability: Handles small test suites to thousands of tests with parallel execution
Flexibility: Supports unit, integration, and end-to-end testing in one framework
Core Competencies
1. Fixtures and Dependency Injection
Fixtures are pytest's killer feature, providing reusable setup/teardown logic and dependency injection. For customer support systems, fixtures manage databases, API clients, test data, and external service mocks.
Basic Fixtures
import pytest
from app.models import Ticket, Customer, Agent
@pytest.fixture
def support_ticket():
"""Provide a basic support ticket dictionary."""
return {
"id": 1,
"title": "Cannot access account",
"description": "User unable to login after password reset",
"status": "open",
"priority": "high",
"customer_email": "user@example.com",
"category": "authentication"
}
def test_ticket_structure(support_ticket):
"""Test ticket has required fields."""
assert support_ticket["status"] == "open"
assert support_ticket["priority"] in ["low", "medium", "high", "critical"]
assert "@" in support_ticket["customer_email"]
Fixture Scopes
Control when fixtures are created and destroyed using scopes:
function (default): Created once per test function, destroyed after test completes
class: Created once per test class, shared across all methods
module: Created once per module file, shared across all tests in file
package: Created once per package, shared across all tests in package
session: Created once per entire test session, shared across all tests
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from typing import Generator
@pytest.fixture(scope="session")
def database_engine():
"""Create database engine once per session."""
engine = create_engine(
"postgresql://support_user:password@localhost/support_test",
echo=False,
pool_pre_ping=True,
pool_size=10,
max_overflow=20
)
# Create all tables
from app.database import Base
Base.metadata.create_all(engine)
yield engine
# Cleanup: drop all tables and dispose engine
Base.metadata.drop_all(engine)
engine.dispose()
@pytest.fixture(scope="function")
def db_session(database_engine) -> Generator[Session, None, None]:
"""Create a new database session for each test with automatic rollback."""
SessionLocal = sessionmaker(bind=database_engine)
session = SessionLocal()
try:
yield session
finally:
session.rollback() # Rollback any changes
session.close()
Autouse Fixtures
Fixtures that run automatically without being explicitly requested:
@pytest.fixture(autouse=True)
def reset_caches():
"""Clear all caches before each test."""
from app.cache import ticket_cache, customer_cache, agent_cache
ticket_cache.clear()
customer_cache.clear()
agent_cache.clear()
yield
# Optional cleanup after test
ticket_cache.clear()
customer_cache.clear()
agent_cache.clear()
@pytest.fixture(autouse=True, scope="session")
def configure_logging():
"""Configure logging for test session."""
import logging
logging.basicConfig(level=logging.WARNING)
logging.getLogger("sqlalchemy.engine").setLevel(logging.ERROR)
Fixture Dependencies and Composition
Fixtures can depend on other fixtures, creating dependency chains:
Create fixtures that return factory functions for flexible test data creation:
@pytest.fixture
def ticket_factory(db_session):
"""Factory for creating tickets with custom attributes."""
created_tickets = []
def _create_ticket(
title="Default Ticket",
description="Default description",
priority="medium",
status="open",
**kwargs
):
ticket = Ticket(
title=title,
description=description,
priority=priority,
status=status,
**kwargs
)
db_session.add(ticket)
db_session.commit()
created_tickets.append(ticket)
return ticket
yield _create_ticket
# Cleanup: delete all created tickets
for ticket in created_tickets:
db_session.delete(ticket)
db_session.commit()
def test_multiple_tickets(ticket_factory):
"""Test using factory to create multiple tickets."""
high_priority = ticket_factory(priority="high", title="Urgent Issue")
low_priority = ticket_factory(priority="low", title="Minor Request")
assert high_priority.priority == "high"
assert low_priority.priority == "low"
2. Parametrization
Parametrization allows running the same test with different inputs, essential for comprehensive testing of customer support systems with various priority levels, statuses, user roles, and edge cases.
Mocking is essential for isolating units of code from external dependencies like email services, payment gateways, CRM systems, and third-party APIs. The pytest-mock plugin provides a clean interface to Python's unittest.mock.
Basic Mocking
def test_ticket_notification(mocker, db_session):
"""Test that creating a ticket sends email notification."""
# Mock the email service
mock_send = mocker.patch('app.services.email.EmailService.send_email')
mock_send.return_value = {"status": "sent", "message_id": "msg_123"}
# Create ticket
from app.services.ticket import TicketService
service = TicketService(db_session)
ticket = service.create_ticket(
title="Login Issue",
description="Cannot access account",
customer_email="customer@example.com",
priority="high"
)
# Verify email was sent
mock_send.assert_called_once()
call_args = mock_send.call_args
assert call_args.kwargs['to'] == "customer@example.com"
assert "Login Issue" in call_args.kwargs['subject']
assert ticket.id is not None
import pytest
from unittest.mock import AsyncMock
@pytest.mark.asyncio
async def test_async_notification_service(mocker):
"""Test async notification service with proper async mocking."""
mock_send = mocker.patch(
'app.services.notification.send_push_notification',
new_callable=AsyncMock
)
mock_send.return_value = {
"delivered": True,
"notification_id": "notif_789"
}
from app.services.notification import NotificationService
service = NotificationService()
result = await service.notify_agent(
agent_id=456,
message="New high-priority ticket assigned to you",
priority="high"
)
assert result["delivered"] is True
assert result["notification_id"] is not None
mock_send.assert_awaited_once()
# Verify call arguments
call_kwargs = mock_send.call_args.kwargs
assert call_kwargs["agent_id"] == 456
assert "high-priority" in call_kwargs["message"]
4. Coverage Reporting
Code coverage measures what percentage of your code is executed during testing. For customer support systems, comprehensive coverage ensures reliability and helps identify untested edge cases.
Basic Coverage Setup
# Install pytest-cov
pip install pytest-cov
# Run tests with coverage
pytest --cov=app tests/
# Generate HTML report
pytest --cov=app --cov-report=html tests/
# Show lines not covered
pytest --cov=app --cov-report=term-missing tests/
# Fail if coverage below threshold
pytest --cov=app --cov-fail-under=80 tests/
# Run with branch coverage
pytest --cov=app --cov-branch --cov-report=html
5. Async Testing
Modern customer support systems use asynchronous operations for better performance and scalability. Testing async code requires special handling to properly await coroutines.
# pytest.ini or pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto" # Automatically detect and run async tests
# Or in pytest.ini
[pytest]
asyncio_mode = auto
6. FastAPI Testing
FastAPI is widely used for customer support APIs. Testing endpoints ensures proper request/response handling, validation, authentication, and business logic.
Basic FastAPI Testing
from fastapi.testclient import TestClient
from app.main import app
@pytest.fixture
def client():
"""Provide FastAPI test client."""
return TestClient(app)
def test_create_ticket_endpoint(client):
"""Test ticket creation via POST endpoint."""
response = client.post(
"/api/v1/tickets",
json={
"title": "Cannot login to application",
"description": "User receives 'invalid credentials' error",
"priority": "high",
"customer_email": "frustrated@example.com",
"category": "authentication"
}
)
assert response.status_code == 201
data = response.json()
assert data["title"] == "Cannot login to application"
assert data["priority"] == "high"
assert "id" in data
assert "created_at" in data
def test_get_ticket_by_id(client, sample_tickets):
"""Test retrieving specific ticket by ID."""
ticket_id = sample_tickets[0].id
response = client.get(f"/api/v1/tickets/{ticket_id}")
assert response.status_code == 200
data = response.json()
assert data["id"] == ticket_id
assert data["title"] == sample_tickets[0].title
def test_update_ticket_status(client, sample_tickets):
"""Test updating ticket status via PATCH."""
ticket_id = sample_tickets[0].id
response = client.patch(
f"/api/v1/tickets/{ticket_id}",
json={"status": "in_progress"}
)
assert response.status_code == 200
data = response.json()
assert data["status"] == "in_progress"
assert data["id"] == ticket_id
Dependency Override
FastAPI's dependency injection can be overridden for testing:
@pytest.fixture
def customer_factory(db_session):
"""Factory for creating customers with custom attributes."""
created = []
def _create_customer(
email=None,
name=None,
tier="basic",
**kwargs
):
import uuid
customer = Customer(
email=email or f"customer{uuid.uuid4().hex[:8]}@example.com",
name=name or f"Customer {uuid.uuid4().hex[:8]}",
tier=tier,
**kwargs
)
db_session.add(customer)
db_session.commit()
created.append(customer)
return customer
yield _create_customer
# Cleanup
for customer in created:
db_session.delete(customer)
db_session.commit()
@pytest.fixture
def ticket_factory(db_session, customer_factory):
"""Factory for creating tickets with automatic customer creation."""
created = []
def _create_ticket(
title=None,
customer=None,
priority="medium",
status="open",
**kwargs
):
import uuid
if customer is None:
customer = customer_factory()
ticket = Ticket(
title=title or f"Ticket {uuid.uuid4().hex[:8]}",
description=kwargs.pop("description", "Test ticket description"),
customer_id=customer.id,
priority=priority,
status=status,
**kwargs
)
db_session.add(ticket)
db_session.commit()
created.append(ticket)
return ticket
yield _create_ticket
# Cleanup
for ticket in created:
db_session.delete(ticket)
db_session.commit()
# Usage in tests
def test_with_factories(customer_factory, ticket_factory):
"""Test using factory fixtures for flexible test data creation."""
# Create premium customer
premium_customer = customer_factory(tier="enterprise", name="Big Corp")
# Create multiple tickets for same customer
critical_ticket = ticket_factory(
customer=premium_customer,
priority="critical",
title="Production outage"
)
normal_ticket = ticket_factory(
customer=premium_customer,
priority="low",
title="Feature request"
)
assert critical_ticket.customer_id == premium_customer.id
assert normal_ticket.customer_id == premium_customer.id
assert premium_customer.tier == "enterprise"
9. Customer Support Testing Scenarios
Real-world testing scenarios specific to customer support systems.
Testing Ticket Assignment Logic
def test_round_robin_assignment(db_session, sample_agents):
"""Test round-robin ticket assignment to available agents."""
from app.services.assignment import assign_ticket_round_robin
tickets = []
for i in range(9):
ticket = Ticket(
title=f"Assignment Test {i}",
priority="medium",
status="open"
)
db_session.add(ticket)
db_session.flush()
assign_ticket_round_robin(ticket, db_session)
tickets.append(ticket)
db_session.commit()
# Verify even distribution
for agent in sample_agents:
assigned_count = len([t for t in tickets if t.assigned_agent_id == agent.id])
assert assigned_count == 3 # 9 tickets / 3 agents = 3 each
def test_priority_based_assignment(db_session, sample_agents):
"""Test that critical tickets go to senior agents."""
from app.services.assignment import assign_ticket_by_priority
critical_ticket = Ticket(
title="Critical Issue",
priority="critical",
status="open"
)
db_session.add(critical_ticket)
db_session.flush()
assign_ticket_by_priority(critical_ticket, db_session)
db_session.commit()
# Find senior agent
senior_agent = next(a for a in sample_agents if a.role == "senior_agent")
assert critical_ticket.assigned_agent_id == senior_agent.id
Testing SLA Compliance
from datetime import datetime, timedelta
def test_sla_breach_detection(db_session):
"""Test detection of SLA breaches for different priorities."""
from app.services.sla import find_sla_breaches
now = datetime.utcnow()
# Critical ticket created 2 hours ago (SLA: 1 hour) - BREACHED
breached_critical = Ticket(
title="Critical - Breached",
priority="critical",
status="open",
created_at=now - timedelta(hours=2)
)
# High priority created 5 hours ago (SLA: 4 hours) - BREACHED
breached_high = Ticket(
title="High - Breached",
priority="high",
status="open",
created_at=now - timedelta(hours=5)
)
# Medium priority created 2 hours ago (SLA: 24 hours) - OK
ok_medium = Ticket(
title="Medium - OK",
priority="medium",
status="open",
created_at=now - timedelta(hours=2)
)
# Critical ticket but already assigned (not breached)
assigned_critical = Ticket(
title="Critical - Assigned",
priority="critical",
status="in_progress",
created_at=now - timedelta(hours=2),
assigned_agent_id=1
)
db_session.add_all([breached_critical, breached_high, ok_medium, assigned_critical])
db_session.commit()
breaches = find_sla_breaches(db_session)
breach_titles = [b.title for b in breaches]
assert len(breaches) == 2
assert "Critical - Breached" in breach_titles
assert "High - Breached" in breach_titles
assert "Medium - OK" not in breach_titles
assert "Critical - Assigned" not in breach_titles
def test_sla_time_remaining_calculation(db_session):
"""Test calculation of remaining SLA time."""
from app.services.sla import calculate_sla_remaining
now = datetime.utcnow()
# High priority ticket created 1 hour ago (SLA: 4 hours)
ticket = Ticket(
title="SLA Test",
priority="high",
status="open",
created_at=now - timedelta(hours=1)
)
db_session.add(ticket)
db_session.commit()
remaining = calculate_sla_remaining(ticket)
# Should have ~3 hours remaining (4 hour SLA - 1 hour elapsed)
assert 2.9 <= remaining.total_seconds() / 3600 <= 3.1
Testing Escalation Rules
def test_automatic_escalation(db_session):
"""Test automatic ticket escalation for overdue tickets."""
from app.services.escalation import escalate_overdue_tickets
now = datetime.utcnow()
# Medium priority ticket open for 3 days without assignment
overdue_ticket = Ticket(
title="Overdue Ticket",
priority="medium",
status="open",
created_at=now - timedelta(days=3)
)
# Recent ticket (should not escalate)
recent_ticket = Ticket(
title="Recent Ticket",
priority="medium",
status="open",
created_at=now - timedelta(hours=2)
)
db_session.add_all([overdue_ticket, recent_ticket])
db_session.commit()
# Run escalation logic
escalated_count = escalate_overdue_tickets(db_session)
db_session.refresh(overdue_ticket)
db_session.refresh(recent_ticket)
assert escalated_count == 1
assert overdue_ticket.priority == "high" # Escalated from medium
assert overdue_ticket.escalated is True
assert recent_ticket.priority == "medium" # Unchanged
assert recent_ticket.escalated is False
Testing Customer Notifications
def test_status_change_notification(mocker, db_session):
"""Test email notification sent when ticket status changes."""
from app.services.ticket import update_ticket_status
mock_send_email = mocker.patch('app.services.email.send_email')
mock_send_email.return_value = {"status": "sent"}
ticket = Ticket(
title="Notification Test",
customer_email="customer@example.com",
status="open"
)
db_session.add(ticket)
db_session.commit()
# Update status
update_ticket_status(ticket.id, "resolved", db_session)
# Verify notification sent
mock_send_email.assert_called_once()
call_kwargs = mock_send_email.call_args.kwargs
assert call_kwargs["to"] == "customer@example.com"
assert "resolved" in call_kwargs["template_name"].lower()
assert call_kwargs["template_data"]["ticket_id"] == ticket.id
def test_assignment_notification(mocker, db_session, sample_agents):
"""Test notification sent to agent when ticket assigned."""
from app.services.ticket import assign_ticket
mock_notify = mocker.patch('app.services.notification.notify_agent')
ticket = Ticket(title="Assignment Notification Test", status="open")
db_session.add(ticket)
db_session.commit()
agent = sample_agents[0]
assign_ticket(ticket.id, agent.id, db_session)
mock_notify.assert_called_once()
call_args = mock_notify.call_args
assert call_args[0][0] == agent.id # First arg is agent_id
assert "assigned" in call_args[0][1].lower() # Second arg is message
10. Integration Patterns
Testing integrations with external services and systems.
# Good - clear, specific test names
def test_high_priority_ticket_sends_immediate_notification_to_assigned_agent():
pass
def test_ticket_auto_closes_after_7_days_without_customer_response():
pass
def test_sla_breach_alert_sent_to_team_lead_when_critical_ticket_unassigned_for_1_hour():
pass
# Bad - vague test names
def test_ticket_notification():
pass
def test_auto_close():
pass
def test_sla():
pass
AAA Pattern (Arrange-Act-Assert)
def test_ticket_assignment():
# Arrange: Set up test data and conditions
agent = create_agent(name="Test Agent", role="agent")
ticket = create_ticket(title="Test", priority="high", status="open")
# Act: Perform the action being tested
result = assign_ticket(ticket.id, agent.id)
# Assert: Verify expected outcomes
assert result.assigned_agent_id == agent.id
assert result.status == "assigned"
assert result.assigned_at is not None
Test Isolation
# Each test should be independent and not rely on others
# Bad - tests depend on execution order
def test_create_user():
create_user("test@example.com")
def test_get_user():
user = get_user("test@example.com") # Relies on test_create_user
assert user is not None
# Good - each test is self-contained
@pytest.fixture
def test_user(db_session):
user = User(email="test@example.com")
db_session.add(user)
db_session.commit()
return user
def test_get_user_by_email(db_session, test_user):
found_user = get_user(test_user.email, db_session)
assert found_user.email == test_user.email
Testing Guidelines
Test one thing per test:
# Good
def test_ticket_creation_sets_default_status():
ticket = create_ticket(title="Test")
assert ticket.status == "open"
def test_ticket_creation_sets_created_timestamp():
ticket = create_ticket(title="Test")
assert ticket.created_at is not None
# Bad
def test_ticket_creation():
ticket = create_ticket(title="Test")
assert ticket.status == "open"
assert ticket.created_at is not None
assert ticket.priority == "medium"
assert ticket.category is None
Use fixtures for setup:
# Good
@pytest.fixture
def authenticated_user(db_session):
user = User(email="auth@example.com", role="agent")
db_session.add(user)
db_session.commit()
return user
def test_authenticated_endpoint(authenticated_user, api_client):
response = api_client.get("/api/tickets", user=authenticated_user)
assert response.status_code == 200
# Less ideal - setup in test
def test_authenticated_endpoint(api_client):
user = User(email="auth@example.com", role="agent")
# ... more setup ...
response = api_client.get("/api/tickets", user=user)
assert response.status_code == 200
Mock external dependencies:
def test_email_notification_on_ticket_creation(mocker):
"""Always mock external services like email."""
mock_send = mocker.patch('app.services.email.send_email')
mock_send.return_value = {"status": "sent"}
ticket = create_ticket(title="Test", customer_email="user@example.com")
mock_send.assert_called_once()
assert ticket.id is not None
12. Common Pitfalls and Solutions
Problem: Database State Leakage
Tests affecting each other due to shared database state.
Solution: Use transaction rollback in fixtures.
@pytest.fixture
def db_session(db_engine):
"""Isolate database state per test with rollback."""
connection = db_engine.connect()
transaction = connection.begin()
session = sessionmaker(bind=connection)()
yield session
session.close()
transaction.rollback() # Rollback all changes
connection.close()
Problem: Async Test Hanging
Forgetting to mark async tests or improper event loop handling.
Solution: Always use @pytest.mark.asyncio and await coroutines.
# Wrong - test will hang
async def test_async_operation():
result = await async_function()
assert result is not None
# Correct
@pytest.mark.asyncio
async def test_async_operation():
result = await async_function()
assert result is not None
Problem: Fixture Scope Issues
Session-scoped fixtures with mutable state causing test pollution.
Solution: Use appropriate scope and reset state.
# Problematic
@pytest.fixture(scope="session")
def user_cache():
return {} # Shared dict across all tests
# Better
@pytest.fixture(scope="function")
def user_cache():
return {} # New dict per test
# Or reset state
@pytest.fixture(scope="session")
def user_cache():
cache = {}
yield cache
cache.clear() # Clean up after each test
Problem: Over-Mocking
Mocking too much makes tests meaningless.
Solution: Only mock external boundaries.
# Over-mocked - testing nothing real
def test_ticket_creation(mocker):
mocker.patch('app.services.create_ticket', return_value=Ticket(id=1))
ticket = create_ticket(title="Test")
assert ticket.id == 1 # Just testing the mock
# Better - test actual logic
def test_ticket_creation(db_session):
ticket = create_ticket_service(db_session, title="Test", priority="high")
assert ticket.id is not None
assert ticket.status == "open"
assert ticket.priority == "high"
Problem: Slow Test Suite
Tests taking too long to run.
Solution: Use appropriate markers and parallel execution.
# Run only fast tests
pytest -m "not slow"
# Run tests in parallel
pytest -n auto # Requires pytest-xdist
# Run specific test file
pytest tests/unit/test_validators.py -v
13. PostgreSQL Integration
Testing with Real PostgreSQL
@pytest.fixture(scope="session")
def postgres_engine():
"""Create PostgreSQL engine with test database."""
from sqlalchemy import create_engine
# Connect to default database to create test database
admin_url = "postgresql://postgres:password@localhost/postgres"
admin_engine = create_engine(admin_url, isolation_level="AUTOCOMMIT")
with admin_engine.connect() as conn:
# Drop and recreate test database
conn.execute("DROP DATABASE IF EXISTS support_test")
conn.execute("CREATE DATABASE support_test")
# Connect to test database
test_url = "postgresql://postgres:password@localhost/support_test"
test_engine = create_engine(test_url)
from app.database import Base
Base.metadata.create_all(test_engine)
yield test_engine
# Cleanup
test_engine.dispose()
with admin_engine.connect() as conn:
conn.execute("DROP DATABASE support_test")
admin_engine.dispose()
Testing PostgreSQL-Specific Features
def test_full_text_search(db_session):
"""Test PostgreSQL full-text search on tickets."""
from sqlalchemy import func
tickets = [
Ticket(title="Cannot login", description="Password reset not working"),
Ticket(title="Slow performance", description="Dashboard loading slowly"),
Ticket(title="Email issues", description="Not receiving notifications"),
]
db_session.add_all(tickets)
db_session.commit()
# Search using PostgreSQL full-text search
search_term = "login password"
results = db_session.query(Ticket).filter(
func.to_tsvector('english', Ticket.title + ' ' + Ticket.description).match(search_term)
).all()
assert len(results) >= 1
assert any("login" in t.title.lower() for t in results)
def test_jsonb_field_operations(db_session):
"""Test JSONB field operations in PostgreSQL."""
from sqlalchemy.dialects.postgresql import JSONB
ticket = Ticket(
title="JSONB Test",
metadata_={"tags": ["urgent", "billing"], "source": "email"}
)
db_session.add(ticket)
db_session.commit()
# Query using JSONB operations
from sqlalchemy import cast
results = db_session.query(Ticket).filter(
Ticket.metadata_['tags'].astext.contains('urgent')
).all()
assert len(results) >= 1
assert results[0].metadata_["tags"] == ["urgent", "billing"]
