Project-Specific Testing Configuration

MANDATORY: Read this file BEFORE writing ANY test.

This document provides project-specific testing patterns, authentication requirements, fixtures, and common pitfalls for Chaverim ALPR Platform.

---

Table of Contents

1. Authentication in Tests
2. Required Fixtures
3. Database Testing Patterns
4. Mock/Stub Patterns
5. Component-Specific Patterns
6. Common Pitfalls
7. Test Examples

---

Authentication in Tests

Test User Roles

Define all user roles used in the application:

Roles: - admin - Full system access (user management, collector management, all features) - operator - Standard access (search, alerts, watchlists, read-only collector status) - viewer - Read-only access (search, view detections, no alert management)

Authentication Fixtures

Backend Example:

# tests/fixtures/auth_fixtures.py
import pytest
from src.models import User
from src.services.auth import create_access_token

@pytest.fixture
def admin_user(db_session):
    """Create admin user for testing."""
    user = User(
        email="admin@test.com",
        name="Admin User",
        role="admin",
        password_hash=hash_password("adminpass")
    )
    db_session.add(user)
    db_session.commit()
    return user

@pytest.fixture
def regular_user(db_session):
    """Create regular user for testing."""
    user = User(
        email="user@test.com",
        name="Regular User",
        role="user",
        password_hash=hash_password("userpass")
    )
    db_session.add(user)
    db_session.commit()
    return user

@pytest.fixture
def admin_token(admin_user):
    """Generate auth token for admin user."""
    return create_access_token(user_id=admin_user.id)

@pytest.fixture
def user_token(regular_user):
    """Generate auth token for regular user."""
    return create_access_token(user_id=regular_user.id)

Testing Authenticated Endpoints

Pattern:

def test_admin_only_endpoint(client, admin_token, regular_user):
    """Test that only admins can access admin endpoint."""

    # Should succeed with admin token
    response = client.get(
        "/api/admin/users",
        headers={"Authorization": f"Bearer {admin_token}"}
    )
    assert response.status_code == 200

    # Should fail with user token
    user_token = create_access_token(user_id=regular_user.id)
    response = client.get(
        "/api/admin/users",
        headers={"Authorization": f"Bearer {user_token}"}
    )
    assert response.status_code == 403

    # Should fail without token
    response = client.get("/api/admin/users")
    assert response.status_code == 401

---

Required Fixtures

Database Fixtures

Session Fixture:

@pytest.fixture(scope="function")
def db_session():
    """
    Provide a database session with automatic rollback.
    Scope: function - creates new session for each test.
    """
    connection = engine.connect()
    transaction = connection.begin()
    session = SessionLocal(bind=connection)

    yield session

    session.close()
    transaction.rollback()
    connection.close()

Client Fixture (FastAPI example):

@pytest.fixture
def client(db_session):
    """Provide test client with database session."""
    from fastapi.testclient import TestClient
    from src.main import app

    # Override database dependency
    def override_get_db():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db

    with TestClient(app) as test_client:
        yield test_client

    app.dependency_overrides.clear()

ALPR-Specific Fixtures

Collector Fixture:

@pytest.fixture
def sample_collector(db_session):
    """Create a sample edge collector for testing."""
    collector = Collector(
        id="test-collector-001",
        name="Test Location",
        api_key="alpr_test_" + secrets.token_urlsafe(32),
        is_active=True,
        last_heartbeat=datetime.utcnow()
    )
    db_session.add(collector)
    db_session.commit()
    return collector

@pytest.fixture
def collector_api_key(sample_collector):
    """Get API key for collector authentication."""
    return sample_collector.api_key

detection Fixture:

@pytest.fixture
def sample_detection(db_session, sample_collector):
    """Create a sample LPR detection for testing."""
    detection = Detection(
        detection_id=str(uuid.uuid4()),
        collector_id=sample_collector.id,
        timestamp=datetime.utcnow(),
        plate_number="ABC1234",
        plate_confidence=0.95,
        camera_id="cam-001",
        camera_mac="08a189ff6e1a"
    )
    db_session.add(detection)
    db_session.commit()
    return detection

@pytest.fixture
def detection_batch(db_session, sample_collector):
    """Create multiple detections for testing pagination/search."""
    plates = ["ABC1234", "XYZ7890", "DEF5678", "GHI9012"]
    detections = []
    for i, plate in enumerate(plates):
        detection = Detection(
            detection_id=str(uuid.uuid4()),
            collector_id=sample_collector.id,
            timestamp=datetime.utcnow() - timedelta(hours=i),
            plate_number=plate,
            plate_confidence=0.9 + (i * 0.02),
            camera_id=f"cam-{i:03d}",
            camera_mac=f"08a189ff6e{i:02x}"
        )
        detections.append(detection)
    db_session.add_all(detections)
    db_session.commit()
    return detections

