45 lines
1.5 KiB
Python
45 lines
1.5 KiB
Python
import sys
|
|
import os
|
|
sys.path.append('/app')
|
|
sys.path.append('/home/josmar/dev/anexo76/backend')
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
import asyncio
|
|
import logging
|
|
|
|
# Disable logging to keep output clean
|
|
logging.basicConfig(level=logging.ERROR)
|
|
|
|
from api.v1.modules.a76.general_catalogs.fractions.tariff_fractions.service import TariffFractionService
|
|
from api.v1.modules.a76.general_catalogs.fractions.tariff_fractions.dto import TariffFractionResponseDTO
|
|
|
|
async def test():
|
|
# Use the database URL from the environment or default
|
|
db_url = "postgresql://postgres:postgres@anexo76-postgres-a76:5432/anexo76_core"
|
|
engine = create_engine(db_url)
|
|
Session = sessionmaker(bind=engine)
|
|
db = Session()
|
|
|
|
try:
|
|
print("Starting test for catalog='american'...")
|
|
items, total = await TariffFractionService.get_all(
|
|
db, skip=0, limit=10, filters=None, catalog="american", tenant_id=1, company_id=1
|
|
)
|
|
print(f"Service Success! Total: {total}")
|
|
|
|
print("Validating items with TariffFractionResponseDTO...")
|
|
for item in items:
|
|
dto = TariffFractionResponseDTO.model_validate(item)
|
|
print(f"DTO: ID={dto.id}, Code={dto.code}, Fraction={dto.fraction}")
|
|
|
|
except Exception as e:
|
|
print(f"Error caught: {type(e).__name__}: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
finally:
|
|
db.close()
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(test())
|