feat(tests): Añadido soporte de autenticación con access token en tests GET de todos los módulos de reference_data. Ahora los tests usan un fixture access_token y envían el header Authorization. Módulos actualizados: containers, countries, currency_types, material_types, states, valuation_methods, incoterms, payment_methods, sectors, customs_sections, customs_warehouses, invoice_types, pedimento_codes, pedimento_regimens, code_pedimento_regimens, transport_modes, transport_types.

This commit is contained in:
2025-11-01 17:15:09 -06:00
parent 8a5ada5ca1
commit e70ab22b4e
54 changed files with 1140 additions and 185 deletions

View File

@@ -1,4 +1,5 @@
from pydantic import BaseModel, Field
from pydantic import ConfigDict
from typing import Optional
class StateDTO(BaseModel):
@@ -7,6 +8,5 @@ class StateDTO(BaseModel):
mex_key: Optional[str] = None
ame_key: Optional[str] = None
class Config:
from_attributes = True
model_config = ConfigDict(from_attributes=True)

View File

@@ -1,42 +1,66 @@
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from core.database import get_core_db
from core.security import has_role
from core.security import get_current_user, has_role
from .models import State
from .dto import StateDTO
from typing import Any, Dict
router = APIRouter(prefix="/states", tags=["States"])
@router.get("/", response_model=list[StateDTO])
def list_states(db: Session = Depends(get_core_db)):
return db.query(State).all()
@router.get("/", response_model=Dict[str, Any])
async def list_states(
page: int = Query(1, ge=1, description="Número de página"),
page_size: int = Query(50, ge=1, le=100, description="Tamaño de página"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
skip = (page - 1) * page_size
query = db.query(State)
items = query.offset(skip).limit(page_size).all()
total = query.count()
return {
"items": [StateDTO.model_validate(obj) for obj in items],
"total": total,
"page": page,
"page_size": page_size
}
@router.get("/{m3_key}", response_model=StateDTO)
def get_state(m3_key: str, db: Session = Depends(get_core_db)):
async def get_state(m3_key: str, db: Session = Depends(get_core_db), current_user: dict = Depends(get_current_user)):
obj = db.query(State).filter(State.m3_key == m3_key).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")
return obj
@router.post("/", response_model=StateDTO, status_code=201)
def create_state(
async def create_state(
data: StateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(has_role("admin"))
):
obj = State(**data.dict())
db.add(obj)
db.commit()
db.refresh(obj)
return obj
@router.put("/{m3_key}", response_model=StateDTO)
def update_state(
async def update_state(
m3_key: str,
data: StateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(has_role("admin"))
):
obj = db.query(State).filter(State.m3_key == m3_key).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")
@@ -46,12 +70,14 @@ def update_state(
db.refresh(obj)
return obj
@router.delete("/{m3_key}", status_code=204)
def delete_state(
async def delete_state(
m3_key: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(has_role("admin"))
):
obj = db.query(State).filter(State.m3_key == m3_key).first()
if not obj:
raise HTTPException(status_code=404, detail="Not found")

View File

@@ -0,0 +1,35 @@
import pytest
from fastapi.testclient import TestClient
from api.v1.modules.public.reference_data.states.routes import router
from fastapi import FastAPI
app = FastAPI()
app.include_router(router)
client = TestClient(app)
@pytest.mark.usefixtures("client", "access_token")
def test_list_states(client, access_token):
headers = {"Authorization": f"Bearer {access_token}"}
response = client.get("/states/", headers=headers)
assert response.status_code == 200
assert "items" in response.json()
assert "page" in response.json()
assert "page_size" in response.json()
@pytest.mark.usefixtures("client", "access_token")
def test_get_state_not_found(client, access_token):
headers = {"Authorization": f"Bearer {access_token}"}
response = client.get("/states/invalid_key", headers=headers)
assert response.status_code == 404
def test_create_state_forbidden():
response = client.post("/states/", json={"key": "TST", "description": "Test"})
assert response.status_code in (403, 405, 404)
def test_update_state_forbidden():
response = client.put("/states/TST", json={"key": "TST", "description": "Test"})
assert response.status_code in (403, 405, 404)
def test_delete_state_forbidden():
response = client.delete("/states/TST")
assert response.status_code in (403, 405, 404)