Watchlist Fixture:

@pytest.fixture
def sample_watchlist(db_session, admin_user):
    """Create a sample watchlist for testing."""
    watchlist = Watchlist(
        name="Test Watchlist",
        description="Watchlist for testing",
        created_by=admin_user.id
    )
    db_session.add(watchlist)
    db_session.commit()
    return watchlist

@pytest.fixture
def watchlist_entry(db_session, sample_watchlist):
    """Create a watchlist entry for testing alerts."""
    entry = WatchlistEntry(
        watchlist_id=sample_watchlist.id,
        plate_number="ABC1234",
        notes="Test entry",
        priority="high"
    )
    db_session.add(entry)
    db_session.commit()
    return entry

---

Database Testing Patterns

Pattern: Testing Transactions

Problem: Ensure multi-step database operations complete atomically.

Solution:

def test_order_creation_transaction(db_session, regular_user):
    """Test that order creation is transactional."""

    with pytest.raises(ValidationError):
        # This should rollback everything if validation fails
        with db_session.begin_nested():
            order = Order(user_id=regular_user.id, total=100)
            db_session.add(order)

            # This should fail validation
            item = OrderItem(order_id=order.id, quantity=-1)
            db_session.add(item)
            db_session.flush()  # Trigger validation

    # Verify nothing was committed
    assert db_session.query(Order).count() == 0

Pattern: Testing Constraints

Test database-level constraints:

def test_unique_constraint(db_session):
    """Test unique email constraint."""
    user1 = User(email="test@example.com", name="User 1")
    db_session.add(user1)
    db_session.commit()

    # Duplicate email should fail
    user2 = User(email="test@example.com", name="User 2")
    db_session.add(user2)

    with pytest.raises(IntegrityError):
        db_session.commit()

Pattern: Testing Cascades

Test foreign key cascades:

def test_delete_cascade(db_session, regular_user):
    """Test that deleting user deletes related orders."""
    order = Order(user_id=regular_user.id, total=100)
    db_session.add(order)
    db_session.commit()

    order_id = order.id

    # Delete user
    db_session.delete(regular_user)
    db_session.commit()

    # Verify order was also deleted (or nullified, depending on cascade)
    assert db_session.query(Order).filter_by(id=order_id).first() is None

---

Mock/Stub Patterns

External Service Mocks

[Example: Payment Processor Mock]

@pytest.fixture
def mock_payment_processor():
    """Mock payment processor for testing."""
    with patch('src.services.payment.PaymentProcessor') as mock:
        # Configure mock behavior
        mock.return_value.charge.return_value = {
            'success': True,
            'transaction_id': 'test_txn_12345',
            'amount': 99.99
        }
        yield mock

def test_order_payment(db_session, sample_order, mock_payment_processor):
    """Test order payment processing."""
    result = process_payment(sample_order)

    assert result['success'] is True
    mock_payment_processor.return_value.charge.assert_called_once_with(
        amount=sample_order.total,
        currency='USD'
    )

[Example: Email Service Mock]

@pytest.fixture
def mock_email_service():
    """Mock email service to prevent sending real emails."""
    with patch('src.services.email.EmailService') as mock:
        mock.return_value.send.return_value = True
        yield mock

def test_user_registration_sends_email(
    client, mock_email_service
):
    """Test that user registration sends welcome email."""
    response = client.post("/api/register", json={
        "email": "newuser@test.com",
        "password": "securepass123",
        "name": "New User"
    })

    assert response.status_code == 201
    mock_email_service.return_value.send.assert_called_once()

MinIO Image Storage Mock

@pytest.fixture
def mock_minio():
    """Mock MinIO client for image storage tests."""
    with patch('src.services.storage.MinioClient') as mock:
        mock.return_value.put_object.return_value = "images/test-uuid.jpg"
        mock.return_value.get_presigned_url.return_value = "https://minio/images/test-uuid.jpg?token=xxx"
        yield mock

def test_detection_image_storage(client, collector_api_key, mock_minio):
    """Test that detection images are stored in MinIO."""
    detection_data = {
        "detection_id": str(uuid.uuid4()),
        "plate": {"text": "ABC1234", "confidence": 0.95},
        "images": {"full": "base64-encoded-jpeg-data"}
    }
    response = client.post(
        "/api/v1/detections/batch",
        headers={"X-API-Key": collector_api_key},
        json={"detections": [detection_data]}
    )
    assert response.status_code == 201
    mock_minio.return_value.put_object.assert_called_once()

