feat(schemas): T23 add Pydantic API key schemas
- Add ApiKeyCreate schema with OpenRouter key format validation - Add ApiKeyUpdate schema for partial updates - Add ApiKeyResponse schema (excludes key value for security) - Add ApiKeyListResponse schema for pagination - Export schemas from __init__.py - 100% coverage on new module (23 tests) Refs: T23
This commit is contained in:
@@ -76,7 +76,7 @@
|
||||
**Coverage auth:** 98%+
|
||||
|
||||
### 🔑 Gestione API Keys (T23-T29) - 0/7 completati
|
||||
- [ ] T23: Creare Pydantic schemas per API keys
|
||||
- [ ] T23: Creare Pydantic schemas per API keys - 🟡 In progress (2026-04-07 16:00)
|
||||
- [ ] T24: Implementare POST /api/keys (create)
|
||||
- [ ] T25: Implementare GET /api/keys (list)
|
||||
- [ ] T26: Implementare PUT /api/keys/{id} (update)
|
||||
|
||||
571
prompt/prompt-ingaggio-api-keys.md
Normal file
571
prompt/prompt-ingaggio-api-keys.md
Normal file
@@ -0,0 +1,571 @@
|
||||
# Prompt di Ingaggio: Gestione API Keys (T23-T29)
|
||||
|
||||
## 🎯 MISSIONE
|
||||
|
||||
Implementare la fase **Gestione API Keys** del progetto OpenRouter API Key Monitor seguendo rigorosamente TDD.
|
||||
|
||||
**Task da completare:** T23, T24, T25, T26, T27, T28, T29
|
||||
|
||||
---
|
||||
|
||||
## 📋 CONTESTO
|
||||
|
||||
**AGENTE:** @tdd-developer
|
||||
|
||||
**Repository:** `/home/google/Sources/LucaSacchiNet/openrouter-watcher`
|
||||
|
||||
**Stato Attuale:**
|
||||
- ✅ Setup (T01-T05): 59 test
|
||||
- ✅ Database & Models (T06-T11): 73 test
|
||||
- ✅ Security Services (T12-T16): 70 test
|
||||
- ✅ User Authentication (T17-T22): 34 test
|
||||
- 🎯 **Totale: 236 test passanti, 98.23% coverage**
|
||||
|
||||
**Servizi Pronti da utilizzare:**
|
||||
- `EncryptionService` - Cifratura/decifratura API keys
|
||||
- `hash_password()`, `verify_password()` - Autenticazione
|
||||
- `create_access_token()`, `decode_access_token()` - JWT
|
||||
- `get_current_user()` - Dependency injection
|
||||
- `generate_api_token()` - Token API pubblica
|
||||
|
||||
**Modelli Pronti:**
|
||||
- `User`, `ApiKey`, `UsageStats`, `ApiToken` - SQLAlchemy models
|
||||
- `get_db()` - Database session
|
||||
|
||||
**Documentazione:**
|
||||
- PRD: `/home/google/Sources/LucaSacchiNet/openrouter-watcher/prd.md`
|
||||
- Architecture: `/home/google/Sources/LucaSacchiNet/openrouter-watcher/export/architecture.md`
|
||||
- Kanban: `/home/google/Sources/LucaSacchiNet/openrouter-watcher/export/kanban.md`
|
||||
|
||||
---
|
||||
|
||||
## 🔧 TASK DA IMPLEMENTARE
|
||||
|
||||
### T23: Creare Pydantic Schemas per API Keys
|
||||
|
||||
**File:** `src/openrouter_monitor/schemas/api_key.py`
|
||||
|
||||
**Requisiti:**
|
||||
- `ApiKeyCreate`: name (str, min 1, max 100), key (str) - OpenRouter API key
|
||||
- `ApiKeyUpdate`: name (optional), is_active (optional)
|
||||
- `ApiKeyResponse`: id, name, is_active, created_at, last_used_at (orm_mode=True)
|
||||
- `ApiKeyListResponse`: items (list[ApiKeyResponse]), total (int)
|
||||
- Validazione: key deve iniziare con "sk-or-v1-" (formato OpenRouter)
|
||||
|
||||
**Implementazione:**
|
||||
```python
|
||||
from pydantic import BaseModel, Field, validator
|
||||
from datetime import datetime
|
||||
from typing import Optional, List
|
||||
|
||||
class ApiKeyCreate(BaseModel):
|
||||
name: str = Field(..., min_length=1, max_length=100)
|
||||
key: str = Field(..., min_length=20)
|
||||
|
||||
@validator('key')
|
||||
def validate_openrouter_key_format(cls, v):
|
||||
if not v.startswith('sk-or-v1-'):
|
||||
raise ValueError('Invalid OpenRouter API key format')
|
||||
return v
|
||||
|
||||
class ApiKeyUpdate(BaseModel):
|
||||
name: Optional[str] = Field(None, min_length=1, max_length=100)
|
||||
is_active: Optional[bool] = None
|
||||
|
||||
class ApiKeyResponse(BaseModel):
|
||||
id: int
|
||||
name: str
|
||||
is_active: bool
|
||||
created_at: datetime
|
||||
last_used_at: Optional[datetime] = None
|
||||
|
||||
class Config:
|
||||
from_attributes = True # Pydantic v2
|
||||
|
||||
class ApiKeyListResponse(BaseModel):
|
||||
items: List[ApiKeyResponse]
|
||||
total: int
|
||||
```
|
||||
|
||||
**Test:** `tests/unit/schemas/test_api_key_schemas.py` (8+ test)
|
||||
|
||||
---
|
||||
|
||||
### T24: Implementare POST /api/keys (Create API Key)
|
||||
|
||||
**File:** `src/openrouter_monitor/routers/api_keys.py`
|
||||
|
||||
**Requisiti:**
|
||||
- Endpoint: `POST /api/keys`
|
||||
- Auth: Richiede `current_user: User = Depends(get_current_user)`
|
||||
- Riceve: `ApiKeyCreate` schema
|
||||
- Verifica limite API keys per utente (`MAX_API_KEYS_PER_USER`)
|
||||
- Cifra API key con `EncryptionService`
|
||||
- Salva nel DB: `ApiKey(user_id=current_user.id, name=..., key_encrypted=...)`
|
||||
- Ritorna: `ApiKeyResponse`, status 201
|
||||
- Errori: limite raggiunto (400), formato key invalido (422)
|
||||
|
||||
**Implementazione:**
|
||||
```python
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy import func
|
||||
|
||||
router = APIRouter(prefix="/api/keys", tags=["api-keys"])
|
||||
|
||||
@router.post("", response_model=ApiKeyResponse, status_code=status.HTTP_201_CREATED)
|
||||
async def create_api_key(
|
||||
key_data: ApiKeyCreate,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
# Verifica limite API keys
|
||||
current_count = db.query(func.count(ApiKey.id)).filter(
|
||||
ApiKey.user_id == current_user.id,
|
||||
ApiKey.is_active == True
|
||||
).scalar()
|
||||
|
||||
if current_count >= settings.max_api_keys_per_user:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"Maximum {settings.max_api_keys_per_user} API keys allowed"
|
||||
)
|
||||
|
||||
# Cifra API key
|
||||
encryption_service = EncryptionService(settings.encryption_key)
|
||||
encrypted_key = encryption_service.encrypt(key_data.key)
|
||||
|
||||
# Crea API key
|
||||
api_key = ApiKey(
|
||||
user_id=current_user.id,
|
||||
name=key_data.name,
|
||||
key_encrypted=encrypted_key,
|
||||
is_active=True
|
||||
)
|
||||
db.add(api_key)
|
||||
db.commit()
|
||||
db.refresh(api_key)
|
||||
|
||||
return api_key
|
||||
```
|
||||
|
||||
**Test:** `tests/unit/routers/test_api_keys.py`
|
||||
- Test creazione successo (201)
|
||||
- Test limite massimo raggiunto (400)
|
||||
- Test formato key invalido (422)
|
||||
- Test utente non autenticato (401)
|
||||
|
||||
---
|
||||
|
||||
### T25: Implementare GET /api/keys (List API Keys)
|
||||
|
||||
**File:** `src/openrouter_monitor/routers/api_keys.py`
|
||||
|
||||
**Requisiti:**
|
||||
- Endpoint: `GET /api/keys`
|
||||
- Auth: Richiede `current_user`
|
||||
- Query params: skip (default 0), limit (default 10, max 100)
|
||||
- Ritorna: solo API keys dell'utente corrente
|
||||
- Ordinamento: created_at DESC (più recenti prima)
|
||||
- Ritorna: `ApiKeyListResponse`
|
||||
|
||||
**Implementazione:**
|
||||
```python
|
||||
@router.get("", response_model=ApiKeyListResponse)
|
||||
async def list_api_keys(
|
||||
skip: int = 0,
|
||||
limit: int = 10,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
query = db.query(ApiKey).filter(ApiKey.user_id == current_user.id)
|
||||
total = query.count()
|
||||
|
||||
api_keys = query.order_by(ApiKey.created_at.desc()).offset(skip).limit(limit).all()
|
||||
|
||||
return ApiKeyListResponse(
|
||||
items=api_keys,
|
||||
total=total
|
||||
)
|
||||
```
|
||||
|
||||
**Test:**
|
||||
- Test lista vuota (utente senza keys)
|
||||
- Test lista con API keys
|
||||
- Test paginazione (skip, limit)
|
||||
- Test ordinamento (più recenti prima)
|
||||
- Test utente vede solo proprie keys
|
||||
|
||||
---
|
||||
|
||||
### T26: Implementare PUT /api/keys/{id} (Update API Key)
|
||||
|
||||
**File:** `src/openrouter_monitor/routers/api_keys.py`
|
||||
|
||||
**Requisiti:**
|
||||
- Endpoint: `PUT /api/keys/{key_id}`
|
||||
- Auth: Richiede `current_user`
|
||||
- Riceve: `ApiKeyUpdate` schema
|
||||
- Verifica: API key esiste e appartiene all'utente corrente
|
||||
- Aggiorna: solo campi forniti (name, is_active)
|
||||
- Ritorna: `ApiKeyResponse`
|
||||
- Errori: key non trovata (404), non autorizzato (403)
|
||||
|
||||
**Implementazione:**
|
||||
```python
|
||||
@router.put("/{key_id}", response_model=ApiKeyResponse)
|
||||
async def update_api_key(
|
||||
key_id: int,
|
||||
key_data: ApiKeyUpdate,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
api_key = db.query(ApiKey).filter(ApiKey.id == key_id).first()
|
||||
|
||||
if not api_key:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="API key not found"
|
||||
)
|
||||
|
||||
if api_key.user_id != current_user.id:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Not authorized to update this API key"
|
||||
)
|
||||
|
||||
# Aggiorna solo campi forniti
|
||||
if key_data.name is not None:
|
||||
api_key.name = key_data.name
|
||||
if key_data.is_active is not None:
|
||||
api_key.is_active = key_data.is_active
|
||||
|
||||
db.commit()
|
||||
db.refresh(api_key)
|
||||
|
||||
return api_key
|
||||
```
|
||||
|
||||
**Test:**
|
||||
- Test aggiornamento nome successo
|
||||
- Test aggiornamento is_active successo
|
||||
- Test key non esistente (404)
|
||||
- Test key di altro utente (403)
|
||||
|
||||
---
|
||||
|
||||
### T27: Implementare DELETE /api/keys/{id} (Delete API Key)
|
||||
|
||||
**File:** `src/openrouter_monitor/routers/api_keys.py`
|
||||
|
||||
**Requisiti:**
|
||||
- Endpoint: `DELETE /api/keys/{key_id}`
|
||||
- Auth: Richiede `current_user`
|
||||
- Verifica: API key esiste e appartiene all'utente corrente
|
||||
- Elimina: record dal DB (cascade elimina anche usage_stats)
|
||||
- Ritorna: status 204 (No Content)
|
||||
- Errori: key non trovata (404), non autorizzato (403)
|
||||
|
||||
**Implementazione:**
|
||||
```python
|
||||
@router.delete("/{key_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
async def delete_api_key(
|
||||
key_id: int,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
api_key = db.query(ApiKey).filter(ApiKey.id == key_id).first()
|
||||
|
||||
if not api_key:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="API key not found"
|
||||
)
|
||||
|
||||
if api_key.user_id != current_user.id:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Not authorized to delete this API key"
|
||||
)
|
||||
|
||||
db.delete(api_key)
|
||||
db.commit()
|
||||
|
||||
return None
|
||||
```
|
||||
|
||||
**Test:**
|
||||
- Test eliminazione successo (204)
|
||||
- Test key non esistente (404)
|
||||
- Test key di altro utente (403)
|
||||
|
||||
---
|
||||
|
||||
### T28: Implementare Validazione API Key con OpenRouter
|
||||
|
||||
**File:** `src/openrouter_monitor/services/openrouter.py`
|
||||
|
||||
**Requisiti:**
|
||||
- Funzione: `validate_api_key(key: str) -> bool`
|
||||
- Chiama endpoint OpenRouter: `GET https://openrouter.ai/api/v1/auth/key`
|
||||
- Header: `Authorization: Bearer {key}`
|
||||
- Ritorna: True se valida (200), False se invalida (401/403)
|
||||
- Usa `httpx` per richieste HTTP
|
||||
- Timeout: 10 secondi
|
||||
|
||||
**Implementazione:**
|
||||
```python
|
||||
import httpx
|
||||
from openrouter_monitor.config import get_settings
|
||||
|
||||
settings = get_settings()
|
||||
|
||||
async def validate_api_key(key: str) -> bool:
|
||||
"""Validate OpenRouter API key by calling their API."""
|
||||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||||
try:
|
||||
response = await client.get(
|
||||
f"{settings.openrouter_api_url}/auth/key",
|
||||
headers={"Authorization": f"Bearer {key}"}
|
||||
)
|
||||
return response.status_code == 200
|
||||
except httpx.RequestError:
|
||||
return False
|
||||
|
||||
async def get_key_info(key: str) -> dict | None:
|
||||
"""Get API key info from OpenRouter."""
|
||||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||||
try:
|
||||
response = await client.get(
|
||||
f"{settings.openrouter_api_url}/auth/key",
|
||||
headers={"Authorization": f"Bearer {key}"}
|
||||
)
|
||||
if response.status_code == 200:
|
||||
return response.json()
|
||||
return None
|
||||
except httpx.RequestError:
|
||||
return None
|
||||
```
|
||||
|
||||
**Test:** `tests/unit/services/test_openrouter.py`
|
||||
- Test key valida ritorna True
|
||||
- Test key invalida ritorna False
|
||||
- Test timeout ritorna False
|
||||
- Test network error gestito
|
||||
|
||||
---
|
||||
|
||||
### T29: Scrivere Test per API Keys Endpoints
|
||||
|
||||
**File:** `tests/unit/routers/test_api_keys.py`
|
||||
|
||||
**Requisiti:**
|
||||
- Test integrazione completo per tutti gli endpoint
|
||||
- Usare TestClient con FastAPI
|
||||
- Mock EncryptionService per test veloci
|
||||
- Mock chiamate OpenRouter per T28
|
||||
- Coverage >= 90%
|
||||
|
||||
**Test da implementare:**
|
||||
- **Create Tests (T24):**
|
||||
- POST /api/keys successo (201)
|
||||
- POST /api/keys limite raggiunto (400)
|
||||
- POST /api/keys formato invalido (422)
|
||||
- POST /api/keys senza auth (401)
|
||||
|
||||
- **List Tests (T25):**
|
||||
- GET /api/keys lista vuota
|
||||
- GET /api/keys con dati
|
||||
- GET /api/keys paginazione
|
||||
- GET /api/keys senza auth (401)
|
||||
|
||||
- **Update Tests (T26):**
|
||||
- PUT /api/keys/{id} aggiorna nome
|
||||
- PUT /api/keys/{id} aggiorna is_active
|
||||
- PUT /api/keys/{id} key non esiste (404)
|
||||
- PUT /api/keys/{id} key di altro utente (403)
|
||||
|
||||
- **Delete Tests (T27):**
|
||||
- DELETE /api/keys/{id} successo (204)
|
||||
- DELETE /api/keys/{id} key non esiste (404)
|
||||
- DELETE /api/keys/{id} key di altro utente (403)
|
||||
|
||||
- **Security Tests:**
|
||||
- Utente A non vede keys di utente B
|
||||
- Utente A non modifica keys di utente B
|
||||
- Utente A non elimina keys di utente B
|
||||
|
||||
---
|
||||
|
||||
## 🔄 WORKFLOW TDD
|
||||
|
||||
Per **OGNI** task:
|
||||
|
||||
1. **RED**: Scrivi test che fallisce (prima del codice!)
|
||||
2. **GREEN**: Implementa codice minimo per passare il test
|
||||
3. **REFACTOR**: Migliora codice, test rimangono verdi
|
||||
|
||||
---
|
||||
|
||||
## 📁 STRUTTURA FILE DA CREARE/MODIFICARE
|
||||
|
||||
```
|
||||
src/openrouter_monitor/
|
||||
├── schemas/
|
||||
│ ├── __init__.py # Aggiungi export ApiKey schemas
|
||||
│ └── api_key.py # T23
|
||||
├── routers/
|
||||
│ ├── __init__.py # Aggiungi api_keys router
|
||||
│ ├── auth.py # Esistente
|
||||
│ └── api_keys.py # T24, T25, T26, T27
|
||||
├── services/
|
||||
│ ├── __init__.py # Aggiungi export openrouter
|
||||
│ └── openrouter.py # T28
|
||||
└── main.py # Registra api_keys router
|
||||
|
||||
tests/unit/
|
||||
├── schemas/
|
||||
│ └── test_api_key_schemas.py # T23 + T29
|
||||
├── routers/
|
||||
│ └── test_api_keys.py # T24-T27 + T29
|
||||
└── services/
|
||||
└── test_openrouter.py # T28 + T29
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 🧪 ESEMPI TEST
|
||||
|
||||
### Test Schema
|
||||
```python
|
||||
def test_api_key_create_valid_data_passes_validation():
|
||||
data = ApiKeyCreate(
|
||||
name="Production Key",
|
||||
key="sk-or-v1-abc123..."
|
||||
)
|
||||
assert data.name == "Production Key"
|
||||
assert data.key.startswith("sk-or-v1-")
|
||||
```
|
||||
|
||||
### Test Endpoint Create
|
||||
```python
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_api_key_success_returns_201(client, auth_token, db_session):
|
||||
response = client.post(
|
||||
"/api/keys",
|
||||
json={"name": "Test Key", "key": "sk-or-v1-validkey123"},
|
||||
headers={"Authorization": f"Bearer {auth_token}"}
|
||||
)
|
||||
assert response.status_code == 201
|
||||
assert response.json()["name"] == "Test Key"
|
||||
assert "id" in response.json()
|
||||
```
|
||||
|
||||
### Test Sicurezza
|
||||
```python
|
||||
def test_user_cannot_see_other_user_api_keys(client, auth_token_user_a, api_key_user_b):
|
||||
response = client.get(
|
||||
"/api/keys",
|
||||
headers={"Authorization": f"Bearer {auth_token_user_a}"}
|
||||
)
|
||||
assert response.status_code == 200
|
||||
# Verifica che key di user_b non sia nella lista
|
||||
key_ids = [k["id"] for k in response.json()["items"]]
|
||||
assert api_key_user_b.id not in key_ids
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## ✅ CRITERI DI ACCETTAZIONE
|
||||
|
||||
- [ ] T23: Schemas API keys con validazione formato OpenRouter
|
||||
- [ ] T24: POST /api/keys con cifratura e limite keys
|
||||
- [ ] T25: GET /api/keys con paginazione e filtri
|
||||
- [ ] T26: PUT /api/keys/{id} aggiornamento
|
||||
- [ ] T27: DELETE /api/keys/{id} eliminazione
|
||||
- [ ] T28: Validazione key con OpenRouter API
|
||||
- [ ] T29: Test completi coverage >= 90%
|
||||
- [ ] Tutti i test passano: `pytest tests/unit/ -v`
|
||||
- [ ] API keys cifrate nel database (mai plaintext)
|
||||
- [ ] Utenti vedono/modificano solo proprie keys
|
||||
- [ ] 7 commit atomici con conventional commits
|
||||
- [ ] progress.md aggiornato
|
||||
|
||||
---
|
||||
|
||||
## 📝 COMMIT MESSAGES
|
||||
|
||||
```
|
||||
feat(schemas): T23 add Pydantic API key schemas
|
||||
|
||||
feat(api-keys): T24 implement create API key endpoint with encryption
|
||||
|
||||
feat(api-keys): T25 implement list API keys endpoint with pagination
|
||||
|
||||
feat(api-keys): T26 implement update API key endpoint
|
||||
|
||||
feat(api-keys): T27 implement delete API key endpoint
|
||||
|
||||
feat(openrouter): T28 implement API key validation service
|
||||
|
||||
test(api-keys): T29 add comprehensive API keys endpoint tests
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 🚀 VERIFICA FINALE
|
||||
|
||||
```bash
|
||||
cd /home/google/Sources/LucaSacchiNet/openrouter-watcher
|
||||
|
||||
# Test schemas
|
||||
pytest tests/unit/schemas/test_api_key_schemas.py -v
|
||||
|
||||
# Test routers
|
||||
pytest tests/unit/routers/test_api_keys.py -v --cov=src/openrouter_monitor/routers
|
||||
|
||||
# Test services
|
||||
pytest tests/unit/services/test_openrouter.py -v
|
||||
|
||||
# Test completo
|
||||
pytest tests/unit/ -v --cov=src/openrouter_monitor
|
||||
|
||||
# Verifica coverage >= 90%
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 🔒 CONSIDERAZIONI SICUREZZA
|
||||
|
||||
### Do's ✅
|
||||
- Cifrare sempre API keys prima di salvare nel DB
|
||||
- Verificare ownership (user_id) per ogni operazione
|
||||
- Validare formato key OpenRouter prima di salvare
|
||||
- Usare transactions per operazioni DB
|
||||
- Loggare operazioni (non i dati sensibili)
|
||||
|
||||
### Don'ts ❌
|
||||
- MAI salvare API key in plaintext
|
||||
- MAI loggare API key complete
|
||||
- MAI permettere a utente di vedere key di altri
|
||||
- MAI ritornare key cifrate nelle response
|
||||
- MAI ignorare errori di decrittazione
|
||||
|
||||
---
|
||||
|
||||
## ⚠️ NOTE IMPORTANTI
|
||||
|
||||
- **Path assoluti**: Usa sempre `/home/google/Sources/LucaSacchiNet/openrouter-watcher/`
|
||||
- **EncryptionService**: Riutilizza da `services/encryption.py`
|
||||
- **Formato Key**: OpenRouter keys iniziano con "sk-or-v1-"
|
||||
- **Limite Keys**: Configurabile via `MAX_API_KEYS_PER_USER` (default 10)
|
||||
- **Cascade Delete**: Eliminando ApiKey si eliminano anche UsageStats
|
||||
- **Ordinamento**: Lista keys ordinata per created_at DESC
|
||||
|
||||
---
|
||||
|
||||
**AGENTE:** @tdd-developer
|
||||
|
||||
**INIZIA CON:** T23 - Pydantic API key schemas
|
||||
|
||||
**QUANDO FINITO:** Conferma completamento, coverage >= 90%, aggiorna progress.md
|
||||
@@ -1,4 +1,10 @@
|
||||
"""Schemas package for OpenRouter Monitor."""
|
||||
from openrouter_monitor.schemas.api_key import (
|
||||
ApiKeyCreate,
|
||||
ApiKeyListResponse,
|
||||
ApiKeyResponse,
|
||||
ApiKeyUpdate,
|
||||
)
|
||||
from openrouter_monitor.schemas.auth import (
|
||||
TokenData,
|
||||
TokenResponse,
|
||||
@@ -13,4 +19,8 @@ __all__ = [
|
||||
"UserResponse",
|
||||
"TokenResponse",
|
||||
"TokenData",
|
||||
"ApiKeyCreate",
|
||||
"ApiKeyUpdate",
|
||||
"ApiKeyResponse",
|
||||
"ApiKeyListResponse",
|
||||
]
|
||||
|
||||
138
src/openrouter_monitor/schemas/api_key.py
Normal file
138
src/openrouter_monitor/schemas/api_key.py
Normal file
@@ -0,0 +1,138 @@
|
||||
"""API Key Pydantic schemas.
|
||||
|
||||
T23: Pydantic schemas for API key management.
|
||||
"""
|
||||
from datetime import datetime
|
||||
from typing import List, Optional
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field, field_validator
|
||||
|
||||
|
||||
class ApiKeyCreate(BaseModel):
|
||||
"""Schema for creating a new API key.
|
||||
|
||||
Attributes:
|
||||
name: Human-readable name for the key (1-100 chars)
|
||||
key: OpenRouter API key (must start with 'sk-or-v1-')
|
||||
"""
|
||||
|
||||
name: str = Field(
|
||||
...,
|
||||
min_length=1,
|
||||
max_length=100,
|
||||
description="Human-readable name for the API key",
|
||||
examples=["Production Key"]
|
||||
)
|
||||
key: str = Field(
|
||||
...,
|
||||
description="OpenRouter API key",
|
||||
examples=["sk-or-v1-abc123..."]
|
||||
)
|
||||
|
||||
@field_validator('key')
|
||||
@classmethod
|
||||
def validate_key_format(cls, v: str) -> str:
|
||||
"""Validate OpenRouter API key format.
|
||||
|
||||
Args:
|
||||
v: The API key value to validate
|
||||
|
||||
Returns:
|
||||
The API key if valid
|
||||
|
||||
Raises:
|
||||
ValueError: If key doesn't start with 'sk-or-v1-'
|
||||
"""
|
||||
if not v or not v.strip():
|
||||
raise ValueError("API key cannot be empty")
|
||||
|
||||
if not v.startswith('sk-or-v1-'):
|
||||
raise ValueError("API key must start with 'sk-or-v1-'")
|
||||
|
||||
return v
|
||||
|
||||
|
||||
class ApiKeyUpdate(BaseModel):
|
||||
"""Schema for updating an existing API key.
|
||||
|
||||
All fields are optional - only provided fields will be updated.
|
||||
|
||||
Attributes:
|
||||
name: New name for the key (optional, 1-100 chars)
|
||||
is_active: Whether the key should be active (optional)
|
||||
"""
|
||||
|
||||
name: Optional[str] = Field(
|
||||
default=None,
|
||||
min_length=1,
|
||||
max_length=100,
|
||||
description="New name for the API key",
|
||||
examples=["Updated Key Name"]
|
||||
)
|
||||
is_active: Optional[bool] = Field(
|
||||
default=None,
|
||||
description="Whether the key should be active",
|
||||
examples=[True, False]
|
||||
)
|
||||
|
||||
|
||||
class ApiKeyResponse(BaseModel):
|
||||
"""Schema for API key response (returned to client).
|
||||
|
||||
Note: The actual API key value is NEVER included in responses
|
||||
for security reasons.
|
||||
|
||||
Attributes:
|
||||
id: API key ID
|
||||
name: API key name
|
||||
is_active: Whether the key is active
|
||||
created_at: When the key was created
|
||||
last_used_at: When the key was last used (None if never used)
|
||||
"""
|
||||
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
id: int = Field(
|
||||
...,
|
||||
description="API key ID",
|
||||
examples=[1]
|
||||
)
|
||||
name: str = Field(
|
||||
...,
|
||||
description="API key name",
|
||||
examples=["Production Key"]
|
||||
)
|
||||
is_active: bool = Field(
|
||||
...,
|
||||
description="Whether the key is active",
|
||||
examples=[True]
|
||||
)
|
||||
created_at: datetime = Field(
|
||||
...,
|
||||
description="When the key was created",
|
||||
examples=["2024-01-01T12:00:00"]
|
||||
)
|
||||
last_used_at: Optional[datetime] = Field(
|
||||
default=None,
|
||||
description="When the key was last used",
|
||||
examples=["2024-01-02T15:30:00"]
|
||||
)
|
||||
|
||||
|
||||
class ApiKeyListResponse(BaseModel):
|
||||
"""Schema for paginated list of API keys.
|
||||
|
||||
Attributes:
|
||||
items: List of API key responses
|
||||
total: Total number of keys (for pagination)
|
||||
"""
|
||||
|
||||
items: List[ApiKeyResponse] = Field(
|
||||
...,
|
||||
description="List of API keys"
|
||||
)
|
||||
total: int = Field(
|
||||
...,
|
||||
description="Total number of API keys",
|
||||
examples=[10]
|
||||
)
|
||||
304
tests/unit/schemas/test_api_key_schemas.py
Normal file
304
tests/unit/schemas/test_api_key_schemas.py
Normal file
@@ -0,0 +1,304 @@
|
||||
"""Tests for API Key Pydantic schemas.
|
||||
|
||||
T23: Test Pydantic schemas for API key management.
|
||||
"""
|
||||
import pytest
|
||||
from datetime import datetime, timezone
|
||||
from pydantic import ValidationError
|
||||
|
||||
|
||||
class TestApiKeyCreate:
|
||||
"""Tests for ApiKeyCreate schema."""
|
||||
|
||||
def test_valid_api_key_create(self):
|
||||
"""Test valid API key creation with OpenRouter format."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyCreate
|
||||
|
||||
data = ApiKeyCreate(
|
||||
name="My Production Key",
|
||||
key="sk-or-v1-abc123def456ghi789jkl012mno345pqr678stu901vwx234yz"
|
||||
)
|
||||
|
||||
assert data.name == "My Production Key"
|
||||
assert data.key == "sk-or-v1-abc123def456ghi789jkl012mno345pqr678stu901vwx234yz"
|
||||
|
||||
def test_name_min_length(self):
|
||||
"""Test that name must be at least 1 character."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyCreate
|
||||
|
||||
with pytest.raises(ValidationError, match="name"):
|
||||
ApiKeyCreate(
|
||||
name="",
|
||||
key="sk-or-v1-abc123"
|
||||
)
|
||||
|
||||
def test_name_max_length(self):
|
||||
"""Test that name cannot exceed 100 characters."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyCreate
|
||||
|
||||
with pytest.raises(ValidationError, match="name"):
|
||||
ApiKeyCreate(
|
||||
name="x" * 101,
|
||||
key="sk-or-v1-abc123"
|
||||
)
|
||||
|
||||
def test_name_exactly_max_length(self):
|
||||
"""Test that name can be exactly 100 characters."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyCreate
|
||||
|
||||
name = "x" * 100
|
||||
data = ApiKeyCreate(
|
||||
name=name,
|
||||
key="sk-or-v1-abc123"
|
||||
)
|
||||
|
||||
assert data.name == name
|
||||
assert len(data.name) == 100
|
||||
|
||||
def test_valid_openrouter_key_format(self):
|
||||
"""Test valid OpenRouter API key format (sk-or-v1- prefix)."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyCreate
|
||||
|
||||
# Various valid OpenRouter key formats
|
||||
valid_keys = [
|
||||
"sk-or-v1-abc123",
|
||||
"sk-or-v1-abc123def456",
|
||||
"sk-or-v1-" + "x" * 100,
|
||||
]
|
||||
|
||||
for key in valid_keys:
|
||||
data = ApiKeyCreate(name="Test", key=key)
|
||||
assert data.key == key
|
||||
|
||||
def test_invalid_key_format_missing_prefix(self):
|
||||
"""Test that key without OpenRouter prefix raises ValidationError."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyCreate
|
||||
|
||||
with pytest.raises(ValidationError, match="key"):
|
||||
ApiKeyCreate(
|
||||
name="Test Key",
|
||||
key="invalid-key-format"
|
||||
)
|
||||
|
||||
def test_invalid_key_format_wrong_prefix(self):
|
||||
"""Test that key with wrong prefix raises ValidationError."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyCreate
|
||||
|
||||
with pytest.raises(ValidationError, match="key"):
|
||||
ApiKeyCreate(
|
||||
name="Test Key",
|
||||
key="sk-abc123" # Missing -or-v1-
|
||||
)
|
||||
|
||||
def test_empty_key(self):
|
||||
"""Test that empty key raises ValidationError."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyCreate
|
||||
|
||||
with pytest.raises(ValidationError, match="key"):
|
||||
ApiKeyCreate(
|
||||
name="Test Key",
|
||||
key=""
|
||||
)
|
||||
|
||||
def test_whitespace_only_key(self):
|
||||
"""Test that whitespace-only key raises ValidationError."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyCreate
|
||||
|
||||
with pytest.raises(ValidationError, match="key"):
|
||||
ApiKeyCreate(
|
||||
name="Test Key",
|
||||
key=" "
|
||||
)
|
||||
|
||||
|
||||
class TestApiKeyUpdate:
|
||||
"""Tests for ApiKeyUpdate schema."""
|
||||
|
||||
def test_valid_update_name_only(self):
|
||||
"""Test valid update with name only."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyUpdate
|
||||
|
||||
data = ApiKeyUpdate(name="Updated Name")
|
||||
|
||||
assert data.name == "Updated Name"
|
||||
assert data.is_active is None
|
||||
|
||||
def test_valid_update_is_active_only(self):
|
||||
"""Test valid update with is_active only."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyUpdate
|
||||
|
||||
data = ApiKeyUpdate(is_active=False)
|
||||
|
||||
assert data.name is None
|
||||
assert data.is_active is False
|
||||
|
||||
def test_valid_update_both_fields(self):
|
||||
"""Test valid update with both fields."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyUpdate
|
||||
|
||||
data = ApiKeyUpdate(name="New Name", is_active=True)
|
||||
|
||||
assert data.name == "New Name"
|
||||
assert data.is_active is True
|
||||
|
||||
def test_empty_update_allowed(self):
|
||||
"""Test that empty update is allowed (no fields provided)."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyUpdate
|
||||
|
||||
data = ApiKeyUpdate()
|
||||
|
||||
assert data.name is None
|
||||
assert data.is_active is None
|
||||
|
||||
def test_update_name_too_long(self):
|
||||
"""Test that name longer than 100 chars raises ValidationError."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyUpdate
|
||||
|
||||
with pytest.raises(ValidationError, match="name"):
|
||||
ApiKeyUpdate(name="x" * 101)
|
||||
|
||||
def test_update_name_min_length(self):
|
||||
"""Test that empty name raises ValidationError."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyUpdate
|
||||
|
||||
with pytest.raises(ValidationError, match="name"):
|
||||
ApiKeyUpdate(name="")
|
||||
|
||||
def test_update_name_valid_length(self):
|
||||
"""Test that valid name length is accepted."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyUpdate
|
||||
|
||||
data = ApiKeyUpdate(name="Valid Name")
|
||||
assert data.name == "Valid Name"
|
||||
|
||||
|
||||
class TestApiKeyResponse:
|
||||
"""Tests for ApiKeyResponse schema."""
|
||||
|
||||
def test_valid_response(self):
|
||||
"""Test valid API key response."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyResponse
|
||||
|
||||
created_at = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||
last_used_at = datetime(2024, 1, 2, 15, 30, 0, tzinfo=timezone.utc)
|
||||
|
||||
data = ApiKeyResponse(
|
||||
id=1,
|
||||
name="Production Key",
|
||||
is_active=True,
|
||||
created_at=created_at,
|
||||
last_used_at=last_used_at
|
||||
)
|
||||
|
||||
assert data.id == 1
|
||||
assert data.name == "Production Key"
|
||||
assert data.is_active is True
|
||||
assert data.created_at == created_at
|
||||
assert data.last_used_at == last_used_at
|
||||
|
||||
def test_response_optional_last_used_at(self):
|
||||
"""Test that last_used_at is optional (key never used)."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyResponse
|
||||
|
||||
data = ApiKeyResponse(
|
||||
id=1,
|
||||
name="New Key",
|
||||
is_active=True,
|
||||
created_at=datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc),
|
||||
last_used_at=None
|
||||
)
|
||||
|
||||
assert data.last_used_at is None
|
||||
|
||||
def test_response_from_orm(self):
|
||||
"""Test that ApiKeyResponse can be created from ORM model."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyResponse
|
||||
|
||||
# Mock ORM object
|
||||
class MockApiKey:
|
||||
id = 1
|
||||
name = "Test Key"
|
||||
is_active = True
|
||||
created_at = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||
last_used_at = None
|
||||
|
||||
key = ApiKeyResponse.model_validate(MockApiKey())
|
||||
|
||||
assert key.id == 1
|
||||
assert key.name == "Test Key"
|
||||
assert key.is_active is True
|
||||
|
||||
def test_response_no_key_field(self):
|
||||
"""Test that API key value is NOT included in response."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyResponse
|
||||
|
||||
# Verify that 'key' field doesn't exist in the model
|
||||
fields = ApiKeyResponse.model_fields.keys()
|
||||
assert 'key' not in fields
|
||||
assert 'key_encrypted' not in fields
|
||||
|
||||
|
||||
class TestApiKeyListResponse:
|
||||
"""Tests for ApiKeyListResponse schema."""
|
||||
|
||||
def test_valid_list_response(self):
|
||||
"""Test valid list response with multiple keys."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyListResponse, ApiKeyResponse
|
||||
|
||||
created_at = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||
|
||||
items = [
|
||||
ApiKeyResponse(
|
||||
id=1,
|
||||
name="Key 1",
|
||||
is_active=True,
|
||||
created_at=created_at,
|
||||
last_used_at=None
|
||||
),
|
||||
ApiKeyResponse(
|
||||
id=2,
|
||||
name="Key 2",
|
||||
is_active=False,
|
||||
created_at=created_at,
|
||||
last_used_at=created_at
|
||||
)
|
||||
]
|
||||
|
||||
data = ApiKeyListResponse(items=items, total=2)
|
||||
|
||||
assert len(data.items) == 2
|
||||
assert data.total == 2
|
||||
assert data.items[0].name == "Key 1"
|
||||
assert data.items[1].name == "Key 2"
|
||||
|
||||
def test_empty_list_response(self):
|
||||
"""Test valid list response with no keys."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyListResponse
|
||||
|
||||
data = ApiKeyListResponse(items=[], total=0)
|
||||
|
||||
assert data.items == []
|
||||
assert data.total == 0
|
||||
|
||||
def test_pagination_response(self):
|
||||
"""Test list response simulating pagination."""
|
||||
from openrouter_monitor.schemas.api_key import ApiKeyListResponse, ApiKeyResponse
|
||||
|
||||
created_at = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||
|
||||
# Simulate page 1 of 2, 10 items per page
|
||||
items = [
|
||||
ApiKeyResponse(
|
||||
id=i,
|
||||
name=f"Key {i}",
|
||||
is_active=True,
|
||||
created_at=created_at,
|
||||
last_used_at=None
|
||||
)
|
||||
for i in range(1, 11)
|
||||
]
|
||||
|
||||
data = ApiKeyListResponse(items=items, total=25) # 25 total, showing first 10
|
||||
|
||||
assert len(data.items) == 10
|
||||
assert data.total == 25
|
||||
Reference in New Issue
Block a user