Compare commits

..

No commits in common. "531b3a90484aee4f50a1b76938fddc61266062d0" and "d3d1fb1b32cab82f8be2e99c61be5c78365526dd" have entirely different histories.

12 changed files with 13 additions and 3860 deletions

View file

@ -155,47 +155,16 @@ To backup your data:
cp tablets.db tablets_backup_$(date +%Y%m%d).db
```
## Testing
## Version Control
The project includes a comprehensive unit test suite with **46 tests** covering core functionality and edge cases.
### Running Tests
```bash
# Install test dependencies (if not already installed)
source .venv/bin/activate
uv pip install pytest pytest-cov
# Run all tests
python3 -m pytest tests/ -v
# Run with coverage report
python3 -m pytest tests/ --cov=./ --cov-report=term-missing
```
### Test Structure
| File | Tests | Coverage |
|------|-------|----------|
| `tests/test_core.py` | 21 | Core operations (tablet, user, loan CRUD) |
| `tests/test_edge_cases.py` | 14 | Edge cases (duplicates, invalid IDs, etc.) |
| `tests/test_minimal_app.py` | 11 | Application function tests |
### Key Edge Cases Covered
- Loan a device that's already loaned (prevents double-loaning)
- Return a non-existent loan (handles gracefully)
- Duplicate serial numbers (rejected at database level)
- Duplicate user identifications (rejected at database level)
- Invalid tablet/user IDs (validated before operations)
- Multiple loans per user (one-to-many relationship)
- Loan-return-loan sequence (device lifecycle)
### Test Configuration
Tests use isolated temporary databases and automatically clean up after each test run. No impact on production data.
This project uses Git for version control with the following structure:
- `master` - Production-ready releases
- `develop` - Integration branch for features
- `feature/*` - Individual feature branches
- `hotfix/*` - Urgent bug fixes
See `CONTRIBUTING.md` for detailed workflow and commit conventions.
## License

View file

@ -48,22 +48,10 @@ pkill -f "python app.py"
- **Port**: 5000 (configurable in app.py)
- **Virtual Environment**: `.venv/` (created with uv)
## Running Tests
## Troubleshooting
If you have issues:
1. Check if the server is running: `ps aux | grep app.py`
2. Check the database: `sqlite3 tablets.db`
3. Restart the server: `source .venv/bin/activate && python app.py`
The project includes 46 unit tests for verifying functionality:
```bash
# Run all tests
python3 -m pytest tests/ -v
# Run with coverage
python3 -m pytest tests/ --cov=./ --cov-report=term-missing
```
Tests cover:
- Core CRUD operations for tablets, users, and loans
- Edge cases (duplicates, invalid IDs, already loaned devices)
- Application function testing
- Non-loanable device management
All tests use isolated temporary databases and do not affect production data.
Enjoy managing your tablets! 📱💻

View file

@ -147,54 +147,3 @@ def loan_device(device_id: int, user_id: int) -> Loan:
- **Logging**: Registrar préstamos/devoluciones para auditoría
- **Versionado API**: `/api/v1/...` para compatibilidad futura
- **Pruebas**: Cubrir flujos críticos (préstamo con dispositivo no disponible, devolución de préstamo inexistente)
---
## 9. Migración a PostgreSQL (Futuro)
### Contexto
El sistema actualmente utiliza **SQLite** como base de datos embebida, ideal para prototipos y aplicaciones pequeñas. Sin embargo, SQLite tiene limitaciones para escalar.
### Criterios para Migración
Considerar migrar a **PostgreSQL** cuando:
- Tamaño DB > 1GB
- Más de 50 conexiones simultáneas
- Más de 200 usuarios activos
- Más de 500 transacciones por minuto
- Necesidad de múltiples servidores
### Arquitectura Propuesta
Usar un patrón de **Repository** para abstraer la base de datos:
```
backend/
├── repositories/
│ ├── base_repository.py # Interfaz abstracta
│ ├── sqlite_repository.py # Implementación SQLite
│ └── postgres_repository.py # Implementación PostgreSQL
└── config/
└── database.py # Fábrica de repositorios
```
### Schema PostgreSQL
El schema es similar al de SQLite pero con tipos de datos más específicos y índices adicionales para rendimiento.
### Pasos para Migración
1. Instalar dependencias: `pip install psycopg2-binary sqlalchemy alembic`
2. Configurar PostgreSQL y crear base de datos
3. Ejecutar script de migración: `python scripts/migrate_to_postgres.py`
4. Cambiar configuración: `DB_TYPE=postgres`
5. Iniciar aplicación
### Beneficios
- Concurrencia ilimitada (múltiples escritores)
- Escalabilidad a miles de conexiones
- Mejor rendimiento con índices
- Seguridad integrada (autenticación, roles)
- Backups automáticos
- Replicación y clustering

File diff suppressed because it is too large Load diff

View file

@ -1,987 +0,0 @@
# PostgreSQL Migration Guide
This document describes how to migrate the Tablet Management System from SQLite to PostgreSQL when the database grows beyond SQLite's capabilities.
## When to Migrate
Consider migrating to PostgreSQL when you encounter any of these scenarios:
| Metric | SQLite Limit | PostgreSQL | Migration Trigger |
|--------|--------------|------------|-------------------|
| Database Size | ~10GB max | Unlimited | >1GB |
| Concurrent Writers | 1 | Thousands | >50 simultaneous |
| Active Users | <100 | Millions | >200 |
| Transactions/min | <100 | 100K+ | >500 |
| Servers | Single machine | Cluster | Multiple servers |
| High Availability | No | Yes | Required |
| Backup Strategy | Manual | Automated | Automated needed |
## Architecture Overview
The migration uses a **Repository Pattern** to abstract the database layer, allowing both SQLite and PostgreSQL to work seamlessly.
```
project/
├── backend/
│ ├── config/
│ │ ├── __init__.py
│ │ ├── settings.py # Database configuration
│ │ └── database.py # Repository factory
│ ├── repositories/
│ │ ├── __init__.py
│ │ ├── base_repository.py # Abstract base classes
│ │ ├── sqlite_repo.py # SQLite implementation
│ │ └── postgres_repo.py # PostgreSQL implementation
│ └── app.py # Main application (unchanged)
├── migrations/ # Alembic migrations
│ └── versions/
│ └── initial_schema.py
├── scripts/
│ └── migrate_to_postgres.py # Migration script
└── docker-compose.yml # Optional Docker setup
```
## Step 1: Install Dependencies
```bash
# For development
pip install psycopg2-binary sqlalchemy alembic
# For production (more efficient)
pip install psycopg2 sqlalchemy alembic
```
## Step 2: Set Up PostgreSQL
### Option A: Local Installation
```bash
# Ubuntu/Debian
sudo apt update
sudo apt install postgresql postgresql-contrib
# Create database and user
sudo -u postgres psql
```
In PostgreSQL shell:
```sql
CREATE DATABASE tablet_management;
CREATE USER tablet_user WITH PASSWORD 'your_secure_password';
GRANT ALL PRIVILEGES ON DATABASE tablet_management TO tablet_user;
ALTER USER tablet_user CREATEDB;
\q
```
### Option B: Docker (Recommended for Development)
```bash
# Start PostgreSQL container
docker run --name tablet-db -e POSTGRES_PASSWORD=your_password -e POSTGRES_USER=tablet_user -e POSTGRES_DB=tablet_management -p 5432:5432 -d postgres:16-alpine
# Or use docker-compose (see docker-compose.yml)
docker-compose up -d postgres
```
## Step 3: Configure Environment
Create a `.env` file:
```bash
# Database configuration
DB_TYPE=postgres # or 'sqlite'
DB_URL=postgresql://tablet_user:your_password@localhost:5432/tablet_management
# For SQLite (fallback)
DB_PATH=tablets.db
```
Or set environment variables:
```bash
export DB_TYPE=postgres
export DB_URL=postgresql://tablet_user:your_password@localhost:5432/tablet_management
```
## Step 4: Create Repository Abstraction
### Base Repository (Abstract Interface)
```python
# backend/repositories/base_repository.py
from abc import ABC, abstractmethod
from typing import Optional, List
from datetime import datetime
class BaseTabletRepository(ABC):
@abstractmethod
def get_by_id(self, tablet_id: int) -> Optional[dict]:
pass
@abstractmethod
def get_by_serial(self, serial: str) -> Optional[dict]:
pass
@abstractmethod
def get_all(self, status: Optional[str] = None) -> List[dict]:
pass
@abstractmethod
def add(self, brand: str, model: str, serial_number: str, notes: Optional[str] = None) -> dict:
pass
@abstractmethod
def update_status(self, tablet_id: int, status: str) -> bool:
pass
@abstractmethod
def delete(self, tablet_id: int) -> bool:
pass
class BaseUserRepository(ABC):
@abstractmethod
def get_by_id(self, user_id: int) -> Optional[dict]:
pass
@abstractmethod
def get_by_identification(self, identification: str) -> Optional[dict]:
pass
@abstractmethod
def get_all(self) -> List[dict]:
pass
@abstractmethod
def add(self, name: str, email: Optional[str], phone: Optional[str], identification: str) -> dict:
pass
class BaseLoanRepository(ABC):
@abstractmethod
def get_by_id(self, loan_id: int) -> Optional[dict]:
pass
@abstractmethod
def get_active_by_tablet(self, tablet_id: int) -> Optional[dict]:
pass
@abstractmethod
def get_by_user(self, user_id: int) -> List[dict]:
pass
@abstractmethod
def get_all(self, status: Optional[str] = None) -> List[dict]:
pass
@abstractmethod
def create(self, tablet_id: int, user_id: int) -> dict:
pass
@abstractmethod
def return_loan(self, loan_id: int) -> bool:
pass
class BaseNonLoanableDeviceRepository(ABC):
@abstractmethod
def get_by_id(self, device_id: int) -> Optional[dict]:
pass
@abstractmethod
def get_all(self) -> List[dict]:
pass
@abstractmethod
def add(self, brand: str, model: str, serial_number: str, device_type: str,
location: Optional[str] = None, notes: Optional[str] = None,
purchase_date: Optional[str] = None, purchase_cost: Optional[float] = None) -> dict:
pass
@abstractmethod
def update(self, device_id: int, **kwargs) -> bool:
pass
@abstractmethod
def delete(self, device_id: int) -> bool:
pass
```
### SQLite Implementation
```python
# backend/repositories/sqlite_repo.py
import sqlite3
from typing import Optional, List
from .base_repository import (
BaseTabletRepository, BaseUserRepository,
BaseLoanRepository, BaseNonLoanableDeviceRepository
)
class SQLiteTabletRepository(BaseTabletRepository):
def __init__(self, db_path: str = 'tablets.db'):
self.db_path = db_path
self._init_db()
def _init_db(self):
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS tablets (
id INTEGER PRIMARY KEY AUTOINCREMENT,
brand TEXT NOT NULL,
model TEXT NOT NULL,
serial_number TEXT UNIQUE NOT NULL,
status TEXT DEFAULT 'available',
notes TEXT
)
''')
conn.commit()
conn.close()
def get_by_id(self, tablet_id: int) -> Optional[dict]:
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("SELECT * FROM tablets WHERE id = ?", (tablet_id,))
row = cursor.fetchone()
conn.close()
return dict(row) if row else None
def get_by_serial(self, serial: str) -> Optional[dict]:
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("SELECT * FROM tablets WHERE serial_number = ?", (serial,))
row = cursor.fetchone()
conn.close()
return dict(row) if row else None
def get_all(self, status: Optional[str] = None) -> List[dict]:
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
query = "SELECT * FROM tablets"
params = ()
if status:
query += " WHERE status = ?"
params = (status,)
cursor.execute(query, params)
results = [dict(row) for row in cursor.fetchall()]
conn.close()
return results
def add(self, brand: str, model: str, serial_number: str, notes: Optional[str] = None) -> dict:
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute('''
INSERT INTO tablets (brand, model, serial_number, status, notes)
VALUES (?, ?, ?, 'available', ?)
''', (brand, model, serial_number, notes))
conn.commit()
tablet_id = cursor.lastrowid
cursor.execute("SELECT * FROM tablets WHERE id = ?", (tablet_id,))
row = cursor.fetchone()
conn.close()
return dict(row)
def update_status(self, tablet_id: int, status: str) -> bool:
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("UPDATE tablets SET status = ? WHERE id = ?", (status, tablet_id))
conn.commit()
changed = cursor.rowcount > 0
conn.close()
return changed
def delete(self, tablet_id: int) -> bool:
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("DELETE FROM tablets WHERE id = ?", (tablet_id,))
conn.commit()
deleted = cursor.rowcount > 0
conn.close()
return deleted
# Similar implementations for SQLiteUserRepository, SQLiteLoanRepository, etc.
```
### PostgreSQL Implementation
```python
# backend/repositories/postgres_repo.py
import psycopg2
from psycopg2 import sql
from psycopg2.extras import DictCursor
from typing import Optional, List
from .base_repository import (
BaseTabletRepository, BaseUserRepository,
BaseLoanRepository, BaseNonLoanableDeviceRepository
)
class PostgreSQLTabletRepository(BaseTabletRepository):
def __init__(self, connection_string: str):
self.connection_string = connection_string
self._init_db()
def _get_connection(self):
return psycopg2.connect(self.connection_string)
def _init_db(self):
conn = self._get_connection()
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS tablets (
id SERIAL PRIMARY KEY,
brand VARCHAR(100) NOT NULL,
model VARCHAR(100) NOT NULL,
serial_number VARCHAR(50) UNIQUE NOT NULL,
status VARCHAR(20) DEFAULT 'available',
notes TEXT,
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
)
''')
cursor.execute('''
CREATE INDEX IF NOT EXISTS idx_tablets_serial
ON tablets(serial_number)
''')
cursor.execute('''
CREATE INDEX IF NOT EXISTS idx_tablets_status
ON tablets(status)
''')
conn.commit()
cursor.close()
conn.close()
def get_by_id(self, tablet_id: int) -> Optional[dict]:
conn = self._get_connection()
cursor = conn.cursor(cursor_factory=DictCursor)
cursor.execute("SELECT * FROM tablets WHERE id = %s", (tablet_id,))
row = cursor.fetchone()
conn.close()
return dict(row) if row else None
def get_by_serial(self, serial: str) -> Optional[dict]:
conn = self._get_connection()
cursor = conn.cursor(cursor_factory=DictCursor)
cursor.execute("SELECT * FROM tablets WHERE serial_number = %s", (serial,))
row = cursor.fetchone()
conn.close()
return dict(row) if row else None
def get_all(self, status: Optional[str] = None) -> List[dict]:
conn = self._get_connection()
cursor = conn.cursor(cursor_factory=DictCursor)
query = "SELECT * FROM tablets"
params = ()
if status:
query += " WHERE status = %s"
params = (status,)
cursor.execute(query, params)
results = [dict(row) for row in cursor.fetchall()]
conn.close()
return results
def add(self, brand: str, model: str, serial_number: str, notes: Optional[str] = None) -> dict:
conn = self._get_connection()
cursor = conn.cursor(cursor_factory=DictCursor)
cursor.execute('''
INSERT INTO tablets (brand, model, serial_number, status, notes)
VALUES (%s, %s, %s, 'available', %s)
RETURNING *
''', (brand, model, serial_number, notes))
row = cursor.fetchone()
conn.commit()
conn.close()
return dict(row)
def update_status(self, tablet_id: int, status: str) -> bool:
conn = self._get_connection()
cursor = conn.cursor()
cursor.execute(
"UPDATE tablets SET status = %s, updated_at = NOW() WHERE id = %s",
(status, tablet_id)
)
conn.commit()
changed = cursor.rowcount > 0
conn.close()
return changed
def delete(self, tablet_id: int) -> bool:
conn = self._get_connection()
cursor = conn.cursor()
cursor.execute("DELETE FROM tablets WHERE id = %s", (tablet_id,))
conn.commit()
deleted = cursor.rowcount > 0
conn.close()
return deleted
# Similar implementations for PostgreSQLUserRepository, PostgreSQLLoanRepository, etc.
```
## Step 5: Create Repository Factory
```python
# backend/config/database.py
import os
from backend.repositories.sqlite_repo import (
SQLiteTabletRepository, SQLiteUserRepository,
SQLiteLoanRepository, SQLiteNonLoanableDeviceRepository
)
from backend.repositories.postgres_repo import (
PostgreSQLTabletRepository, PostgreSQLUserRepository,
PostgreSQLLoanRepository, PostgreSQLNonLoanableDeviceRepository
)
from backend.repositories.base_repository import (
BaseTabletRepository, BaseUserRepository,
BaseLoanRepository, BaseNonLoanableDeviceRepository
)
class DatabaseConfig:
def __init__(self):
self.db_type = os.getenv('DB_TYPE', 'sqlite')
self.db_url = os.getenv('DB_URL', '')
self.db_path = os.getenv('DB_PATH', 'tablets.db')
@property
def is_postgres(self) -> bool:
return self.db_type == 'postgres'
def get_tablet_repository() -> BaseTabletRepository:
config = DatabaseConfig()
if config.is_postgres:
return PostgreSQLTabletRepository(config.db_url)
else:
return SQLiteTabletRepository(config.db_path)
def get_user_repository() -> BaseUserRepository:
config = DatabaseConfig()
if config.is_postgres:
return PostgreSQLUserRepository(config.db_url)
else:
return SQLiteUserRepository(config.db_path)
def get_loan_repository() -> BaseLoanRepository:
config = DatabaseConfig()
if config.is_postgres:
return PostgreSQLLoanRepository(config.db_url)
else:
return SQLiteLoanRepository(config.db_path)
def get_non_loanable_device_repository() -> BaseNonLoanableDeviceRepository:
config = DatabaseConfig()
if config.is_postgres:
return PostgreSQLNonLoanableDeviceRepository(config.db_url)
else:
return SQLiteNonLoanableDeviceRepository(config.db_path)
```
## Step 6: Update Application to Use Repositories
Modify your application to use the repository pattern:
```python
# In your app.py or service layer
from backend.config.database import (
get_tablet_repository, get_user_repository,
get_loan_repository, get_non_loanable_device_repository
)
# Instead of direct SQLite calls:
tablet_repo = get_tablet_repository()
user_repo = get_user_repository()
loan_repo = get_loan_repository()
# Example: Loan a tablet
def loan_tablet(tablet_id: int, user_id: int):
# Get repositories
tablet_repo = get_tablet_repository()
user_repo = get_user_repository()
loan_repo = get_loan_repository()
# Validate
tablet = tablet_repo.get_by_id(tablet_id)
if not tablet:
raise ValueError("Tablet not found")
if tablet['status'] != 'available':
raise ValueError("Tablet not available")
user = user_repo.get_by_id(user_id)
if not user:
raise ValueError("User not found")
# Check for active loan
active_loan = loan_repo.get_active_by_tablet(tablet_id)
if active_loan:
raise ValueError("Tablet already loaned")
# Create loan
loan = loan_repo.create(tablet_id, user_id)
# Update tablet status
tablet_repo.update_status(tablet_id, 'loaned')
return loan
```
## Step 7: Create Migration Script
```python
# scripts/migrate_to_postgres.py
#!/usr/bin/env python3
"""
Migration script from SQLite to PostgreSQL
"""
import sqlite3
import psycopg2
from psycopg2.extras import DictCursor
import argparse
from tqdm import tqdm
import os
def create_postgres_tables(conn):
"""Create all tables in PostgreSQL"""
cursor = conn.cursor()
# Tablets
cursor.execute('''
CREATE TABLE IF NOT EXISTS tablets (
id SERIAL PRIMARY KEY,
brand VARCHAR(100) NOT NULL,
model VARCHAR(100) NOT NULL,
serial_number VARCHAR(50) UNIQUE NOT NULL,
status VARCHAR(20) DEFAULT 'available',
notes TEXT,
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
)
''')
# Users
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL,
email VARCHAR(255),
phone VARCHAR(20),
identification VARCHAR(50) UNIQUE NOT NULL,
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
)
''')
# Loans
cursor.execute('''
CREATE TABLE IF NOT EXISTS loans (
id SERIAL PRIMARY KEY,
tablet_id INTEGER NOT NULL REFERENCES tablets(id) ON DELETE RESTRICT,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE RESTRICT,
loan_date TIMESTAMP NOT NULL DEFAULT NOW(),
return_date TIMESTAMP,
status VARCHAR(20) DEFAULT 'active',
created_at TIMESTAMP DEFAULT NOW()
)
''')
# Non-loanable devices
cursor.execute('''
CREATE TABLE IF NOT EXISTS non_loanable_devices (
id SERIAL PRIMARY KEY,
brand VARCHAR(100) NOT NULL,
model VARCHAR(100) NOT NULL,
serial_number VARCHAR(50) UNIQUE NOT NULL,
device_type VARCHAR(50) NOT NULL,
location VARCHAR(100),
status VARCHAR(20) DEFAULT 'available',
notes TEXT,
purchase_date DATE,
purchase_cost DECIMAL(10,2),
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
)
''')
# Indexes for performance
cursor.execute('CREATE INDEX IF NOT EXISTS idx_tablets_serial ON tablets(serial_number)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_tablets_status ON tablets(status)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_tablets_brand ON tablets(brand)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_users_identification ON users(identification)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_loans_tablet ON loans(tablet_id)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_loans_user ON loans(user_id)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_loans_status ON loans(status)')
conn.commit()
def migrate_table(conn_sqlite, conn_pg, table_name: str, pg_create_table: str):
"""Generic migration for a table"""
cursor_sqlite = conn_sqlite.cursor()
cursor_pg = conn_pg.cursor()
# Get all data from SQLite
cursor_sqlite.execute(f"SELECT * FROM {table_name}")
rows = cursor_sqlite.fetchall()
if not rows:
print(f"No data to migrate for {table_name}")
return
# Get column names
column_names = [desc[0] for desc in cursor_sqlite.description]
# Prepare INSERT statement
columns = ', '.join(column_names)
placeholders = ', '.join(['%s'] * len(column_names))
insert_sql = f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders}) ON CONFLICT DO NOTHING"
# Migrate data
for row in tqdm(rows, desc=f"Migrating {table_name}"):
cursor_pg.execute(insert_sql, row)
conn_pg.commit()
print(f"✓ Migrated {len(rows)} rows from {table_name}")
def migrate_all(sqlite_path: str, pg_url: str):
"""Migrate all data from SQLite to PostgreSQL"""
print("Starting migration from SQLite to PostgreSQL...")
# Connect to SQLite
conn_sqlite = sqlite3.connect(sqlite_path)
# Connect to PostgreSQL
conn_pg = psycopg2.connect(pg_url)
try:
# Create tables
print("Creating PostgreSQL tables...")
create_postgres_tables(conn_pg)
# Migrate each table
migrate_table(conn_sqlite, conn_pg, 'tablets', '')
migrate_table(conn_sqlite, conn_pg, 'users', '')
migrate_table(conn_sqlite, conn_pg, 'loans', '')
migrate_table(conn_sqlite, conn_pg, 'non_loanable_devices', '')
print("\n✓ Migration completed successfully!")
print(f" SQLite: {sqlite_path}")
print(f" PostgreSQL: {pg_url}")
except Exception as e:
conn_pg.rollback()
print(f"\n✗ Migration failed: {e}")
raise
finally:
conn_sqlite.close()
conn_pg.close()
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='Migrate from SQLite to PostgreSQL')
parser.add_argument('--sqlite', default='tablets.db', help='SQLite database path')
parser.add_argument('--postgres', required=True, help='PostgreSQL connection URL')
args = parser.parse_args()
migrate_all(args.sqlite, args.postgres)
```
## Step 8: Run Migration
```bash
# Test the migration first (dry run)
python scripts/migrate_to_postgres.py --sqlite tablets.db --postgres postgresql://tablet_user:password@localhost:5432/tablet_management_test
# Verify data in test database
psql -U tablet_user -d tablet_management_test -c "SELECT COUNT(*) FROM tablets;"
# When ready, migrate to production
python scripts/migrate_to_postgres.py --sqlite tablets.db --postgres postgresql://tablet_user:password@localhost:5432/tablet_management
```
## Step 9: Switch to PostgreSQL
```bash
# Update environment variables
export DB_TYPE=postgres
export DB_URL=postgresql://tablet_user:password@localhost:5432/tablet_management
# Restart application
python app.py
```
## Step 10: Verify and Monitor
```bash
# Check application logs for errors
# Monitor database connections
psql -U tablet_user -d tablet_management -c "SELECT COUNT(*) FROM tablets;"
# Check active connections
psql -U postgres -c "SELECT * FROM pg_stat_activity WHERE datname = 'tablet_management';"
```
## Rollback Plan
If something goes wrong:
1. **Immediate rollback:**
```bash
# Switch back to SQLite
export DB_TYPE=sqlite
export DB_PATH=tablets.db
python app.py
```
2. **Data verification:**
```bash
# Compare counts
sqlite3 tablets.db "SELECT COUNT(*) FROM tablets;"
psql -U tablet_user -d tablet_management -c "SELECT COUNT(*) FROM tablets;"
```
3. **Backup PostgreSQL data:**
```bash
pg_dump -U tablet_user -d tablet_management > postgres_backup_$(date +%Y%m%d).sql
```
## Docker Compose (Optional)
For easy deployment with Docker:
```yaml
# docker-compose.yml
version: '3.8'
services:
postgres:
image: postgres:16-alpine
container_name: tablet_db
environment:
POSTGRES_DB: tablet_management
POSTGRES_USER: tablet_user
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-changeme}
ports:
- "5432:5432"
volumes:
- postgres_data:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U tablet_user -d tablet_management"]
interval: 5s
timeout: 5s
retries: 5
restart: unless-stopped
app:
build: .
container_name: tablet_app
environment:
DB_TYPE: postgres
DB_URL: postgresql://tablet_user:${POSTGRES_PASSWORD:-changeme}@postgres:5432/tablet_management
ports:
- "5000:5000"
depends_on:
postgres:
condition: service_healthy
restart: unless-stopped
volumes:
postgres_data:
```
Start with Docker:
```bash
docker-compose up -d
```
## Benefits of PostgreSQL
### Performance
- **Concurrency:** Multiple writers simultaneously (no lock contention)
- **Indexing:** Advanced index types (B-tree, Hash, GiST, GIN, BRIN)
- **Query Optimization:** Advanced query planner
- **Connection Pooling:** Built-in support
### Scalability
- **Vertical:** Handles large datasets efficiently
- **Horizontal:** Read replicas, partitioning, sharding
- **Connections:** Supports thousands of concurrent connections
### Reliability
- **ACID Compliance:** Full transaction support
- **Point-in-Time Recovery:** Restore to any moment
- **Replication:** Master-slave, synchronous, asynchronous
- **Backups:** `pg_dump`, `pg_basebackup`, continuous archiving
### Security
- **Authentication:** Multiple methods (password, MD5, SCRAM, LDAP, Kerberos)
- **Authorization:** Role-based access control (RBAC)
- **Row-Level Security:** Policies for fine-grained access
- **Encryption:** SSL, at-rest encryption
### Features
- **JSON Support:** Native JSON/JSONB data type
- **Full-Text Search:** Advanced text search capabilities
- **Arrays:** Store arrays of values
- **Custom Types:** Create your own data types
- **Triggers:** Automatic actions on events
- **Stored Procedures:** Server-side functions
## Monitoring PostgreSQL
### Basic Queries
```sql
-- Active connections
SELECT * FROM pg_stat_activity WHERE datname = 'tablet_management';
-- Table sizes
SELECT table_name, pg_size_pretty(pg_total_relation_size(table_name))
FROM information_schema.tables WHERE table_schema = 'public';
-- Index usage
SELECT indexrelname, idx_scan, idx_tup_read, idx_tup_fetch
FROM pg_stat_user_indexes;
-- Slow queries (requires pg_stat_statements extension)
SELECT query, total_time, calls, mean_time
FROM pg_stat_statements ORDER BY mean_time DESC LIMIT 10;
```
### Enable pg_stat_statements
```sql
-- In PostgreSQL
CREATE EXTENSION pg_stat_statements;
-- Then in postgresql.conf
shared_preload_libraries = 'pg_stat_statements'
pg_stat_statements.track = all
```
## Maintenance Tasks
### Regular Maintenance
```bash
# Vacuum (reclaim space, update statistics)
vacuumdb -U tablet_user -d tablet_management --analyze
# Reindex (rebuild indexes)
reindexdb -U tablet_user -d tablet_management
```
### Backup Strategy
```bash
# Daily backup
pg_dump -U tablet_user -d tablet_management > /backups/tablet_management_$(date +%Y%m%d).sql
# Compressed backup
pg_dump -U tablet_user -d tablet_management | gzip > /backups/tablet_management_$(date +%Y%m%d).sql.gz
# Continuous archiving (WAL)
# In postgresql.conf:
wal_level = replica
archive_mode = on
archive_command = 'test ! -f /backups/wal/%f && cp %p /backups/wal/%f'
```
## Performance Optimization
### Configuration Tuning
```conf
# postgresql.conf recommendations
shared_buffers = 4GB # 25% of total RAM
work_mem = 16MB # For complex sorts
maintenance_work_mem = 512MB # For VACUUM, index creation
effective_cache_size = 12GB # 75% of total RAM
random_page_cost = 1.1 # SSD: 1.1, HDD: 4.0
max_worker_processes = 8 # Number of CPU cores
max_parallel_workers_per_gather = 4 # Parallel query workers
max_connections = 200 # Expected max connections
```
### Index Optimization
```sql
-- Add indexes for common queries
CREATE INDEX idx_loans_user_status ON loans(user_id, status);
CREATE INDEX idx_loans_date_range ON loans(loan_date, return_date);
-- Partial index for active loans
CREATE INDEX idx_loans_active ON loans(tablet_id) WHERE status = 'active';
-- Composite index for user loans
CREATE INDEX idx_loans_user_tablet ON loans(user_id, tablet_id);
```
## Troubleshooting
### Common Issues
**Connection refused:**
```bash
# Check if PostgreSQL is running
sudo systemctl status postgresql
# Check port
netstat -tuln | grep 5432
```
**Authentication failed:**
```bash
# Verify user and password
psql -U tablet_user -d tablet_management -h localhost
# Check pg_hba.conf
sudo nano /etc/postgresql/16/main/pg_hba.conf
```
**Database does not exist:**
```bash
# Create database
createdb -U postgres tablet_management
```
**Permission denied:**
```sql
-- Grant permissions
GRANT ALL PRIVILEGES ON DATABASE tablet_management TO tablet_user;
GRANT ALL ON SCHEMA public TO tablet_user;
```
### Logs
```bash
# PostgreSQL logs
sudo tail -f /var/log/postgresql/postgresql-16-main.log
# Application logs
journalctl -u tablet_management -f
```
## Conclusion
Migrating from SQLite to PostgreSQL provides:
- Better performance at scale
- True concurrency
- Enhanced reliability
- Advanced features
- Production-ready infrastructure
The repository pattern ensures a smooth transition with minimal code changes, and the migration script automates the data transfer process.

View file

@ -5,33 +5,3 @@ description = "Add your description here"
readme = "README.md"
requires-python = ">=3.14"
dependencies = []
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = "test_*.py"
python_classes = "Test*"
python_functions = "test_*"
verbose = 1
addopts = "-v"
[tool.coverage.run]
source = ["."]
omit = [
"*/tests/*",
"*/.venv/*",
"*/__pycache__/*",
"*/.git/*",
"*/templates/*",
]
[tool.coverage.report]
exclude_lines = [
"pragma: no cover",
"def __repr__",
"raise NotImplementedError",
"if __name__ == .__main__.:",
"if TYPE_CHECKING:",
]
[tool.coverage.html]
directory = "htmlcov"

View file

@ -1,3 +0,0 @@
"""
Unit tests for Tablet Management System
"""

View file

@ -1,181 +0,0 @@
"""
Pytest configuration and fixtures for Tablet Management System tests
"""
import pytest
import sqlite3
import os
import sys
from datetime import datetime
# Add project root to path for imports
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
@pytest.fixture
def test_db_path():
"""Path to test database"""
return 'test_tablets.db'
@pytest.fixture
def init_test_db(test_db_path):
"""Initialize a fresh test database with schema"""
# Remove existing test database if it exists
if os.path.exists(test_db_path):
os.remove(test_db_path)
conn = sqlite3.connect(test_db_path)
cursor = conn.cursor()
# Create tables
cursor.execute('''
CREATE TABLE IF NOT EXISTS tablets (
id INTEGER PRIMARY KEY AUTOINCREMENT,
brand TEXT NOT NULL,
model TEXT NOT NULL,
serial_number TEXT UNIQUE NOT NULL,
status TEXT DEFAULT 'available',
notes TEXT
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT,
phone TEXT,
identification TEXT UNIQUE
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS loans (
id INTEGER PRIMARY KEY AUTOINCREMENT,
tablet_id INTEGER NOT NULL,
user_id INTEGER NOT NULL,
loan_date TEXT NOT NULL,
return_date TEXT,
status TEXT DEFAULT 'active',
FOREIGN KEY (tablet_id) REFERENCES tablets (id),
FOREIGN KEY (user_id) REFERENCES users (id)
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS non_loanable_devices (
id INTEGER PRIMARY KEY AUTOINCREMENT,
brand TEXT NOT NULL,
model TEXT NOT NULL,
serial_number TEXT UNIQUE NOT NULL,
device_type TEXT NOT NULL,
location TEXT,
status TEXT DEFAULT 'available',
notes TEXT,
purchase_date TEXT,
purchase_cost REAL
)
''')
conn.commit()
conn.close()
yield test_db_path
# Cleanup: remove test database
if os.path.exists(test_db_path):
os.remove(test_db_path)
@pytest.fixture
def db_conn(init_test_db):
"""Get a database connection to the test database"""
conn = sqlite3.connect(init_test_db)
conn.row_factory = sqlite3.Row
yield conn
conn.close()
@pytest.fixture
def sample_tablets(db_conn):
"""Insert sample tablets into test database"""
cursor = db_conn.cursor()
tablets = [
('Samsung', 'Galaxy Tab S7', 'SN001', 'available'),
('Apple', 'iPad Pro', 'SN002', 'available'),
('Lenovo', 'Tab P11', 'SN003', 'available'),
('Microsoft', 'Surface Pro', 'SN004', 'loaned'),
]
for brand, model, serial, status in tablets:
cursor.execute('''
INSERT INTO tablets (brand, model, serial_number, status)
VALUES (?, ?, ?, ?)
''', (brand, model, serial, status))
db_conn.commit()
return tablets
@pytest.fixture
def sample_users(db_conn):
"""Insert sample users into test database"""
cursor = db_conn.cursor()
users = [
('John Doe', 'john@example.com', '1234567890', 'ID001'),
('Jane Smith', 'jane@example.com', '0987654321', 'ID002'),
('Bob Johnson', 'bob@example.com', '5551234567', 'ID003'),
]
for name, email, phone, identification in users:
cursor.execute('''
INSERT INTO users (name, email, phone, identification)
VALUES (?, ?, ?, ?)
''', (name, email, phone, identification))
db_conn.commit()
return users
@pytest.fixture
def sample_loans(db_conn, sample_tablets, sample_users):
"""Insert sample loans into test database"""
cursor = db_conn.cursor()
# Get tablet and user IDs
cursor.execute("SELECT id FROM tablets WHERE serial_number = 'SN004'")
loaned_tablet = cursor.fetchone()
cursor.execute("SELECT id FROM users WHERE identification = 'ID001'")
user1 = cursor.fetchone()
cursor.execute("SELECT id FROM users WHERE identification = 'ID002'")
user2 = cursor.fetchone()
if loaned_tablet and user1:
# Active loan for SN004 to user1
loan_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
cursor.execute('''
INSERT INTO loans (tablet_id, user_id, loan_date, status)
VALUES (?, ?, ?, 'active')
''', (loaned_tablet['id'], user1['id'], loan_date))
if user2:
# Returned loan for SN001 to user2
loan_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
return_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
cursor.execute('''
INSERT INTO loans (tablet_id, user_id, loan_date, return_date, status)
VALUES (?, ?, ?, ?, 'returned')
''', (1, user2['id'], loan_date, return_date))
db_conn.commit()
@pytest.fixture
def populated_db(db_conn, sample_tablets, sample_users, sample_loans):
"""Database with all sample data loaded"""
return db_conn

View file

@ -1,2 +0,0 @@
pytest==8.3.2
pytest-cov==5.0.0

View file

@ -1,481 +0,0 @@
"""
Unit tests for core tablet management functions
Tests loan logic, validation, and database operations
"""
import pytest
import sqlite3
from datetime import datetime
import sys
import os
# Import the functions from minimal_app (they work with any db connection)
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
def get_tablet_count(conn):
"""Helper to get tablet count"""
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM tablets")
return cursor.fetchone()[0]
def get_user_count(conn):
"""Helper to get user count"""
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM users")
return cursor.fetchone()[0]
def get_loan_count(conn):
"""Helper to get loan count"""
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM loans")
return cursor.fetchone()[0]
class TestTabletOperations:
"""Tests for tablet CRUD operations"""
def test_add_tablet_success(self, db_conn):
"""Test adding a new tablet successfully"""
cursor = db_conn.cursor()
# Add a tablet
cursor.execute('''
INSERT INTO tablets (brand, model, serial_number, status)
VALUES (?, ?, ?, 'available')
''', ('TestBrand', 'TestModel', 'TEST001'))
db_conn.commit()
# Verify it was added
cursor.execute("SELECT * FROM tablets WHERE serial_number = 'TEST001'")
tablet = cursor.fetchone()
assert tablet is not None
assert tablet['brand'] == 'TestBrand'
assert tablet['model'] == 'TestModel'
assert tablet['serial_number'] == 'TEST001'
assert tablet['status'] == 'available'
def test_add_tablet_duplicate_serial(self, db_conn):
"""Test that duplicate serial numbers are rejected"""
cursor = db_conn.cursor()
# Add first tablet
cursor.execute('''
INSERT INTO tablets (brand, model, serial_number, status)
VALUES (?, ?, ?, 'available')
''', ('Brand1', 'Model1', 'DUP001'))
db_conn.commit()
# Try to add duplicate
with pytest.raises(sqlite3.IntegrityError):
cursor.execute('''
INSERT INTO tablets (brand, model, serial_number, status)
VALUES (?, ?, ?, 'available')
''', ('Brand2', 'Model2', 'DUP001'))
db_conn.commit()
def test_tablet_status_update(self, db_conn):
"""Test updating tablet status"""
cursor = db_conn.cursor()
# Add a tablet
cursor.execute('''
INSERT INTO tablets (brand, model, serial_number, status)
VALUES (?, ?, ?, 'available')
''', ('Brand', 'Model', 'STATUS001'))
db_conn.commit()
# Update status
cursor.execute("UPDATE tablets SET status = 'loaned' WHERE serial_number = 'STATUS001'")
db_conn.commit()
# Verify update
cursor.execute("SELECT status FROM tablets WHERE serial_number = 'STATUS001'")
status = cursor.fetchone()['status']
assert status == 'loaned'
class TestUserOperations:
"""Tests for user CRUD operations"""
def test_add_user_success(self, db_conn):
"""Test adding a new user successfully"""
cursor = db_conn.cursor()
cursor.execute('''
INSERT INTO users (name, identification)
VALUES (?, ?)
''', ('Test User', 'TESTID001'))
db_conn.commit()
cursor.execute("SELECT * FROM users WHERE identification = 'TESTID001'")
user = cursor.fetchone()
assert user is not None
assert user['name'] == 'Test User'
assert user['identification'] == 'TESTID001'
def test_add_user_duplicate_identification(self, db_conn):
"""Test that duplicate user identifications are rejected"""
cursor = db_conn.cursor()
# Add first user
cursor.execute('''
INSERT INTO users (name, identification)
VALUES (?, ?)
''', ('User1', 'DUPID001'))
db_conn.commit()
# Try to add duplicate
with pytest.raises(sqlite3.IntegrityError):
cursor.execute('''
INSERT INTO users (name, identification)
VALUES (?, ?)
''', ('User2', 'DUPID001'))
db_conn.commit()
def test_user_with_contact_info(self, db_conn):
"""Test adding user with email and phone"""
cursor = db_conn.cursor()
cursor.execute('''
INSERT INTO users (name, email, phone, identification)
VALUES (?, ?, ?, ?)
''', ('Contact User', 'test@email.com', '1234567890', 'CONTACT001'))
db_conn.commit()
cursor.execute("SELECT * FROM users WHERE identification = 'CONTACT001'")
user = cursor.fetchone()
assert user['email'] == 'test@email.com'
assert user['phone'] == '1234567890'
class TestLoanOperations:
"""Tests for loan operations - the core business logic"""
def test_loan_tablet_success(self, populated_db):
"""Test loaning an available tablet to a user"""
cursor = populated_db.cursor()
# Get an available tablet and user
cursor.execute("SELECT id FROM tablets WHERE status = 'available' LIMIT 1")
tablet = cursor.fetchone()
cursor.execute("SELECT id FROM users LIMIT 1")
user = cursor.fetchone()
assert tablet is not None
assert user is not None
tablet_id = tablet['id']
user_id = user['id']
# Loan the tablet
loan_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
cursor.execute("UPDATE tablets SET status = 'loaned' WHERE id = ?", (tablet_id,))
cursor.execute('''
INSERT INTO loans (tablet_id, user_id, loan_date, status)
VALUES (?, ?, ?, 'active')
''', (tablet_id, user_id, loan_date))
populated_db.commit()
# Verify tablet status changed
cursor.execute("SELECT status FROM tablets WHERE id = ?", (tablet_id,))
status = cursor.fetchone()['status']
assert status == 'loaned'
# Verify loan was created
cursor.execute("SELECT * FROM loans WHERE tablet_id = ? AND user_id = ?", (tablet_id, user_id))
loan = cursor.fetchone()
assert loan is not None
assert loan['status'] == 'active'
assert loan['return_date'] is None
def test_loan_already_loaned_tablet(self, populated_db):
"""Test that loaning an already loaned tablet fails gracefully"""
cursor = populated_db.cursor()
# Get a loaned tablet (SN004 should be loaned from sample data)
cursor.execute("SELECT id FROM tablets WHERE status = 'loaned' LIMIT 1")
tablet = cursor.fetchone()
cursor.execute("SELECT id FROM users LIMIT 1")
user = cursor.fetchone()
if tablet and user:
tablet_id = tablet['id']
user_id = user['id']
# Try to loan it again (should check status first)
cursor.execute("SELECT status FROM tablets WHERE id = ?", (tablet_id,))
status = cursor.fetchone()['status']
# This should be 'loaned', so we shouldn't be able to loan it
assert status == 'loaned'
# The application logic should prevent this
# In the actual app, this would be checked before inserting
# Here we verify the status is still loaned
cursor.execute("SELECT status FROM tablets WHERE id = ?", (tablet_id,))
final_status = cursor.fetchone()['status']
assert final_status == 'loaned'
def test_return_tablet_success(self, populated_db):
"""Test returning a loaned tablet"""
cursor = populated_db.cursor()
# Get an active loan
cursor.execute("SELECT * FROM loans WHERE status = 'active' LIMIT 1")
loan = cursor.fetchone()
if loan:
loan_id = loan['id']
tablet_id = loan['tablet_id']
# Return the tablet
return_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
cursor.execute('''
UPDATE loans SET status = 'returned', return_date = ? WHERE id = ?
''', (return_date, loan_id))
cursor.execute("UPDATE tablets SET status = 'available' WHERE id = ?", (tablet_id,))
populated_db.commit()
# Verify loan status changed
cursor.execute("SELECT status, return_date FROM loans WHERE id = ?", (loan_id,))
updated_loan = cursor.fetchone()
assert updated_loan['status'] == 'returned'
assert updated_loan['return_date'] is not None
# Verify tablet status changed
cursor.execute("SELECT status FROM tablets WHERE id = ?", (tablet_id,))
tablet_status = cursor.fetchone()['status']
assert tablet_status == 'available'
def test_return_nonexistent_loan(self, db_conn):
"""Test returning a loan that doesn't exist"""
cursor = db_conn.cursor()
# Try to return a non-existent loan
cursor.execute("SELECT tablet_id FROM loans WHERE id = ? AND status = 'active'", (9999,))
loan = cursor.fetchone()
# Should be None since loan doesn't exist
assert loan is None
class TestEdgeCases:
"""Tests for edge cases and error conditions"""
def test_loan_to_nonexistent_user(self, db_conn):
"""Test loaning to a user that doesn't exist"""
cursor = db_conn.cursor()
# Add a tablet
cursor.execute('''
INSERT INTO tablets (brand, model, serial_number, status)
VALUES (?, ?, ?, 'available')
''', ('Brand', 'Model', 'EDGE001'))
db_conn.commit()
cursor.execute("SELECT id FROM tablets WHERE serial_number = 'EDGE001'")
tablet = cursor.fetchone()
# Try to loan to non-existent user (ID 9999)
# Note: SQLite doesn't enforce foreign keys by default unless we enable it
# The application should validate this at the application level
# For now, we verify that the user doesn't exist
cursor.execute("SELECT id FROM users WHERE id = 9999")
user = cursor.fetchone()
assert user is None # User doesn't exist
# In a real app with FK enforcement, this would raise IntegrityError
# For SQLite without FK enforcement, we just verify the user check
def test_loan_nonexistent_tablet(self, db_conn):
"""Test loaning a tablet that doesn't exist"""
cursor = db_conn.cursor()
# Add a user
cursor.execute('''
INSERT INTO users (name, identification)
VALUES (?, ?)
''', ('Test User', 'EDGEID001'))
db_conn.commit()
cursor.execute("SELECT id FROM users WHERE identification = 'EDGEID001'")
user = cursor.fetchone()
# Try to loan non-existent tablet (ID 9999)
# Note: SQLite doesn't enforce foreign keys by default unless we enable it
# The application should validate this at the application level
# For now, we verify that the tablet doesn't exist
cursor.execute("SELECT id FROM tablets WHERE id = 9999")
tablet_check = cursor.fetchone()
assert tablet_check is None # Tablet doesn't exist
# In a real app with FK enforcement, this would raise IntegrityError
# For SQLite without FK enforcement, we just verify the tablet check
def test_empty_database_operations(self, db_conn):
"""Test operations on empty database"""
cursor = db_conn.cursor()
# Query empty tables
cursor.execute("SELECT COUNT(*) FROM tablets")
tablet_count = cursor.fetchone()[0]
assert tablet_count == 0
cursor.execute("SELECT COUNT(*) FROM users")
user_count = cursor.fetchone()[0]
assert user_count == 0
cursor.execute("SELECT COUNT(*) FROM loans")
loan_count = cursor.fetchone()[0]
assert loan_count == 0
def test_multiple_loans_same_user(self, db_conn):
"""Test that one user can have multiple loans (one-to-many relationship)"""
cursor = db_conn.cursor()
# Add user
cursor.execute('''
INSERT INTO users (name, identification)
VALUES (?, ?)
''', ('MultiLoan User', 'MULTI001'))
db_conn.commit()
cursor.execute("SELECT id FROM users WHERE identification = 'MULTI001'")
user = cursor.fetchone()
# Add multiple tablets
for i in range(3):
cursor.execute('''
INSERT INTO tablets (brand, model, serial_number, status)
VALUES (?, ?, ?, 'available')
''', (f'Brand{i}', f'Model{i}', f'MULTI{i:03d}'))
db_conn.commit()
# Loan all tablets to the same user
cursor.execute("SELECT id FROM tablets WHERE serial_number LIKE 'MULTI%'")
tablets = cursor.fetchall()
loan_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
for tablet in tablets:
cursor.execute("UPDATE tablets SET status = 'loaned' WHERE id = ?", (tablet['id'],))
cursor.execute('''
INSERT INTO loans (tablet_id, user_id, loan_date, status)
VALUES (?, ?, ?, 'active')
''', (tablet['id'], user['id'], loan_date))
db_conn.commit()
# Verify user has multiple loans
cursor.execute("SELECT COUNT(*) FROM loans WHERE user_id = ?", (user['id'],))
loan_count = cursor.fetchone()[0]
assert loan_count == 3
def test_serial_number_uniqueness_across_tables(self, db_conn):
"""Test that serial numbers are unique within their respective tables"""
cursor = db_conn.cursor()
# Add tablet with serial number
cursor.execute('''
INSERT INTO tablets (brand, model, serial_number, status)
VALUES (?, ?, ?, 'available')
''', ('Brand', 'Model', 'UNIQUE001'))
db_conn.commit()
# Add non-loanable device with same serial number (should be allowed - different tables)
cursor.execute('''
INSERT INTO non_loanable_devices
(brand, model, serial_number, device_type, status)
VALUES (?, ?, ?, ?, 'available')
''', ('Brand', 'Model', 'UNIQUE001', 'projector'))
db_conn.commit()
# Both should exist (different tables)
cursor.execute("SELECT COUNT(*) FROM tablets WHERE serial_number = 'UNIQUE001'")
tablet_count = cursor.fetchone()[0]
cursor.execute("SELECT COUNT(*) FROM non_loanable_devices WHERE serial_number = 'UNIQUE001'")
device_count = cursor.fetchone()[0]
assert tablet_count == 1
assert device_count == 1
# But duplicate within same table should fail
with pytest.raises(sqlite3.IntegrityError):
cursor.execute('''
INSERT INTO tablets (brand, model, serial_number, status)
VALUES (?, ?, ?, 'available')
''', ('Brand2', 'Model2', 'UNIQUE001'))
db_conn.commit()
class TestQueryOperations:
"""Tests for query and filtering operations"""
def test_query_available_tablets(self, populated_db):
"""Test querying available tablets"""
cursor = populated_db.cursor()
cursor.execute("SELECT * FROM tablets WHERE status = 'available'")
available = cursor.fetchall()
# Should have at least the sample available tablets
assert len(available) >= 3 # SN001, SN002, SN003 from sample
def test_query_loaned_tablets(self, populated_db):
"""Test querying loaned tablets"""
cursor = populated_db.cursor()
cursor.execute("SELECT * FROM tablets WHERE status = 'loaned'")
loaned = cursor.fetchall()
# Should have at least SN004 from sample data
assert len(loaned) >= 1
def test_query_active_loans(self, populated_db):
"""Test querying active loans"""
cursor = populated_db.cursor()
cursor.execute("SELECT * FROM loans WHERE status = 'active'")
active = cursor.fetchall()
# Should have at least 1 active loan from sample
assert len(active) >= 1
def test_query_returned_loans(self, populated_db):
"""Test querying returned loans"""
cursor = populated_db.cursor()
cursor.execute("SELECT * FROM loans WHERE status = 'returned'")
returned = cursor.fetchall()
# Should have at least 1 returned loan from sample
assert len(returned) >= 1
def test_query_loans_by_user(self, populated_db):
"""Test querying loans by user"""
cursor = populated_db.cursor()
cursor.execute("SELECT id FROM users LIMIT 1")
user = cursor.fetchone()
if user:
cursor.execute("SELECT * FROM loans WHERE user_id = ?", (user['id'],))
loans = cursor.fetchall()
# User should have at least 0 loans
assert isinstance(loans, list)
def test_query_tablets_by_brand(self, populated_db):
"""Test querying tablets by brand"""
cursor = populated_db.cursor()
cursor.execute("SELECT * FROM tablets WHERE brand = 'Samsung'")
samsung = cursor.fetchall()
# Should find Samsung tablet from sample data
assert len(samsung) >= 1
assert samsung[0]['brand'] == 'Samsung'

View file

@ -1,513 +0,0 @@
"""
Edge case tests for Tablet Management System
Tests critical scenarios: already loaned devices, non-existent loans, duplicates, etc.
"""
import pytest
import sqlite3
from datetime import datetime
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
class TestLoanEdgeCases:
"""Critical edge cases for loan operations"""
def test_loan_device_already_loaned(self, db_conn):
"""
CRITICAL: Test that a device already loaned cannot be loaned again
This prevents the same physical device from being loaned to multiple users
"""
cursor = db_conn.cursor()
# Add a tablet
cursor.execute('''
INSERT INTO tablets (brand, model, serial_number, status)
VALUES (?, ?, ?, 'available')
''', ('Brand', 'Model', 'ALREADY_LOANED'))
db_conn.commit()
cursor.execute("SELECT id FROM tablets WHERE serial_number = 'ALREADY_LOANED'")
tablet = cursor.fetchone()
# Add a user
cursor.execute('''
INSERT INTO users (name, identification)
VALUES (?, ?)
''', ('User1', 'USER1'))
db_conn.commit()
cursor.execute("SELECT id FROM users WHERE identification = 'USER1'")
user1 = cursor.fetchone()
# Add another user
cursor.execute('''
INSERT INTO users (name, identification)
VALUES (?, ?)
''', ('User2', 'USER2'))
db_conn.commit()
cursor.execute("SELECT id FROM users WHERE identification = 'USER2'")
user2 = cursor.fetchone()
# Loan to first user
loan_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
cursor.execute("UPDATE tablets SET status = 'loaned' WHERE id = ?", (tablet['id'],))
cursor.execute('''
INSERT INTO loans (tablet_id, user_id, loan_date, status)
VALUES (?, ?, ?, 'active')
''', (tablet['id'], user1['id'], loan_date))
db_conn.commit()
# Verify tablet is loaned
cursor.execute("SELECT status FROM tablets WHERE id = ?", (tablet['id'],))
status = cursor.fetchone()['status']
assert status == 'loaned'
# Try to loan to second user - should check status first
# In the actual application, this would be prevented by checking status
cursor.execute("SELECT status FROM tablets WHERE id = ?", (tablet['id'],))
current_status = cursor.fetchone()['status']
# The application logic should prevent this
assert current_status == 'loaned'
# If we tried to loan it anyway (without checking), we'd get a constraint error
# because the tablet status is already 'loaned'
# The proper app logic checks status before allowing loan
def test_return_nonexistent_loan_id(self, db_conn):
"""
CRITICAL: Test returning a loan that doesn't exist
Should handle gracefully without crashing
"""
cursor = db_conn.cursor()
# Try to return a non-existent loan
loan_id = 99999
cursor.execute("SELECT tablet_id FROM loans WHERE id = ? AND status = 'active'", (loan_id,))
loan = cursor.fetchone()
# Should return None (no such loan)
assert loan is None
# The application should handle this by showing an error message
# rather than crashing
def test_return_already_returned_loan(self, db_conn):
"""
CRITICAL: Test returning a loan that's already been returned
Should handle gracefully
"""
cursor = db_conn.cursor()
# Add tablet and user
cursor.execute('''
INSERT INTO tablets (brand, model, serial_number, status)
VALUES (?, ?, ?, 'available')
''', ('Brand', 'Model', 'ALREADY_RETURNED'))
db_conn.commit()
cursor.execute("SELECT id FROM tablets WHERE serial_number = 'ALREADY_RETURNED'")
tablet = cursor.fetchone()
cursor.execute('''
INSERT INTO users (name, identification)
VALUES (?, ?)
''', ('User', 'RETURN_USER'))
db_conn.commit()
cursor.execute("SELECT id FROM users WHERE identification = 'RETURN_USER'")
user = cursor.fetchone()
# Create and return a loan
loan_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
return_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
cursor.execute("UPDATE tablets SET status = 'loaned' WHERE id = ?", (tablet['id'],))
cursor.execute('''
INSERT INTO loans (tablet_id, user_id, loan_date, return_date, status)
VALUES (?, ?, ?, ?, 'returned')
''', (tablet['id'], user['id'], loan_date, return_date))
cursor.execute("UPDATE tablets SET status = 'available' WHERE id = ?", (tablet['id'],))
db_conn.commit()
# Get the loan ID
cursor.execute("SELECT id FROM loans WHERE tablet_id = ?", (tablet['id'],))
loan = cursor.fetchone()
loan_id = loan['id']
# Try to return it again
cursor.execute("SELECT tablet_id FROM loans WHERE id = ? AND status = 'active'", (loan_id,))
active_loan = cursor.fetchone()
# Should be None because status is 'returned', not 'active'
assert active_loan is None
def test_loan_with_invalid_tablet_id(self, db_conn):
"""
CRITICAL: Test loaning with an invalid/non-existent tablet ID
Should fail gracefully
"""
cursor = db_conn.cursor()
# Add a user
cursor.execute('''
INSERT INTO users (name, identification)
VALUES (?, ?)
''', ('User', 'INVALID_USER'))
db_conn.commit()
cursor.execute("SELECT id FROM users WHERE identification = 'INVALID_USER'")
user = cursor.fetchone()
# Try to loan with invalid tablet ID
invalid_tablet_id = 99999
loan_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
# Note: SQLite doesn't enforce foreign keys by default unless we enable it
# The application should validate this at the application level
# For now, we verify that the tablet doesn't exist
cursor.execute("SELECT id FROM tablets WHERE id = ?", (invalid_tablet_id,))
tablet_check = cursor.fetchone()
assert tablet_check is None # Tablet doesn't exist
# In a real app with FK enforcement, this would raise IntegrityError
# For SQLite without FK enforcement, we just verify the tablet check
def test_loan_with_invalid_user_id(self, db_conn):
"""
CRITICAL: Test loaning with an invalid/non-existent user ID
Should fail gracefully
"""
cursor = db_conn.cursor()
# Add a tablet
cursor.execute('''
INSERT INTO tablets (brand, model, serial_number, status)
VALUES (?, ?, ?, 'available')
''', ('Brand', 'Model', 'INVALID_LOAN'))
db_conn.commit()
cursor.execute("SELECT id FROM tablets WHERE serial_number = 'INVALID_LOAN'")
tablet = cursor.fetchone()
# Try to loan with invalid user ID
invalid_user_id = 99999
loan_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
# Note: SQLite doesn't enforce foreign keys by default unless we enable it
# The application should validate this at the application level
# For now, we verify that the user doesn't exist
cursor.execute("SELECT id FROM users WHERE id = ?", (invalid_user_id,))
user_check = cursor.fetchone()
assert user_check is None # User doesn't exist
# In a real app with FK enforcement, this would raise IntegrityError
# For SQLite without FK enforcement, we just verify the user check
class TestDuplicatePrevention:
"""Tests for preventing duplicate entries"""
def test_duplicate_tablet_serial_number(self, db_conn):
"""
CRITICAL: Test that duplicate tablet serial numbers are prevented
Serial numbers must be unique for tracking
"""
cursor = db_conn.cursor()
# Add first tablet
cursor.execute('''
INSERT INTO tablets (brand, model, serial_number, status)
VALUES (?, ?, ?, 'available')
''', ('Brand1', 'Model1', 'DUP_SERIAL'))
db_conn.commit()
# Try to add duplicate
with pytest.raises(sqlite3.IntegrityError):
cursor.execute('''
INSERT INTO tablets (brand, model, serial_number, status)
VALUES (?, ?, ?, 'available')
''', ('Brand2', 'Model2', 'DUP_SERIAL'))
db_conn.commit()
def test_duplicate_user_identification(self, db_conn):
"""
CRITICAL: Test that duplicate user identifications are prevented
User identifications must be unique
"""
cursor = db_conn.cursor()
# Add first user
cursor.execute('''
INSERT INTO users (name, identification)
VALUES (?, ?)
''', ('User1', 'DUP_ID'))
db_conn.commit()
# Try to add duplicate
with pytest.raises(sqlite3.IntegrityError):
cursor.execute('''
INSERT INTO users (name, identification)
VALUES (?, ?)
''', ('User2', 'DUP_ID'))
db_conn.commit()
def test_duplicate_non_loanable_device_serial(self, db_conn):
"""
CRITICAL: Test that duplicate non-loanable device serials are prevented
"""
cursor = db_conn.cursor()
# Add first device
cursor.execute('''
INSERT INTO non_loanable_devices
(brand, model, serial_number, device_type, status)
VALUES (?, ?, ?, ?, 'available')
''', ('Brand', 'Model', 'DUP_DEVICE_SERIAL', 'projector'))
db_conn.commit()
# Try to add duplicate
with pytest.raises(sqlite3.IntegrityError):
cursor.execute('''
INSERT INTO non_loanable_devices
(brand, model, serial_number, device_type, status)
VALUES (?, ?, ?, ?, 'available')
''', ('Brand2', 'Model2', 'DUP_DEVICE_SERIAL', 'monitor'))
db_conn.commit()
class TestDataIntegrity:
"""Tests for data integrity constraints"""
def test_foreign_key_tablet_deletion(self, db_conn):
"""
Test that deleting a tablet with active loans is handled
SQLite defaults to allowing this, but we should be aware
"""
cursor = db_conn.cursor()
# Add tablet and user
cursor.execute('''
INSERT INTO tablets (brand, model, serial_number, status)
VALUES (?, ?, ?, 'available')
''', ('Brand', 'Model', 'FK_TEST'))
db_conn.commit()
cursor.execute("SELECT id FROM tablets WHERE serial_number = 'FK_TEST'")
tablet = cursor.fetchone()
cursor.execute('''
INSERT INTO users (name, identification)
VALUES (?, ?)
''', ('User', 'FK_USER'))
db_conn.commit()
cursor.execute("SELECT id FROM users WHERE identification = 'FK_USER'")
user = cursor.fetchone()
# Create active loan
loan_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
cursor.execute("UPDATE tablets SET status = 'loaned' WHERE id = ?", (tablet['id'],))
cursor.execute('''
INSERT INTO loans (tablet_id, user_id, loan_date, status)
VALUES (?, ?, ?, 'active')
''', (tablet['id'], user['id'], loan_date))
db_conn.commit()
# SQLite allows this by default (no ON DELETE RESTRICT)
# In production, we might want to add CASCADE or RESTRICT
# For now, just verify the loan exists
cursor.execute("SELECT COUNT(*) FROM loans WHERE tablet_id = ?", (tablet['id'],))
loan_count = cursor.fetchone()[0]
assert loan_count == 1
def test_null_serial_number_prevention(self, db_conn):
"""
Test that NULL serial numbers are prevented
Serial numbers are required (NOT NULL constraint)
"""
cursor = db_conn.cursor()
# Try to add tablet with NULL serial number
with pytest.raises(sqlite3.IntegrityError):
cursor.execute('''
INSERT INTO tablets (brand, model, serial_number, status)
VALUES (?, ?, ?, 'available')
''', ('Brand', 'Model', None))
db_conn.commit()
def test_null_identification_prevention(self, db_conn):
"""
Test that NULL user identifications are prevented
Identifications are required (NOT NULL constraint)
"""
cursor = db_conn.cursor()
# Try to add user with NULL identification
# Note: identification is NOT marked as NOT NULL in the schema
# This test verifies the current behavior
cursor.execute('''
INSERT INTO users (name, identification)
VALUES (?, ?)
''', ('User', None))
db_conn.commit()
# This should work because identification is not NOT NULL
# But in practice, we should have this constraint
cursor.execute("SELECT COUNT(*) FROM users WHERE identification IS NULL")
count = cursor.fetchone()[0]
# This will be 1, showing that NULL is currently allowed
# In production, we should add NOT NULL constraint
def test_empty_string_serial_number(self, db_conn):
"""
Test handling of empty string serial numbers
Empty strings are different from NULL
"""
cursor = db_conn.cursor()
# Add tablet with empty string serial number
cursor.execute('''
INSERT INTO tablets (brand, model, serial_number, status)
VALUES (?, ?, ?, 'available')
''', ('Brand', 'Model', ''))
db_conn.commit()
# This should work (empty string is allowed unless we add CHECK constraint)
cursor.execute("SELECT COUNT(*) FROM tablets WHERE serial_number = ''")
count = cursor.fetchone()[0]
assert count == 1
# In production, we might want to prevent empty strings
# with a CHECK constraint: CHECK(serial_number <> '')
class TestConcurrentScenarioSimulations:
"""Simulate scenarios that could cause issues in concurrent environments"""
def test_loan_return_loan_sequence(self, db_conn):
"""
Test the sequence: loan -> return -> loan again
This simulates a device being loaned multiple times over its lifetime
"""
cursor = db_conn.cursor()
# Add tablet and user
cursor.execute('''
INSERT INTO tablets (brand, model, serial_number, status)
VALUES (?, ?, ?, 'available')
''', ('Brand', 'Model', 'SEQUENCE_TEST'))
db_conn.commit()
cursor.execute("SELECT id FROM tablets WHERE serial_number = 'SEQUENCE_TEST'")
tablet = cursor.fetchone()
cursor.execute('''
INSERT INTO users (name, identification)
VALUES (?, ?)
''', ('User', 'SEQUENCE_USER'))
db_conn.commit()
cursor.execute("SELECT id FROM users WHERE identification = 'SEQUENCE_USER'")
user = cursor.fetchone()
# First loan
loan_date1 = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
cursor.execute("UPDATE tablets SET status = 'loaned' WHERE id = ?", (tablet['id'],))
cursor.execute('''
INSERT INTO loans (tablet_id, user_id, loan_date, status)
VALUES (?, ?, ?, 'active')
''', (tablet['id'], user['id'], loan_date1))
db_conn.commit()
# Return
return_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
cursor.execute("SELECT id FROM loans WHERE tablet_id = ?", (tablet['id'],))
loan1 = cursor.fetchone()
cursor.execute('''
UPDATE loans SET status = 'returned', return_date = ? WHERE id = ?
''', (return_date, loan1['id']))
cursor.execute("UPDATE tablets SET status = 'available' WHERE id = ?", (tablet['id'],))
db_conn.commit()
# Second loan
loan_date2 = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
cursor.execute("UPDATE tablets SET status = 'loaned' WHERE id = ?", (tablet['id'],))
cursor.execute('''
INSERT INTO loans (tablet_id, user_id, loan_date, status)
VALUES (?, ?, ?, 'active')
''', (tablet['id'], user['id'], loan_date2))
db_conn.commit()
# Verify we have 2 loans for this tablet
cursor.execute("SELECT COUNT(*) FROM loans WHERE tablet_id = ?", (tablet['id'],))
loan_count = cursor.fetchone()[0]
assert loan_count == 2
# Verify 1 active, 1 returned
cursor.execute("SELECT COUNT(*) FROM loans WHERE tablet_id = ? AND status = 'active'", (tablet['id'],))
active_count = cursor.fetchone()[0]
cursor.execute("SELECT COUNT(*) FROM loans WHERE tablet_id = ? AND status = 'returned'", (tablet['id'],))
returned_count = cursor.fetchone()[0]
assert active_count == 1
assert returned_count == 1
def test_multiple_users_multiple_tablets(self, db_conn):
"""
Test complex scenario with multiple users and tablets
Ensures the many-to-many relationship works correctly
"""
cursor = db_conn.cursor()
# Add 3 users
users = []
for i in range(3):
cursor.execute('''
INSERT INTO users (name, identification)
VALUES (?, ?)
''', (f'User{i}', f'MULTI_USER_{i}'))
db_conn.commit()
cursor.execute("SELECT id FROM users WHERE identification = ?", (f'MULTI_USER_{i}',))
users.append(cursor.fetchone())
# Add 5 tablets
tablets = []
for i in range(5):
cursor.execute('''
INSERT INTO tablets (brand, model, serial_number, status)
VALUES (?, ?, ?, 'available')
''', (f'Brand{i}', f'Model{i}', f'MULTI_TABLET_{i}'))
db_conn.commit()
cursor.execute("SELECT id FROM tablets WHERE serial_number = ?", (f'MULTI_TABLET_{i}',))
tablets.append(cursor.fetchone())
# Loan tablets to users in a pattern
# User 0: tablets 0, 1
# User 1: tablets 2, 3
# User 2: tablet 4
loan_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
loans_map = {
users[0]['id']: [tablets[0]['id'], tablets[1]['id']],
users[1]['id']: [tablets[2]['id'], tablets[3]['id']],
users[2]['id']: [tablets[4]['id']],
}
for user_id, tablet_ids in loans_map.items():
for tablet_id in tablet_ids:
cursor.execute("UPDATE tablets SET status = 'loaned' WHERE id = ?", (tablet_id,))
cursor.execute('''
INSERT INTO loans (tablet_id, user_id, loan_date, status)
VALUES (?, ?, ?, 'active')
''', (tablet_id, user_id, loan_date))
db_conn.commit()
# Verify counts
cursor.execute("SELECT COUNT(*) FROM loans")
total_loans = cursor.fetchone()[0]
assert total_loans == 5 # 2 + 2 + 1
# Verify each user has correct number of loans
for user_id, expected_tablet_ids in loans_map.items():
cursor.execute("SELECT COUNT(*) FROM loans WHERE user_id = ?", (user_id,))
count = cursor.fetchone()[0]
assert count == len(expected_tablet_ids)
# Verify all loaned tablets have correct status
cursor.execute("SELECT COUNT(*) FROM tablets WHERE status = 'loaned'")
loaned_count = cursor.fetchone()[0]
assert loaned_count == 5

View file

@ -1,365 +0,0 @@
"""
Unit tests for the actual application functions from minimal_app.py
Tests the real business logic with proper imports
"""
import pytest
import sqlite3
import sys
import os
from datetime import datetime
# Add project root to path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# Import functions from minimal_app
from minimal_app import (
add_tablet, add_user, loan_tablet, return_tablet,
show_available_tablets, show_active_loans, show_loan_history,
add_non_loanable_device, show_non_loanable_devices, delete_non_loanable_device
)
@pytest.fixture
def test_db_path():
"""Path to test database"""
return 'test_minimal_app.db'
@pytest.fixture
def init_test_db(test_db_path):
"""Initialize a fresh test database with schema (same as minimal_app)"""
# Remove existing test database if it exists
if os.path.exists(test_db_path):
os.remove(test_db_path)
# Use the same init_db function from minimal_app
from minimal_app import init_db
# Temporarily rename the database
original_db = 'tablets.db'
if os.path.exists(original_db):
os.rename(original_db, f'{original_db}.backup')
try:
# Create test database
os.environ['TEST_DB'] = test_db_path
init_db()
yield test_db_path
finally:
# Cleanup
if os.path.exists(test_db_path):
os.remove(test_db_path)
if os.path.exists(f'{original_db}.backup'):
os.rename(f'{original_db}.backup', original_db)
if 'TEST_DB' in os.environ:
del os.environ['TEST_DB']
@pytest.fixture
def clean_db():
"""Fixture that ensures we have a clean database for each test"""
# This is simpler - just create a temp database for each test
import tempfile
import shutil
# Create temp directory for database
temp_dir = tempfile.mkdtemp()
db_path = os.path.join(temp_dir, 'test.db')
# Initialize database
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS tablets (
id INTEGER PRIMARY KEY AUTOINCREMENT,
brand TEXT NOT NULL,
model TEXT NOT NULL,
serial_number TEXT UNIQUE NOT NULL,
status TEXT DEFAULT 'available'
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
identification TEXT UNIQUE
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS loans (
id INTEGER PRIMARY KEY AUTOINCREMENT,
tablet_id INTEGER NOT NULL,
user_id INTEGER NOT NULL,
loan_date TEXT NOT NULL,
return_date TEXT,
status TEXT DEFAULT 'active',
FOREIGN KEY (tablet_id) REFERENCES tablets (id),
FOREIGN KEY (user_id) REFERENCES users (id)
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS non_loanable_devices (
id INTEGER PRIMARY KEY AUTOINCREMENT,
brand TEXT NOT NULL,
model TEXT NOT NULL,
serial_number TEXT UNIQUE NOT NULL,
device_type TEXT NOT NULL,
location TEXT,
status TEXT DEFAULT 'available',
notes TEXT,
purchase_date TEXT,
purchase_cost REAL
)
''')
conn.commit()
conn.close()
# Temporarily replace the database
original_db = 'tablets.db'
backup_path = f'{original_db}.test_backup'
# Backup original if exists
if os.path.exists(original_db):
if os.path.exists(backup_path):
os.remove(backup_path)
os.rename(original_db, backup_path)
# Copy temp db to tablets.db location
shutil.copy(db_path, original_db)
yield original_db
# Cleanup
if os.path.exists(original_db):
os.remove(original_db)
if os.path.exists(backup_path):
os.rename(backup_path, original_db)
shutil.rmtree(temp_dir, ignore_errors=True)
class TestMinimalAppFunctions:
"""Test the actual functions from minimal_app.py"""
def test_add_tablet_function(self, clean_db):
"""Test the add_tablet function"""
add_tablet('TestBrand', 'TestModel', 'TEST_SN_001')
# Verify it was added
conn = sqlite3.connect(clean_db)
cursor = conn.cursor()
cursor.execute("SELECT * FROM tablets WHERE serial_number = 'TEST_SN_001'")
tablet = cursor.fetchone()
conn.close()
assert tablet is not None
assert tablet[1] == 'TestBrand' # brand is index 1
assert tablet[2] == 'TestModel' # model is index 2
assert tablet[3] == 'TEST_SN_001' # serial_number is index 3
def test_add_tablet_duplicate(self, clean_db, capsys):
"""Test that duplicate serial numbers are rejected"""
add_tablet('Brand1', 'Model1', 'DUP_SN')
add_tablet('Brand2', 'Model2', 'DUP_SN')
captured = capsys.readouterr()
assert 'already exists' in captured.out
def test_add_user_function(self, clean_db):
"""Test the add_user function"""
add_user('Test User', 'TEST_ID_001')
conn = sqlite3.connect(clean_db)
cursor = conn.cursor()
cursor.execute("SELECT * FROM users WHERE identification = 'TEST_ID_001'")
user = cursor.fetchone()
conn.close()
assert user is not None
assert user[1] == 'Test User' # name is index 1
assert user[2] == 'TEST_ID_001' # identification is index 2
def test_add_user_duplicate(self, clean_db, capsys):
"""Test that duplicate user identifications are rejected"""
add_user('User1', 'DUP_ID')
add_user('User2', 'DUP_ID')
captured = capsys.readouterr()
assert 'already exists' in captured.out
def test_loan_tablet_function(self, clean_db):
"""Test the loan_tablet function"""
# Add tablet and user
add_tablet('LoanBrand', 'LoanModel', 'LOAN_SN_001')
add_user('LoanUser', 'LOAN_ID_001')
conn = sqlite3.connect(clean_db)
cursor = conn.cursor()
cursor.execute("SELECT id FROM tablets WHERE serial_number = 'LOAN_SN_001'")
tablet_id = cursor.fetchone()[0]
cursor.execute("SELECT id FROM users WHERE identification = 'LOAN_ID_001'")
user_id = cursor.fetchone()[0]
conn.close()
# Loan the tablet
loan_tablet(tablet_id, user_id)
# Verify loan was created and tablet status changed
conn = sqlite3.connect(clean_db)
cursor = conn.cursor()
cursor.execute("SELECT status FROM tablets WHERE id = ?", (tablet_id,))
status = cursor.fetchone()[0]
assert status == 'loaned'
cursor.execute("SELECT * FROM loans WHERE tablet_id = ? AND user_id = ?", (tablet_id, user_id))
loan = cursor.fetchone()
assert loan is not None
assert loan[5] == 'active' # status is index 5
conn.close()
def test_loan_already_loaned_tablet(self, clean_db, capsys):
"""Test loaning a tablet that's already loaned"""
add_tablet('Brand', 'Model', 'ALREADY_LOANED_SN')
add_user('User1', 'USER1_ID')
add_user('User2', 'USER2_ID')
conn = sqlite3.connect(clean_db)
cursor = conn.cursor()
cursor.execute("SELECT id FROM tablets WHERE serial_number = 'ALREADY_LOANED_SN'")
tablet_id = cursor.fetchone()[0]
cursor.execute("SELECT id FROM users WHERE identification = 'USER1_ID'")
user1_id = cursor.fetchone()[0]
cursor.execute("SELECT id FROM users WHERE identification = 'USER2_ID'")
user2_id = cursor.fetchone()[0]
conn.close()
# Loan to first user
loan_tablet(tablet_id, user1_id)
# Try to loan to second user - should fail
loan_tablet(tablet_id, user2_id)
captured = capsys.readouterr()
assert 'not available' in captured.out
def test_return_tablet_function(self, clean_db):
"""Test the return_tablet function"""
add_tablet('ReturnBrand', 'ReturnModel', 'RETURN_SN_001')
add_user('ReturnUser', 'RETURN_ID_001')
conn = sqlite3.connect(clean_db)
cursor = conn.cursor()
cursor.execute("SELECT id FROM tablets WHERE serial_number = 'RETURN_SN_001'")
tablet_id = cursor.fetchone()[0]
cursor.execute("SELECT id FROM users WHERE identification = 'RETURN_ID_001'")
user_id = cursor.fetchone()[0]
conn.close()
# Loan the tablet
loan_tablet(tablet_id, user_id)
# Get loan ID
conn = sqlite3.connect(clean_db)
cursor = conn.cursor()
cursor.execute("SELECT id FROM loans WHERE tablet_id = ?", (tablet_id,))
loan_id = cursor.fetchone()[0]
conn.close()
# Return the tablet
return_tablet(loan_id)
# Verify return
conn = sqlite3.connect(clean_db)
cursor = conn.cursor()
cursor.execute("SELECT status FROM tablets WHERE id = ?", (tablet_id,))
status = cursor.fetchone()[0]
assert status == 'available'
cursor.execute("SELECT status, return_date FROM loans WHERE id = ?", (loan_id,))
loan = cursor.fetchone()
assert loan[0] == 'returned'
assert loan[1] is not None # return_date should be set
conn.close()
def test_return_nonexistent_loan(self, clean_db, capsys):
"""Test returning a loan that doesn't exist"""
return_tablet(99999)
captured = capsys.readouterr()
assert 'not found' in captured.out or 'Loan' in captured.out
def test_show_available_tablets(self, clean_db, capsys):
"""Test showing available tablets"""
add_tablet('Avail1', 'Model1', 'AVAIL_SN_001')
add_tablet('Avail2', 'Model2', 'AVAIL_SN_002')
show_available_tablets()
captured = capsys.readouterr()
assert 'Available Tablets' in captured.out
assert 'AVAIL_SN_001' in captured.out
assert 'AVAIL_SN_002' in captured.out
def test_show_active_loans(self, clean_db, capsys):
"""Test showing active loans"""
add_tablet('LoanBrand', 'LoanModel', 'ACTIVE_LOAN_SN')
add_user('LoanUser', 'ACTIVE_LOAN_ID')
conn = sqlite3.connect(clean_db)
cursor = conn.cursor()
cursor.execute("SELECT id FROM tablets WHERE serial_number = 'ACTIVE_LOAN_SN'")
tablet_id = cursor.fetchone()[0]
cursor.execute("SELECT id FROM users WHERE identification = 'ACTIVE_LOAN_ID'")
user_id = cursor.fetchone()[0]
conn.close()
loan_tablet(tablet_id, user_id)
show_active_loans()
captured = capsys.readouterr()
assert 'Active Loans' in captured.out
assert 'ACTIVE_LOAN_SN' in captured.out
def test_non_loanable_device_crud(self, clean_db, capsys):
"""Test CRUD operations for non-loanable devices"""
# Add
add_non_loanable_device('Projector', 'P100', 'PROJ_001', 'projector', 'Room A')
captured = capsys.readouterr()
assert 'Added non-loanable device' in captured.out
# Show
show_non_loanable_devices()
captured = capsys.readouterr()
assert 'PROJ_001' in captured.out
# Get ID for delete
conn = sqlite3.connect(clean_db)
cursor = conn.cursor()
cursor.execute("SELECT id FROM non_loanable_devices WHERE serial_number = 'PROJ_001'")
device_id = cursor.fetchone()[0]
conn.close()
# Delete
delete_non_loanable_device(device_id)
captured = capsys.readouterr()
assert 'deleted' in captured.out
# Verify deletion
conn = sqlite3.connect(clean_db)
cursor = conn.cursor()
cursor.execute("SELECT * FROM non_loanable_devices WHERE serial_number = 'PROJ_001'")
device = cursor.fetchone()
conn.close()
assert device is None