Command Queue Testing

def test_send_reboot_command(client, admin_token, sample_collector, db_session):
    """Test queuing a reboot command for a collector."""
    response = client.post(
        f"/api/v1/collectors/{sample_collector.id}/commands",
        headers={"Authorization": f"Bearer {admin_token}"},
        json={"cmd": "reboot", "delay_sec": 5}
    )
    assert response.status_code == 202

    # Verify command is queued
    cmd = db_session.query(PendingCommand).filter_by(
        collector_id=sample_collector.id
    ).first()
    assert cmd is not None
    assert cmd.cmd == "reboot"

def test_heartbeat_returns_pending_commands(client, collector_api_key, sample_collector, db_session):
    """Test that heartbeat returns pending commands."""
    # Queue a command
    cmd = PendingCommand(
        collector_id=sample_collector.id,
        cmd="config_update",
        payload={"cameras": []}
    )
    db_session.add(cmd)
    db_session.commit()

    # Send heartbeat
    response = client.post(
        "/api/v1/heartbeat",
        headers={"X-API-Key": collector_api_key},
        json={"uptime_sec": 100, "queue_depth": 0}
    )
    assert response.status_code == 200
    assert len(response.json()["pending_commands"]) == 1
    assert response.json()["pending_commands"][0]["cmd"] == "config_update"

def test_command_acknowledgment(client, collector_api_key, db_session, sample_collector):
    """Test that command acknowledgment updates status."""
    cmd = PendingCommand(
        id="cmd-123",
        collector_id=sample_collector.id,
        cmd="reboot"
    )
    db_session.add(cmd)
    db_session.commit()

    response = client.post(
        "/api/v1/commands/cmd-123/ack",
        headers={"X-API-Key": collector_api_key},
        json={"status": "success", "message": "Rebooting..."}
    )
    assert response.status_code == 200

    # Verify command is marked as acknowledged
    db_session.refresh(cmd)
    assert cmd.status == "acknowledged"

---

Component-Specific Patterns

Detection ingestion API

Required Setup:

@pytest.fixture
def event_ingestion_client(client, sample_collector):
    """Client configured for Detection ingestion tests."""
    return client, sample_collector.api_key

Common Test Patterns:

def test_detection_ingestion_valid(client, collector_api_key):
    """Test valid detection is accepted and stored."""
    detection = {
        "detection_id": str(uuid.uuid4()),
        "timestamp": datetime.utcnow().isoformat(),
        "camera": {"id": "cam-001", "mac": "08a189ff6e1a"},
        "plate": {"text": "ABC1234", "confidence": 0.95},
        "vehicle": {"direction": "entering"}
    }
    response = client.post(
        "/api/v1/detections",
        headers={"X-API-Key": collector_api_key},
        json={"detections": [detection]}
    )
    assert response.status_code == 201

def test_detection_ingestion_invalid_api_key(client):
    """Test that invalid API key is rejected."""
    response = client.post(
        "/api/v1/detections",
        headers={"X-API-Key": "invalid_key"},
        json={"detections": []}
    )
    assert response.status_code == 401

def test_detection_ingestion_inactive_collector(client, db_session, sample_collector):
    """Test that inactive collector is rejected."""
    sample_collector.is_active = False
    db_session.commit()

    response = client.post(
        "/api/v1/detections",
        headers={"X-API-Key": sample_collector.api_key},
        json={"detections": []}
    )
    assert response.status_code == 401

Gotchas: - Always normalize plate numbers before comparison in tests - Detection timestamps must be valid ISO 8601 format - Test both single detection and batch (up to 100 detections)

---

Plate Search API

Required Setup:

@pytest.fixture
def searchable_detections(detection_batch):
    """Ensure detection batch is indexed for search tests."""
    return detection_batch

Common Test Patterns:

def test_plate_search_exact_match(client, user_token, searchable_detections):
    """Test exact plate number search."""
    response = client.get(
        "/api/v1/search?plate=ABC1234",
        headers={"Authorization": f"Bearer {user_token}"}
    )
    assert response.status_code == 200
    assert len(response.json()["data"]) >= 1
    assert response.json()["data"][0]["plate_number"] == "ABC1234"

def test_plate_search_partial_match(client, user_token, searchable_detections):
    """Test partial plate number search."""
    response = client.get(
        "/api/v1/search?plate=ABC&partial=true",
        headers={"Authorization": f"Bearer {user_token}"}
    )
    assert response.status_code == 200
    # Should find ABC1234
    plates = [d["plate_number"] for d in response.json()["data"]]
    assert any("ABC" in p for p in plates)

