diff --git a/.env.test b/.env.test
new file mode 100644
index 0000000..d1d3996
--- /dev/null
+++ b/.env.test
@@ -0,0 +1,21 @@
+# Database Test Environment Configuration
+# ========================================
+# This file is used by VS Code when running tests
+# Copy and modify as needed for your local setup
+
+# SQLite In-Memory (Default - No Setup Required)
+# Use this for quick local testing without external dependencies
+TEST_DATABASE_URL=sqlite+aiosqlite:///:memory:
+TEST_SYNC_DATABASE_URL=sqlite:///:memory:
+
+# PostgreSQL via Docker
+# Uncomment these lines if you want to test with PostgreSQL
+# First start the container: docker run -d --name nerdlearn-test-db -e POSTGRES_USER=test -e POSTGRES_PASSWORD=test -e POSTGRES_DB=nerdlearn_test -p 5433:5432 postgres:15
+# TEST_DATABASE_URL=postgresql+asyncpg://test:test@localhost:5433/nerdlearn_test
+# TEST_SYNC_DATABASE_URL=postgresql://test:test@localhost:5433/nerdlearn_test
+
+# Logging
+LOG_LEVEL=WARNING
+
+# Test Configuration
+PYTEST_TIMEOUT=60
diff --git a/.github/workflows/database-tests.yml b/.github/workflows/database-tests.yml
new file mode 100644
index 0000000..e0d6f46
--- /dev/null
+++ b/.github/workflows/database-tests.yml
@@ -0,0 +1,539 @@
+name: Database Tests & Reports
+
+on:
+ push:
+ branches: [main, develop]
+ paths:
+ - 'apps/api/app/models/**'
+ - 'packages/db/**'
+ - 'tests/database/**'
+ - '.github/workflows/database-tests.yml'
+ pull_request:
+ branches: [main, develop]
+ paths:
+ - 'apps/api/app/models/**'
+ - 'packages/db/**'
+ - 'tests/database/**'
+ schedule:
+ # Run daily at 2 AM UTC for regression detection
+ - cron: '0 2 * * *'
+ workflow_dispatch:
+ inputs:
+ run_benchmarks:
+ description: 'Run performance benchmarks'
+ required: false
+ default: 'true'
+ type: boolean
+ run_stress_tests:
+ description: 'Run stress/load tests'
+ required: false
+ default: 'false'
+ type: boolean
+
+env:
+ PYTHON_VERSION: '3.11'
+ NODE_VERSION: '20'
+
+jobs:
+ schema-integrity:
+ name: Schema Integrity Tests
+ runs-on: ubuntu-latest
+ services:
+ postgres:
+ image: postgres:15
+ env:
+ POSTGRES_USER: test
+ POSTGRES_PASSWORD: test
+ POSTGRES_DB: nerdlearn_test
+ ports:
+ - 5432:5432
+ options: >-
+ --health-cmd pg_isready
+ --health-interval 10s
+ --health-timeout 5s
+ --health-retries 5
+
+ steps:
+ - uses: actions/checkout@v4
+
+ - name: Set up Python
+ uses: actions/setup-python@v5
+ with:
+ python-version: ${{ env.PYTHON_VERSION }}
+
+ - name: Install dependencies
+ run: |
+ python -m pip install --upgrade pip
+ pip install -r apps/api/requirements.txt
+ pip install -r apps/api/requirements-test.txt
+ pip install pytest-html pytest-json-report
+
+ - name: Run schema integrity tests
+ env:
+ TEST_DATABASE_URL: postgresql+asyncpg://test:test@localhost:5432/nerdlearn_test
+ TEST_SYNC_DATABASE_URL: postgresql://test:test@localhost:5432/nerdlearn_test
+ run: |
+ pytest tests/database/test_schema_integrity.py \
+ -v \
+ --tb=short \
+ --json-report \
+ --json-report-file=reports/schema_tests.json \
+ --html=reports/schema_tests.html \
+ --self-contained-html
+
+ - name: Upload schema test reports
+ uses: actions/upload-artifact@v4
+ if: always()
+ with:
+ name: schema-test-reports
+ path: reports/
+ retention-days: 30
+
+ data-integrity:
+ name: Data Integrity Tests
+ runs-on: ubuntu-latest
+ needs: schema-integrity
+ services:
+ postgres:
+ image: postgres:15
+ env:
+ POSTGRES_USER: test
+ POSTGRES_PASSWORD: test
+ POSTGRES_DB: nerdlearn_test
+ ports:
+ - 5432:5432
+ options: >-
+ --health-cmd pg_isready
+ --health-interval 10s
+ --health-timeout 5s
+ --health-retries 5
+
+ steps:
+ - uses: actions/checkout@v4
+
+ - name: Set up Python
+ uses: actions/setup-python@v5
+ with:
+ python-version: ${{ env.PYTHON_VERSION }}
+
+ - name: Install dependencies
+ run: |
+ python -m pip install --upgrade pip
+ pip install -r apps/api/requirements.txt
+ pip install -r apps/api/requirements-test.txt
+ pip install pytest-html pytest-json-report
+
+ - name: Run data integrity tests
+ env:
+ TEST_DATABASE_URL: postgresql+asyncpg://test:test@localhost:5432/nerdlearn_test
+ TEST_SYNC_DATABASE_URL: postgresql://test:test@localhost:5432/nerdlearn_test
+ run: |
+ pytest tests/database/test_data_integrity.py \
+ -v \
+ --tb=short \
+ --json-report \
+ --json-report-file=reports/data_integrity_tests.json \
+ --html=reports/data_integrity_tests.html \
+ --self-contained-html
+
+ - name: Upload data integrity reports
+ uses: actions/upload-artifact@v4
+ if: always()
+ with:
+ name: data-integrity-reports
+ path: reports/
+ retention-days: 30
+
+ crud-operations:
+ name: CRUD Operations Tests
+ runs-on: ubuntu-latest
+ needs: schema-integrity
+ services:
+ postgres:
+ image: postgres:15
+ env:
+ POSTGRES_USER: test
+ POSTGRES_PASSWORD: test
+ POSTGRES_DB: nerdlearn_test
+ ports:
+ - 5432:5432
+ options: >-
+ --health-cmd pg_isready
+ --health-interval 10s
+ --health-timeout 5s
+ --health-retries 5
+
+ steps:
+ - uses: actions/checkout@v4
+
+ - name: Set up Python
+ uses: actions/setup-python@v5
+ with:
+ python-version: ${{ env.PYTHON_VERSION }}
+
+ - name: Install dependencies
+ run: |
+ python -m pip install --upgrade pip
+ pip install -r apps/api/requirements.txt
+ pip install -r apps/api/requirements-test.txt
+ pip install pytest-html pytest-json-report
+
+ - name: Run CRUD tests
+ env:
+ TEST_DATABASE_URL: postgresql+asyncpg://test:test@localhost:5432/nerdlearn_test
+ run: |
+ pytest tests/database/test_crud_operations.py \
+ -v \
+ --tb=short \
+ --json-report \
+ --json-report-file=reports/crud_tests.json \
+ --html=reports/crud_tests.html \
+ --self-contained-html
+
+ - name: Upload CRUD test reports
+ uses: actions/upload-artifact@v4
+ if: always()
+ with:
+ name: crud-test-reports
+ path: reports/
+ retention-days: 30
+
+ relationships:
+ name: Relationship & Cascade Tests
+ runs-on: ubuntu-latest
+ needs: [data-integrity, crud-operations]
+ services:
+ postgres:
+ image: postgres:15
+ env:
+ POSTGRES_USER: test
+ POSTGRES_PASSWORD: test
+ POSTGRES_DB: nerdlearn_test
+ ports:
+ - 5432:5432
+ options: >-
+ --health-cmd pg_isready
+ --health-interval 10s
+ --health-timeout 5s
+ --health-retries 5
+
+ steps:
+ - uses: actions/checkout@v4
+
+ - name: Set up Python
+ uses: actions/setup-python@v5
+ with:
+ python-version: ${{ env.PYTHON_VERSION }}
+
+ - name: Install dependencies
+ run: |
+ python -m pip install --upgrade pip
+ pip install -r apps/api/requirements.txt
+ pip install -r apps/api/requirements-test.txt
+ pip install pytest-html pytest-json-report
+
+ - name: Run relationship tests
+ env:
+ TEST_DATABASE_URL: postgresql+asyncpg://test:test@localhost:5432/nerdlearn_test
+ run: |
+ pytest tests/database/test_relationships.py \
+ -v \
+ --tb=short \
+ --json-report \
+ --json-report-file=reports/relationship_tests.json \
+ --html=reports/relationship_tests.html \
+ --self-contained-html
+
+ - name: Upload relationship test reports
+ uses: actions/upload-artifact@v4
+ if: always()
+ with:
+ name: relationship-test-reports
+ path: reports/
+ retention-days: 30
+
+ performance-benchmarks:
+ name: Performance Benchmarks
+ runs-on: ubuntu-latest
+ needs: relationships
+ if: ${{ github.event_name == 'schedule' || github.event.inputs.run_benchmarks == 'true' || github.event_name == 'push' }}
+ services:
+ postgres:
+ image: postgres:15
+ env:
+ POSTGRES_USER: test
+ POSTGRES_PASSWORD: test
+ POSTGRES_DB: nerdlearn_test
+ ports:
+ - 5432:5432
+ options: >-
+ --health-cmd pg_isready
+ --health-interval 10s
+ --health-timeout 5s
+ --health-retries 5
+
+ steps:
+ - uses: actions/checkout@v4
+
+ - name: Set up Python
+ uses: actions/setup-python@v5
+ with:
+ python-version: ${{ env.PYTHON_VERSION }}
+
+ - name: Install dependencies
+ run: |
+ python -m pip install --upgrade pip
+ pip install -r apps/api/requirements.txt
+ pip install -r apps/api/requirements-test.txt
+ pip install pytest-html pytest-json-report pytest-benchmark
+
+ - name: Run performance benchmarks
+ env:
+ TEST_DATABASE_URL: postgresql+asyncpg://test:test@localhost:5432/nerdlearn_test
+ run: |
+ pytest tests/database/test_performance.py \
+ -v \
+ --tb=short \
+ --json-report \
+ --json-report-file=reports/performance_tests.json \
+ --html=reports/performance_tests.html \
+ --self-contained-html \
+ -m "benchmark or slow"
+
+ - name: Upload performance reports
+ uses: actions/upload-artifact@v4
+ if: always()
+ with:
+ name: performance-reports
+ path: reports/
+ retention-days: 90
+
+ - name: Comment benchmark results on PR
+ if: github.event_name == 'pull_request'
+ uses: actions/github-script@v7
+ with:
+ script: |
+ const fs = require('fs');
+ const report = JSON.parse(fs.readFileSync('reports/performance_tests.json', 'utf8'));
+
+ let comment = '## Performance Benchmark Results\n\n';
+ comment += '| Test | Duration | Status |\n';
+ comment += '|------|----------|--------|\n';
+
+ for (const test of report.tests || []) {
+ const status = test.outcome === 'passed' ? '✅' : '❌';
+ comment += `| ${test.nodeid.split('::').pop()} | ${test.call?.duration?.toFixed(4) || 'N/A'}s | ${status} |\n`;
+ }
+
+ github.rest.issues.createComment({
+ issue_number: context.issue.number,
+ owner: context.repo.owner,
+ repo: context.repo.repo,
+ body: comment
+ });
+
+ concurrency-tests:
+ name: Concurrency Tests
+ runs-on: ubuntu-latest
+ needs: relationships
+ if: ${{ github.event_name == 'schedule' || github.event.inputs.run_stress_tests == 'true' }}
+ services:
+ postgres:
+ image: postgres:15
+ env:
+ POSTGRES_USER: test
+ POSTGRES_PASSWORD: test
+ POSTGRES_DB: nerdlearn_test
+ ports:
+ - 5432:5432
+ options: >-
+ --health-cmd pg_isready
+ --health-interval 10s
+ --health-timeout 5s
+ --health-retries 5
+
+ steps:
+ - uses: actions/checkout@v4
+
+ - name: Set up Python
+ uses: actions/setup-python@v5
+ with:
+ python-version: ${{ env.PYTHON_VERSION }}
+
+ - name: Install dependencies
+ run: |
+ python -m pip install --upgrade pip
+ pip install -r apps/api/requirements.txt
+ pip install -r apps/api/requirements-test.txt
+ pip install pytest-html pytest-json-report
+
+ - name: Run concurrency tests
+ env:
+ TEST_DATABASE_URL: postgresql+asyncpg://test:test@localhost:5432/nerdlearn_test
+ run: |
+ pytest tests/database/test_concurrency.py \
+ -v \
+ --tb=short \
+ --json-report \
+ --json-report-file=reports/concurrency_tests.json \
+ --html=reports/concurrency_tests.html \
+ --self-contained-html
+
+ - name: Upload concurrency test reports
+ uses: actions/upload-artifact@v4
+ if: always()
+ with:
+ name: concurrency-reports
+ path: reports/
+ retention-days: 30
+
+ generate-summary-report:
+ name: Generate Summary Report
+ runs-on: ubuntu-latest
+ needs: [schema-integrity, data-integrity, crud-operations, relationships]
+ if: always()
+
+ steps:
+ - uses: actions/checkout@v4
+
+ - name: Download all reports
+ uses: actions/download-artifact@v4
+ with:
+ path: all-reports/
+
+ - name: Set up Python
+ uses: actions/setup-python@v5
+ with:
+ python-version: ${{ env.PYTHON_VERSION }}
+
+ - name: Generate combined report
+ run: |
+ python << 'EOF'
+ import json
+ import os
+ from datetime import datetime
+ from pathlib import Path
+
+ # Collect all JSON reports
+ all_tests = []
+ for report_dir in Path('all-reports').iterdir():
+ for json_file in report_dir.glob('*.json'):
+ try:
+ with open(json_file) as f:
+ data = json.load(f)
+ all_tests.extend(data.get('tests', []))
+ except Exception as e:
+ print(f"Error reading {json_file}: {e}")
+
+ # Calculate summary
+ total = len(all_tests)
+ passed = sum(1 for t in all_tests if t.get('outcome') == 'passed')
+ failed = sum(1 for t in all_tests if t.get('outcome') == 'failed')
+ skipped = sum(1 for t in all_tests if t.get('outcome') == 'skipped')
+
+ summary = {
+ 'timestamp': datetime.utcnow().isoformat(),
+ 'total_tests': total,
+ 'passed': passed,
+ 'failed': failed,
+ 'skipped': skipped,
+ 'pass_rate': (passed / total * 100) if total > 0 else 0,
+ 'tests': all_tests
+ }
+
+ os.makedirs('reports', exist_ok=True)
+ with open('reports/combined_summary.json', 'w') as f:
+ json.dump(summary, f, indent=2)
+
+ # Print summary
+ print(f"\n{'='*60}")
+ print("DATABASE TEST SUMMARY")
+ print(f"{'='*60}")
+ print(f"Total: {total}")
+ print(f"Passed: {passed}")
+ print(f"Failed: {failed}")
+ print(f"Skipped: {skipped}")
+ print(f"Pass Rate: {summary['pass_rate']:.1f}%")
+ print(f"{'='*60}")
+ EOF
+
+ - name: Upload combined report
+ uses: actions/upload-artifact@v4
+ with:
+ name: combined-database-report
+ path: reports/combined_summary.json
+ retention-days: 90
+
+ - name: Set job summary
+ run: |
+ python << 'EOF'
+ import json
+
+ with open('reports/combined_summary.json') as f:
+ summary = json.load(f)
+
+ md = f"""
+ ## Database Test Results
+
+ | Metric | Value |
+ |--------|-------|
+ | Total Tests | {summary['total_tests']} |
+ | Passed | {summary['passed']} |
+ | Failed | {summary['failed']} |
+ | Skipped | {summary['skipped']} |
+ | Pass Rate | {summary['pass_rate']:.1f}% |
+
+ ### Status: {'✅ All tests passed!' if summary['failed'] == 0 else '❌ Some tests failed'}
+ """
+
+ with open('$GITHUB_STEP_SUMMARY', 'a') as f:
+ f.write(md)
+ EOF
+
+ notify-on-failure:
+ name: Notify on Failure
+ runs-on: ubuntu-latest
+ needs: [schema-integrity, data-integrity, crud-operations, relationships]
+ if: failure() && github.event_name == 'schedule'
+
+ steps:
+ - name: Create issue on failure
+ uses: actions/github-script@v7
+ with:
+ script: |
+ const title = `Database Tests Failed - ${new Date().toISOString().split('T')[0]}`;
+ const body = `
+ ## Scheduled Database Tests Failed
+
+ The nightly database test run has failed.
+
+ **Workflow run:** ${context.serverUrl}/${context.repo.owner}/${context.repo.repo}/actions/runs/${context.runId}
+
+ Please investigate the failing tests and fix any issues.
+
+ ### Categories to check:
+ - Schema Integrity
+ - Data Integrity
+ - CRUD Operations
+ - Relationships & Cascades
+
+ cc: @database-team
+ `;
+
+ // Check for existing open issue
+ const issues = await github.rest.issues.listForRepo({
+ owner: context.repo.owner,
+ repo: context.repo.repo,
+ labels: 'database-tests,automated',
+ state: 'open'
+ });
+
+ if (issues.data.length === 0) {
+ await github.rest.issues.create({
+ owner: context.repo.owner,
+ repo: context.repo.repo,
+ title: title,
+ body: body,
+ labels: ['database-tests', 'automated', 'bug']
+ });
+ }
diff --git a/scripts/run_db_tests.sh b/scripts/run_db_tests.sh
new file mode 100755
index 0000000..005e33a
--- /dev/null
+++ b/scripts/run_db_tests.sh
@@ -0,0 +1,202 @@
+#!/bin/bash
+# Database Test Runner Script
+# Runs database tests with reporting and optionally starts required services
+
+set -e
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+PROJECT_ROOT="$(dirname "$SCRIPT_DIR")"
+REPORTS_DIR="$PROJECT_ROOT/reports/database"
+
+# Colors for output
+RED='\033[0;31m'
+GREEN='\033[0;32m'
+YELLOW='\033[1;33m'
+BLUE='\033[0;34m'
+NC='\033[0m' # No Color
+
+# Default values
+USE_DOCKER=false
+TEST_CATEGORY=""
+VERBOSE=false
+COVERAGE=false
+BENCHMARKS=false
+
+# Help message
+show_help() {
+ echo "Database Test Runner"
+ echo ""
+ echo "Usage: $0 [OPTIONS]"
+ echo ""
+ echo "Options:"
+ echo " -d, --docker Start PostgreSQL in Docker before running tests"
+ echo " -c, --category CAT Run specific test category:"
+ echo " schema, data, crud, relationships,"
+ echo " performance, concurrency, all (default)"
+ echo " -v, --verbose Verbose output"
+ echo " --coverage Run with coverage reporting"
+ echo " --benchmarks Include benchmark tests"
+ echo " -h, --help Show this help message"
+ echo ""
+ echo "Examples:"
+ echo " $0 -d -c schema # Run schema tests with Docker DB"
+ echo " $0 -c crud --coverage # Run CRUD tests with coverage"
+ echo " $0 --benchmarks # Run all tests including benchmarks"
+}
+
+# Parse arguments
+while [[ $# -gt 0 ]]; do
+ case $1 in
+ -d|--docker)
+ USE_DOCKER=true
+ shift
+ ;;
+ -c|--category)
+ TEST_CATEGORY="$2"
+ shift 2
+ ;;
+ -v|--verbose)
+ VERBOSE=true
+ shift
+ ;;
+ --coverage)
+ COVERAGE=true
+ shift
+ ;;
+ --benchmarks)
+ BENCHMARKS=true
+ shift
+ ;;
+ -h|--help)
+ show_help
+ exit 0
+ ;;
+ *)
+ echo -e "${RED}Unknown option: $1${NC}"
+ show_help
+ exit 1
+ ;;
+ esac
+done
+
+# Create reports directory
+mkdir -p "$REPORTS_DIR"
+
+# Start Docker PostgreSQL if requested
+if [ "$USE_DOCKER" = true ]; then
+ echo -e "${BLUE}Starting PostgreSQL container...${NC}"
+
+ docker run -d \
+ --name nerdlearn-test-db \
+ -e POSTGRES_USER=test \
+ -e POSTGRES_PASSWORD=test \
+ -e POSTGRES_DB=nerdlearn_test \
+ -p 5433:5432 \
+ postgres:15 \
+ || true # Ignore if already running
+
+ # Wait for PostgreSQL to be ready
+ echo -e "${YELLOW}Waiting for PostgreSQL to be ready...${NC}"
+ for i in {1..30}; do
+ if docker exec nerdlearn-test-db pg_isready -U test > /dev/null 2>&1; then
+ echo -e "${GREEN}PostgreSQL is ready!${NC}"
+ break
+ fi
+ sleep 1
+ done
+
+ export TEST_DATABASE_URL="postgresql+asyncpg://test:test@localhost:5433/nerdlearn_test"
+ export TEST_SYNC_DATABASE_URL="postgresql://test:test@localhost:5433/nerdlearn_test"
+fi
+
+# Build pytest command
+PYTEST_CMD="pytest"
+
+# Add test path based on category
+case $TEST_CATEGORY in
+ schema)
+ PYTEST_CMD="$PYTEST_CMD tests/database/test_schema_integrity.py"
+ ;;
+ data)
+ PYTEST_CMD="$PYTEST_CMD tests/database/test_data_integrity.py"
+ ;;
+ crud)
+ PYTEST_CMD="$PYTEST_CMD tests/database/test_crud_operations.py"
+ ;;
+ relationships)
+ PYTEST_CMD="$PYTEST_CMD tests/database/test_relationships.py"
+ ;;
+ performance)
+ PYTEST_CMD="$PYTEST_CMD tests/database/test_performance.py"
+ ;;
+ concurrency)
+ PYTEST_CMD="$PYTEST_CMD tests/database/test_concurrency.py"
+ ;;
+ all|"")
+ PYTEST_CMD="$PYTEST_CMD tests/database/"
+ ;;
+ *)
+ echo -e "${RED}Unknown category: $TEST_CATEGORY${NC}"
+ exit 1
+ ;;
+esac
+
+# Add options
+PYTEST_CMD="$PYTEST_CMD --tb=short"
+
+if [ "$VERBOSE" = true ]; then
+ PYTEST_CMD="$PYTEST_CMD -v"
+fi
+
+if [ "$COVERAGE" = true ]; then
+ PYTEST_CMD="$PYTEST_CMD --cov=app.models --cov-report=html:$REPORTS_DIR/coverage"
+fi
+
+if [ "$BENCHMARKS" = false ]; then
+ PYTEST_CMD="$PYTEST_CMD -m 'not benchmark'"
+fi
+
+# Add report generation
+TIMESTAMP=$(date +%Y%m%d_%H%M%S)
+PYTEST_CMD="$PYTEST_CMD --html=$REPORTS_DIR/report_$TIMESTAMP.html --self-contained-html"
+PYTEST_CMD="$PYTEST_CMD --json-report --json-report-file=$REPORTS_DIR/report_$TIMESTAMP.json"
+
+# Run tests
+echo -e "${BLUE}Running database tests...${NC}"
+echo -e "${YELLOW}Command: $PYTEST_CMD${NC}"
+echo ""
+
+cd "$PROJECT_ROOT"
+
+# Execute
+if eval "$PYTEST_CMD"; then
+ echo ""
+ echo -e "${GREEN}========================================${NC}"
+ echo -e "${GREEN}All tests passed!${NC}"
+ echo -e "${GREEN}========================================${NC}"
+ EXIT_CODE=0
+else
+ echo ""
+ echo -e "${RED}========================================${NC}"
+ echo -e "${RED}Some tests failed!${NC}"
+ echo -e "${RED}========================================${NC}"
+ EXIT_CODE=1
+fi
+
+echo ""
+echo -e "${BLUE}Reports saved to: $REPORTS_DIR${NC}"
+echo " - HTML: report_$TIMESTAMP.html"
+echo " - JSON: report_$TIMESTAMP.json"
+
+# Cleanup Docker if we started it
+if [ "$USE_DOCKER" = true ]; then
+ read -p "Stop and remove test database container? (y/n) " -n 1 -r
+ echo
+ if [[ $REPLY =~ ^[Yy]$ ]]; then
+ docker stop nerdlearn-test-db
+ docker rm nerdlearn-test-db
+ echo -e "${GREEN}Container removed${NC}"
+ fi
+fi
+
+exit $EXIT_CODE
diff --git a/tests/database/README.md b/tests/database/README.md
new file mode 100644
index 0000000..a6f98d6
--- /dev/null
+++ b/tests/database/README.md
@@ -0,0 +1,494 @@
+# Database Testing Guide for VS Code
+
+This guide explains how to run the NerdLearn database tests in VS Code, including setup, configuration, and interpreting results.
+
+## Table of Contents
+
+1. [Prerequisites](#prerequisites)
+2. [VS Code Setup](#vs-code-setup)
+3. [Running Tests](#running-tests)
+4. [Using the Test Runner Script](#using-the-test-runner-script)
+5. [VS Code Test Explorer](#vs-code-test-explorer)
+6. [Running with Docker](#running-with-docker)
+7. [Understanding Test Reports](#understanding-test-reports)
+8. [Troubleshooting](#troubleshooting)
+
+---
+
+## Prerequisites
+
+### 1. Python Environment
+
+Ensure you have Python 3.11+ installed:
+
+```bash
+python --version
+# Should show Python 3.11.x or higher
+```
+
+### 2. Install Python Dependencies
+
+From the project root, install the required packages:
+
+```bash
+# Install API dependencies
+pip install -r apps/api/requirements.txt
+
+# Install test dependencies
+pip install -r apps/api/requirements-test.txt
+
+# Install additional test packages
+pip install pytest-html pytest-json-report pytest-asyncio aiosqlite
+```
+
+If `requirements-test.txt` doesn't exist, install these manually:
+
+```bash
+pip install pytest pytest-asyncio pytest-cov pytest-html pytest-json-report aiosqlite
+```
+
+### 3. Database Options
+
+You have two options for running tests:
+
+#### Option A: In-Memory SQLite (Default - No Setup Required)
+Tests will automatically use SQLite in-memory database. This is the simplest option.
+
+#### Option B: PostgreSQL (Recommended for Full Testing)
+For tests that require PostgreSQL-specific features:
+
+```bash
+# Using Docker
+docker run -d \
+ --name nerdlearn-test-db \
+ -e POSTGRES_USER=test \
+ -e POSTGRES_PASSWORD=test \
+ -e POSTGRES_DB=nerdlearn_test \
+ -p 5433:5432 \
+ postgres:15
+
+# Set environment variable
+export TEST_DATABASE_URL="postgresql+asyncpg://test:test@localhost:5433/nerdlearn_test"
+```
+
+---
+
+## VS Code Setup
+
+### 1. Install Required Extensions
+
+Open VS Code and install these extensions:
+
+| Extension | ID | Purpose |
+|-----------|-----|---------|
+| Python | `ms-python.python` | Python language support |
+| Pylance | `ms-python.vscode-pylance` | Python IntelliSense |
+| Python Test Explorer | `littlefoxteam.vscode-python-test-adapter` | Visual test runner |
+
+To install from command line:
+```bash
+code --install-extension ms-python.python
+code --install-extension ms-python.vscode-pylance
+```
+
+### 2. Configure Python Interpreter
+
+1. Open Command Palette: `Ctrl+Shift+P` (or `Cmd+Shift+P` on Mac)
+2. Type: `Python: Select Interpreter`
+3. Choose your Python 3.11+ interpreter
+
+### 3. Configure pytest in VS Code
+
+Create or update `.vscode/settings.json`:
+
+```json
+{
+ "python.testing.pytestEnabled": true,
+ "python.testing.unittestEnabled": false,
+ "python.testing.pytestArgs": [
+ "tests/database",
+ "-v",
+ "--tb=short"
+ ],
+ "python.testing.cwd": "${workspaceFolder}",
+ "python.envFile": "${workspaceFolder}/.env.test"
+}
+```
+
+### 4. Create Test Environment File (Optional)
+
+Create `.env.test` in the project root:
+
+```bash
+# For SQLite (default)
+TEST_DATABASE_URL=sqlite+aiosqlite:///:memory:
+TEST_SYNC_DATABASE_URL=sqlite:///:memory:
+
+# For PostgreSQL (uncomment if using Docker)
+# TEST_DATABASE_URL=postgresql+asyncpg://test:test@localhost:5433/nerdlearn_test
+# TEST_SYNC_DATABASE_URL=postgresql://test:test@localhost:5433/nerdlearn_test
+```
+
+---
+
+## Running Tests
+
+### Method 1: VS Code Terminal
+
+Open the integrated terminal (`Ctrl+`` ` or `View > Terminal`) and run:
+
+```bash
+# Run all database tests
+pytest tests/database/ -v
+
+# Run specific test file
+pytest tests/database/test_schema_integrity.py -v
+
+# Run specific test class
+pytest tests/database/test_crud_operations.py::TestUserCRUD -v
+
+# Run specific test
+pytest tests/database/test_crud_operations.py::TestUserCRUD::test_create_user -v
+
+# Run with coverage
+pytest tests/database/ -v --cov=apps/api/app/models --cov-report=html
+
+# Run only fast tests (exclude benchmarks)
+pytest tests/database/ -v -m "not benchmark"
+
+# Run with HTML report
+pytest tests/database/ -v --html=reports/database/report.html --self-contained-html
+```
+
+### Method 2: VS Code Test Explorer UI
+
+1. Click the **Testing** icon in the Activity Bar (flask icon on left sidebar)
+2. Click **Refresh** to discover tests
+3. You'll see a tree of all database tests organized by file and class
+4. Click the **Play** button next to any test, class, or file to run it
+5. Green checkmarks = passed, red X = failed
+
+
+
+### Method 3: Run Tests from Editor
+
+When viewing a test file:
+1. You'll see **Run Test | Debug Test** links above each test function
+2. Click **Run Test** to execute that specific test
+3. Results appear in the **Python Test Log** output panel
+
+---
+
+## Using the Test Runner Script
+
+We've provided a convenient shell script for running tests:
+
+### Make Script Executable (First Time Only)
+
+```bash
+chmod +x scripts/run_db_tests.sh
+```
+
+### Basic Usage
+
+```bash
+# Run all tests with SQLite
+./scripts/run_db_tests.sh
+
+# Run with Docker PostgreSQL
+./scripts/run_db_tests.sh -d
+
+# Run specific category
+./scripts/run_db_tests.sh -c schema
+./scripts/run_db_tests.sh -c data
+./scripts/run_db_tests.sh -c crud
+./scripts/run_db_tests.sh -c relationships
+./scripts/run_db_tests.sh -c performance
+./scripts/run_db_tests.sh -c concurrency
+
+# Verbose output
+./scripts/run_db_tests.sh -v
+
+# With coverage
+./scripts/run_db_tests.sh --coverage
+
+# Include benchmark tests
+./scripts/run_db_tests.sh --benchmarks
+
+# Combine options
+./scripts/run_db_tests.sh -d -c crud -v --coverage
+```
+
+### Script Options Reference
+
+| Option | Description |
+|--------|-------------|
+| `-d, --docker` | Start PostgreSQL in Docker before running tests |
+| `-c, --category` | Run specific test category (schema, data, crud, relationships, performance, concurrency, all) |
+| `-v, --verbose` | Verbose output |
+| `--coverage` | Generate coverage report |
+| `--benchmarks` | Include performance benchmark tests |
+| `-h, --help` | Show help message |
+
+---
+
+## VS Code Test Explorer
+
+### Configuring Test Discovery
+
+If tests aren't appearing in Test Explorer, check:
+
+1. **Python extension is activated**: Look for Python version in status bar
+2. **pytest is installed**: `pip install pytest pytest-asyncio`
+3. **Correct workspace**: Ensure you opened the NerdLearn folder
+4. **Refresh tests**: Click refresh button in Test Explorer
+
+### Running Tests by Category
+
+You can filter tests in Test Explorer:
+- Type in the search box to filter by name
+- Use markers like `@pytest.mark.requires_db` to group tests
+
+### Debugging Tests
+
+1. Set breakpoints by clicking left of line numbers
+2. Right-click a test in Test Explorer
+3. Select **Debug Test**
+4. Use Debug toolbar to step through code
+
+---
+
+## Running with Docker
+
+### Start PostgreSQL Container
+
+```bash
+# Start container
+docker run -d \
+ --name nerdlearn-test-db \
+ -e POSTGRES_USER=test \
+ -e POSTGRES_PASSWORD=test \
+ -e POSTGRES_DB=nerdlearn_test \
+ -p 5433:5432 \
+ postgres:15
+
+# Verify it's running
+docker ps
+
+# Check logs if needed
+docker logs nerdlearn-test-db
+```
+
+### Set Environment Variables
+
+**Option 1: Terminal**
+```bash
+export TEST_DATABASE_URL="postgresql+asyncpg://test:test@localhost:5433/nerdlearn_test"
+export TEST_SYNC_DATABASE_URL="postgresql://test:test@localhost:5433/nerdlearn_test"
+pytest tests/database/ -v
+```
+
+**Option 2: VS Code launch.json**
+
+Create `.vscode/launch.json`:
+```json
+{
+ "version": "0.2.0",
+ "configurations": [
+ {
+ "name": "Python: pytest",
+ "type": "python",
+ "request": "launch",
+ "module": "pytest",
+ "args": ["tests/database/", "-v"],
+ "env": {
+ "TEST_DATABASE_URL": "postgresql+asyncpg://test:test@localhost:5433/nerdlearn_test",
+ "TEST_SYNC_DATABASE_URL": "postgresql://test:test@localhost:5433/nerdlearn_test"
+ },
+ "console": "integratedTerminal"
+ }
+ ]
+}
+```
+
+### Stop Container When Done
+
+```bash
+docker stop nerdlearn-test-db
+docker rm nerdlearn-test-db
+```
+
+---
+
+## Understanding Test Reports
+
+### HTML Reports
+
+After running tests with `--html` flag, open the report:
+
+```bash
+# Generate report
+pytest tests/database/ -v --html=reports/database/report.html --self-contained-html
+
+# Open in browser (Linux)
+xdg-open reports/database/report.html
+
+# Open in browser (Mac)
+open reports/database/report.html
+
+# Open in browser (Windows)
+start reports/database/report.html
+```
+
+The HTML report shows:
+- **Summary**: Total, passed, failed, skipped counts
+- **Environment**: Python version, platform, plugins
+- **Results Table**: Each test with status, duration, and error details
+
+### JSON Reports
+
+For programmatic access:
+
+```bash
+pytest tests/database/ -v --json-report --json-report-file=reports/database/report.json
+```
+
+### Coverage Reports
+
+```bash
+# Generate coverage
+pytest tests/database/ --cov=apps/api/app/models --cov-report=html:reports/coverage
+
+# Open coverage report
+open reports/coverage/index.html
+```
+
+---
+
+## Troubleshooting
+
+### Common Issues
+
+#### 1. "No tests discovered"
+
+**Solution:**
+```bash
+# Verify pytest can find tests
+pytest tests/database/ --collect-only
+
+# Check for import errors
+python -c "import tests.database.test_schema_integrity"
+```
+
+#### 2. "ModuleNotFoundError: No module named 'app'"
+
+**Solution:** Add the project to Python path:
+```bash
+# In terminal
+export PYTHONPATH="${PYTHONPATH}:${PWD}/apps/api"
+
+# Or in .vscode/settings.json
+{
+ "python.analysis.extraPaths": ["./apps/api"]
+}
+```
+
+#### 3. "Database connection failed"
+
+**Solution:**
+```bash
+# Check if using SQLite (should work without setup)
+echo $TEST_DATABASE_URL
+
+# If empty, tests will use SQLite automatically
+unset TEST_DATABASE_URL
+
+# For PostgreSQL, verify container is running
+docker ps | grep nerdlearn-test-db
+```
+
+#### 4. "pytest-asyncio error"
+
+**Solution:**
+```bash
+pip install pytest-asyncio
+
+# Ensure asyncio_mode is set
+# Check tests/database/conftest.py has proper async fixtures
+```
+
+#### 5. Tests hanging or timing out
+
+**Solution:**
+```bash
+# Run with timeout
+pytest tests/database/ -v --timeout=30
+
+# Run specific test to isolate issue
+pytest tests/database/test_schema_integrity.py::TestTableExistence -v
+```
+
+### Getting Help
+
+If you encounter issues:
+
+1. Check the test output for specific error messages
+2. Run a single test to isolate the problem
+3. Verify all dependencies are installed
+4. Check the GitHub Issues for known problems
+
+### Debug Mode
+
+For detailed debugging:
+
+```bash
+# Maximum verbosity
+pytest tests/database/ -vvv --tb=long
+
+# Show print statements
+pytest tests/database/ -v -s
+
+# Stop on first failure
+pytest tests/database/ -v -x
+
+# Run last failed tests
+pytest tests/database/ -v --lf
+```
+
+---
+
+## Quick Reference Card
+
+```bash
+# === SETUP ===
+pip install pytest pytest-asyncio pytest-cov pytest-html aiosqlite
+
+# === RUN ALL TESTS ===
+pytest tests/database/ -v
+
+# === RUN BY CATEGORY ===
+pytest tests/database/test_schema_integrity.py -v # Schema tests
+pytest tests/database/test_data_integrity.py -v # Data integrity
+pytest tests/database/test_crud_operations.py -v # CRUD operations
+pytest tests/database/test_relationships.py -v # Relationships
+pytest tests/database/test_performance.py -v # Benchmarks
+pytest tests/database/test_concurrency.py -v # Concurrency
+
+# === WITH REPORTS ===
+pytest tests/database/ -v --html=report.html --self-contained-html
+
+# === WITH COVERAGE ===
+pytest tests/database/ -v --cov=apps/api/app/models --cov-report=html
+
+# === USING SCRIPT ===
+./scripts/run_db_tests.sh -d -c all -v --coverage
+```
+
+---
+
+## Next Steps
+
+1. Run the tests to verify everything works: `pytest tests/database/ -v`
+2. Set up the VS Code Test Explorer for visual test running
+3. Configure pre-commit hooks to run tests before commits
+4. Check the CI workflow (`.github/workflows/database-tests.yml`) for automated testing
diff --git a/tests/database/__init__.py b/tests/database/__init__.py
new file mode 100644
index 0000000..1c06893
--- /dev/null
+++ b/tests/database/__init__.py
@@ -0,0 +1,12 @@
+"""
+Database Tests Package
+
+This package contains comprehensive database tests for the NerdLearn platform.
+Tests are organized into categories:
+- Schema integrity tests
+- Data integrity and constraint tests
+- CRUD operation tests
+- Relationship and cascade tests
+- Performance benchmark tests
+- Concurrent access tests
+"""
diff --git a/tests/database/conftest.py b/tests/database/conftest.py
new file mode 100644
index 0000000..512b120
--- /dev/null
+++ b/tests/database/conftest.py
@@ -0,0 +1,347 @@
+"""
+Database Test Fixtures and Configuration
+
+Provides shared fixtures for database testing including:
+- In-memory SQLite for unit tests
+- PostgreSQL test containers for integration tests
+- Test data factories
+- Database session management
+"""
+import os
+import pytest
+import asyncio
+from datetime import datetime, timedelta
+from typing import AsyncGenerator, Generator
+from unittest.mock import MagicMock, AsyncMock, patch
+
+from sqlalchemy import create_engine, event, text
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+from sqlalchemy.orm import sessionmaker, Session
+from sqlalchemy.pool import StaticPool
+
+
+# Test database URLs
+TEST_DATABASE_URL = os.getenv(
+ "TEST_DATABASE_URL",
+ "sqlite+aiosqlite:///:memory:"
+)
+TEST_SYNC_DATABASE_URL = os.getenv(
+ "TEST_SYNC_DATABASE_URL",
+ "sqlite:///:memory:"
+)
+
+
+@pytest.fixture(scope="session")
+def event_loop():
+ """Create event loop for async tests."""
+ policy = asyncio.get_event_loop_policy()
+ loop = policy.new_event_loop()
+ yield loop
+ loop.close()
+
+
+@pytest.fixture(scope="session")
+def sync_engine():
+ """Create synchronous test engine for schema tests."""
+ engine = create_engine(
+ TEST_SYNC_DATABASE_URL,
+ connect_args={"check_same_thread": False},
+ poolclass=StaticPool,
+ echo=False
+ )
+ return engine
+
+
+@pytest.fixture(scope="session")
+async def async_engine():
+ """Create async test engine."""
+ engine = create_async_engine(
+ TEST_DATABASE_URL,
+ connect_args={"check_same_thread": False},
+ poolclass=StaticPool,
+ echo=False
+ )
+ return engine
+
+
+@pytest.fixture
+async def async_session(async_engine) -> AsyncGenerator[AsyncSession, None]:
+ """Provide async database session for tests."""
+ # Import models to ensure they are registered
+ from app.core.database import Base
+ import app.models # noqa: F401
+
+ async with async_engine.begin() as conn:
+ await conn.run_sync(Base.metadata.create_all)
+
+ async_session_factory = sessionmaker(
+ async_engine,
+ class_=AsyncSession,
+ expire_on_commit=False
+ )
+
+ async with async_session_factory() as session:
+ yield session
+ await session.rollback()
+
+ async with async_engine.begin() as conn:
+ await conn.run_sync(Base.metadata.drop_all)
+
+
+@pytest.fixture
+def sync_session(sync_engine) -> Generator[Session, None, None]:
+ """Provide sync database session for tests."""
+ from app.core.database import Base
+ import app.models # noqa: F401
+
+ Base.metadata.create_all(bind=sync_engine)
+
+ SessionLocal = sessionmaker(bind=sync_engine, autoflush=False, autocommit=False)
+ session = SessionLocal()
+
+ try:
+ yield session
+ session.rollback()
+ finally:
+ session.close()
+ Base.metadata.drop_all(bind=sync_engine)
+
+
+# ============================================================================
+# Test Data Factories
+# ============================================================================
+
+class UserFactory:
+ """Factory for creating test User records."""
+
+ _counter = 0
+
+ @classmethod
+ def create(cls, **kwargs) -> dict:
+ """Create user data dictionary."""
+ cls._counter += 1
+ defaults = {
+ "email": f"testuser{cls._counter}@example.com",
+ "username": f"testuser{cls._counter}",
+ "hashed_password": "hashed_password_placeholder",
+ "full_name": f"Test User {cls._counter}",
+ "is_active": True,
+ "is_instructor": False,
+ "total_xp": 0,
+ "level": 1,
+ "streak_days": 0,
+ }
+ defaults.update(kwargs)
+ return defaults
+
+ @classmethod
+ def reset(cls):
+ """Reset counter for clean test runs."""
+ cls._counter = 0
+
+
+class ConceptFactory:
+ """Factory for creating test Concept records."""
+
+ _counter = 0
+ DOMAINS = ["Mathematics", "Computer Science", "Physics", "Biology"]
+
+ @classmethod
+ def create(cls, **kwargs) -> dict:
+ """Create concept data dictionary."""
+ cls._counter += 1
+ defaults = {
+ "name": f"Test Concept {cls._counter}",
+ "description": f"Description for concept {cls._counter}",
+ "domain": cls.DOMAINS[cls._counter % len(cls.DOMAINS)],
+ "subdomain": f"Subdomain {cls._counter}",
+ "avg_difficulty": 5.0,
+ }
+ defaults.update(kwargs)
+ return defaults
+
+ @classmethod
+ def reset(cls):
+ cls._counter = 0
+
+
+class CourseFactory:
+ """Factory for creating test Course records."""
+
+ _counter = 0
+
+ @classmethod
+ def create(cls, **kwargs) -> dict:
+ """Create course data dictionary."""
+ cls._counter += 1
+ defaults = {
+ "title": f"Test Course {cls._counter}",
+ "description": f"Description for course {cls._counter}",
+ "domain": "Computer Science",
+ "is_published": False,
+ }
+ defaults.update(kwargs)
+ return defaults
+
+ @classmethod
+ def reset(cls):
+ cls._counter = 0
+
+
+class SpacedRepetitionCardFactory:
+ """Factory for creating test SpacedRepetitionCard records."""
+
+ _counter = 0
+
+ @classmethod
+ def create(cls, user_id: int, concept_id: int, **kwargs) -> dict:
+ """Create spaced repetition card data dictionary."""
+ cls._counter += 1
+ defaults = {
+ "user_id": user_id,
+ "concept_id": concept_id,
+ "difficulty": 5.0,
+ "stability": 2.5,
+ "retrievability": 0.9,
+ "review_count": 0,
+ "next_review_at": datetime.utcnow() + timedelta(days=1),
+ }
+ defaults.update(kwargs)
+ return defaults
+
+ @classmethod
+ def reset(cls):
+ cls._counter = 0
+
+
+@pytest.fixture(autouse=True)
+def reset_factories():
+ """Reset all factories before each test."""
+ UserFactory.reset()
+ ConceptFactory.reset()
+ CourseFactory.reset()
+ SpacedRepetitionCardFactory.reset()
+ yield
+
+
+# ============================================================================
+# Mock Fixtures
+# ============================================================================
+
+@pytest.fixture
+def mock_redis():
+ """Mock Redis client for testing."""
+ mock = MagicMock()
+ mock.get = AsyncMock(return_value=None)
+ mock.set = AsyncMock(return_value=True)
+ mock.delete = AsyncMock(return_value=1)
+ mock.expire = AsyncMock(return_value=True)
+ return mock
+
+
+@pytest.fixture
+def mock_neo4j():
+ """Mock Neo4j driver for testing."""
+ mock = MagicMock()
+ mock.execute_query = MagicMock(return_value=([], None, None))
+ return mock
+
+
+# ============================================================================
+# Performance Test Utilities
+# ============================================================================
+
+class QueryTimer:
+ """Context manager for timing database queries."""
+
+ def __init__(self):
+ self.queries = []
+ self.total_time = 0
+
+ def __enter__(self):
+ self.start_time = datetime.utcnow()
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ self.total_time = (datetime.utcnow() - self.start_time).total_seconds()
+ return False
+
+ def record_query(self, query: str, duration: float):
+ """Record a query execution."""
+ self.queries.append({
+ "query": query,
+ "duration": duration
+ })
+
+
+@pytest.fixture
+def query_timer():
+ """Provide query timer for performance tests."""
+ return QueryTimer()
+
+
+# ============================================================================
+# Test Report Data Collection
+# ============================================================================
+
+class TestReportCollector:
+ """Collects test results for reporting."""
+
+ def __init__(self):
+ self.results = []
+ self.start_time = None
+ self.end_time = None
+
+ def start(self):
+ self.start_time = datetime.utcnow()
+
+ def end(self):
+ self.end_time = datetime.utcnow()
+
+ def record(self, test_name: str, category: str, passed: bool,
+ duration: float, details: dict = None):
+ """Record a test result."""
+ self.results.append({
+ "test_name": test_name,
+ "category": category,
+ "passed": passed,
+ "duration": duration,
+ "details": details or {},
+ "timestamp": datetime.utcnow().isoformat()
+ })
+
+ def get_summary(self) -> dict:
+ """Get test summary statistics."""
+ total = len(self.results)
+ passed = sum(1 for r in self.results if r["passed"])
+ failed = total - passed
+
+ by_category = {}
+ for result in self.results:
+ cat = result["category"]
+ if cat not in by_category:
+ by_category[cat] = {"passed": 0, "failed": 0, "total": 0}
+ by_category[cat]["total"] += 1
+ if result["passed"]:
+ by_category[cat]["passed"] += 1
+ else:
+ by_category[cat]["failed"] += 1
+
+ return {
+ "total_tests": total,
+ "passed": passed,
+ "failed": failed,
+ "pass_rate": (passed / total * 100) if total > 0 else 0,
+ "by_category": by_category,
+ "duration": (self.end_time - self.start_time).total_seconds()
+ if self.end_time and self.start_time else 0
+ }
+
+
+@pytest.fixture(scope="session")
+def report_collector():
+ """Provide test report collector."""
+ collector = TestReportCollector()
+ collector.start()
+ yield collector
+ collector.end()
diff --git a/tests/database/pytest.ini b/tests/database/pytest.ini
new file mode 100644
index 0000000..18efa62
--- /dev/null
+++ b/tests/database/pytest.ini
@@ -0,0 +1,42 @@
+[pytest]
+# Database tests configuration
+testpaths = .
+python_files = test_*.py
+python_classes = Test*
+python_functions = test_*
+
+# Async configuration
+asyncio_mode = auto
+
+# Default options
+addopts =
+ -v
+ --tb=short
+ --strict-markers
+ -p tests.database.reporting
+
+# Markers
+markers =
+ requires_db: Tests that require a database connection
+ unit: Fast unit tests (mocked DB)
+ integration: Integration tests (real DB)
+ benchmark: Performance benchmark tests
+ slow: Slow-running tests
+ concurrency: Tests for concurrent access patterns
+
+# Timeout (requires pytest-timeout)
+timeout = 60
+
+# Coverage (requires pytest-cov)
+# Run with: pytest --cov=app.models --cov-report=html
+
+# Logging
+log_cli = true
+log_cli_level = WARNING
+log_cli_format = %(asctime)s [%(levelname)s] %(message)s
+log_cli_date_format = %H:%M:%S
+
+# Filter warnings
+filterwarnings =
+ ignore::DeprecationWarning
+ ignore::PendingDeprecationWarning
diff --git a/tests/database/reporting.py b/tests/database/reporting.py
new file mode 100644
index 0000000..4274917
--- /dev/null
+++ b/tests/database/reporting.py
@@ -0,0 +1,481 @@
+"""
+Database Test Reporting Infrastructure
+
+Provides utilities for generating comprehensive test reports including:
+- Test execution summaries
+- Performance metrics
+- Trend analysis
+- HTML and JSON report generation
+"""
+import json
+import os
+from datetime import datetime
+from pathlib import Path
+from typing import Dict, List, Optional, Any
+from dataclasses import dataclass, field, asdict
+
+
+@dataclass
+class TestResult:
+ """Individual test result."""
+ name: str
+ category: str
+ status: str # passed, failed, skipped, error
+ duration: float
+ message: Optional[str] = None
+ details: Dict[str, Any] = field(default_factory=dict)
+ timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())
+
+
+@dataclass
+class BenchmarkResult:
+ """Performance benchmark result."""
+ name: str
+ metric: str
+ value: float
+ unit: str
+ threshold: Optional[float] = None
+ passed: bool = True
+ details: Dict[str, Any] = field(default_factory=dict)
+ timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())
+
+
+@dataclass
+class TestReport:
+ """Complete test report."""
+ run_id: str
+ timestamp: str
+ duration: float
+ environment: Dict[str, str]
+ summary: Dict[str, int]
+ tests: List[TestResult]
+ benchmarks: List[BenchmarkResult]
+ categories: Dict[str, Dict[str, int]]
+
+ def to_dict(self) -> Dict[str, Any]:
+ """Convert report to dictionary."""
+ return {
+ "run_id": self.run_id,
+ "timestamp": self.timestamp,
+ "duration": self.duration,
+ "environment": self.environment,
+ "summary": self.summary,
+ "tests": [asdict(t) for t in self.tests],
+ "benchmarks": [asdict(b) for b in self.benchmarks],
+ "categories": self.categories
+ }
+
+
+class DatabaseTestReporter:
+ """Reporter for database test results."""
+
+ def __init__(self, output_dir: str = "reports/database"):
+ self.output_dir = Path(output_dir)
+ self.output_dir.mkdir(parents=True, exist_ok=True)
+ self.tests: List[TestResult] = []
+ self.benchmarks: List[BenchmarkResult] = []
+ self.start_time: Optional[datetime] = None
+ self.end_time: Optional[datetime] = None
+ self.run_id: str = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
+
+ def start_run(self):
+ """Mark the start of a test run."""
+ self.start_time = datetime.utcnow()
+ self.tests = []
+ self.benchmarks = []
+
+ def end_run(self):
+ """Mark the end of a test run."""
+ self.end_time = datetime.utcnow()
+
+ def record_test(
+ self,
+ name: str,
+ category: str,
+ status: str,
+ duration: float,
+ message: Optional[str] = None,
+ details: Optional[Dict[str, Any]] = None
+ ):
+ """Record a test result."""
+ self.tests.append(TestResult(
+ name=name,
+ category=category,
+ status=status,
+ duration=duration,
+ message=message,
+ details=details or {}
+ ))
+
+ def record_benchmark(
+ self,
+ name: str,
+ metric: str,
+ value: float,
+ unit: str,
+ threshold: Optional[float] = None,
+ details: Optional[Dict[str, Any]] = None
+ ):
+ """Record a benchmark result."""
+ passed = threshold is None or value <= threshold
+ self.benchmarks.append(BenchmarkResult(
+ name=name,
+ metric=metric,
+ value=value,
+ unit=unit,
+ threshold=threshold,
+ passed=passed,
+ details=details or {}
+ ))
+
+ def get_summary(self) -> Dict[str, int]:
+ """Get test summary statistics."""
+ return {
+ "total": len(self.tests),
+ "passed": sum(1 for t in self.tests if t.status == "passed"),
+ "failed": sum(1 for t in self.tests if t.status == "failed"),
+ "skipped": sum(1 for t in self.tests if t.status == "skipped"),
+ "errors": sum(1 for t in self.tests if t.status == "error")
+ }
+
+ def get_categories(self) -> Dict[str, Dict[str, int]]:
+ """Get results by category."""
+ categories: Dict[str, Dict[str, int]] = {}
+ for test in self.tests:
+ if test.category not in categories:
+ categories[test.category] = {
+ "total": 0, "passed": 0, "failed": 0, "skipped": 0
+ }
+ categories[test.category]["total"] += 1
+ if test.status in categories[test.category]:
+ categories[test.category][test.status] += 1
+ return categories
+
+ def get_environment(self) -> Dict[str, str]:
+ """Get environment information."""
+ return {
+ "python_version": os.popen("python --version").read().strip(),
+ "database": os.getenv("TEST_DATABASE_URL", "sqlite:///:memory:"),
+ "os": os.name,
+ "timestamp": datetime.utcnow().isoformat()
+ }
+
+ def generate_report(self) -> TestReport:
+ """Generate complete test report."""
+ duration = 0.0
+ if self.start_time and self.end_time:
+ duration = (self.end_time - self.start_time).total_seconds()
+
+ return TestReport(
+ run_id=self.run_id,
+ timestamp=datetime.utcnow().isoformat(),
+ duration=duration,
+ environment=self.get_environment(),
+ summary=self.get_summary(),
+ tests=self.tests,
+ benchmarks=self.benchmarks,
+ categories=self.get_categories()
+ )
+
+ def save_json_report(self, filename: Optional[str] = None) -> str:
+ """Save report as JSON."""
+ report = self.generate_report()
+ filename = filename or f"db_test_report_{self.run_id}.json"
+ filepath = self.output_dir / filename
+
+ with open(filepath, "w") as f:
+ json.dump(report.to_dict(), f, indent=2)
+
+ return str(filepath)
+
+ def save_html_report(self, filename: Optional[str] = None) -> str:
+ """Generate and save HTML report."""
+ report = self.generate_report()
+ filename = filename or f"db_test_report_{self.run_id}.html"
+ filepath = self.output_dir / filename
+
+ html = self._generate_html(report)
+ with open(filepath, "w") as f:
+ f.write(html)
+
+ return str(filepath)
+
+ def _generate_html(self, report: TestReport) -> str:
+ """Generate HTML report content."""
+ summary = report.summary
+ pass_rate = (
+ (summary["passed"] / summary["total"] * 100)
+ if summary["total"] > 0 else 0
+ )
+
+ # Generate test rows
+ test_rows = ""
+ for test in report.tests:
+ status_class = {
+ "passed": "success",
+ "failed": "danger",
+ "skipped": "warning",
+ "error": "danger"
+ }.get(test.status, "secondary")
+
+ test_rows += f"""
+
+ | {test.name} |
+ {test.category} |
+ {test.status} |
+ {test.duration:.4f}s |
+ {test.message or '-'} |
+
+ """
+
+ # Generate benchmark rows
+ benchmark_rows = ""
+ for bench in report.benchmarks:
+ status_class = "success" if bench.passed else "danger"
+ threshold_str = f"{bench.threshold}{bench.unit}" if bench.threshold else "-"
+ benchmark_rows += f"""
+
+ | {bench.name} |
+ {bench.metric} |
+ {bench.value:.4f} {bench.unit} |
+ {threshold_str} |
+
+ {"PASS" if bench.passed else "FAIL"}
+ |
+
+ """
+
+ # Generate category summary
+ category_rows = ""
+ for cat, stats in report.categories.items():
+ cat_pass_rate = (
+ (stats["passed"] / stats["total"] * 100)
+ if stats["total"] > 0 else 0
+ )
+ category_rows += f"""
+
+ | {cat} |
+ {stats["total"]} |
+ {stats["passed"]} |
+ {stats["failed"]} |
+ {stats["skipped"]} |
+ {cat_pass_rate:.1f}% |
+
+ """
+
+ html = f"""
+
+
+
+
+
+ Database Test Report - {report.run_id}
+
+
+
+
+
+
Database Test Report
+
Run ID: {report.run_id} | Generated: {report.timestamp}
+
+
+
+
+
+
+
Total Tests
+
{summary["total"]}
+
+
+
+
+
+
+
Passed
+
{summary["passed"]}
+
+
+
+
+
+
+
Failed
+
{summary["failed"]}
+
+
+
+
+
= 80 else "bg-warning"}">
+
+
Pass Rate
+
{pass_rate:.1f}%
+
+
+
+
+
+
+
Results by Category
+
+
+
+ | Category |
+ Total |
+ Passed |
+ Failed |
+ Skipped |
+ Pass Rate |
+
+
+
+ {category_rows}
+
+
+
+
+
Test Details
+
+
+
+ | Test Name |
+ Category |
+ Status |
+ Duration |
+ Message |
+
+
+
+ {test_rows}
+
+
+
+
+
Performance Benchmarks
+
+
+
+ | Benchmark |
+ Metric |
+ Value |
+ Threshold |
+ Status |
+
+
+
+ {benchmark_rows if benchmark_rows else "| No benchmarks recorded |
"}
+
+
+
+
+
Environment
+
+
+ {"".join(f"| {k} | {v} |
" for k, v in report.environment.items())}
+
+
+
+
+
+
+
+
+ """
+ return html
+
+ def print_summary(self):
+ """Print a summary to console."""
+ summary = self.get_summary()
+ pass_rate = (
+ (summary["passed"] / summary["total"] * 100)
+ if summary["total"] > 0 else 0
+ )
+
+ print("\n" + "=" * 60)
+ print("DATABASE TEST SUMMARY")
+ print("=" * 60)
+ print(f"Total: {summary['total']}")
+ print(f"Passed: {summary['passed']}")
+ print(f"Failed: {summary['failed']}")
+ print(f"Skipped: {summary['skipped']}")
+ print(f"Errors: {summary['errors']}")
+ print(f"Pass Rate: {pass_rate:.1f}%")
+ print("=" * 60)
+
+ if self.benchmarks:
+ print("\nBENCHMARKS:")
+ for b in self.benchmarks:
+ status = "PASS" if b.passed else "FAIL"
+ print(f" {b.name}: {b.value:.4f}{b.unit} [{status}]")
+
+
+# Pytest plugin hooks for automatic reporting
+class DatabaseTestPlugin:
+ """Pytest plugin for database test reporting."""
+
+ def __init__(self):
+ self.reporter = DatabaseTestReporter()
+
+ def pytest_sessionstart(self, session):
+ """Called before test session starts."""
+ self.reporter.start_run()
+
+ def pytest_runtest_logreport(self, report):
+ """Called for each test phase (setup, call, teardown)."""
+ if report.when == "call":
+ # Determine category from test path
+ category = "general"
+ if "schema" in report.nodeid:
+ category = "schema_integrity"
+ elif "data" in report.nodeid:
+ category = "data_integrity"
+ elif "crud" in report.nodeid:
+ category = "crud_operations"
+ elif "relationship" in report.nodeid:
+ category = "relationships"
+ elif "performance" in report.nodeid:
+ category = "performance"
+ elif "concurren" in report.nodeid:
+ category = "concurrency"
+
+ # Determine status
+ if report.passed:
+ status = "passed"
+ elif report.failed:
+ status = "failed"
+ elif report.skipped:
+ status = "skipped"
+ else:
+ status = "error"
+
+ self.reporter.record_test(
+ name=report.nodeid.split("::")[-1],
+ category=category,
+ status=status,
+ duration=report.duration,
+ message=str(report.longrepr) if report.longrepr else None
+ )
+
+ def pytest_sessionfinish(self, session, exitstatus):
+ """Called after test session finishes."""
+ self.reporter.end_run()
+ self.reporter.print_summary()
+
+ # Generate reports
+ json_path = self.reporter.save_json_report()
+ html_path = self.reporter.save_html_report()
+
+ print(f"\nReports generated:")
+ print(f" JSON: {json_path}")
+ print(f" HTML: {html_path}")
+
+
+def pytest_configure(config):
+ """Register the plugin."""
+ config.pluginmanager.register(DatabaseTestPlugin(), "database_reporter")
diff --git a/tests/database/test_concurrency.py b/tests/database/test_concurrency.py
new file mode 100644
index 0000000..e84d5d1
--- /dev/null
+++ b/tests/database/test_concurrency.py
@@ -0,0 +1,576 @@
+"""
+Concurrent Access Tests
+
+Tests to verify database behavior under concurrent access conditions.
+Ensures data integrity is maintained during parallel operations.
+
+Test Categories:
+- Concurrent reads
+- Concurrent writes
+- Read-write conflicts
+- Transaction isolation
+- Deadlock handling
+- Race condition detection
+"""
+import pytest
+import asyncio
+from datetime import datetime, timedelta
+from typing import List, Tuple
+from sqlalchemy import select, update, text
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import StaticPool
+from sqlalchemy.exc import OperationalError
+
+from .conftest import UserFactory, TEST_DATABASE_URL
+
+pytestmark = [pytest.mark.requires_db, pytest.mark.integration]
+
+
+@pytest.fixture
+async def session_factory():
+ """Create a session factory for concurrent session creation."""
+ engine = create_async_engine(
+ TEST_DATABASE_URL,
+ connect_args={"check_same_thread": False},
+ poolclass=StaticPool,
+ echo=False
+ )
+
+ from app.core.database import Base
+ import app.models # noqa: F401
+
+ async with engine.begin() as conn:
+ await conn.run_sync(Base.metadata.create_all)
+
+ factory = sessionmaker(
+ engine,
+ class_=AsyncSession,
+ expire_on_commit=False
+ )
+
+ yield factory
+
+ async with engine.begin() as conn:
+ await conn.run_sync(Base.metadata.drop_all)
+
+ await engine.dispose()
+
+
+class TestConcurrentReads:
+ """Test concurrent read operations."""
+
+ @pytest.mark.asyncio
+ async def test_multiple_concurrent_reads(self, session_factory):
+ """Verify multiple concurrent reads don't interfere."""
+ from app.models.user import User
+
+ # Setup: create test data
+ async with session_factory() as session:
+ users = [User(**UserFactory.create()) for _ in range(50)]
+ session.add_all(users)
+ await session.commit()
+
+ # Concurrent reads
+ async def read_users(session_factory) -> Tuple[int, float]:
+ start = datetime.utcnow()
+ async with session_factory() as session:
+ result = await session.execute(select(User))
+ users = result.scalars().all()
+ return len(users), (datetime.utcnow() - start).total_seconds()
+
+ # Run 10 concurrent reads
+ tasks = [read_users(session_factory) for _ in range(10)]
+ results = await asyncio.gather(*tasks)
+
+ # All reads should return the same count
+ counts = [r[0] for r in results]
+ assert all(c == 50 for c in counts), \
+ f"Inconsistent read results: {counts}"
+
+ @pytest.mark.asyncio
+ async def test_read_during_write(self, session_factory):
+ """Verify reads work correctly while writes are happening."""
+ from app.models.user import User
+
+ # Setup
+ async with session_factory() as session:
+ users = [User(**UserFactory.create()) for _ in range(20)]
+ session.add_all(users)
+ await session.commit()
+
+ read_counts: List[int] = []
+ write_complete = asyncio.Event()
+
+ async def continuous_reads():
+ """Perform continuous reads until write completes."""
+ while not write_complete.is_set():
+ async with session_factory() as session:
+ result = await session.execute(select(User))
+ users = result.scalars().all()
+ read_counts.append(len(users))
+ await asyncio.sleep(0.01)
+
+ async def perform_writes():
+ """Add more users."""
+ async with session_factory() as session:
+ for i in range(10):
+ user = User(**UserFactory.create())
+ session.add(user)
+ await session.commit()
+ await asyncio.sleep(0.02)
+ write_complete.set()
+
+ # Run concurrently
+ read_task = asyncio.create_task(continuous_reads())
+ write_task = asyncio.create_task(perform_writes())
+
+ await write_task
+ read_task.cancel()
+ try:
+ await read_task
+ except asyncio.CancelledError:
+ pass
+
+ # Reads should show increasing counts (depending on isolation level)
+ # At minimum, counts should be >= initial (20) and <= final (30)
+ assert all(20 <= c <= 30 for c in read_counts), \
+ f"Unexpected counts during concurrent read/write: {read_counts}"
+
+
+class TestConcurrentWrites:
+ """Test concurrent write operations."""
+
+ @pytest.mark.asyncio
+ async def test_concurrent_inserts_no_conflict(self, session_factory):
+ """Verify concurrent inserts with unique data succeed."""
+ from app.models.user import User
+
+ async def insert_user(session_factory, index: int):
+ async with session_factory() as session:
+ user = User(**UserFactory.create(
+ email=f"concurrent{index}@test.com",
+ username=f"concurrent_user_{index}"
+ ))
+ session.add(user)
+ await session.commit()
+ return user.id
+
+ # Insert 20 users concurrently
+ tasks = [insert_user(session_factory, i) for i in range(20)]
+ ids = await asyncio.gather(*tasks)
+
+ # All inserts should succeed with unique IDs
+ assert len(set(ids)) == 20, "Some inserts failed or returned duplicate IDs"
+
+ # Verify all users exist
+ async with session_factory() as session:
+ result = await session.execute(select(User))
+ users = result.scalars().all()
+ assert len(users) == 20
+
+ @pytest.mark.asyncio
+ async def test_concurrent_updates_same_record(self, session_factory):
+ """Test concurrent updates to the same record."""
+ from app.models.user import User
+
+ # Setup: create a single user
+ async with session_factory() as session:
+ user = User(**UserFactory.create(total_xp=0))
+ session.add(user)
+ await session.commit()
+ user_id = user.id
+
+ async def increment_xp(session_factory, user_id: int, amount: int):
+ """Increment user XP."""
+ async with session_factory() as session:
+ result = await session.execute(
+ select(User).where(User.id == user_id)
+ )
+ user = result.scalar_one()
+ # Simulate read-modify-write
+ current_xp = user.total_xp
+ await asyncio.sleep(0.01) # Simulate processing
+ await session.execute(
+ update(User)
+ .where(User.id == user_id)
+ .values(total_xp=current_xp + amount)
+ )
+ await session.commit()
+
+ # Run concurrent updates
+ tasks = [increment_xp(session_factory, user_id, 10) for _ in range(10)]
+ await asyncio.gather(*tasks)
+
+ # Check final XP
+ # Note: Without proper locking, we might have lost updates
+ async with session_factory() as session:
+ result = await session.execute(
+ select(User).where(User.id == user_id)
+ )
+ user = result.scalar_one()
+
+ # Document expected vs actual
+ # With SQLite and this pattern, we likely lost updates
+ # This demonstrates why atomic operations are needed
+ expected_xp = 100 # 10 updates * 10 XP
+ actual_xp = user.total_xp
+
+ # This test documents the race condition
+ # In a real app, use atomic updates or SELECT FOR UPDATE
+ if actual_xp != expected_xp:
+ pytest.skip(
+ f"Race condition detected (expected): "
+ f"expected {expected_xp}, got {actual_xp}. "
+ f"Use atomic operations in production."
+ )
+
+ @pytest.mark.asyncio
+ async def test_atomic_increment(self, session_factory):
+ """Test atomic increment operation."""
+ from app.models.user import User
+
+ # Setup
+ async with session_factory() as session:
+ user = User(**UserFactory.create(total_xp=0))
+ session.add(user)
+ await session.commit()
+ user_id = user.id
+
+ async def atomic_increment(session_factory, user_id: int, amount: int):
+ """Atomically increment XP using SQL expression."""
+ async with session_factory() as session:
+ await session.execute(
+ update(User)
+ .where(User.id == user_id)
+ .values(total_xp=User.total_xp + amount)
+ )
+ await session.commit()
+
+ # Run concurrent atomic updates
+ tasks = [atomic_increment(session_factory, user_id, 10) for _ in range(10)]
+ await asyncio.gather(*tasks)
+
+ # With atomic operations, all updates should be applied
+ async with session_factory() as session:
+ result = await session.execute(
+ select(User).where(User.id == user_id)
+ )
+ user = result.scalar_one()
+
+ assert user.total_xp == 100, \
+ f"Atomic increment failed: expected 100, got {user.total_xp}"
+
+
+class TestTransactionIsolation:
+ """Test transaction isolation behavior."""
+
+ @pytest.mark.asyncio
+ async def test_dirty_read_prevention(self, session_factory):
+ """Verify uncommitted changes are not visible to other transactions."""
+ from app.models.user import User
+
+ # Setup
+ async with session_factory() as session:
+ user = User(**UserFactory.create(total_xp=100))
+ session.add(user)
+ await session.commit()
+ user_id = user.id
+
+ read_before_commit = None
+
+ async def update_without_commit(session_factory, user_id: int):
+ """Update but don't commit."""
+ async with session_factory() as session:
+ await session.execute(
+ update(User)
+ .where(User.id == user_id)
+ .values(total_xp=999)
+ )
+ # Don't commit, let session close (rollback)
+ await asyncio.sleep(0.5)
+
+ async def read_value(session_factory, user_id: int):
+ """Read the value during the other transaction."""
+ nonlocal read_before_commit
+ await asyncio.sleep(0.1) # Let update run first
+ async with session_factory() as session:
+ result = await session.execute(
+ select(User).where(User.id == user_id)
+ )
+ user = result.scalar_one()
+ read_before_commit = user.total_xp
+
+ # Run concurrently
+ await asyncio.gather(
+ update_without_commit(session_factory, user_id),
+ read_value(session_factory, user_id)
+ )
+
+ # Read should see original value (100), not uncommitted (999)
+ # Note: Depends on isolation level
+ assert read_before_commit == 100 or read_before_commit == 999, \
+ f"Unexpected value: {read_before_commit}"
+
+ @pytest.mark.asyncio
+ async def test_repeatable_read(self, session_factory):
+ """Test repeatable read behavior within a transaction."""
+ from app.models.user import User
+
+ # Setup
+ async with session_factory() as session:
+ user = User(**UserFactory.create(total_xp=100))
+ session.add(user)
+ await session.commit()
+ user_id = user.id
+
+ reads_in_transaction: List[int] = []
+
+ async def read_twice_in_transaction(session_factory, user_id: int):
+ """Read the same record twice within one transaction."""
+ async with session_factory() as session:
+ # First read
+ result = await session.execute(
+ select(User).where(User.id == user_id)
+ )
+ user = result.scalar_one()
+ reads_in_transaction.append(user.total_xp)
+
+ await asyncio.sleep(0.2) # Allow external update
+
+ # Second read in same transaction
+ session.expire_all() # Force re-read
+ result = await session.execute(
+ select(User).where(User.id == user_id)
+ )
+ user = result.scalar_one()
+ reads_in_transaction.append(user.total_xp)
+
+ async def external_update(session_factory, user_id: int):
+ """Update the record from another session."""
+ await asyncio.sleep(0.1)
+ async with session_factory() as session:
+ await session.execute(
+ update(User)
+ .where(User.id == user_id)
+ .values(total_xp=200)
+ )
+ await session.commit()
+
+ await asyncio.gather(
+ read_twice_in_transaction(session_factory, user_id),
+ external_update(session_factory, user_id)
+ )
+
+ # Document behavior (depends on isolation level)
+ first_read, second_read = reads_in_transaction
+ # In REPEATABLE READ: first_read == second_read
+ # In READ COMMITTED: second_read might differ
+ assert first_read == 100, f"First read unexpected: {first_read}"
+
+
+class TestDeadlockHandling:
+ """Test deadlock detection and handling."""
+
+ @pytest.mark.asyncio
+ async def test_potential_deadlock_scenario(self, session_factory):
+ """Test a scenario that could cause deadlock."""
+ from app.models.user import User
+
+ # Setup: create two users
+ async with session_factory() as session:
+ user1 = User(**UserFactory.create())
+ user2 = User(**UserFactory.create())
+ session.add_all([user1, user2])
+ await session.commit()
+ user1_id, user2_id = user1.id, user2.id
+
+ errors: List[Exception] = []
+
+ async def update_order_1(session_factory, id1: int, id2: int):
+ """Update user1 then user2."""
+ try:
+ async with session_factory() as session:
+ await session.execute(
+ update(User).where(User.id == id1).values(total_xp=100)
+ )
+ await asyncio.sleep(0.1)
+ await session.execute(
+ update(User).where(User.id == id2).values(total_xp=200)
+ )
+ await session.commit()
+ except Exception as e:
+ errors.append(e)
+
+ async def update_order_2(session_factory, id1: int, id2: int):
+ """Update user2 then user1 (opposite order)."""
+ try:
+ async with session_factory() as session:
+ await session.execute(
+ update(User).where(User.id == id2).values(total_xp=300)
+ )
+ await asyncio.sleep(0.1)
+ await session.execute(
+ update(User).where(User.id == id1).values(total_xp=400)
+ )
+ await session.commit()
+ except Exception as e:
+ errors.append(e)
+
+ # Run concurrently - may cause deadlock
+ await asyncio.gather(
+ update_order_1(session_factory, user1_id, user2_id),
+ update_order_2(session_factory, user1_id, user2_id),
+ return_exceptions=True
+ )
+
+ # SQLite handles this differently than PostgreSQL
+ # Document any errors
+ if errors:
+ pytest.skip(
+ f"Deadlock-like error occurred (expected in some cases): {errors[0]}"
+ )
+
+
+class TestConnectionPool:
+ """Test connection pool behavior under load."""
+
+ @pytest.mark.asyncio
+ async def test_pool_exhaustion_handling(self, session_factory):
+ """Test behavior when connection pool is exhausted."""
+ from app.models.user import User
+
+ async def long_running_query(session_factory, duration: float):
+ """Simulate a long-running query."""
+ async with session_factory() as session:
+ result = await session.execute(select(User))
+ _ = result.scalars().all()
+ await asyncio.sleep(duration)
+
+ # Try to open many concurrent sessions
+ tasks = [long_running_query(session_factory, 0.5) for _ in range(20)]
+
+ # Should complete without errors (pool should handle)
+ try:
+ await asyncio.wait_for(
+ asyncio.gather(*tasks),
+ timeout=30.0
+ )
+ except asyncio.TimeoutError:
+ pytest.fail("Connection pool exhaustion caused timeout")
+ except Exception as e:
+ # Some pool exhaustion error is acceptable
+ assert "pool" in str(e).lower() or "connection" in str(e).lower()
+
+
+class TestRaceConditions:
+ """Test and document potential race conditions."""
+
+ @pytest.mark.asyncio
+ async def test_check_then_act_race(self, session_factory):
+ """Demonstrate check-then-act race condition."""
+ from app.models.user import User
+ from app.models.course import Course, Enrollment
+
+ # Setup
+ async with session_factory() as session:
+ user = User(**UserFactory.create())
+ course = Course(
+ title="Limited Course",
+ description="Only 1 spot",
+ domain="Test",
+ is_published=True
+ )
+ session.add_all([user, course])
+ await session.commit()
+ user_id, course_id = user.id, course.id
+
+ enrollments_created = []
+ max_enrollments = 1 # Simulate limited spots
+
+ async def try_enroll(session_factory, user_id: int, course_id: int, idx: int):
+ """Try to enroll if spots available."""
+ async with session_factory() as session:
+ # Check current enrollment count
+ result = await session.execute(
+ select(Enrollment).where(Enrollment.course_id == course_id)
+ )
+ current_count = len(result.scalars().all())
+
+ if current_count < max_enrollments:
+ # Simulate delay between check and act
+ await asyncio.sleep(0.05)
+
+ # Create enrollment
+ enrollment = Enrollment(
+ user_id=user_id,
+ course_id=course_id,
+ progress=0.0
+ )
+ session.add(enrollment)
+ try:
+ await session.commit()
+ enrollments_created.append(idx)
+ except Exception:
+ pass # Unique constraint might prevent this
+
+ # Multiple concurrent enrollment attempts
+ tasks = [
+ try_enroll(session_factory, user_id, course_id, i)
+ for i in range(5)
+ ]
+ await asyncio.gather(*tasks)
+
+ # Document: without proper locking, multiple enrollments may be created
+ async with session_factory() as session:
+ result = await session.execute(
+ select(Enrollment).where(Enrollment.course_id == course_id)
+ )
+ final_count = len(result.scalars().all())
+
+ # This demonstrates the race condition
+ if final_count > max_enrollments:
+ pytest.skip(
+ f"Race condition demonstrated: {final_count} enrollments "
+ f"created when max was {max_enrollments}. "
+ f"Use SELECT FOR UPDATE or application-level locking."
+ )
+
+ @pytest.mark.asyncio
+ async def test_concurrent_streak_update(self, session_factory):
+ """Test concurrent streak updates don't cause issues."""
+ from app.models.user import User
+
+ # Setup
+ async with session_factory() as session:
+ user = User(**UserFactory.create(streak_days=5))
+ session.add(user)
+ await session.commit()
+ user_id = user.id
+
+ async def update_streak(session_factory, user_id: int):
+ """Update streak atomically."""
+ async with session_factory() as session:
+ await session.execute(
+ update(User)
+ .where(User.id == user_id)
+ .values(
+ streak_days=User.streak_days + 1,
+ last_activity_date=datetime.utcnow()
+ )
+ )
+ await session.commit()
+
+ # Multiple concurrent streak updates (shouldn't happen in real app)
+ tasks = [update_streak(session_factory, user_id) for _ in range(5)]
+ await asyncio.gather(*tasks)
+
+ async with session_factory() as session:
+ result = await session.execute(
+ select(User).where(User.id == user_id)
+ )
+ user = result.scalar_one()
+
+ # With atomic update, all should apply
+ assert user.streak_days == 10, \
+ f"Expected streak 10, got {user.streak_days}"
diff --git a/tests/database/test_crud_operations.py b/tests/database/test_crud_operations.py
new file mode 100644
index 0000000..5bf8041
--- /dev/null
+++ b/tests/database/test_crud_operations.py
@@ -0,0 +1,666 @@
+"""
+CRUD Operation Tests
+
+Tests for Create, Read, Update, Delete operations across all major entities.
+Verifies that basic database operations work correctly and data is persisted
+as expected.
+
+Test Categories:
+- User CRUD operations
+- Course CRUD operations
+- Enrollment CRUD operations
+- Spaced Repetition CRUD operations
+- Achievement CRUD operations
+"""
+import pytest
+from datetime import datetime, timedelta
+from sqlalchemy import select, update, delete
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from .conftest import UserFactory, CourseFactory, ConceptFactory
+
+pytestmark = [pytest.mark.requires_db, pytest.mark.unit]
+
+
+class TestUserCRUD:
+ """Test User entity CRUD operations."""
+
+ @pytest.mark.asyncio
+ async def test_create_user(self, async_session: AsyncSession):
+ """Test creating a new user."""
+ from app.models.user import User
+
+ user_data = UserFactory.create()
+ user = User(**user_data)
+ async_session.add(user)
+ await async_session.commit()
+
+ assert user.id is not None
+ assert user.email == user_data["email"]
+ assert user.username == user_data["username"]
+ assert user.created_at is not None
+
+ @pytest.mark.asyncio
+ async def test_read_user_by_id(self, async_session: AsyncSession):
+ """Test reading a user by ID."""
+ from app.models.user import User
+
+ # Create user
+ user_data = UserFactory.create()
+ user = User(**user_data)
+ async_session.add(user)
+ await async_session.commit()
+ user_id = user.id
+
+ # Clear session cache
+ async_session.expire_all()
+
+ # Read user
+ result = await async_session.execute(
+ select(User).where(User.id == user_id)
+ )
+ fetched_user = result.scalar_one()
+
+ assert fetched_user.id == user_id
+ assert fetched_user.email == user_data["email"]
+
+ @pytest.mark.asyncio
+ async def test_read_user_by_email(self, async_session: AsyncSession):
+ """Test reading a user by email address."""
+ from app.models.user import User
+
+ user_data = UserFactory.create(email="findme@example.com")
+ user = User(**user_data)
+ async_session.add(user)
+ await async_session.commit()
+
+ async_session.expire_all()
+
+ result = await async_session.execute(
+ select(User).where(User.email == "findme@example.com")
+ )
+ fetched_user = result.scalar_one_or_none()
+
+ assert fetched_user is not None
+ assert fetched_user.email == "findme@example.com"
+
+ @pytest.mark.asyncio
+ async def test_update_user(self, async_session: AsyncSession):
+ """Test updating user fields."""
+ from app.models.user import User
+
+ # Create user
+ user_data = UserFactory.create()
+ user = User(**user_data)
+ async_session.add(user)
+ await async_session.commit()
+ user_id = user.id
+
+ # Update user
+ await async_session.execute(
+ update(User)
+ .where(User.id == user_id)
+ .values(full_name="Updated Name", total_xp=500)
+ )
+ await async_session.commit()
+
+ async_session.expire_all()
+
+ # Verify update
+ result = await async_session.execute(
+ select(User).where(User.id == user_id)
+ )
+ updated_user = result.scalar_one()
+
+ assert updated_user.full_name == "Updated Name"
+ assert updated_user.total_xp == 500
+
+ @pytest.mark.asyncio
+ async def test_update_user_partial(self, async_session: AsyncSession):
+ """Test partial update (only specific fields)."""
+ from app.models.user import User
+
+ user_data = UserFactory.create(total_xp=100, level=5)
+ user = User(**user_data)
+ async_session.add(user)
+ await async_session.commit()
+ user_id = user.id
+
+ # Update only XP, level should remain
+ await async_session.execute(
+ update(User)
+ .where(User.id == user_id)
+ .values(total_xp=200)
+ )
+ await async_session.commit()
+
+ async_session.expire_all()
+
+ result = await async_session.execute(
+ select(User).where(User.id == user_id)
+ )
+ updated_user = result.scalar_one()
+
+ assert updated_user.total_xp == 200
+ assert updated_user.level == 5 # Unchanged
+
+ @pytest.mark.asyncio
+ async def test_delete_user(self, async_session: AsyncSession):
+ """Test deleting a user."""
+ from app.models.user import User
+
+ user_data = UserFactory.create()
+ user = User(**user_data)
+ async_session.add(user)
+ await async_session.commit()
+ user_id = user.id
+
+ # Delete user
+ await async_session.execute(
+ delete(User).where(User.id == user_id)
+ )
+ await async_session.commit()
+
+ # Verify deletion
+ result = await async_session.execute(
+ select(User).where(User.id == user_id)
+ )
+ deleted_user = result.scalar_one_or_none()
+
+ assert deleted_user is None
+
+ @pytest.mark.asyncio
+ async def test_create_multiple_users(self, async_session: AsyncSession):
+ """Test creating multiple users in batch."""
+ from app.models.user import User
+
+ users = [User(**UserFactory.create()) for _ in range(5)]
+ async_session.add_all(users)
+ await async_session.commit()
+
+ # Verify all users created
+ result = await async_session.execute(select(User))
+ all_users = result.scalars().all()
+
+ assert len(all_users) == 5
+
+
+class TestCourseCRUD:
+ """Test Course entity CRUD operations."""
+
+ @pytest.mark.asyncio
+ async def test_create_course(self, async_session: AsyncSession):
+ """Test creating a new course."""
+ from app.models.course import Course
+
+ course_data = CourseFactory.create()
+ course = Course(**course_data)
+ async_session.add(course)
+ await async_session.commit()
+
+ assert course.id is not None
+ assert course.title == course_data["title"]
+
+ @pytest.mark.asyncio
+ async def test_read_courses_by_domain(self, async_session: AsyncSession):
+ """Test reading courses filtered by domain."""
+ from app.models.course import Course
+
+ # Create courses in different domains
+ course1 = Course(**CourseFactory.create(domain="Mathematics"))
+ course2 = Course(**CourseFactory.create(domain="Computer Science"))
+ course3 = Course(**CourseFactory.create(domain="Mathematics"))
+ async_session.add_all([course1, course2, course3])
+ await async_session.commit()
+
+ # Query by domain
+ result = await async_session.execute(
+ select(Course).where(Course.domain == "Mathematics")
+ )
+ math_courses = result.scalars().all()
+
+ assert len(math_courses) == 2
+
+ @pytest.mark.asyncio
+ async def test_update_course_publish_status(self, async_session: AsyncSession):
+ """Test publishing a course."""
+ from app.models.course import Course
+
+ course = Course(**CourseFactory.create(is_published=False))
+ async_session.add(course)
+ await async_session.commit()
+ course_id = course.id
+
+ # Publish course
+ await async_session.execute(
+ update(Course)
+ .where(Course.id == course_id)
+ .values(is_published=True)
+ )
+ await async_session.commit()
+
+ async_session.expire_all()
+
+ result = await async_session.execute(
+ select(Course).where(Course.id == course_id)
+ )
+ updated_course = result.scalar_one()
+
+ assert updated_course.is_published is True
+
+ @pytest.mark.asyncio
+ async def test_delete_course(self, async_session: AsyncSession):
+ """Test deleting a course."""
+ from app.models.course import Course
+
+ course = Course(**CourseFactory.create())
+ async_session.add(course)
+ await async_session.commit()
+ course_id = course.id
+
+ await async_session.execute(
+ delete(Course).where(Course.id == course_id)
+ )
+ await async_session.commit()
+
+ result = await async_session.execute(
+ select(Course).where(Course.id == course_id)
+ )
+ assert result.scalar_one_or_none() is None
+
+
+class TestEnrollmentCRUD:
+ """Test Enrollment entity CRUD operations."""
+
+ @pytest.mark.asyncio
+ async def test_create_enrollment(self, async_session: AsyncSession):
+ """Test enrolling a user in a course."""
+ from app.models.user import User
+ from app.models.course import Course, Enrollment
+
+ user = User(**UserFactory.create())
+ course = Course(**CourseFactory.create())
+ async_session.add_all([user, course])
+ await async_session.commit()
+
+ enrollment = Enrollment(
+ user_id=user.id,
+ course_id=course.id,
+ progress=0.0
+ )
+ async_session.add(enrollment)
+ await async_session.commit()
+
+ assert enrollment.id is not None
+ assert enrollment.user_id == user.id
+ assert enrollment.course_id == course.id
+
+ @pytest.mark.asyncio
+ async def test_update_enrollment_progress(self, async_session: AsyncSession):
+ """Test updating enrollment progress."""
+ from app.models.user import User
+ from app.models.course import Course, Enrollment
+
+ user = User(**UserFactory.create())
+ course = Course(**CourseFactory.create())
+ async_session.add_all([user, course])
+ await async_session.commit()
+
+ enrollment = Enrollment(
+ user_id=user.id,
+ course_id=course.id,
+ progress=0.0
+ )
+ async_session.add(enrollment)
+ await async_session.commit()
+ enrollment_id = enrollment.id
+
+ # Update progress
+ await async_session.execute(
+ update(Enrollment)
+ .where(Enrollment.id == enrollment_id)
+ .values(progress=0.5)
+ )
+ await async_session.commit()
+
+ async_session.expire_all()
+
+ result = await async_session.execute(
+ select(Enrollment).where(Enrollment.id == enrollment_id)
+ )
+ updated_enrollment = result.scalar_one()
+
+ assert updated_enrollment.progress == 0.5
+
+ @pytest.mark.asyncio
+ async def test_complete_enrollment(self, async_session: AsyncSession):
+ """Test marking enrollment as completed."""
+ from app.models.user import User
+ from app.models.course import Course, Enrollment
+
+ user = User(**UserFactory.create())
+ course = Course(**CourseFactory.create())
+ async_session.add_all([user, course])
+ await async_session.commit()
+
+ enrollment = Enrollment(
+ user_id=user.id,
+ course_id=course.id,
+ progress=0.0
+ )
+ async_session.add(enrollment)
+ await async_session.commit()
+ enrollment_id = enrollment.id
+
+ # Complete enrollment
+ completion_time = datetime.utcnow()
+ await async_session.execute(
+ update(Enrollment)
+ .where(Enrollment.id == enrollment_id)
+ .values(progress=1.0, completed_at=completion_time)
+ )
+ await async_session.commit()
+
+ async_session.expire_all()
+
+ result = await async_session.execute(
+ select(Enrollment).where(Enrollment.id == enrollment_id)
+ )
+ completed_enrollment = result.scalar_one()
+
+ assert completed_enrollment.progress == 1.0
+ assert completed_enrollment.completed_at is not None
+
+
+class TestSpacedRepetitionCRUD:
+ """Test Spaced Repetition entities CRUD operations."""
+
+ @pytest.mark.asyncio
+ async def test_create_spaced_repetition_card(self, async_session: AsyncSession):
+ """Test creating a new spaced repetition card."""
+ from app.models.user import User
+ from app.models.spaced_repetition import SpacedRepetitionCard, Concept
+
+ user = User(**UserFactory.create())
+ concept = Concept(name="Test Concept", description="Test description")
+ async_session.add_all([user, concept])
+ await async_session.commit()
+
+ card = SpacedRepetitionCard(
+ user_id=user.id,
+ concept_id=concept.id,
+ difficulty=5.0,
+ stability=2.5,
+ retrievability=0.9,
+ review_count=0,
+ next_review_at=datetime.utcnow() + timedelta(days=1)
+ )
+ async_session.add(card)
+ await async_session.commit()
+
+ assert card.id is not None
+ assert card.difficulty == 5.0
+
+ @pytest.mark.asyncio
+ async def test_update_fsrs_parameters(self, async_session: AsyncSession):
+ """Test updating FSRS parameters after review."""
+ from app.models.user import User
+ from app.models.spaced_repetition import SpacedRepetitionCard, Concept
+
+ user = User(**UserFactory.create())
+ concept = Concept(name="Test", description="Test")
+ async_session.add_all([user, concept])
+ await async_session.commit()
+
+ card = SpacedRepetitionCard(
+ user_id=user.id,
+ concept_id=concept.id,
+ difficulty=5.0,
+ stability=2.5,
+ retrievability=0.9,
+ review_count=0,
+ next_review_at=datetime.utcnow() + timedelta(days=1)
+ )
+ async_session.add(card)
+ await async_session.commit()
+ card_id = card.id
+
+ # Simulate FSRS update after review
+ new_stability = 5.0
+ new_difficulty = 4.5
+ new_next_review = datetime.utcnow() + timedelta(days=7)
+
+ await async_session.execute(
+ update(SpacedRepetitionCard)
+ .where(SpacedRepetitionCard.id == card_id)
+ .values(
+ stability=new_stability,
+ difficulty=new_difficulty,
+ review_count=SpacedRepetitionCard.review_count + 1,
+ next_review_at=new_next_review
+ )
+ )
+ await async_session.commit()
+
+ async_session.expire_all()
+
+ result = await async_session.execute(
+ select(SpacedRepetitionCard).where(SpacedRepetitionCard.id == card_id)
+ )
+ updated_card = result.scalar_one()
+
+ assert updated_card.stability == new_stability
+ assert updated_card.difficulty == new_difficulty
+ assert updated_card.review_count == 1
+
+ @pytest.mark.asyncio
+ async def test_create_review_log(self, async_session: AsyncSession):
+ """Test logging a review."""
+ from app.models.user import User
+ from app.models.spaced_repetition import SpacedRepetitionCard, Concept, ReviewLog
+
+ user = User(**UserFactory.create())
+ concept = Concept(name="Test", description="Test")
+ async_session.add_all([user, concept])
+ await async_session.commit()
+
+ card = SpacedRepetitionCard(
+ user_id=user.id,
+ concept_id=concept.id,
+ difficulty=5.0,
+ stability=2.5,
+ retrievability=0.9,
+ review_count=0,
+ next_review_at=datetime.utcnow() + timedelta(days=1)
+ )
+ async_session.add(card)
+ await async_session.commit()
+
+ review = ReviewLog(
+ card_id=card.id,
+ rating=4,
+ reviewed_at=datetime.utcnow(),
+ response_time_ms=2500
+ )
+ async_session.add(review)
+ await async_session.commit()
+
+ assert review.id is not None
+ assert review.rating == 4
+
+ @pytest.mark.asyncio
+ async def test_get_due_cards(self, async_session: AsyncSession):
+ """Test querying cards due for review."""
+ from app.models.user import User
+ from app.models.spaced_repetition import SpacedRepetitionCard, Concept
+
+ user = User(**UserFactory.create())
+ concept = Concept(name="Test", description="Test")
+ async_session.add_all([user, concept])
+ await async_session.commit()
+
+ # Create cards with different due dates
+ past_card = SpacedRepetitionCard(
+ user_id=user.id,
+ concept_id=concept.id,
+ difficulty=5.0,
+ stability=2.5,
+ retrievability=0.9,
+ review_count=1,
+ next_review_at=datetime.utcnow() - timedelta(days=1) # Overdue
+ )
+ future_card = SpacedRepetitionCard(
+ user_id=user.id,
+ concept_id=concept.id,
+ difficulty=5.0,
+ stability=2.5,
+ retrievability=0.9,
+ review_count=1,
+ next_review_at=datetime.utcnow() + timedelta(days=7) # Not due
+ )
+ async_session.add_all([past_card, future_card])
+ await async_session.commit()
+
+ # Query due cards
+ result = await async_session.execute(
+ select(SpacedRepetitionCard)
+ .where(SpacedRepetitionCard.next_review_at <= datetime.utcnow())
+ )
+ due_cards = result.scalars().all()
+
+ assert len(due_cards) == 1
+
+
+class TestAchievementCRUD:
+ """Test Achievement entity CRUD operations."""
+
+ @pytest.mark.asyncio
+ async def test_create_achievement(self, async_session: AsyncSession):
+ """Test unlocking an achievement."""
+ from app.models.user import User
+ from app.models.gamification import UserAchievement
+
+ user = User(**UserFactory.create())
+ async_session.add(user)
+ await async_session.commit()
+
+ achievement = UserAchievement(
+ user_id=user.id,
+ achievement_type="STREAK_MILESTONE",
+ name="First Week Streak",
+ description="Maintained a 7-day streak",
+ xp_reward=100
+ )
+ async_session.add(achievement)
+ await async_session.commit()
+
+ assert achievement.id is not None
+ assert achievement.xp_reward == 100
+
+ @pytest.mark.asyncio
+ async def test_get_user_achievements(self, async_session: AsyncSession):
+ """Test retrieving all achievements for a user."""
+ from app.models.user import User
+ from app.models.gamification import UserAchievement
+
+ user = User(**UserFactory.create())
+ async_session.add(user)
+ await async_session.commit()
+
+ # Create multiple achievements
+ achievements = [
+ UserAchievement(
+ user_id=user.id,
+ achievement_type="STREAK_MILESTONE",
+ name="7-Day Streak",
+ description="7-day streak",
+ xp_reward=100
+ ),
+ UserAchievement(
+ user_id=user.id,
+ achievement_type="XP_MILESTONE",
+ name="1000 XP",
+ description="Earned 1000 XP",
+ xp_reward=50
+ ),
+ ]
+ async_session.add_all(achievements)
+ await async_session.commit()
+
+ # Query achievements
+ result = await async_session.execute(
+ select(UserAchievement).where(UserAchievement.user_id == user.id)
+ )
+ user_achievements = result.scalars().all()
+
+ assert len(user_achievements) == 2
+ total_xp = sum(a.xp_reward for a in user_achievements)
+ assert total_xp == 150
+
+
+class TestBulkOperations:
+ """Test bulk database operations."""
+
+ @pytest.mark.asyncio
+ async def test_bulk_insert_users(self, async_session: AsyncSession):
+ """Test bulk inserting multiple users."""
+ from app.models.user import User
+
+ users = [User(**UserFactory.create()) for _ in range(100)]
+ async_session.add_all(users)
+ await async_session.commit()
+
+ result = await async_session.execute(select(User))
+ all_users = result.scalars().all()
+
+ assert len(all_users) == 100
+
+ @pytest.mark.asyncio
+ async def test_bulk_update(self, async_session: AsyncSession):
+ """Test bulk updating multiple records."""
+ from app.models.user import User
+
+ # Create users with level 1
+ users = [User(**UserFactory.create(level=1)) for _ in range(50)]
+ async_session.add_all(users)
+ await async_session.commit()
+
+ # Bulk update all to level 2
+ await async_session.execute(
+ update(User)
+ .where(User.level == 1)
+ .values(level=2)
+ )
+ await async_session.commit()
+
+ async_session.expire_all()
+
+ result = await async_session.execute(
+ select(User).where(User.level == 2)
+ )
+ updated_users = result.scalars().all()
+
+ assert len(updated_users) == 50
+
+ @pytest.mark.asyncio
+ async def test_bulk_delete(self, async_session: AsyncSession):
+ """Test bulk deleting multiple records."""
+ from app.models.user import User
+
+ # Create active and inactive users
+ active_users = [User(**UserFactory.create(is_active=True)) for _ in range(30)]
+ inactive_users = [User(**UserFactory.create(is_active=False)) for _ in range(20)]
+ async_session.add_all(active_users + inactive_users)
+ await async_session.commit()
+
+ # Delete inactive users
+ await async_session.execute(
+ delete(User).where(User.is_active == False) # noqa: E712
+ )
+ await async_session.commit()
+
+ result = await async_session.execute(select(User))
+ remaining_users = result.scalars().all()
+
+ assert len(remaining_users) == 30
+ assert all(u.is_active for u in remaining_users)
diff --git a/tests/database/test_data_integrity.py b/tests/database/test_data_integrity.py
new file mode 100644
index 0000000..df98859
--- /dev/null
+++ b/tests/database/test_data_integrity.py
@@ -0,0 +1,470 @@
+"""
+Data Integrity and Constraint Tests
+
+Tests to verify data integrity constraints are enforced correctly:
+- Unique constraint violations
+- Foreign key constraint enforcement
+- NOT NULL constraint enforcement
+- Check constraint enforcement
+- Data validation rules
+"""
+import pytest
+from datetime import datetime, timedelta
+from sqlalchemy import text
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from .conftest import UserFactory, CourseFactory, ConceptFactory
+
+pytestmark = [pytest.mark.requires_db, pytest.mark.unit]
+
+
+class TestUniqueConstraints:
+ """Test unique constraint enforcement."""
+
+ @pytest.mark.asyncio
+ async def test_duplicate_email_rejected(self, async_session: AsyncSession):
+ """Verify duplicate email addresses are rejected."""
+ from app.models.user import User
+
+ # Create first user
+ user1_data = UserFactory.create(email="unique@example.com")
+ user1 = User(**user1_data)
+ async_session.add(user1)
+ await async_session.commit()
+
+ # Attempt to create second user with same email
+ user2_data = UserFactory.create(email="unique@example.com")
+ user2 = User(**user2_data)
+ async_session.add(user2)
+
+ with pytest.raises(IntegrityError):
+ await async_session.commit()
+
+ @pytest.mark.asyncio
+ async def test_duplicate_username_rejected(self, async_session: AsyncSession):
+ """Verify duplicate usernames are rejected."""
+ from app.models.user import User
+
+ # Create first user
+ user1_data = UserFactory.create(username="uniqueuser")
+ user1 = User(**user1_data)
+ async_session.add(user1)
+ await async_session.commit()
+
+ # Attempt to create second user with same username
+ user2_data = UserFactory.create(username="uniqueuser")
+ user2 = User(**user2_data)
+ async_session.add(user2)
+
+ with pytest.raises(IntegrityError):
+ await async_session.commit()
+
+ @pytest.mark.asyncio
+ async def test_case_sensitivity_email(self, async_session: AsyncSession):
+ """Test email case sensitivity handling."""
+ from app.models.user import User
+
+ user1_data = UserFactory.create(email="Test@Example.com")
+ user1 = User(**user1_data)
+ async_session.add(user1)
+ await async_session.commit()
+
+ # Note: This behavior depends on database collation
+ # Some databases treat email as case-insensitive
+ user2_data = UserFactory.create(email="test@example.com")
+ user2 = User(**user2_data)
+ async_session.add(user2)
+
+ try:
+ await async_session.commit()
+ # If commit succeeds, emails are case-sensitive
+ assert True
+ except IntegrityError:
+ # If fails, emails are case-insensitive (good for email uniqueness)
+ assert True
+
+
+class TestNotNullConstraints:
+ """Test NOT NULL constraint enforcement."""
+
+ @pytest.mark.asyncio
+ async def test_user_email_required(self, async_session: AsyncSession):
+ """Verify user email cannot be null."""
+ from app.models.user import User
+
+ user_data = UserFactory.create()
+ user_data["email"] = None
+ user = User(**user_data)
+ async_session.add(user)
+
+ with pytest.raises(IntegrityError):
+ await async_session.commit()
+
+ @pytest.mark.asyncio
+ async def test_user_username_required(self, async_session: AsyncSession):
+ """Verify user username cannot be null."""
+ from app.models.user import User
+
+ user_data = UserFactory.create()
+ user_data["username"] = None
+ user = User(**user_data)
+ async_session.add(user)
+
+ with pytest.raises(IntegrityError):
+ await async_session.commit()
+
+ @pytest.mark.asyncio
+ async def test_user_password_required(self, async_session: AsyncSession):
+ """Verify user password hash cannot be null."""
+ from app.models.user import User
+
+ user_data = UserFactory.create()
+ user_data["hashed_password"] = None
+ user = User(**user_data)
+ async_session.add(user)
+
+ with pytest.raises(IntegrityError):
+ await async_session.commit()
+
+
+class TestForeignKeyConstraints:
+ """Test foreign key constraint enforcement."""
+
+ @pytest.mark.asyncio
+ async def test_enrollment_requires_valid_user(self, async_session: AsyncSession):
+ """Verify enrollment cannot reference non-existent user."""
+ from app.models.course import Course, Enrollment
+
+ # Create course
+ course_data = CourseFactory.create()
+ course = Course(**course_data)
+ async_session.add(course)
+ await async_session.commit()
+
+ # Attempt enrollment with invalid user_id
+ enrollment = Enrollment(
+ user_id=99999, # Non-existent user
+ course_id=course.id,
+ progress=0.0
+ )
+ async_session.add(enrollment)
+
+ with pytest.raises(IntegrityError):
+ await async_session.commit()
+
+ @pytest.mark.asyncio
+ async def test_enrollment_requires_valid_course(self, async_session: AsyncSession):
+ """Verify enrollment cannot reference non-existent course."""
+ from app.models.user import User
+ from app.models.course import Enrollment
+
+ # Create user
+ user_data = UserFactory.create()
+ user = User(**user_data)
+ async_session.add(user)
+ await async_session.commit()
+
+ # Attempt enrollment with invalid course_id
+ enrollment = Enrollment(
+ user_id=user.id,
+ course_id=99999, # Non-existent course
+ progress=0.0
+ )
+ async_session.add(enrollment)
+
+ with pytest.raises(IntegrityError):
+ await async_session.commit()
+
+ @pytest.mark.asyncio
+ async def test_spaced_repetition_card_requires_valid_user(
+ self, async_session: AsyncSession
+ ):
+ """Verify spaced repetition card requires valid user."""
+ from app.models.spaced_repetition import SpacedRepetitionCard, Concept
+
+ # Create concept
+ concept = Concept(name="Test Concept", description="Test")
+ async_session.add(concept)
+ await async_session.commit()
+
+ # Attempt to create card with invalid user
+ card = SpacedRepetitionCard(
+ user_id=99999, # Non-existent user
+ concept_id=concept.id,
+ difficulty=5.0,
+ stability=2.5,
+ retrievability=0.9,
+ review_count=0,
+ next_review_at=datetime.utcnow() + timedelta(days=1)
+ )
+ async_session.add(card)
+
+ with pytest.raises(IntegrityError):
+ await async_session.commit()
+
+
+class TestDataValidation:
+ """Test application-level data validation rules."""
+
+ @pytest.mark.asyncio
+ async def test_user_xp_non_negative(self, async_session: AsyncSession):
+ """Verify XP cannot be negative (application rule)."""
+ from app.models.user import User
+
+ user_data = UserFactory.create(total_xp=-100)
+ user = User(**user_data)
+ async_session.add(user)
+
+ # Note: This depends on whether there's a CHECK constraint
+ # If no constraint, this tests application validation layer
+ try:
+ await async_session.commit()
+ # If commit succeeds, there's no DB constraint - app should validate
+ assert user.total_xp >= 0 or True # App should handle
+ except IntegrityError:
+ # DB constraint exists
+ assert True
+
+ @pytest.mark.asyncio
+ async def test_user_level_positive(self, async_session: AsyncSession):
+ """Verify user level must be positive."""
+ from app.models.user import User
+
+ user_data = UserFactory.create(level=0)
+ user = User(**user_data)
+ async_session.add(user)
+
+ # Level should be >= 1
+ try:
+ await async_session.commit()
+ # No DB constraint - app should validate
+ assert True
+ except IntegrityError:
+ # DB constraint exists
+ assert True
+
+ @pytest.mark.asyncio
+ async def test_progress_percentage_range(self, async_session: AsyncSession):
+ """Verify progress is between 0 and 1."""
+ from app.models.user import User
+ from app.models.course import Course, Enrollment
+
+ # Create user and course
+ user = User(**UserFactory.create())
+ course = Course(**CourseFactory.create())
+ async_session.add_all([user, course])
+ await async_session.commit()
+
+ # Test with invalid progress > 1
+ enrollment = Enrollment(
+ user_id=user.id,
+ course_id=course.id,
+ progress=1.5 # Invalid: > 1
+ )
+ async_session.add(enrollment)
+
+ try:
+ await async_session.commit()
+ # No DB constraint - app should validate
+ pytest.skip("No DB constraint on progress range")
+ except IntegrityError:
+ # DB constraint exists
+ assert True
+
+ @pytest.mark.asyncio
+ async def test_fsrs_difficulty_range(self, async_session: AsyncSession):
+ """Verify FSRS difficulty is within expected range (1-10)."""
+ from app.models.user import User
+ from app.models.spaced_repetition import SpacedRepetitionCard, Concept
+
+ # Create user and concept
+ user = User(**UserFactory.create())
+ concept = Concept(name="Test", description="Test")
+ async_session.add_all([user, concept])
+ await async_session.commit()
+
+ # Test with out-of-range difficulty
+ card = SpacedRepetitionCard(
+ user_id=user.id,
+ concept_id=concept.id,
+ difficulty=15.0, # Out of typical range
+ stability=2.5,
+ retrievability=0.9,
+ review_count=0,
+ next_review_at=datetime.utcnow() + timedelta(days=1)
+ )
+ async_session.add(card)
+
+ try:
+ await async_session.commit()
+ # No DB constraint - app should validate
+ pytest.skip("No DB constraint on FSRS difficulty range")
+ except IntegrityError:
+ assert True
+
+
+class TestDataConsistency:
+ """Test data consistency rules across related tables."""
+
+ @pytest.mark.asyncio
+ async def test_user_stats_matches_achievements(self, async_session: AsyncSession):
+ """Verify user stats are consistent with achievements."""
+ from app.models.user import User
+ from app.models.gamification import UserStats, UserAchievement
+
+ # Create user with stats
+ user = User(**UserFactory.create())
+ async_session.add(user)
+ await async_session.commit()
+
+ # Create stats
+ stats = UserStats(
+ user_id=user.id,
+ total_achievements=0,
+ total_reviews=0,
+ total_correct=0
+ )
+ async_session.add(stats)
+ await async_session.commit()
+
+ # Create achievement
+ achievement = UserAchievement(
+ user_id=user.id,
+ achievement_type="STREAK_MILESTONE",
+ name="First Week Streak",
+ description="Maintained a 7-day streak",
+ xp_reward=100
+ )
+ async_session.add(achievement)
+ await async_session.commit()
+
+ # Stats should be updated (this tests business logic)
+ # In real app, this would be done via service layer
+ await async_session.refresh(stats)
+
+ # This is a consistency check reminder
+ achievements = await async_session.execute(
+ text("SELECT COUNT(*) FROM user_achievements WHERE user_id = :user_id"),
+ {"user_id": user.id}
+ )
+ count = achievements.scalar()
+
+ # Note: This test documents expected consistency
+ # Real enforcement should be in application layer or triggers
+ assert count >= 0 # At minimum, count should be valid
+
+ @pytest.mark.asyncio
+ async def test_review_log_timestamp_consistency(self, async_session: AsyncSession):
+ """Verify review timestamps are logically consistent."""
+ from app.models.user import User
+ from app.models.spaced_repetition import SpacedRepetitionCard, Concept, ReviewLog
+
+ # Create user and concept
+ user = User(**UserFactory.create())
+ concept = Concept(name="Test", description="Test")
+ async_session.add_all([user, concept])
+ await async_session.commit()
+
+ # Create card
+ card = SpacedRepetitionCard(
+ user_id=user.id,
+ concept_id=concept.id,
+ difficulty=5.0,
+ stability=2.5,
+ retrievability=0.9,
+ review_count=0,
+ next_review_at=datetime.utcnow() + timedelta(days=1)
+ )
+ async_session.add(card)
+ await async_session.commit()
+
+ # Create review log with future timestamp
+ future_review = ReviewLog(
+ card_id=card.id,
+ rating=4,
+ reviewed_at=datetime.utcnow() + timedelta(days=365), # Future
+ response_time_ms=2000
+ )
+ async_session.add(future_review)
+
+ # Note: Without a CHECK constraint, this will succeed
+ # App should validate timestamps are not in the future
+ try:
+ await async_session.commit()
+ pytest.skip("No DB constraint preventing future review timestamps")
+ except IntegrityError:
+ assert True
+
+
+class TestEdgeCases:
+ """Test edge cases and boundary conditions."""
+
+ @pytest.mark.asyncio
+ async def test_empty_string_vs_null(self, async_session: AsyncSession):
+ """Verify distinction between empty string and NULL."""
+ from app.models.user import User
+
+ # Create user with empty full_name (not NULL)
+ user_data = UserFactory.create(full_name="")
+ user = User(**user_data)
+ async_session.add(user)
+ await async_session.commit()
+
+ await async_session.refresh(user)
+
+ # Empty string should be preserved, not converted to NULL
+ assert user.full_name == "" or user.full_name is None
+
+ @pytest.mark.asyncio
+ async def test_very_long_strings(self, async_session: AsyncSession):
+ """Test handling of very long string values."""
+ from app.models.user import User
+
+ # Test with very long username (may exceed column limit)
+ long_username = "a" * 1000
+ user_data = UserFactory.create(username=long_username)
+ user = User(**user_data)
+ async_session.add(user)
+
+ try:
+ await async_session.commit()
+ # If succeeds, no length constraint or VARCHAR is large enough
+ await async_session.refresh(user)
+ # Verify it was stored (possibly truncated)
+ assert len(user.username) > 0
+ except Exception:
+ # String too long for column
+ assert True
+
+ @pytest.mark.asyncio
+ async def test_unicode_characters(self, async_session: AsyncSession):
+ """Verify unicode characters are handled correctly."""
+ from app.models.user import User
+
+ # Test with various unicode characters
+ unicode_name = "José García 日本語 🎓"
+ user_data = UserFactory.create(full_name=unicode_name)
+ user = User(**user_data)
+ async_session.add(user)
+ await async_session.commit()
+
+ await async_session.refresh(user)
+ assert user.full_name == unicode_name
+
+ @pytest.mark.asyncio
+ async def test_timestamp_precision(self, async_session: AsyncSession):
+ """Verify timestamp precision is maintained."""
+ from app.models.user import User
+
+ user_data = UserFactory.create()
+ user = User(**user_data)
+ async_session.add(user)
+ await async_session.commit()
+
+ await async_session.refresh(user)
+
+ # created_at should be set automatically
+ assert user.created_at is not None
+ # Should be recent (within last minute)
+ assert (datetime.utcnow() - user.created_at.replace(tzinfo=None)).seconds < 60
diff --git a/tests/database/test_performance.py b/tests/database/test_performance.py
new file mode 100644
index 0000000..b7f2789
--- /dev/null
+++ b/tests/database/test_performance.py
@@ -0,0 +1,590 @@
+"""
+Performance Benchmark Tests
+
+Tests to measure and track database performance metrics.
+These tests establish baselines and detect performance regressions.
+
+Test Categories:
+- Query execution time benchmarks
+- Bulk operation performance
+- Index effectiveness
+- Connection pool behavior
+- Memory usage patterns
+"""
+import pytest
+import time
+import statistics
+from datetime import datetime, timedelta
+from typing import List
+from sqlalchemy import select, text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from .conftest import UserFactory, CourseFactory
+
+pytestmark = [pytest.mark.requires_db, pytest.mark.benchmark, pytest.mark.slow]
+
+
+# Performance thresholds (in seconds)
+THRESHOLDS = {
+ "single_insert": 0.1,
+ "single_select": 0.05,
+ "bulk_insert_100": 1.0,
+ "bulk_insert_1000": 5.0,
+ "bulk_select_100": 0.2,
+ "complex_query": 0.5,
+ "join_query": 0.3,
+}
+
+
+def measure_time(func):
+ """Decorator to measure function execution time."""
+ async def wrapper(*args, **kwargs):
+ start = time.perf_counter()
+ result = await func(*args, **kwargs)
+ end = time.perf_counter()
+ return result, end - start
+ return wrapper
+
+
+class TestInsertPerformance:
+ """Test insert operation performance."""
+
+ @pytest.mark.asyncio
+ async def test_single_user_insert_time(self, async_session: AsyncSession):
+ """Benchmark single user insert time."""
+ from app.models.user import User
+
+ times: List[float] = []
+
+ for _ in range(10):
+ user_data = UserFactory.create()
+ user = User(**user_data)
+
+ start = time.perf_counter()
+ async_session.add(user)
+ await async_session.commit()
+ end = time.perf_counter()
+
+ times.append(end - start)
+
+ avg_time = statistics.mean(times)
+ max_time = max(times)
+
+ assert avg_time < THRESHOLDS["single_insert"], \
+ f"Average insert time {avg_time:.4f}s exceeds threshold"
+
+ # Store for reporting
+ pytest.benchmark_result = {
+ "test": "single_user_insert",
+ "avg_time": avg_time,
+ "max_time": max_time,
+ "min_time": min(times),
+ "std_dev": statistics.stdev(times) if len(times) > 1 else 0,
+ "samples": len(times)
+ }
+
+ @pytest.mark.asyncio
+ async def test_bulk_insert_100_users(self, async_session: AsyncSession):
+ """Benchmark inserting 100 users in batch."""
+ from app.models.user import User
+
+ users = [User(**UserFactory.create()) for _ in range(100)]
+
+ start = time.perf_counter()
+ async_session.add_all(users)
+ await async_session.commit()
+ end = time.perf_counter()
+
+ elapsed = end - start
+ rate = 100 / elapsed # users per second
+
+ assert elapsed < THRESHOLDS["bulk_insert_100"], \
+ f"Bulk insert 100 took {elapsed:.4f}s, exceeds threshold"
+
+ pytest.benchmark_result = {
+ "test": "bulk_insert_100",
+ "total_time": elapsed,
+ "records": 100,
+ "rate": rate,
+ "per_record": elapsed / 100
+ }
+
+ @pytest.mark.asyncio
+ async def test_bulk_insert_1000_users(self, async_session: AsyncSession):
+ """Benchmark inserting 1000 users in batch."""
+ from app.models.user import User
+
+ users = [User(**UserFactory.create()) for _ in range(1000)]
+
+ start = time.perf_counter()
+ async_session.add_all(users)
+ await async_session.commit()
+ end = time.perf_counter()
+
+ elapsed = end - start
+ rate = 1000 / elapsed
+
+ assert elapsed < THRESHOLDS["bulk_insert_1000"], \
+ f"Bulk insert 1000 took {elapsed:.4f}s, exceeds threshold"
+
+ pytest.benchmark_result = {
+ "test": "bulk_insert_1000",
+ "total_time": elapsed,
+ "records": 1000,
+ "rate": rate,
+ "per_record": elapsed / 1000
+ }
+
+
+class TestSelectPerformance:
+ """Test select/read operation performance."""
+
+ @pytest.mark.asyncio
+ async def test_single_user_select_by_id(self, async_session: AsyncSession):
+ """Benchmark single user select by primary key."""
+ from app.models.user import User
+
+ # Setup: create users
+ users = [User(**UserFactory.create()) for _ in range(100)]
+ async_session.add_all(users)
+ await async_session.commit()
+
+ user_ids = [u.id for u in users]
+ times: List[float] = []
+
+ for user_id in user_ids[:20]: # Sample 20 queries
+ start = time.perf_counter()
+ result = await async_session.execute(
+ select(User).where(User.id == user_id)
+ )
+ _ = result.scalar_one()
+ end = time.perf_counter()
+
+ times.append(end - start)
+ async_session.expire_all()
+
+ avg_time = statistics.mean(times)
+
+ assert avg_time < THRESHOLDS["single_select"], \
+ f"Average select time {avg_time:.4f}s exceeds threshold"
+
+ pytest.benchmark_result = {
+ "test": "single_select_by_id",
+ "avg_time": avg_time,
+ "max_time": max(times),
+ "min_time": min(times),
+ "samples": len(times)
+ }
+
+ @pytest.mark.asyncio
+ async def test_select_by_indexed_column(self, async_session: AsyncSession):
+ """Benchmark select using indexed email column."""
+ from app.models.user import User
+
+ # Setup
+ users = [User(**UserFactory.create()) for _ in range(100)]
+ async_session.add_all(users)
+ await async_session.commit()
+
+ emails = [u.email for u in users]
+ times: List[float] = []
+
+ for email in emails[:20]:
+ start = time.perf_counter()
+ result = await async_session.execute(
+ select(User).where(User.email == email)
+ )
+ _ = result.scalar_one_or_none()
+ end = time.perf_counter()
+
+ times.append(end - start)
+ async_session.expire_all()
+
+ avg_time = statistics.mean(times)
+
+ assert avg_time < THRESHOLDS["single_select"], \
+ f"Average indexed select time {avg_time:.4f}s exceeds threshold"
+
+ pytest.benchmark_result = {
+ "test": "select_by_email_index",
+ "avg_time": avg_time,
+ "max_time": max(times),
+ "samples": len(times)
+ }
+
+ @pytest.mark.asyncio
+ async def test_bulk_select_all_users(self, async_session: AsyncSession):
+ """Benchmark selecting all users."""
+ from app.models.user import User
+
+ # Setup
+ users = [User(**UserFactory.create()) for _ in range(100)]
+ async_session.add_all(users)
+ await async_session.commit()
+
+ async_session.expire_all()
+
+ start = time.perf_counter()
+ result = await async_session.execute(select(User))
+ all_users = result.scalars().all()
+ end = time.perf_counter()
+
+ elapsed = end - start
+
+ assert len(all_users) == 100
+ assert elapsed < THRESHOLDS["bulk_select_100"], \
+ f"Bulk select 100 took {elapsed:.4f}s, exceeds threshold"
+
+ pytest.benchmark_result = {
+ "test": "bulk_select_100",
+ "total_time": elapsed,
+ "records": len(all_users),
+ "per_record": elapsed / len(all_users)
+ }
+
+
+class TestJoinPerformance:
+ """Test join query performance."""
+
+ @pytest.mark.asyncio
+ async def test_user_enrollment_join(self, async_session: AsyncSession):
+ """Benchmark user-enrollment join query."""
+ from app.models.user import User
+ from app.models.course import Course, Enrollment
+ from sqlalchemy.orm import selectinload
+
+ # Setup: create users with enrollments
+ users = [User(**UserFactory.create()) for _ in range(50)]
+ courses = [Course(**CourseFactory.create()) for _ in range(10)]
+ async_session.add_all(users + courses)
+ await async_session.commit()
+
+ enrollments = [
+ Enrollment(
+ user_id=users[i % 50].id,
+ course_id=courses[i % 10].id,
+ progress=0.5
+ )
+ for i in range(200)
+ ]
+ async_session.add_all(enrollments)
+ await async_session.commit()
+
+ async_session.expire_all()
+
+ # Benchmark join query
+ start = time.perf_counter()
+ result = await async_session.execute(
+ select(User)
+ .options(selectinload(User.enrollments))
+ .where(User.is_active == True) # noqa: E712
+ )
+ users_with_enrollments = result.scalars().all()
+ end = time.perf_counter()
+
+ elapsed = end - start
+
+ assert elapsed < THRESHOLDS["join_query"], \
+ f"Join query took {elapsed:.4f}s, exceeds threshold"
+
+ pytest.benchmark_result = {
+ "test": "user_enrollment_join",
+ "total_time": elapsed,
+ "users": len(users_with_enrollments),
+ "total_enrollments": sum(len(u.enrollments) for u in users_with_enrollments)
+ }
+
+ @pytest.mark.asyncio
+ async def test_spaced_repetition_due_cards_query(
+ self, async_session: AsyncSession
+ ):
+ """Benchmark query for due spaced repetition cards."""
+ from app.models.user import User
+ from app.models.spaced_repetition import SpacedRepetitionCard, Concept
+
+ # Setup
+ user = User(**UserFactory.create())
+ async_session.add(user)
+ await async_session.commit()
+
+ concepts = [
+ Concept(name=f"Concept {i}", description=f"Desc {i}")
+ for i in range(100)
+ ]
+ async_session.add_all(concepts)
+ await async_session.commit()
+
+ cards = [
+ SpacedRepetitionCard(
+ user_id=user.id,
+ concept_id=concepts[i].id,
+ difficulty=5.0,
+ stability=2.5,
+ retrievability=0.9,
+ review_count=i,
+ next_review_at=datetime.utcnow() + timedelta(days=i - 50)
+ )
+ for i in range(100)
+ ]
+ async_session.add_all(cards)
+ await async_session.commit()
+
+ async_session.expire_all()
+
+ # Benchmark due cards query
+ start = time.perf_counter()
+ result = await async_session.execute(
+ select(SpacedRepetitionCard)
+ .where(SpacedRepetitionCard.user_id == user.id)
+ .where(SpacedRepetitionCard.next_review_at <= datetime.utcnow())
+ .order_by(SpacedRepetitionCard.next_review_at)
+ )
+ due_cards = result.scalars().all()
+ end = time.perf_counter()
+
+ elapsed = end - start
+
+ assert elapsed < THRESHOLDS["complex_query"], \
+ f"Due cards query took {elapsed:.4f}s, exceeds threshold"
+
+ pytest.benchmark_result = {
+ "test": "due_cards_query",
+ "total_time": elapsed,
+ "cards_due": len(due_cards),
+ "total_cards": 100
+ }
+
+
+class TestComplexQueryPerformance:
+ """Test complex query performance."""
+
+ @pytest.mark.asyncio
+ async def test_aggregate_user_statistics(self, async_session: AsyncSession):
+ """Benchmark aggregate statistics query."""
+ from app.models.user import User
+ from sqlalchemy import func
+
+ # Setup
+ users = [
+ User(**UserFactory.create(
+ total_xp=i * 100,
+ level=i % 10 + 1,
+ streak_days=i % 30
+ ))
+ for i in range(100)
+ ]
+ async_session.add_all(users)
+ await async_session.commit()
+
+ # Benchmark aggregate query
+ start = time.perf_counter()
+ result = await async_session.execute(
+ select(
+ func.count(User.id).label("total_users"),
+ func.avg(User.total_xp).label("avg_xp"),
+ func.max(User.total_xp).label("max_xp"),
+ func.sum(User.total_xp).label("total_xp"),
+ func.avg(User.level).label("avg_level")
+ )
+ )
+ stats = result.one()
+ end = time.perf_counter()
+
+ elapsed = end - start
+
+ assert stats.total_users == 100
+ assert elapsed < THRESHOLDS["complex_query"], \
+ f"Aggregate query took {elapsed:.4f}s, exceeds threshold"
+
+ pytest.benchmark_result = {
+ "test": "aggregate_statistics",
+ "total_time": elapsed,
+ "total_users": stats.total_users,
+ "avg_xp": float(stats.avg_xp) if stats.avg_xp else 0
+ }
+
+ @pytest.mark.asyncio
+ async def test_leaderboard_query(self, async_session: AsyncSession):
+ """Benchmark leaderboard query (top users by XP)."""
+ from app.models.user import User
+
+ # Setup
+ users = [
+ User(**UserFactory.create(
+ total_xp=i * 100,
+ level=(i // 100) + 1
+ ))
+ for i in range(500)
+ ]
+ async_session.add_all(users)
+ await async_session.commit()
+
+ async_session.expire_all()
+
+ # Benchmark leaderboard query
+ start = time.perf_counter()
+ result = await async_session.execute(
+ select(User)
+ .where(User.is_active == True) # noqa: E712
+ .order_by(User.total_xp.desc())
+ .limit(100)
+ )
+ top_users = result.scalars().all()
+ end = time.perf_counter()
+
+ elapsed = end - start
+
+ assert len(top_users) == 100
+ # Verify ordering
+ xps = [u.total_xp for u in top_users]
+ assert xps == sorted(xps, reverse=True)
+
+ assert elapsed < THRESHOLDS["complex_query"], \
+ f"Leaderboard query took {elapsed:.4f}s, exceeds threshold"
+
+ pytest.benchmark_result = {
+ "test": "leaderboard_top_100",
+ "total_time": elapsed,
+ "returned": len(top_users),
+ "total_users": 500
+ }
+
+
+class TestIndexEffectiveness:
+ """Test that indexes are being used effectively."""
+
+ @pytest.mark.asyncio
+ async def test_email_index_scan_vs_full_scan(self, async_session: AsyncSession):
+ """Compare indexed vs non-indexed query performance."""
+ from app.models.user import User
+
+ # Setup: create many users
+ users = [User(**UserFactory.create()) for _ in range(500)]
+ async_session.add_all(users)
+ await async_session.commit()
+
+ target_email = users[250].email
+ target_name = users[250].full_name
+
+ async_session.expire_all()
+
+ # Query by indexed column (email)
+ start_indexed = time.perf_counter()
+ for _ in range(10):
+ result = await async_session.execute(
+ select(User).where(User.email == target_email)
+ )
+ _ = result.scalar_one()
+ async_session.expire_all()
+ end_indexed = time.perf_counter()
+ indexed_time = (end_indexed - start_indexed) / 10
+
+ # Query by non-indexed column (full_name)
+ start_non_indexed = time.perf_counter()
+ for _ in range(10):
+ result = await async_session.execute(
+ select(User).where(User.full_name == target_name)
+ )
+ _ = result.scalar_one_or_none()
+ async_session.expire_all()
+ end_non_indexed = time.perf_counter()
+ non_indexed_time = (end_non_indexed - start_non_indexed) / 10
+
+ # Indexed query should be faster (in a real DB, might not matter for SQLite)
+ pytest.benchmark_result = {
+ "test": "index_effectiveness",
+ "indexed_avg": indexed_time,
+ "non_indexed_avg": non_indexed_time,
+ "speedup": non_indexed_time / indexed_time if indexed_time > 0 else 0
+ }
+
+ # Note: In SQLite with small data, difference may be negligible
+ # This test is more meaningful with PostgreSQL
+ assert True # Document results
+
+
+class TestTransactionPerformance:
+ """Test transaction handling performance."""
+
+ @pytest.mark.asyncio
+ async def test_commit_frequency_impact(self, async_session: AsyncSession):
+ """Compare single commit vs multiple commits performance."""
+ from app.models.user import User
+
+ # Test 1: Single commit after all inserts
+ users1 = [User(**UserFactory.create()) for _ in range(100)]
+
+ start_single = time.perf_counter()
+ async_session.add_all(users1)
+ await async_session.commit()
+ end_single = time.perf_counter()
+ single_commit_time = end_single - start_single
+
+ # Test 2: Commit after each insert (anti-pattern)
+ start_multi = time.perf_counter()
+ for _ in range(100):
+ user = User(**UserFactory.create())
+ async_session.add(user)
+ await async_session.commit()
+ end_multi = time.perf_counter()
+ multi_commit_time = end_multi - start_multi
+
+ pytest.benchmark_result = {
+ "test": "commit_frequency",
+ "single_commit_100_records": single_commit_time,
+ "multi_commit_100_records": multi_commit_time,
+ "overhead_factor": multi_commit_time / single_commit_time
+ if single_commit_time > 0 else 0
+ }
+
+ # Single commit should be significantly faster
+ assert single_commit_time < multi_commit_time, \
+ "Batch commits should be faster than individual commits"
+
+
+class BenchmarkReporter:
+ """Utility class to collect and report benchmark results."""
+
+ results: List[dict] = []
+
+ @classmethod
+ def record(cls, result: dict):
+ """Record a benchmark result."""
+ cls.results.append({
+ **result,
+ "timestamp": datetime.utcnow().isoformat()
+ })
+
+ @classmethod
+ def generate_report(cls) -> dict:
+ """Generate summary report of all benchmarks."""
+ if not cls.results:
+ return {"status": "no_results"}
+
+ return {
+ "total_benchmarks": len(cls.results),
+ "timestamp": datetime.utcnow().isoformat(),
+ "results": cls.results,
+ "summary": {
+ "passed": sum(1 for r in cls.results if r.get("passed", True)),
+ "failed": sum(1 for r in cls.results if not r.get("passed", True))
+ }
+ }
+
+
+@pytest.fixture(scope="module", autouse=True)
+def collect_benchmark_results(request):
+ """Collect benchmark results at module level."""
+ yield
+ # After all tests in module, generate report
+ report = BenchmarkReporter.generate_report()
+ if report.get("results"):
+ print("\n" + "=" * 60)
+ print("BENCHMARK RESULTS SUMMARY")
+ print("=" * 60)
+ for result in report.get("results", []):
+ print(f" {result.get('test', 'unknown')}: ", end="")
+ if "avg_time" in result:
+ print(f"{result['avg_time']:.4f}s avg")
+ elif "total_time" in result:
+ print(f"{result['total_time']:.4f}s total")
diff --git a/tests/database/test_relationships.py b/tests/database/test_relationships.py
new file mode 100644
index 0000000..9d7d853
--- /dev/null
+++ b/tests/database/test_relationships.py
@@ -0,0 +1,561 @@
+"""
+Relationship and Cascade Tests
+
+Tests to verify database relationships and cascading behaviors work correctly.
+Ensures referential integrity is maintained and cascade operations propagate
+as expected.
+
+Test Categories:
+- One-to-Many relationships
+- Many-to-Many relationships
+- Cascade delete operations
+- Orphan record handling
+- Relationship loading (eager/lazy)
+"""
+import pytest
+from datetime import datetime, timedelta
+from sqlalchemy import select, delete
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+
+from .conftest import UserFactory, CourseFactory
+
+pytestmark = [pytest.mark.requires_db, pytest.mark.unit]
+
+
+class TestUserRelationships:
+ """Test User entity relationships."""
+
+ @pytest.mark.asyncio
+ async def test_user_has_enrollments(self, async_session: AsyncSession):
+ """Verify user can have multiple enrollments."""
+ from app.models.user import User
+ from app.models.course import Course, Enrollment
+
+ user = User(**UserFactory.create())
+ course1 = Course(**CourseFactory.create())
+ course2 = Course(**CourseFactory.create())
+ async_session.add_all([user, course1, course2])
+ await async_session.commit()
+
+ # Create enrollments
+ enrollment1 = Enrollment(user_id=user.id, course_id=course1.id, progress=0.0)
+ enrollment2 = Enrollment(user_id=user.id, course_id=course2.id, progress=0.0)
+ async_session.add_all([enrollment1, enrollment2])
+ await async_session.commit()
+
+ # Load user with enrollments
+ result = await async_session.execute(
+ select(User)
+ .options(selectinload(User.enrollments))
+ .where(User.id == user.id)
+ )
+ loaded_user = result.scalar_one()
+
+ assert len(loaded_user.enrollments) == 2
+
+ @pytest.mark.asyncio
+ async def test_user_has_achievements(self, async_session: AsyncSession):
+ """Verify user can have multiple achievements."""
+ from app.models.user import User
+ from app.models.gamification import UserAchievement
+
+ user = User(**UserFactory.create())
+ async_session.add(user)
+ await async_session.commit()
+
+ achievements = [
+ UserAchievement(
+ user_id=user.id,
+ achievement_type="STREAK_MILESTONE",
+ name=f"Achievement {i}",
+ description=f"Description {i}",
+ xp_reward=100
+ )
+ for i in range(3)
+ ]
+ async_session.add_all(achievements)
+ await async_session.commit()
+
+ result = await async_session.execute(
+ select(User)
+ .options(selectinload(User.achievements))
+ .where(User.id == user.id)
+ )
+ loaded_user = result.scalar_one()
+
+ assert len(loaded_user.achievements) == 3
+
+ @pytest.mark.asyncio
+ async def test_user_has_spaced_repetition_cards(self, async_session: AsyncSession):
+ """Verify user can have multiple spaced repetition cards."""
+ from app.models.user import User
+ from app.models.spaced_repetition import SpacedRepetitionCard, Concept
+
+ user = User(**UserFactory.create())
+ async_session.add(user)
+ await async_session.commit()
+
+ # Create concepts and cards
+ for i in range(5):
+ concept = Concept(name=f"Concept {i}", description=f"Description {i}")
+ async_session.add(concept)
+ await async_session.commit()
+
+ card = SpacedRepetitionCard(
+ user_id=user.id,
+ concept_id=concept.id,
+ difficulty=5.0,
+ stability=2.5,
+ retrievability=0.9,
+ review_count=0,
+ next_review_at=datetime.utcnow() + timedelta(days=1)
+ )
+ async_session.add(card)
+ await async_session.commit()
+
+ result = await async_session.execute(
+ select(User)
+ .options(selectinload(User.spaced_repetition_cards))
+ .where(User.id == user.id)
+ )
+ loaded_user = result.scalar_one()
+
+ assert len(loaded_user.spaced_repetition_cards) == 5
+
+
+class TestCourseRelationships:
+ """Test Course entity relationships."""
+
+ @pytest.mark.asyncio
+ async def test_course_has_enrollments(self, async_session: AsyncSession):
+ """Verify course can have multiple enrolled students."""
+ from app.models.user import User
+ from app.models.course import Course, Enrollment
+
+ course = Course(**CourseFactory.create())
+ users = [User(**UserFactory.create()) for _ in range(5)]
+ async_session.add(course)
+ async_session.add_all(users)
+ await async_session.commit()
+
+ enrollments = [
+ Enrollment(user_id=user.id, course_id=course.id, progress=0.0)
+ for user in users
+ ]
+ async_session.add_all(enrollments)
+ await async_session.commit()
+
+ result = await async_session.execute(
+ select(Course)
+ .options(selectinload(Course.enrollments))
+ .where(Course.id == course.id)
+ )
+ loaded_course = result.scalar_one()
+
+ assert len(loaded_course.enrollments) == 5
+
+ @pytest.mark.asyncio
+ async def test_course_has_instructor(self, async_session: AsyncSession):
+ """Verify course can be assigned to an instructor."""
+ from app.models.user import User, Instructor
+ from app.models.course import Course
+
+ user = User(**UserFactory.create(is_instructor=True))
+ async_session.add(user)
+ await async_session.commit()
+
+ instructor = Instructor(
+ user_id=user.id,
+ bio="Test instructor",
+ expertise_areas="Python, AI"
+ )
+ async_session.add(instructor)
+ await async_session.commit()
+
+ course = Course(**CourseFactory.create(instructor_id=instructor.id))
+ async_session.add(course)
+ await async_session.commit()
+
+ result = await async_session.execute(
+ select(Course)
+ .options(selectinload(Course.instructor))
+ .where(Course.id == course.id)
+ )
+ loaded_course = result.scalar_one()
+
+ assert loaded_course.instructor_id == instructor.id
+
+
+class TestCascadeDelete:
+ """Test cascade delete operations."""
+
+ @pytest.mark.asyncio
+ async def test_delete_user_cascades_to_enrollments(
+ self, async_session: AsyncSession
+ ):
+ """Verify deleting user removes their enrollments."""
+ from app.models.user import User
+ from app.models.course import Course, Enrollment
+
+ user = User(**UserFactory.create())
+ course = Course(**CourseFactory.create())
+ async_session.add_all([user, course])
+ await async_session.commit()
+
+ enrollment = Enrollment(user_id=user.id, course_id=course.id, progress=0.5)
+ async_session.add(enrollment)
+ await async_session.commit()
+ enrollment_id = enrollment.id
+
+ # Delete user
+ await async_session.execute(
+ delete(User).where(User.id == user.id)
+ )
+ await async_session.commit()
+
+ # Verify enrollment was deleted
+ result = await async_session.execute(
+ select(Enrollment).where(Enrollment.id == enrollment_id)
+ )
+ deleted_enrollment = result.scalar_one_or_none()
+
+ assert deleted_enrollment is None
+
+ @pytest.mark.asyncio
+ async def test_delete_user_cascades_to_achievements(
+ self, async_session: AsyncSession
+ ):
+ """Verify deleting user removes their achievements."""
+ from app.models.user import User
+ from app.models.gamification import UserAchievement
+
+ user = User(**UserFactory.create())
+ async_session.add(user)
+ await async_session.commit()
+
+ achievement = UserAchievement(
+ user_id=user.id,
+ achievement_type="STREAK_MILESTONE",
+ name="Test",
+ description="Test",
+ xp_reward=100
+ )
+ async_session.add(achievement)
+ await async_session.commit()
+ achievement_id = achievement.id
+
+ # Delete user
+ await async_session.execute(
+ delete(User).where(User.id == user.id)
+ )
+ await async_session.commit()
+
+ # Verify achievement was deleted
+ result = await async_session.execute(
+ select(UserAchievement).where(UserAchievement.id == achievement_id)
+ )
+ deleted_achievement = result.scalar_one_or_none()
+
+ assert deleted_achievement is None
+
+ @pytest.mark.asyncio
+ async def test_delete_user_cascades_to_spaced_repetition(
+ self, async_session: AsyncSession
+ ):
+ """Verify deleting user removes their spaced repetition cards."""
+ from app.models.user import User
+ from app.models.spaced_repetition import SpacedRepetitionCard, Concept, ReviewLog
+
+ user = User(**UserFactory.create())
+ concept = Concept(name="Test", description="Test")
+ async_session.add_all([user, concept])
+ await async_session.commit()
+
+ card = SpacedRepetitionCard(
+ user_id=user.id,
+ concept_id=concept.id,
+ difficulty=5.0,
+ stability=2.5,
+ retrievability=0.9,
+ review_count=0,
+ next_review_at=datetime.utcnow() + timedelta(days=1)
+ )
+ async_session.add(card)
+ await async_session.commit()
+
+ review = ReviewLog(
+ card_id=card.id,
+ rating=4,
+ reviewed_at=datetime.utcnow(),
+ response_time_ms=2000
+ )
+ async_session.add(review)
+ await async_session.commit()
+ card_id = card.id
+ review_id = review.id
+
+ # Delete user
+ await async_session.execute(
+ delete(User).where(User.id == user.id)
+ )
+ await async_session.commit()
+
+ # Verify card was deleted
+ result = await async_session.execute(
+ select(SpacedRepetitionCard).where(SpacedRepetitionCard.id == card_id)
+ )
+ assert result.scalar_one_or_none() is None
+
+ # Verify review log was also deleted (cascade through card)
+ result = await async_session.execute(
+ select(ReviewLog).where(ReviewLog.id == review_id)
+ )
+ assert result.scalar_one_or_none() is None
+
+ @pytest.mark.asyncio
+ async def test_delete_course_cascades_to_enrollments(
+ self, async_session: AsyncSession
+ ):
+ """Verify deleting course removes all enrollments."""
+ from app.models.user import User
+ from app.models.course import Course, Enrollment
+
+ course = Course(**CourseFactory.create())
+ users = [User(**UserFactory.create()) for _ in range(3)]
+ async_session.add(course)
+ async_session.add_all(users)
+ await async_session.commit()
+
+ enrollments = [
+ Enrollment(user_id=user.id, course_id=course.id, progress=0.0)
+ for user in users
+ ]
+ async_session.add_all(enrollments)
+ await async_session.commit()
+ course_id = course.id
+
+ # Delete course
+ await async_session.execute(
+ delete(Course).where(Course.id == course_id)
+ )
+ await async_session.commit()
+
+ # Verify all enrollments were deleted
+ result = await async_session.execute(
+ select(Enrollment).where(Enrollment.course_id == course_id)
+ )
+ remaining_enrollments = result.scalars().all()
+
+ assert len(remaining_enrollments) == 0
+
+ @pytest.mark.asyncio
+ async def test_delete_card_cascades_to_review_logs(
+ self, async_session: AsyncSession
+ ):
+ """Verify deleting a spaced repetition card removes its review logs."""
+ from app.models.user import User
+ from app.models.spaced_repetition import SpacedRepetitionCard, Concept, ReviewLog
+
+ user = User(**UserFactory.create())
+ concept = Concept(name="Test", description="Test")
+ async_session.add_all([user, concept])
+ await async_session.commit()
+
+ card = SpacedRepetitionCard(
+ user_id=user.id,
+ concept_id=concept.id,
+ difficulty=5.0,
+ stability=2.5,
+ retrievability=0.9,
+ review_count=0,
+ next_review_at=datetime.utcnow() + timedelta(days=1)
+ )
+ async_session.add(card)
+ await async_session.commit()
+
+ # Create multiple review logs
+ reviews = [
+ ReviewLog(
+ card_id=card.id,
+ rating=i,
+ reviewed_at=datetime.utcnow(),
+ response_time_ms=2000
+ )
+ for i in range(1, 5)
+ ]
+ async_session.add_all(reviews)
+ await async_session.commit()
+ card_id = card.id
+
+ # Delete card
+ await async_session.execute(
+ delete(SpacedRepetitionCard).where(SpacedRepetitionCard.id == card_id)
+ )
+ await async_session.commit()
+
+ # Verify all review logs were deleted
+ result = await async_session.execute(
+ select(ReviewLog).where(ReviewLog.card_id == card_id)
+ )
+ remaining_logs = result.scalars().all()
+
+ assert len(remaining_logs) == 0
+
+
+class TestOrphanRecords:
+ """Test handling of orphaned records."""
+
+ @pytest.mark.asyncio
+ async def test_delete_enrollment_preserves_user(
+ self, async_session: AsyncSession
+ ):
+ """Verify deleting enrollment does not affect user."""
+ from app.models.user import User
+ from app.models.course import Course, Enrollment
+
+ user = User(**UserFactory.create())
+ course = Course(**CourseFactory.create())
+ async_session.add_all([user, course])
+ await async_session.commit()
+ user_id = user.id
+
+ enrollment = Enrollment(user_id=user.id, course_id=course.id, progress=0.5)
+ async_session.add(enrollment)
+ await async_session.commit()
+
+ # Delete enrollment
+ await async_session.execute(
+ delete(Enrollment).where(Enrollment.user_id == user_id)
+ )
+ await async_session.commit()
+
+ # Verify user still exists
+ result = await async_session.execute(
+ select(User).where(User.id == user_id)
+ )
+ preserved_user = result.scalar_one_or_none()
+
+ assert preserved_user is not None
+ assert preserved_user.id == user_id
+
+ @pytest.mark.asyncio
+ async def test_delete_enrollment_preserves_course(
+ self, async_session: AsyncSession
+ ):
+ """Verify deleting enrollment does not affect course."""
+ from app.models.user import User
+ from app.models.course import Course, Enrollment
+
+ user = User(**UserFactory.create())
+ course = Course(**CourseFactory.create())
+ async_session.add_all([user, course])
+ await async_session.commit()
+ course_id = course.id
+
+ enrollment = Enrollment(user_id=user.id, course_id=course.id, progress=0.5)
+ async_session.add(enrollment)
+ await async_session.commit()
+
+ # Delete enrollment
+ await async_session.execute(
+ delete(Enrollment).where(Enrollment.course_id == course_id)
+ )
+ await async_session.commit()
+
+ # Verify course still exists
+ result = await async_session.execute(
+ select(Course).where(Course.id == course_id)
+ )
+ preserved_course = result.scalar_one_or_none()
+
+ assert preserved_course is not None
+
+
+class TestRelationshipLoading:
+ """Test relationship loading strategies."""
+
+ @pytest.mark.asyncio
+ async def test_eager_load_enrollments(self, async_session: AsyncSession):
+ """Test eager loading of enrollments with selectinload."""
+ from app.models.user import User
+ from app.models.course import Course, Enrollment
+
+ user = User(**UserFactory.create())
+ courses = [Course(**CourseFactory.create()) for _ in range(3)]
+ async_session.add(user)
+ async_session.add_all(courses)
+ await async_session.commit()
+
+ enrollments = [
+ Enrollment(user_id=user.id, course_id=course.id, progress=0.0)
+ for course in courses
+ ]
+ async_session.add_all(enrollments)
+ await async_session.commit()
+
+ async_session.expire_all()
+
+ # Load with eager loading
+ result = await async_session.execute(
+ select(User)
+ .options(selectinload(User.enrollments).selectinload(Enrollment.course))
+ .where(User.id == user.id)
+ )
+ loaded_user = result.scalar_one()
+
+ # Verify enrollments and courses are loaded
+ assert len(loaded_user.enrollments) == 3
+ for enrollment in loaded_user.enrollments:
+ assert enrollment.course is not None
+
+ @pytest.mark.asyncio
+ async def test_nested_relationship_loading(self, async_session: AsyncSession):
+ """Test loading nested relationships."""
+ from app.models.user import User
+ from app.models.spaced_repetition import SpacedRepetitionCard, Concept, ReviewLog
+
+ user = User(**UserFactory.create())
+ concept = Concept(name="Test", description="Test")
+ async_session.add_all([user, concept])
+ await async_session.commit()
+
+ card = SpacedRepetitionCard(
+ user_id=user.id,
+ concept_id=concept.id,
+ difficulty=5.0,
+ stability=2.5,
+ retrievability=0.9,
+ review_count=0,
+ next_review_at=datetime.utcnow() + timedelta(days=1)
+ )
+ async_session.add(card)
+ await async_session.commit()
+
+ reviews = [
+ ReviewLog(
+ card_id=card.id,
+ rating=i,
+ reviewed_at=datetime.utcnow(),
+ response_time_ms=2000
+ )
+ for i in range(1, 4)
+ ]
+ async_session.add_all(reviews)
+ await async_session.commit()
+
+ async_session.expire_all()
+
+ # Load user with nested relationships
+ result = await async_session.execute(
+ select(User)
+ .options(
+ selectinload(User.spaced_repetition_cards)
+ .selectinload(SpacedRepetitionCard.review_logs)
+ )
+ .where(User.id == user.id)
+ )
+ loaded_user = result.scalar_one()
+
+ assert len(loaded_user.spaced_repetition_cards) == 1
+ assert len(loaded_user.spaced_repetition_cards[0].review_logs) == 3
diff --git a/tests/database/test_schema_integrity.py b/tests/database/test_schema_integrity.py
new file mode 100644
index 0000000..eeb67ce
--- /dev/null
+++ b/tests/database/test_schema_integrity.py
@@ -0,0 +1,319 @@
+"""
+Schema Integrity Tests
+
+Tests to verify the database schema is correctly defined and matches expectations.
+These tests should be run after migrations to ensure schema consistency.
+
+Test Categories:
+- Table existence verification
+- Column definitions (types, nullability, defaults)
+- Index verification
+- Constraint verification (unique, foreign keys)
+- Enum validation
+"""
+import pytest
+from sqlalchemy import inspect, text
+from sqlalchemy.engine import Engine
+
+pytestmark = [pytest.mark.requires_db, pytest.mark.unit]
+
+
+class TestTableExistence:
+ """Verify all expected tables exist in the schema."""
+
+ EXPECTED_TABLES = [
+ "users",
+ "instructors",
+ "courses",
+ "enrollments",
+ "concepts",
+ "user_concept_mastery",
+ "spaced_repetition_cards",
+ "review_logs",
+ "user_achievements",
+ "user_stats",
+ "daily_activities",
+ "chat_history",
+ ]
+
+ def test_all_required_tables_exist(self, sync_engine: Engine):
+ """Verify all required tables are present in the database."""
+ inspector = inspect(sync_engine)
+ existing_tables = inspector.get_table_names()
+
+ for table in self.EXPECTED_TABLES:
+ assert table in existing_tables, f"Missing required table: {table}"
+
+ def test_no_orphan_tables(self, sync_engine: Engine):
+ """Warn about unexpected tables that might be orphaned."""
+ inspector = inspect(sync_engine)
+ existing_tables = set(inspector.get_table_names())
+ expected_tables = set(self.EXPECTED_TABLES)
+
+ # SQLite internal tables
+ internal_tables = {"sqlite_sequence", "alembic_version"}
+ orphan_tables = existing_tables - expected_tables - internal_tables
+
+ # This is a warning, not a failure - unexpected tables might be intentional
+ if orphan_tables:
+ pytest.skip(f"Found unexpected tables (may be intentional): {orphan_tables}")
+
+
+class TestUserTableSchema:
+ """Verify User table schema correctness."""
+
+ def test_user_table_columns(self, sync_engine: Engine):
+ """Verify user table has all required columns."""
+ inspector = inspect(sync_engine)
+ columns = {col["name"]: col for col in inspector.get_columns("users")}
+
+ required_columns = [
+ "id", "email", "username", "hashed_password",
+ "full_name", "is_active", "is_instructor",
+ "created_at", "updated_at",
+ "total_xp", "level", "streak_days", "last_activity_date"
+ ]
+
+ for col_name in required_columns:
+ assert col_name in columns, f"Missing column: users.{col_name}"
+
+ def test_user_email_is_unique(self, sync_engine: Engine):
+ """Verify email column has unique constraint."""
+ inspector = inspect(sync_engine)
+ unique_constraints = inspector.get_unique_constraints("users")
+ indexes = inspector.get_indexes("users")
+
+ # Check for unique constraint or unique index on email
+ email_unique = any(
+ "email" in (uc.get("column_names", []) or [])
+ for uc in unique_constraints
+ ) or any(
+ idx.get("unique", False) and "email" in idx.get("column_names", [])
+ for idx in indexes
+ )
+
+ assert email_unique, "users.email should have unique constraint"
+
+ def test_user_username_is_unique(self, sync_engine: Engine):
+ """Verify username column has unique constraint."""
+ inspector = inspect(sync_engine)
+ unique_constraints = inspector.get_unique_constraints("users")
+ indexes = inspector.get_indexes("users")
+
+ username_unique = any(
+ "username" in (uc.get("column_names", []) or [])
+ for uc in unique_constraints
+ ) or any(
+ idx.get("unique", False) and "username" in idx.get("column_names", [])
+ for idx in indexes
+ )
+
+ assert username_unique, "users.username should have unique constraint"
+
+ def test_user_email_not_nullable(self, sync_engine: Engine):
+ """Verify email is a required field."""
+ inspector = inspect(sync_engine)
+ columns = {col["name"]: col for col in inspector.get_columns("users")}
+
+ assert not columns["email"]["nullable"], "users.email should not be nullable"
+
+ def test_user_default_values(self, sync_engine: Engine):
+ """Verify default values are set correctly."""
+ inspector = inspect(sync_engine)
+ columns = {col["name"]: col for col in inspector.get_columns("users")}
+
+ # Check boolean defaults
+ assert columns["is_active"].get("default") is not None, \
+ "users.is_active should have a default"
+ assert columns["is_instructor"].get("default") is not None, \
+ "users.is_instructor should have a default"
+
+
+class TestCourseTableSchema:
+ """Verify Course table schema correctness."""
+
+ def test_course_table_columns(self, sync_engine: Engine):
+ """Verify course table has all required columns."""
+ inspector = inspect(sync_engine)
+ columns = {col["name"]: col for col in inspector.get_columns("courses")}
+
+ required_columns = [
+ "id", "title", "description", "domain",
+ "instructor_id", "is_published", "created_at"
+ ]
+
+ for col_name in required_columns:
+ assert col_name in columns, f"Missing column: courses.{col_name}"
+
+
+class TestConceptTableSchema:
+ """Verify Concept table schema correctness."""
+
+ def test_concept_table_columns(self, sync_engine: Engine):
+ """Verify concept table has all required columns."""
+ inspector = inspect(sync_engine)
+ columns = {col["name"]: col for col in inspector.get_columns("concepts")}
+
+ required_columns = ["id", "name", "description"]
+
+ for col_name in required_columns:
+ assert col_name in columns, f"Missing column: concepts.{col_name}"
+
+
+class TestSpacedRepetitionSchema:
+ """Verify Spaced Repetition tables schema correctness."""
+
+ def test_spaced_repetition_card_columns(self, sync_engine: Engine):
+ """Verify spaced_repetition_cards table has FSRS columns."""
+ inspector = inspect(sync_engine)
+ columns = {col["name"]: col for col in inspector.get_columns("spaced_repetition_cards")}
+
+ # FSRS algorithm requires these columns
+ fsrs_columns = [
+ "difficulty", "stability", "retrievability",
+ "review_count", "next_review_at"
+ ]
+
+ for col_name in fsrs_columns:
+ assert col_name in columns, \
+ f"Missing FSRS column: spaced_repetition_cards.{col_name}"
+
+ def test_review_log_columns(self, sync_engine: Engine):
+ """Verify review_logs table has required tracking columns."""
+ inspector = inspect(sync_engine)
+ columns = {col["name"]: col for col in inspector.get_columns("review_logs")}
+
+ required_columns = [
+ "id", "card_id", "rating", "reviewed_at",
+ "response_time_ms"
+ ]
+
+ for col_name in required_columns:
+ assert col_name in columns, f"Missing column: review_logs.{col_name}"
+
+
+class TestIndexes:
+ """Verify database indexes are correctly defined."""
+
+ def test_users_email_index(self, sync_engine: Engine):
+ """Verify index on users.email for fast lookups."""
+ inspector = inspect(sync_engine)
+ indexes = inspector.get_indexes("users")
+
+ email_indexed = any(
+ "email" in idx.get("column_names", [])
+ for idx in indexes
+ )
+
+ assert email_indexed, "users.email should be indexed"
+
+ def test_users_username_index(self, sync_engine: Engine):
+ """Verify index on users.username for fast lookups."""
+ inspector = inspect(sync_engine)
+ indexes = inspector.get_indexes("users")
+
+ username_indexed = any(
+ "username" in idx.get("column_names", [])
+ for idx in indexes
+ )
+
+ assert username_indexed, "users.username should be indexed"
+
+ def test_courses_domain_index(self, sync_engine: Engine):
+ """Verify index on courses.domain for category filtering."""
+ inspector = inspect(sync_engine)
+ indexes = inspector.get_indexes("courses")
+
+ domain_indexed = any(
+ "domain" in idx.get("column_names", [])
+ for idx in indexes
+ )
+
+ # This is a recommended index, not required
+ if not domain_indexed:
+ pytest.skip("courses.domain index recommended but not required")
+
+
+class TestForeignKeyConstraints:
+ """Verify foreign key relationships are correctly defined."""
+
+ def test_enrollment_user_fk(self, sync_engine: Engine):
+ """Verify enrollments.user_id references users.id."""
+ inspector = inspect(sync_engine)
+ fks = inspector.get_foreign_keys("enrollments")
+
+ user_fk = any(
+ fk.get("referred_table") == "users" and
+ "user_id" in fk.get("constrained_columns", [])
+ for fk in fks
+ )
+
+ assert user_fk, "enrollments.user_id should reference users.id"
+
+ def test_enrollment_course_fk(self, sync_engine: Engine):
+ """Verify enrollments.course_id references courses.id."""
+ inspector = inspect(sync_engine)
+ fks = inspector.get_foreign_keys("enrollments")
+
+ course_fk = any(
+ fk.get("referred_table") == "courses" and
+ "course_id" in fk.get("constrained_columns", [])
+ for fk in fks
+ )
+
+ assert course_fk, "enrollments.course_id should reference courses.id"
+
+ def test_spaced_repetition_card_user_fk(self, sync_engine: Engine):
+ """Verify spaced_repetition_cards.user_id references users.id."""
+ inspector = inspect(sync_engine)
+ fks = inspector.get_foreign_keys("spaced_repetition_cards")
+
+ user_fk = any(
+ fk.get("referred_table") == "users" and
+ "user_id" in fk.get("constrained_columns", [])
+ for fk in fks
+ )
+
+ assert user_fk, "spaced_repetition_cards.user_id should reference users.id"
+
+ def test_review_log_card_fk(self, sync_engine: Engine):
+ """Verify review_logs.card_id references spaced_repetition_cards.id."""
+ inspector = inspect(sync_engine)
+ fks = inspector.get_foreign_keys("review_logs")
+
+ card_fk = any(
+ fk.get("referred_table") == "spaced_repetition_cards" and
+ "card_id" in fk.get("constrained_columns", [])
+ for fk in fks
+ )
+
+ assert card_fk, "review_logs.card_id should reference spaced_repetition_cards.id"
+
+
+class TestSchemaVersioning:
+ """Tests related to schema versioning and migrations."""
+
+ def test_alembic_version_table_exists(self, sync_engine: Engine):
+ """Verify Alembic migration tracking table exists (if using migrations)."""
+ inspector = inspect(sync_engine)
+ tables = inspector.get_table_names()
+
+ # Note: This might not exist in test database if created fresh
+ # Skip if not using Alembic
+ if "alembic_version" not in tables:
+ pytest.skip("alembic_version table not present (may be using fresh schema)")
+
+ def test_schema_is_up_to_date(self, sync_engine: Engine):
+ """Verify schema matches latest migration (if using Alembic)."""
+ inspector = inspect(sync_engine)
+ tables = inspector.get_table_names()
+
+ if "alembic_version" not in tables:
+ pytest.skip("Not using Alembic migrations")
+
+ # This would require comparing against migration scripts
+ # For now, just verify the table is populated
+ with sync_engine.connect() as conn:
+ result = conn.execute(text("SELECT version_num FROM alembic_version"))
+ version = result.scalar()
+ assert version is not None, "No migration version recorded"