def test_plate_search_date_range(client, user_token, searchable_detections):
    """Test search with date range filter."""
    yesterday = (datetime.utcnow() - timedelta(days=1)).isoformat()
    tomorrow = (datetime.utcnow() + timedelta(days=1)).isoformat()

    response = client.get(
        f"/api/v1/search?start={yesterday}&end={tomorrow}",
        headers={"Authorization": f"Bearer {user_token}"}
    )
    assert response.status_code == 200

Gotchas: - Search is case-insensitive - tests should verify this - Pagination defaults to 20 results - test with larger datasets - Date filtering uses UTC timestamps

---

Alerting System

Required Setup:

@pytest.fixture
def alert_setup(db_session, sample_watchlist, watchlist_entry, regular_user):
    """Setup for alert testing with subscription."""
    subscription = AlertSubscription(
        user_id=regular_user.id,
        watchlist_id=sample_watchlist.id,
        is_active=True
    )
    db_session.add(subscription)
    db_session.commit()
    return sample_watchlist, watchlist_entry, subscription

Common Test Patterns:

def test_alert_triggered_on_watchlist_match(
    db_session, alert_setup, sample_collector, collector_api_key, client
):
    """Test that alert is created when plate matches watchlist."""
    watchlist, entry, subscription = alert_setup

    # Ingest detection with matching plate
    detection = {
        "detection_id": str(uuid.uuid4()),
        "timestamp": datetime.utcnow().isoformat(),
        "plate": {"text": entry.plate_number, "confidence": 0.95}
    }
    response = client.post(
        "/api/v1/detections",
        headers={"X-API-Key": collector_api_key},
        json={"detections": [detection]}
    )
    assert response.status_code == 201

    # Verify alert was created
    alert = db_session.query(Alert).filter_by(
        plate_number=entry.plate_number
    ).first()
    assert alert is not None
    assert alert.watchlist_id == watchlist.id

def test_alert_websocket_notification(
    client, user_token, regular_user, alert_setup
):
    """Test that subscribed user receives WebSocket notification."""
    # This requires WebSocket test client
    with client.websocket_connect(
        f"/ws/alerts?token={user_token}"
    ) as websocket:
        # Trigger alert (via separate request)
        # ...
        data = websocket.receive_json()
        assert data["type"] == "alert"

Gotchas: - Alert matching is async - may need small delay in tests - WebSocket tests require special handling - Test both active and inactive subscriptions

---

Common Pitfalls

❌ Pitfall: Not Rolling Back Test Data

Problem: Tests leave data in database, causing failures in other tests.

Symptoms: - Tests pass individually but fail when run together - "Unique constraint violation" errors - Unexpected data in tests

Solution:

# Use function-scoped fixtures that rollback
@pytest.fixture(scope="function")
def db_session():
    # ... setup ...
    yield session
    session.rollback()  # Ensure rollback
    session.close()

---

❌ Pitfall: Testing Implementation Instead of Behavior

Problem: Tests break when refactoring even though behavior unchanged.

Bad Example:

def test_user_service():
    service = UserService()
    # Testing internal implementation
    assert service._internal_cache is not None
    assert service._validate_email("test@test.com") is True

Good Example:

def test_user_service_creates_user():
    service = UserService()
    # Testing public behavior
    user = service.create_user("test@test.com", "Test User")
    assert user.email == "test@test.com"
    assert user.name == "Test User"

---

❌ Pitfall: Async Test Timing Issues

Problem: Tests fail intermittently due to timing.

Symptoms: - Tests pass locally but fail in CI - Random failures - "Element not found" errors in E2E tests

Solution:

# Use explicit waits, not sleep
import asyncio
from tenacity import retry, stop_after_attempt, wait_fixed

@retry(stop=stop_after_attempt(3), wait=wait_fixed(0.1))
async def test_async_operation():
    result = await async_function()
    assert result is not None

---

❌ Pitfall: Not Mocking External Services

Problem: Tests make real API calls, causing slow tests or failures.

Symptoms: - Very slow test suite - Tests fail when offline - Hitting rate limits

Solution:

# Always mock external services
@patch('requests.get')
def test_external_api_call(mock_get):
    mock_get.return_value.json.return_value = {"data": "test"}
    result = fetch_external_data()
    assert result["data"] == "test"

---

❌ Pitfall: Not Normalizing Plate Numbers

Problem: Plate number comparisons fail due to inconsistent formatting.

Symptoms: - Tests pass with exact string but fail with minor variations - "ABC1234" doesn't match "abc 1234" or "ABC-1234"

Solution:

def normalize_plate(plate: str) -> str:
    """Always normalize plates before storage and comparison."""
    return plate.upper().replace(" ", "").replace("-", "")

def test_plate_search_normalized(client, user_token):
    """Test that search normalizes input."""
    # These should all find the same plate
    for query in ["ABC1234", "abc1234", "ABC 1234", "abc-1234"]:
        response = client.get(
            f"/api/v1/search?plate={query}",
            headers={"Authorization": f"Bearer {user_token}"}
        )
        assert response.status_code == 200

---

❌ Pitfall: Testing Edge Collector Without Mocking Network

Problem: Edge collector tests make real HTTP connections to central server.

Symptoms: - Tests hang waiting for connection - Tests fail when central server is not running - Flaky tests due to network conditions

Solution:

@pytest.fixture
def mock_central_connection():
    """Mock all central server connections for edge tests."""
    with patch('lpr_edge.core.uploader.httpx.AsyncClient') as http_mock:
        # Mock detection upload
        http_mock.return_value.post.return_value = Mock(
            status_code=201,
            json=lambda: {"status": "ok"}
        )
        # Mock heartbeat response
        http_mock.return_value.post.return_value = Mock(
            status_code=200,
            json=lambda: {"status": "ok", "pending_commands": []}
        )
        yield http_mock

---

Test Examples

Example: Full Integration Test

def test_user_registration_flow(client, db_session, mock_email_service):
    """
    Test complete user registration flow:
    1. Register new user
    2. Verify user created in database
    3. Verify welcome email sent
    4. Verify user can log in
    """

    # 1. Register user
    response = client.post("/api/register", json={
        "email": "newuser@test.com",
        "password": "SecurePass123!",
        "name": "New User"
    })
    assert response.status_code == 201
    data = response.json()
    assert "id" in data["data"]
    user_id = data["data"]["id"]

    # 2. Verify user in database
    user = db_session.query(User).filter_by(id=user_id).first()
    assert user is not None
    assert user.email == "newuser@test.com"
    assert user.name == "New User"

    # 3. Verify welcome email sent
    mock_email_service.return_value.send.assert_called_once_with(
        to="newuser@test.com",
        subject="Welcome to Our App",
        template="welcome"
    )

    # 4. Verify user can log in
    response = client.post("/api/login", json={
        "email": "newuser@test.com",
        "password": "SecurePass123!"
    })
    assert response.status_code == 200
    assert "access_token" in response.json()["data"]

---

Example: Testing Error Handling

def test_error_handling_invalid_input(client):
    """Test that API returns proper error for invalid input."""

    response = client.post("/api/users", json={
        "email": "invalid-email",  # Invalid format
        "name": ""  # Empty name
    })

    assert response.status_code == 400
    error = response.json()["error"]
    assert error["code"] == "VALIDATION_ERROR"
    assert "email" in error["details"]
    assert "name" in error["details"]

---

Example: Testing Permissions

def test_permissions_user_can_only_update_own_profile(
    client, regular_user, admin_user
):
    """Test that users can only update their own profile."""
    user_token = create_access_token(user_id=regular_user.id)

    # User can update own profile
    response = client.patch(
        f"/api/users/{regular_user.id}",
        headers={"Authorization": f"Bearer {user_token}"},
        json={"name": "Updated Name"}
    )
    assert response.status_code == 200

    # User cannot update other user's profile
    response = client.patch(
        f"/api/users/{admin_user.id}",
        headers={"Authorization": f"Bearer {user_token}"},
        json={"name": "Hacked Name"}
    )
    assert response.status_code == 403

---

Testing Checklist

Before writing tests for a new feature:

• [ ] Read this document (TEST_CONFIG.md)
• [ ] Read TEST_SETUP.md for infrastructure
• [ ] Check global.md for testing patterns
• [ ] Check component-specific PRP for testing guidance
• [ ] Identify required fixtures
• [ ] Identify services to mock
• [ ] Plan test coverage (unit, integration, E2E)
• [ ] Write tests for happy path
• [ ] Write tests for error cases
• [ ] Write tests for edge cases
• [ ] Write tests for permissions/authorization
• [ ] Verify tests are isolated (can run in any order)
• [ ] Verify tests clean up after themselves
• [ ] Run tests to ensure they pass
• [ ] Check coverage meets requirements

---

Updating This Document

When to update: - New authentication pattern emerges - New fixtures are created - Common pitfall is discovered - Component-specific patterns are established

How to update: - Add to appropriate section - Provide code examples - Explain why the pattern is important - Link to related documentation

---

Last Updated: 2025-12-29 Test Coverage: TBD - Update this field once tests are written. Target: 90%+ for business logic, 100% for critical paths. Maintained By: Development Team