47 Commits

Author SHA1 Message Date
128ba215dd #51 feat: registrar categorias en admin y documentar endpoints
Issue #51: discriminacion y administracion de categorias de productos.

- ProductCategoryAdmin con list_display/filtro por active
- doc/requests.org: GET/PATCH product_categories, import_from_external,
  deactivate, activate
2026-10-10 16:47:12 -05:00
436f5300fa #51 feat: categorias anidadas en el listado legacy de productos
Issue #51: discriminacion y administracion de categorias de productos.

- Product.to_list() devuelve categorias como objetos
  (id, name, alternative_name, active) en vez de solo nombres
2026-10-10 16:45:32 -05:00
3b53703c4d #51 feat: API de categorias de productos
Issue #51: discriminacion y administracion de categorias de productos.

- ProductCategoryView en /api/product_categories (list/retrieve IsNotPublico,
  mutaciones admin): filtro ?active, PATCH de description/alternative_name/active
- POST deactivate/activate con body {ids} -> {deactivated|activated, failed}
  (ignora ids inexistentes)
- POST import_from_external admin via CategoryTrytonService
- create/destroy deshabilitados (la fuente de categorias es Tryton)
2026-10-10 16:43:37 -05:00
74cd9738b4 #51 feat: servicio de importacion externa de categorias de productos
Issue #51: discriminacion y administracion de categorias de productos.

- CategoryTrytonService.import_from_external: lee categorias asociadas a
  productos salables, hace upsert por external_id (fallback nombre),
  preserva description/alternative_name/active, reporta
  checked/created/updated/untouched/failed
- upsert_category compartido y reutilizado por ProductTrytonService
2026-10-10 16:38:47 -05:00
73fff016a3 #51 feat: serializadores de categorias y categorias anidadas en productos
Issue #51: discriminacion y administracion de categorias de productos.

- ProductCategorySerializer (name/external_id read_only)
- ProductCategoryBriefSerializer para el JSON de productos
- ProductSerializer.categories ahora devuelve objetos anidados
2026-10-10 16:35:43 -05:00
c0dcb7901a #51 fix: detectar cambios de categorias al sincronizar productos
Issue #51: discriminacion y administracion de categorias de productos.

- _need_update compara las categorias (external_id) local vs Tryton
- _describe_changes reporta categorias agregadas (+Nombre) y removidas (-Nombre)
- Actualiza test que asumia asignacion de categorias sin marcar update
2026-10-10 16:33:18 -05:00
570fec0532 #51 feat: agregar descripcion, nombre alternativo y activo a categorias de productos
Issue #51: discriminacion y administracion de categorias de productos.

- ProductCategory.description (TextField blank)
- ProductCategory.alternative_name (CharField 100 nullable)
- ProductCategory.active (Boolean default True)
- Migracion 0056_productcategory_active_and_more
- Tests de modelo (campo don confiao y active por defecto)
2026-10-10 16:27:49 -05:00
53cc333c19 #58 feat: importar y vincular categorías de Tryton al sincronizar productos
- Campo external_id en ProductCategory para vincular con Tryton (migración 0055)
- Import/upsert de categorías usadas por los productos sincronizados
- Asignación M2M en productos creados, actualizados y sin cambios
- Respuesta del import incluye created_categories y updated_categories con detalle
- Fix: leer las categorías desde la clave agrupada "template." que devuelve el RPC de Tryton
2026-08-22 16:40:40 -05:00
0fd79a2ba7 Merge pull request 'feat/49-provenance Implementa trazabilidad de proveedores #49' (#50) from feat/49-provenance into main
Reviewed-on: #50
2026-08-16 22:17:35 -05:00
b5d057bd1a Merge branch 'main' into feat/49-provenance 2026-08-16 22:15:11 -05:00
792ac4367c chore(#49): track seed script and geography CSV in git
Allowlist scripts/ in .gitignore (was caught by [Ss]cripts venv rule)
so seed_geography.py and colombia_municipios.csv can be committed.
This simplifies deployment since the files no longer need to be
copied separately to the server.
2026-08-16 21:58:12 -05:00
3a0c76ce69 feat(#49): add geolocation (latitude/longitude) to municipalities
- Municipality model with optional latitude/longitude decimal fields
- Municipality and brief serializers expose lat/lng (also in product_provenance of public/authenticated summaries for the Leaflet map)
- seed_geography endpoint reads lat/lng and updates existing municipalities on reseed (idempotent)
- TDD tests for model, serializers, summaries and seed
2026-08-16 04:10:45 -05:00
699983931e feat(#49): return FK details in department and municipality serializers
- DepartmentSerializer adds country_detail (nested country object)
- MunicipalitySerializer adds department_detail and country_detail
- Writable FK id fields kept for create/update
- Move geography brief serializers to serializers/geography.py
2026-08-16 03:23:56 -05:00
29d154e140 feat(#49): add product provenance (suppliers, organizations, geography)
- Country, Department, Municipality models (municipality unique per department)
- Organization and Supplier models, Supplier links to organization and municipality
- Product.suppliers M2M to link products to one or more suppliers
- Authenticated CRUD APIs for organizations, suppliers, countries, departments and municipalities (write only for administrators)
- product_provenance domain data embedded in sale/catalog summaries (public and authenticated)
- seed_geography endpoint (admin, idempotent) and scripts/seed_geography.py using pycountry + DANE municipalities CSV
- TDD tests for models, API permissions/CRUD and summary provenance data
2026-08-15 16:32:20 -05:00
ce556918f2 feat: detalle de fallos y resultados enriquecidos en sincronización Tryton
- failed_products/failed_parties ahora incluyen {external_id, name, error} con causa legible (precio nulo, nombre duplicado, etc.)
- Los fallos de actualización de productos y clientes ya no abortan la sincronización
- created/updated/untouched/checked como objetos {id, name}; updated incluye detail con los cambios (Nombre/Precio/Unidad/Dirección)
- Ventas y catálogo: successful {id, code} y failed {id, code, error} (antes solo IDs, sin detalle)
2026-08-15 16:05:10 -05:00
b088794716 docs: agregar workflow TDD rojo/verde con confirmacion de fallos de tests 2026-08-15 14:56:09 -05:00
965332e0b4 Merge pull request 'feat: incluir code en SaleSerializer y CatalogSaleSerializer' (#48) from feat/serializers-con-code into main
Reviewed-on: #48
2026-08-10 01:13:21 -05:00
177b4c01ca feat: incluir code en SaleSerializer y CatalogSaleSerializer 2026-08-10 00:32:59 -05:00
d7132733fc Update Admin list view 2026-08-09 20:54:16 -05:00
ea3727363b Merge pull request 'feat/46-order-summary-public' (#47) from feat/46-order-summary-public into main
Reviewed-on: #47
2026-08-08 14:25:25 -05:00
bee35ac5cb #51 feat: add editable store logo to StoreSettings 2026-08-08 14:20:07 -05:00
85168e24f0 Merge branch 'main' of ssh://gitea.onecluster.org:6666/OneTeam/don_confiao_backend 2026-08-08 13:57:56 -05:00
56bbc8f9b4 #50 feat: add StoreSettings model and API for store address and map coordinates 2026-08-08 13:50:43 -05:00
mono
805565aa8e feat(#46): show order code in admin 2026-08-08 13:29:14 -05:00
mono
4f41156056 feat(#46): add fill_codes client script 2026-08-08 13:29:14 -05:00
mono
2438cf2576 feat(#46): add public order summary and fill_codes endpoints 2026-08-08 13:28:53 -05:00
mono
332884907c feat(#46): add code to Sale and CatalogSale models 2026-08-08 13:28:49 -05:00
mono
b2312edff5 test(#46): add order summary public and fill_codes tests 2026-08-08 13:28:44 -05:00
4410cc919a Merge pull request 'test(Tryton): add catalog sale test and validate build_tryton_comment' (#44) from fix/EnviarComentarioConDetalleDelPedido into main
Reviewed-on: #44
2026-07-11 18:56:05 -05:00
d1c2b4f1bb test(Tryton): add catalog sale test and validate build_tryton_comment
- Added CatalogSale/CatalogSaleLine to setUp
- Added explicit assertion linking TrytonSale.build_tryton_comment()
  to sale.description
- Rewrote test_send_catalog_sales_to_tryton with proper CatalogSale
  data, validating comment format, description, reference and lines
- Added direct assertion of TrytonCatalogSale.build_tryton_comment()
2026-07-11 14:55:44 -05:00
d57b827e4e Merge pull request 'fix(Tryton): restore send sales test and fix TrytonSale.build_tryton_comment' (#43) from fix/EnviarComentarioConDetalleDelPedido into main
Reviewed-on: #43
2026-07-11 14:38:27 -05:00
1e32cfce7f fix(Tryton): restore send sales test and fix TrytonSale.build_tryton_comment
- Restored test_exportar_ventas_para_tryton.py with full parameter
  validation for client.call (method, context, sale fields, lines)
- Fixed TrytonSale.build_tryton_comment referencing self.catalog_sale
  (nonexistent) instead of self.sale.description
2026-07-11 14:32:41 -05:00
f65b85c7e7 feat: Add custom views to admin registered models 2026-07-05 11:08:33 -05:00
ace5091f8b feat: Update admin product view list. 2026-07-03 19:23:55 -05:00
de7c32b10f Merge pull request 'feat: add script to re-upload catalogue images and force resize' (#42) from feat/reupload-catalogue-images into main
Reviewed-on: #42
2026-07-03 18:38:31 -05:00
mono
b93ee95427 feat: add script to re-upload catalogue images and force resize 2026-07-03 18:26:42 -05:00
4fcc7c2503 fix: CatalogSaleSummary accesible para usuarios 'publico'
- Cambia permission_classes de IsNotPublico a IsAuthenticated
- El usuario publico necesita ver el resumen de su compra en catálogo
2026-06-22 12:49:04 -05:00
5b373bd15f feat: agrega rol 'publico' con UserProfile y permisos diferenciados
- Crea modelo UserProfile (OneToOne con User) con user_type: publico/user/administrator
- Señal post_save que crea UserProfile automáticamente al crear User
- UserSerializer.get_role() lee desde profile.user_type
- Nuevos permisos: IsNotPublico, IsPublico y actualiza IsAdministrator
- IsNotPublico restringe acceso a vistas de gestión (productos, clientes, ventas)
- CatalogSaleView.create permite publico; otras acciones requieren IsNotPublico
- UserProfile inline en admin de Usuarios
- Data migration para backfill de usuarios existentes
2026-06-22 12:27:23 -05:00
27b68a9ac2 chore: configuración de infraestructura y dependencias
- docker-compose.staging.yml: env vars a env_file, limpia defaults duplicados
- docker-compose.prod.yml: descomenta y activa servicio nginx
- staging.py: DEBUG=True, POSTGRES_* env vars, logging DEBUG
- pyproject.toml + poetry.lock: agrega pillow y whitenoise
2026-06-22 11:35:47 -05:00
d6b7fe428f feat: Update dependencies. 2026-06-22 11:28:06 -05:00
d9137404a1 Merge branch 'main' of ssh://gitea.onecluster.org:6666/OneTeam/don_confiao_backend 2026-06-22 11:20:21 -05:00
3b635cf17a feat: catálogo público sin autenticación, solo confirmar compra requiere auth
- Cambia default global de IsAuthenticated a AllowAny en settings/base.py
- ProductView: AllowAny para GET, IsAuthenticated para POST/PUT/DELETE
- CatalogueImageViewSet: AllowAny para reads, IsAuthenticated+Admin para writes
- PaymentMethodView: AllowAny (público)
- SaleView, CatalogSaleView, SaleSummary, CatalogSaleSummary: IsAuthenticated
- CustomerView: IsAuthenticated
2026-06-22 11:18:08 -05:00
d406c734ac feat: Nginx configurated to server images for catalog in production. 2026-06-22 10:20:21 -05:00
193198918b Merge pull request 'feat: add standalone script to upload catalogue images via API' (#41) from feature/upload-catalogue-script into main
Reviewed-on: #41
2026-06-14 00:46:45 -05:00
mono
1007584d3e Merge branch 'main' into feature/upload-catalogue-script 2026-06-14 00:46:21 -05:00
mono
ac22adb558 feat: add standalone script to upload catalogue images via API 2026-06-14 00:39:33 -05:00
2415ed3564 Merge pull request 'feat: add catalogue image management (don_confiao_catalog_generator/issues/1)' (#40) from feature/1-catalogue-images into main
Reviewed-on: #40
2026-06-13 20:31:40 -05:00
73 changed files with 6724 additions and 114 deletions

2
.gitignore vendored
View File

@@ -321,7 +321,7 @@ pyrightconfig.json
[Ll]ib
[Ll]ib64
[Ll]ocal
[Ss]cripts
Scripts/
pyvenv.cfg
pip-selfcheck.json

View File

@@ -159,7 +159,44 @@ Nota: El volumen monta `tienda_ilusion/` en `/app/`, por lo que el path correcto
## Tests
- Framework: Django unittest
- Directorio: don_confiao/tests/
- Ejecutar: `docker-compose -f docker-compose.dev.yml run --rm django python manage.py test`
- Ejecutar: `docker compose -f docker-compose.dev.yml run --rm django python manage.py test`
## Workflow TDD (Semáforo Rojo/Verde)
Todo cambio o nueva funcionalidad debe seguir TDD. **Está prohibido asumir fallos de tests: todo fallo debe ser confirmado ejecutando los tests.**
### Regla de oro: confirmar antes de modificar
Antes de modificar tests (por cambios en comportamiento o por nuevas funcionalidades), **SIEMPRE ejecutar los tests primero** y confirmar que el fallo esperado realmente ocurre. Nunca asumir que un test falla: verificar el fallo con la ejecución real.
### Ciclo Rojo → Verde → Refactor
1. **ROJO (escribir el test y confirmar que falla)**:
- Escribir o modificar el test que describe el comportamiento esperado.
- Ejecutar los tests **antes de tocar el código de producción**.
- Confirmar que el test falla por el motivo esperado (assertion fallida por el comportamiento nuevo, no por error de setup o de importación).
- Copiar el output del fallo como evidencia del semáforo rojo.
2. **VERDE (implementar lo mínimo para que pase)**:
- Implementar el código de producción mínimo necesario para que el test pase.
- Volver a ejecutar los tests y confirmar que el semáforo queda en verde.
3. **REFACTOR (limpiar con los tests en verde)**:
- Refactorizar y limpiar el código sin cambiar comportamiento.
- Re-ejecutar los tests para confirmar que siguen en verde.
### Ejecutar tests
Siempre ejecutar dentro del contenedor de desarrollo:
```bash
# Test específico (durante el ciclo, preferible sobre el suite completo para ser rápido)
docker compose -f docker-compose.dev.yml run --rm django python manage.py test don_confiao.tests.<archivo>
# Suite completa (al finalizar, para confirmar que nada se rompió)
docker compose -f docker-compose.dev.yml run --rm django python manage.py test
```
### Confirmar el fallo, no asumirlo
- Si un test falla o se espera que falle, **ejecutarlo y mostrar el resultado** antes de cambiar cualquier código o test.
- Verificar que el fallo corresponde a la razón esperada (leer el traceback), no a un problema de entorno, migraciones, importaciones o fixtures.
- No modificar tests para "hacerlos pasar" si el fallo no fue previamente confirmado.
## Comandos Útiles (dentro del contenedor)
- Migraciones: `docker-compose -f docker-compose.dev.yml run --rm django python manage.py makemigrations && docker-compose -f docker-compose.dev.yml run --rm django python manage.py migrate`

View File

@@ -137,3 +137,49 @@ post importar_productos_de_tryton
#+begin_src json
[]
#+end_src
*** Categorías de productos
La fuente de categorías es Tryton. name y external_id son solo lectura; Don
Confiao administra description, alternative_name y active.
**** Listar categorías
get product_categories/?active=all
get product_categories/?active=true
get product_categories/?active=false
**** Actualizar campos administrados por Don Confiao
patch product_categories/1/
#+begin_src json
{
"description": "Granos enteros",
"alternative_name": "Cereales",
"active": false
}
#+end_src
**** Importar categorías desde Tryton
Solo importa las categorías asociadas a productos salables; no borra
categorías locales. Requiere administrador.
post product_categories/import_from_external
{}
**** Desactivar categorías
post product_categories/deactivate/
#+begin_src json
{
"ids": [1, 2]
}
#+end_src
**** Activación de categorías
post product_categories/activate/
#+begin_src json
{
"ids": [1, 2]
}
#+end_src

View File

@@ -44,7 +44,7 @@ services:
- media_volume:/app/media
- logs_volume:/app/logs
ports:
- "8000:8000"
- "127.0.0.1:9000:8000"
depends_on:
postgres:
condition: service_healthy
@@ -57,22 +57,20 @@ services:
gunicorn config.wsgi:application --bind 0.0.0.0:8000 --workers 4 --timeout 120"
# Optional: Nginx reverse proxy para servir archivos estáticos
# nginx:
# image: nginx:alpine
# container_name: tienda_ilusion_nginx
# ports:
# - "80:80"
# - "443:443"
# volumes:
# - ./nginx.conf:/etc/nginx/nginx.conf:ro
# - static_volume:/var/www/static:ro
# - media_volume:/var/www/media:ro
# - ./ssl:/etc/nginx/ssl:ro
# depends_on:
# - django
# networks:
# - tienda_network
# restart: unless-stopped
nginx:
image: nginx:alpine
ports:
- "10297:80"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf:ro
- static_volume:/var/www/static
- media_volume:/var/www/media
- ./dist:/var/www/donconfiao
depends_on:
- django
networks:
- tienda_network
restart: unless-stopped
volumes:
postgres_data:

View File

@@ -2,16 +2,12 @@ services:
postgres:
image: postgres:15-alpine
container_name: tienda_ilusion_postgres_staging
environment:
- POSTGRES_USER=${DB_USER:-tienda_ilusion_user}
- POSTGRES_DB=${DB_NAME:-tienda_ilusion_staging}
- POSTGRES_PASSWORD=${DB_PASSWORD:-staging_local_password}
env_file:
- .env.staging
volumes:
- postgres_staging_data:/var/lib/postgresql/data
ports:
- "5433:5432" # Puerto diferente para no conflictuar con prod
- "5433:5432"
networks:
- tienda_staging_network
restart: unless-stopped
@@ -35,9 +31,6 @@ services:
environment:
- DJANGO_ENV=staging
- DB_HOST=postgres
- DB_USER=${DB_USER:-tienda_ilusion_user}
- DB_NAME=${DB_NAME:-tienda_ilusion_staging}
- DB_PASSWORD=${DB_PASSWORD:-staging_local_password}
volumes:
- ./tienda_ilusion:/app/
- static_staging_volume:/app/staticfiles

36
nginx.conf Normal file
View File

@@ -0,0 +1,36 @@
events {
worker_connections 1024;
}
http {
include /etc/nginx/mime.types;
default_type application/octet-stream;
access_log /var/log/nginx/access.log;
error_log /var/log/nginx/error.log;
server {
listen 80;
server_name _;
client_max_body_size 10M;
location /media/ {
alias /var/www/media/;
expires 30d;
add_header Cache-Control "public, immutable";
}
location /static/ {
alias /var/www/static/;
expires 365d;
add_header Cache-Control "public, immutable";
}
location / {
root /var/www/donconfiao/;
try_files $uri /index.html;
expires -1;
add_header Cache-Control "no-cache";
}
}
}

126
poetry.lock generated
View File

@@ -117,6 +117,115 @@ files = [
{file = "mslex-1.3.0.tar.gz", hash = "sha256:641c887d1d3db610eee2af37a8e5abda3f70b3006cdfd2d0d29dc0d1ae28a85d"},
]
[[package]]
name = "pillow"
version = "12.2.0"
description = "Python Imaging Library (fork)"
optional = false
python-versions = ">=3.10"
groups = ["main"]
files = [
{file = "pillow-12.2.0-cp310-cp310-macosx_10_10_x86_64.whl", hash = "sha256:a4e8f36e677d3336f35089648c8955c51c6d386a13cf6ee9c189c5f5bd713a9f"},
{file = "pillow-12.2.0-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:2e589959f10d9824d39b350472b92f0ce3b443c0a3442ebf41c40cb8361c5b97"},
{file = "pillow-12.2.0-cp310-cp310-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:a52edc8bfff4429aaabdf4d9ee0daadbbf8562364f940937b941f87a4290f5ff"},
{file = "pillow-12.2.0-cp310-cp310-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:975385f4776fafde056abb318f612ef6285b10a1f12b8570f3647ad0d74b48ec"},
{file = "pillow-12.2.0-cp310-cp310-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:bd9c0c7a0c681a347b3194c500cb1e6ca9cab053ea4d82a5cf45b6b754560136"},
{file = "pillow-12.2.0-cp310-cp310-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:88d387ff40b3ff7c274947ed3125dedf5262ec6919d83946753b5f3d7c67ea4c"},
{file = "pillow-12.2.0-cp310-cp310-musllinux_1_2_aarch64.whl", hash = "sha256:51c4167c34b0d8ba05b547a3bb23578d0ba17b80a5593f93bd8ecb123dd336a3"},
{file = "pillow-12.2.0-cp310-cp310-musllinux_1_2_x86_64.whl", hash = "sha256:34c0d99ecccea270c04882cb3b86e7b57296079c9a4aff88cb3b33563d95afaa"},
{file = "pillow-12.2.0-cp310-cp310-win32.whl", hash = "sha256:b85f66ae9eb53e860a873b858b789217ba505e5e405a24b85c0464822fe88032"},
{file = "pillow-12.2.0-cp310-cp310-win_amd64.whl", hash = "sha256:673aa32138f3e7531ccdbca7b3901dba9b70940a19ccecc6a37c77d5fdeb05b5"},
{file = "pillow-12.2.0-cp310-cp310-win_arm64.whl", hash = "sha256:3e080565d8d7c671db5802eedfb438e5565ffa40115216eabb8cd52d0ecce024"},
{file = "pillow-12.2.0-cp311-cp311-macosx_10_10_x86_64.whl", hash = "sha256:8be29e59487a79f173507c30ddf57e733a357f67881430449bb32614075a40ab"},
{file = "pillow-12.2.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:71cde9a1e1551df7d34a25462fc60325e8a11a82cc2e2f54578e5e9a1e153d65"},
{file = "pillow-12.2.0-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:f490f9368b6fc026f021db16d7ec2fbf7d89e2edb42e8ec09d2c60505f5729c7"},
{file = "pillow-12.2.0-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:8bd7903a5f2a4545f6fd5935c90058b89d30045568985a71c79f5fd6edf9b91e"},
{file = "pillow-12.2.0-cp311-cp311-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:3997232e10d2920a68d25191392e3a4487d8183039e1c74c2297f00ed1c50705"},
{file = "pillow-12.2.0-cp311-cp311-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:e74473c875d78b8e9d5da2a70f7099549f9eb37ded4e2f6a463e60125bccd176"},
{file = "pillow-12.2.0-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:56a3f9c60a13133a98ecff6197af34d7824de9b7b38c3654861a725c970c197b"},
{file = "pillow-12.2.0-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:90e6f81de50ad6b534cab6e5aef77ff6e37722b2f5d908686f4a5c9eba17a909"},
{file = "pillow-12.2.0-cp311-cp311-win32.whl", hash = "sha256:8c984051042858021a54926eb597d6ee3012393ce9c181814115df4c60b9a808"},
{file = "pillow-12.2.0-cp311-cp311-win_amd64.whl", hash = "sha256:6e6b2a0c538fc200b38ff9eb6628228b77908c319a005815f2dde585a0664b60"},
{file = "pillow-12.2.0-cp311-cp311-win_arm64.whl", hash = "sha256:9a8a34cc89c67a65ea7437ce257cea81a9dad65b29805f3ecee8c8fe8ff25ffe"},
{file = "pillow-12.2.0-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:2d192a155bbcec180f8564f693e6fd9bccff5a7af9b32e2e4bf8c9c69dbad6b5"},
{file = "pillow-12.2.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:f3f40b3c5a968281fd507d519e444c35f0ff171237f4fdde090dd60699458421"},
{file = "pillow-12.2.0-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:03e7e372d5240cc23e9f07deca4d775c0817bffc641b01e9c3af208dbd300987"},
{file = "pillow-12.2.0-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:b86024e52a1b269467a802258c25521e6d742349d760728092e1bc2d135b4d76"},
{file = "pillow-12.2.0-cp312-cp312-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:7371b48c4fa448d20d2714c9a1f775a81155050d383333e0a6c15b1123dda005"},
{file = "pillow-12.2.0-cp312-cp312-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:62f5409336adb0663b7caa0da5c7d9e7bdbaae9ce761d34669420c2a801b2780"},
{file = "pillow-12.2.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:01afa7cf67f74f09523699b4e88c73fb55c13346d212a59a2db1f86b0a63e8c5"},
{file = "pillow-12.2.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:fc3d34d4a8fbec3e88a79b92e5465e0f9b842b628675850d860b8bd300b159f5"},
{file = "pillow-12.2.0-cp312-cp312-win32.whl", hash = "sha256:58f62cc0f00fd29e64b29f4fd923ffdb3859c9f9e6105bfc37ba1d08994e8940"},
{file = "pillow-12.2.0-cp312-cp312-win_amd64.whl", hash = "sha256:7f84204dee22a783350679a0333981df803dac21a0190d706a50475e361c93f5"},
{file = "pillow-12.2.0-cp312-cp312-win_arm64.whl", hash = "sha256:af73337013e0b3b46f175e79492d96845b16126ddf79c438d7ea7ff27783a414"},
{file = "pillow-12.2.0-cp313-cp313-ios_13_0_arm64_iphoneos.whl", hash = "sha256:8297651f5b5679c19968abefd6bb84d95fe30ef712eb1b2d9b2d31ca61267f4c"},
{file = "pillow-12.2.0-cp313-cp313-ios_13_0_arm64_iphonesimulator.whl", hash = "sha256:50d8520da2a6ce0af445fa6d648c4273c3eeefbc32d7ce049f22e8b5c3daecc2"},
{file = "pillow-12.2.0-cp313-cp313-ios_13_0_x86_64_iphonesimulator.whl", hash = "sha256:766cef22385fa1091258ad7e6216792b156dc16d8d3fa607e7545b2b72061f1c"},
{file = "pillow-12.2.0-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:5d2fd0fa6b5d9d1de415060363433f28da8b1526c1c129020435e186794b3795"},
{file = "pillow-12.2.0-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:56b25336f502b6ed02e889f4ece894a72612fe885889a6e8c4c80239ff6e5f5f"},
{file = "pillow-12.2.0-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:f1c943e96e85df3d3478f7b691f229887e143f81fedab9b20205349ab04d73ed"},
{file = "pillow-12.2.0-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:03f6fab9219220f041c74aeaa2939ff0062bd5c364ba9ce037197f4c6d498cd9"},
{file = "pillow-12.2.0-cp313-cp313-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:5cdfebd752ec52bf5bb4e35d9c64b40826bc5b40a13df7c3cda20a2c03a0f5ed"},
{file = "pillow-12.2.0-cp313-cp313-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:eedf4b74eda2b5a4b2b2fb4c006d6295df3bf29e459e198c90ea48e130dc75c3"},
{file = "pillow-12.2.0-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:00a2865911330191c0b818c59103b58a5e697cae67042366970a6b6f1b20b7f9"},
{file = "pillow-12.2.0-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:1e1757442ed87f4912397c6d35a0db6a7b52592156014706f17658ff58bbf795"},
{file = "pillow-12.2.0-cp313-cp313-win32.whl", hash = "sha256:144748b3af2d1b358d41286056d0003f47cb339b8c43a9ea42f5fea4d8c66b6e"},
{file = "pillow-12.2.0-cp313-cp313-win_amd64.whl", hash = "sha256:390ede346628ccc626e5730107cde16c42d3836b89662a115a921f28440e6a3b"},
{file = "pillow-12.2.0-cp313-cp313-win_arm64.whl", hash = "sha256:8023abc91fba39036dbce14a7d6535632f99c0b857807cbbbf21ecc9f4717f06"},
{file = "pillow-12.2.0-cp313-cp313t-macosx_10_13_x86_64.whl", hash = "sha256:042db20a421b9bafecc4b84a8b6e444686bd9d836c7fd24542db3e7df7baad9b"},
{file = "pillow-12.2.0-cp313-cp313t-macosx_11_0_arm64.whl", hash = "sha256:dd025009355c926a84a612fecf58bb315a3f6814b17ead51a8e48d3823d9087f"},
{file = "pillow-12.2.0-cp313-cp313t-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:88ddbc66737e277852913bd1e07c150cc7bb124539f94c4e2df5344494e0a612"},
{file = "pillow-12.2.0-cp313-cp313t-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:d362d1878f00c142b7e1a16e6e5e780f02be8195123f164edf7eddd911eefe7c"},
{file = "pillow-12.2.0-cp313-cp313t-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:2c727a6d53cb0018aadd8018c2b938376af27914a68a492f59dfcaca650d5eea"},
{file = "pillow-12.2.0-cp313-cp313t-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:efd8c21c98c5cc60653bcb311bef2ce0401642b7ce9d09e03a7da87c878289d4"},
{file = "pillow-12.2.0-cp313-cp313t-musllinux_1_2_aarch64.whl", hash = "sha256:9f08483a632889536b8139663db60f6724bfcb443c96f1b18855860d7d5c0fd4"},
{file = "pillow-12.2.0-cp313-cp313t-musllinux_1_2_x86_64.whl", hash = "sha256:dac8d77255a37e81a2efcbd1fc05f1c15ee82200e6c240d7e127e25e365c39ea"},
{file = "pillow-12.2.0-cp313-cp313t-win32.whl", hash = "sha256:ee3120ae9dff32f121610bb08e4313be87e03efeadfc6c0d18f89127e24d0c24"},
{file = "pillow-12.2.0-cp313-cp313t-win_amd64.whl", hash = "sha256:325ca0528c6788d2a6c3d40e3568639398137346c3d6e66bb61db96b96511c98"},
{file = "pillow-12.2.0-cp313-cp313t-win_arm64.whl", hash = "sha256:2e5a76d03a6c6dcef67edabda7a52494afa4035021a79c8558e14af25313d453"},
{file = "pillow-12.2.0-cp314-cp314-ios_13_0_arm64_iphoneos.whl", hash = "sha256:3adc9215e8be0448ed6e814966ecf3d9952f0ea40eb14e89a102b87f450660d8"},
{file = "pillow-12.2.0-cp314-cp314-ios_13_0_arm64_iphonesimulator.whl", hash = "sha256:6a9adfc6d24b10f89588096364cc726174118c62130c817c2837c60cf08a392b"},
{file = "pillow-12.2.0-cp314-cp314-ios_13_0_x86_64_iphonesimulator.whl", hash = "sha256:6a6e67ea2e6feda684ed370f9a1c52e7a243631c025ba42149a2cc5934dec295"},
{file = "pillow-12.2.0-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:2bb4a8d594eacdfc59d9e5ad972aa8afdd48d584ffd5f13a937a664c3e7db0ed"},
{file = "pillow-12.2.0-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:80b2da48193b2f33ed0c32c38140f9d3186583ce7d516526d462645fd98660ae"},
{file = "pillow-12.2.0-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:22db17c68434de69d8ecfc2fe821569195c0c373b25cccb9cbdacf2c6e53c601"},
{file = "pillow-12.2.0-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:7b14cc0106cd9aecda615dd6903840a058b4700fcb817687d0ee4fc8b6e389be"},
{file = "pillow-12.2.0-cp314-cp314-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:8cbeb542b2ebc6fcdacabf8aca8c1a97c9b3ad3927d46b8723f9d4f033288a0f"},
{file = "pillow-12.2.0-cp314-cp314-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:4bfd07bc812fbd20395212969e41931001fd59eb55a60658b0e5710872e95286"},
{file = "pillow-12.2.0-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:9aba9a17b623ef750a4d11b742cbafffeb48a869821252b30ee21b5e91392c50"},
{file = "pillow-12.2.0-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:deede7c263feb25dba4e82ea23058a235dcc2fe1f6021025dc71f2b618e26104"},
{file = "pillow-12.2.0-cp314-cp314-win32.whl", hash = "sha256:632ff19b2778e43162304d50da0181ce24ac5bb8180122cbe1bf4673428328c7"},
{file = "pillow-12.2.0-cp314-cp314-win_amd64.whl", hash = "sha256:4e6c62e9d237e9b65fac06857d511e90d8461a32adcc1b9065ea0c0fa3a28150"},
{file = "pillow-12.2.0-cp314-cp314-win_arm64.whl", hash = "sha256:b1c1fbd8a5a1af3412a0810d060a78b5136ec0836c8a4ef9aa11807f2a22f4e1"},
{file = "pillow-12.2.0-cp314-cp314t-macosx_10_15_x86_64.whl", hash = "sha256:57850958fe9c751670e49b2cecf6294acc99e562531f4bd317fa5ddee2068463"},
{file = "pillow-12.2.0-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:d5d38f1411c0ed9f97bcb49b7bd59b6b7c314e0e27420e34d99d844b9ce3b6f3"},
{file = "pillow-12.2.0-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:5c0a9f29ca8e79f09de89293f82fc9b0270bb4af1d58bc98f540cc4aedf03166"},
{file = "pillow-12.2.0-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:1610dd6c61621ae1cf811bef44d77e149ce3f7b95afe66a4512f8c59f25d9ebe"},
{file = "pillow-12.2.0-cp314-cp314t-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:0a34329707af4f73cf1782a36cd2289c0368880654a2c11f027bcee9052d35dd"},
{file = "pillow-12.2.0-cp314-cp314t-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:8e9c4f5b3c546fa3458a29ab22646c1c6c787ea8f5ef51300e5a60300736905e"},
{file = "pillow-12.2.0-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:fb043ee2f06b41473269765c2feae53fc2e2fbf96e5e22ca94fb5ad677856f06"},
{file = "pillow-12.2.0-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:f278f034eb75b4e8a13a54a876cc4a5ab39173d2cdd93a638e1b467fc545ac43"},
{file = "pillow-12.2.0-cp314-cp314t-win32.whl", hash = "sha256:6bb77b2dcb06b20f9f4b4a8454caa581cd4dd0643a08bacf821216a16d9c8354"},
{file = "pillow-12.2.0-cp314-cp314t-win_amd64.whl", hash = "sha256:6562ace0d3fb5f20ed7290f1f929cae41b25ae29528f2af1722966a0a02e2aa1"},
{file = "pillow-12.2.0-cp314-cp314t-win_arm64.whl", hash = "sha256:aa88ccfe4e32d362816319ed727a004423aab09c5cea43c01a4b435643fa34eb"},
{file = "pillow-12.2.0-pp311-pypy311_pp73-macosx_10_15_x86_64.whl", hash = "sha256:0538bd5e05efec03ae613fd89c4ce0368ecd2ba239cc25b9f9be7ed426b0af1f"},
{file = "pillow-12.2.0-pp311-pypy311_pp73-macosx_11_0_arm64.whl", hash = "sha256:394167b21da716608eac917c60aa9b969421b5dcbbe02ae7f013e7b85811c69d"},
{file = "pillow-12.2.0-pp311-pypy311_pp73-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:5d04bfa02cc2d23b497d1e90a0f927070043f6cbf303e738300532379a4b4e0f"},
{file = "pillow-12.2.0-pp311-pypy311_pp73-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:0c838a5125cee37e68edec915651521191cef1e6aa336b855f495766e77a366e"},
{file = "pillow-12.2.0-pp311-pypy311_pp73-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:4a6c9fa44005fa37a91ebfc95d081e8079757d2e904b27103f4f5fa6f0bf78c0"},
{file = "pillow-12.2.0-pp311-pypy311_pp73-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:25373b66e0dd5905ed63fa3cae13c82fbddf3079f2c8bf15c6fb6a35586324c1"},
{file = "pillow-12.2.0-pp311-pypy311_pp73-win_amd64.whl", hash = "sha256:bfa9c230d2fe991bed5318a5f119bd6780cda2915cca595393649fc118ab895e"},
{file = "pillow-12.2.0.tar.gz", hash = "sha256:a830b1a40919539d07806aa58e1b114df53ddd43213d9c8b75847eee6c0182b5"},
]
[package.extras]
docs = ["furo", "olefile", "sphinx (>=8.2)", "sphinx-autobuild", "sphinx-copybutton", "sphinx-inline-tabs", "sphinxext-opengraph"]
fpx = ["olefile"]
mic = ["olefile"]
test-arrow = ["arro3-compute", "arro3-core", "nanoarrow", "pyarrow"]
tests = ["check-manifest", "coverage (>=7.4.2)", "defusedxml", "markdown2", "olefile", "packaging", "pyroma (>=5)", "pytest", "pytest-cov", "pytest-timeout", "pytest-xdist", "trove-classifiers (>=2024.10.12)"]
xmp = ["defusedxml"]
[[package]]
name = "psutil"
version = "6.1.1"
@@ -282,7 +391,22 @@ files = [
{file = "tzdata-2026.2.tar.gz", hash = "sha256:9173fde7d80d9018e02a662e168e5a2d04f87c41ea174b139fbef642eda62d10"},
]
[[package]]
name = "whitenoise"
version = "6.12.0"
description = "Radically simplified static file serving for WSGI applications"
optional = false
python-versions = ">=3.10"
groups = ["main"]
files = [
{file = "whitenoise-6.12.0-py3-none-any.whl", hash = "sha256:fc5e8c572e33ebf24795b47b6a7da8da3c00cff2349f5b04c02f28d0cc5a3cc2"},
{file = "whitenoise-6.12.0.tar.gz", hash = "sha256:f723ebb76a112e98816ff80fcea0a6c9b8ecde835f8ddda25df7a30a3c2db6ad"},
]
[package.extras]
brotli = ["brotli"]
[metadata]
lock-version = "2.1"
python-versions = ">=3.14,<4.0"
content-hash = "02efc37f4afc4dbe36959374dba58e3b6e61be3f8c4298427009b9ee0d3a1910"
content-hash = "ac185d7c04ec39d561c6cc78d022accbdc80e2f815f39c71aa5f38c6502f70b2"

View File

@@ -12,7 +12,9 @@ dependencies = [
"djangorestframework (>=3.17.1,<4.0.0)",
"django-cors-headers (>=4.9.0,<5.0.0)",
"djangorestframework-simplejwt (>=5.5.1,<6.0.0)",
"sabatron-tryton-rpc-client (>=7.4.0,<8.0.0)"
"sabatron-tryton-rpc-client (>=7.4.0,<8.0.0)",
"pillow (>=12.2.0,<13.0.0)",
"whitenoise (>=6.12.0,<7.0.0)"
]

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,80 @@
#!/usr/bin/env python3
import argparse
import getpass
import sys
import requests
TOKEN_URL = "/api/token/"
FILL_SALES_CODES_URL = "/don_confiao/api/sales/fill_codes/"
FILL_CATALOG_SALES_CODES_URL = "/don_confiao/api/catalog_sales/fill_codes/"
def get_credentials():
username = input("Usuario: ")
password = getpass.getpass("Contraseña: ")
return username, password
def get_token(domain, username, password):
url = domain.rstrip("/") + TOKEN_URL
response = requests.post(
url, json={"username": username, "password": password}
)
if response.status_code != 200:
print(
f"Error al obtener token: {response.status_code} {response.text}",
file=sys.stderr,
)
sys.exit(1)
return response.json()["access"]
def fill_codes(domain, token, url):
headers = {"Authorization": f"Bearer {token}"}
full_url = domain.rstrip("/") + url
response = requests.post(full_url, headers=headers)
if response.status_code != 200:
print(
f"Error al llenar códigos ({url}): {response.status_code} {response.text}",
file=sys.stderr,
)
sys.exit(1)
return response.json()
def main():
parser = argparse.ArgumentParser(
description="Asigna un código único a todas las ventas y pedidos que lo requieran."
)
parser.add_argument(
"domain", help="Dominio del backend (ej: http://localhost:8000)"
)
args = parser.parse_args()
username, password = get_credentials()
token = get_token(args.domain, username, password)
print("Token obtenido correctamente.")
print("Asignando códigos a ventas...")
sales_result = fill_codes(args.domain, token, FILL_SALES_CODES_URL)
print(f" [OK] Ventas actualizadas: {sales_result['sales_updated']}")
print("Asignando códigos a pedidos de catálogo...")
catalog_result = fill_codes(
args.domain, token, FILL_CATALOG_SALES_CODES_URL
)
print(
f" [OK] Pedidos de catálogo actualizados: "
f"{catalog_result['catalog_sales_updated']}"
)
print(
f"\nResumen: {sales_result['sales_updated']} ventas, "
f"{catalog_result['catalog_sales_updated']} pedidos de catálogo actualizados"
)
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,99 @@
#!/usr/bin/env python3
import argparse
import getpass
import sys
import requests
TOKEN_URL = "/api/token/"
CATALOGUE_IMAGES_URL = "/don_confiao/api/catalogue_images/"
def get_credentials():
username = input("Usuario: ")
password = getpass.getpass("Contraseña: ")
return username, password
def get_token(domain, username, password):
url = domain.rstrip("/") + TOKEN_URL
response = requests.post(url, json={"username": username, "password": password})
if response.status_code != 200:
print(f"Error al obtener token: {response.status_code} {response.text}", file=sys.stderr)
sys.exit(1)
data = response.json()
return data["access"]
def get_catalogue_images(domain, token):
url = domain.rstrip("/") + CATALOGUE_IMAGES_URL
headers = {"Authorization": f"Bearer {token}"}
response = requests.get(url, headers=headers)
if response.status_code != 200:
print(f"Error al obtener imágenes de catálogo: {response.status_code} {response.text}", file=sys.stderr)
sys.exit(1)
return response.json()
def download_image(image_url):
response = requests.get(image_url)
if response.status_code != 200:
print(f" [ERR] No se pudo descargar imagen: {response.status_code}", file=sys.stderr)
return None
return response.content
def main():
parser = argparse.ArgumentParser(
description="Vuelve a subir cada imagen de catálogo existente para forzar el redimensionado."
)
parser.add_argument("domain", help="Dominio del backend (ej: http://localhost:8000)")
args = parser.parse_args()
username, password = get_credentials()
token = get_token(args.domain, username, password)
print("Token obtenido correctamente.")
images = get_catalogue_images(args.domain, token)
if not images:
print("No hay imágenes de catálogo para procesar.")
return
headers = {"Authorization": f"Bearer {token}"}
updated = 0
errors = 0
for img in images:
ci_id = img["id"]
product_id = img["product"]
image_url = img["image"]
print(f" [{ci_id}] Descargando {image_url} ...")
content = download_image(image_url)
if content is None:
errors += 1
continue
# Determine content type from URL extension
ext = image_url.rsplit(".", 1)[-1].lower() if "." in image_url else "jpg"
mime = {"jpg": "image/jpeg", "jpeg": "image/jpeg", "png": "image/png"}.get(ext, "application/octet-stream")
filename = f"image.{ext}"
update_url = args.domain.rstrip("/") + CATALOGUE_IMAGES_URL.rstrip("/") + f"/{ci_id}/"
files = {"image": (filename, content, mime)}
data = {"product": product_id}
response = requests.put(update_url, headers=headers, files=files, data=data)
if response.status_code == 200:
print(f" [OK] Imagen {ci_id} (producto {product_id}) actualizada.")
updated += 1
else:
print(f" [ERR] Imagen {ci_id}: {response.status_code} {response.text}")
errors += 1
print(f"\nResumen: {updated} actualizadas, {errors} errores")
if __name__ == "__main__":
main()

188
scripts/seed_geography.py Normal file
View File

@@ -0,0 +1,188 @@
#!/usr/bin/env python3
"""Seed geografía de Colombia (países, departamentos y municipios).
Requisitos (instalar ad-hoc, NO son dependencia del proyecto):
pip install pycountry requests
Uso:
python scripts/seed_geography.py http://localhost:7000
El script obtiene el país y los departamentos de Colombia desde
`pycountry` (ISO 3166-1/3166-2) y los municipios desde el dataset DANE
incluido en `scripts/data/colombia_municipios.csv`. No consulta internet
en tiempo de ejecución.
"""
import argparse
import csv
import getpass
import os
import sys
import unicodedata
import requests
try:
import pycountry
except ModuleNotFoundError:
print(
"pycountry no está instalado. Ejecuta: pip install pycountry",
file=sys.stderr,
)
sys.exit(1)
TOKEN_URL = "/api/token/"
SEED_GEOGRAPHY_URL = "/don_confiao/api/seed_geography"
MUNICIPALITIES_CSV = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
"data",
"colombia_municipios.csv",
)
# Nombres DANE que difieren de los nombres oficiales de pycountry.
DEPARTMENT_ALIASES = {
"BOGOTA": "Distrito Capital de Bogotá",
"SAN ANDRES": "San Andrés, Providencia y Santa Catalina",
}
def get_credentials():
username = input("Usuario: ")
password = getpass.getpass("Contraseña: ")
return username, password
def get_token(domain, username, password):
url = domain.rstrip("/") + TOKEN_URL
response = requests.post(
url, json={"username": username, "password": password}
)
if response.status_code != 200:
print(
f"Error al obtener token: {response.status_code} {response.text}",
file=sys.stderr,
)
sys.exit(1)
return response.json()["access"]
def _normalize(name):
return "".join(
char
for char in unicodedata.normalize("NFD", name)
if unicodedata.category(char) != "Mn"
).upper()
def get_department_mapping(subdivisions):
by_normalized = {
_normalize(subdivision.name): subdivision.name
for subdivision in subdivisions
}
csv_department_names = set()
with open(MUNICIPALITIES_CSV, newline="", encoding="utf-8") as f:
for row in csv.DictReader(f):
csv_department_names.add(row["department"])
mapping = dict(DEPARTMENT_ALIASES)
for name in csv_department_names:
if name in mapping:
continue
mapped = by_normalized.get(_normalize(name))
if mapped is None:
print(
f"Departamento '{name}' no encontrado en pycountry",
file=sys.stderr,
)
sys.exit(1)
mapping[name] = mapped
return mapping
def read_municipalities(mapping):
municipalities = []
with open(MUNICIPALITIES_CSV, newline="", encoding="utf-8") as f:
for row in csv.DictReader(f):
municipality = {
"name": row["municipality"],
"department": mapping[row["department"]],
}
if row.get("latitude"):
municipality["latitude"] = float(row["latitude"])
if row.get("longitude"):
municipality["longitude"] = float(row["longitude"])
municipalities.append(municipality)
return municipalities
def build_payload():
country = pycountry.countries.get(alpha_2="CO")
subdivisions = pycountry.subdivisions.get(country_code="CO")
departments = [
{"name": subdivision.name, "code": subdivision.code}
for subdivision in subdivisions
]
mapping = get_department_mapping(subdivisions)
municipalities = read_municipalities(mapping)
return {
"country": {"name": country.name, "code": country.alpha_2},
"departments": departments,
"municipalities": municipalities,
}
def seed_geography(domain, token, payload):
headers = {"Authorization": f"Bearer {token}"}
full_url = domain.rstrip("/") + SEED_GEOGRAPHY_URL
response = requests.post(full_url, headers=headers, json=payload)
if response.status_code != 200:
print(
f"Error al sembrar geografía: {response.status_code} {response.text}",
file=sys.stderr,
)
sys.exit(1)
return response.json()
def main():
parser = argparse.ArgumentParser(
description=(
"Alimenta países, departamentos y municipios de Colombia "
"usando pycountry y el dataset DANE incluido en scripts/data/."
)
)
parser.add_argument(
"domain", help="Dominio del backend (ej: http://localhost:7000)"
)
args = parser.parse_args()
username, password = get_credentials()
token = get_token(args.domain, username, password)
print("Token obtenido correctamente.")
payload = build_payload()
print(
"Preparando siembra: 1 país, "
f"{len(payload['departments'])} departamentos, "
f"{len(payload['municipalities'])} municipios."
)
print("Enviando geografía...")
result = seed_geography(args.domain, token, payload)
print(
f" [OK] {result['country']}: "
f"{result['total_departments']} departamentos, "
f"{result['total_municipalities']} municipios."
)
print(
f"\nResumen: {result['departments_created']} departamentos y "
f"{result['municipalities_created']} municipios creados "
f"(re-ejecutar es idempotente)."
)
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,126 @@
#!/usr/bin/env python3
import argparse
import getpass
import os
import re
import sys
from pathlib import Path
import requests
TOKEN_URL = "/api/token/"
PRODUCTS_URL = "/don_confiao/api/products/"
CATALOGUE_IMAGES_URL = "/don_confiao/api/catalogue_images/"
def get_credentials():
username = input("Usuario: ")
password = getpass.getpass("Contraseña: ")
return username, password
def get_token(domain, username, password):
url = domain.rstrip("/") + TOKEN_URL
response = requests.post(url, json={"username": username, "password": password})
if response.status_code != 200:
print(f"Error al obtener token: {response.status_code} {response.text}", file=sys.stderr)
sys.exit(1)
data = response.json()
return data["access"]
def get_products(domain, token):
url = domain.rstrip("/") + PRODUCTS_URL
headers = {"Authorization": f"Bearer {token}"}
response = requests.get(url, headers=headers)
if response.status_code != 200:
print(f"Error al obtener productos: {response.status_code} {response.text}", file=sys.stderr)
sys.exit(1)
return response.json()
MIME_TYPES = {
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".png": "image/png",
}
def find_images(image_dir):
images = {}
pattern = re.compile(r"^(\d+)\.(jpg|jpeg|png)$", re.IGNORECASE)
for f in os.listdir(image_dir):
m = pattern.match(f)
if m:
external_id = m.group(1)
images[external_id] = os.path.join(image_dir, f)
return images
def main():
parser = argparse.ArgumentParser(
description="Sube imágenes de catálogo para productos usando el external_id como nombre de archivo."
)
parser.add_argument("image_dir", help="Directorio con imágenes nombradas como ##.jpg")
parser.add_argument("domain", help="Dominio del backend (ej: http://localhost:8000)")
args = parser.parse_args()
if not os.path.isdir(args.image_dir):
print(f"Error: el directorio '{args.image_dir}' no existe.", file=sys.stderr)
sys.exit(1)
username, password = get_credentials()
token = get_token(args.domain, username, password)
print("Token obtenido correctamente.")
products = get_products(args.domain, token)
ext_id_to_product_id = {}
for p in products:
if p.get("external_id"):
ext_id_to_product_id[p["external_id"]] = p["id"]
if not ext_id_to_product_id:
print("No se encontraron productos con external_id.", file=sys.stderr)
sys.exit(1)
images = find_images(args.image_dir)
if not images:
print(f"No se encontraron imágenes con el patrón ##.jpg en '{args.image_dir}'.", file=sys.stderr)
sys.exit(1)
headers = {"Authorization": f"Bearer {token}"}
upload_url = args.domain.rstrip("/") + CATALOGUE_IMAGES_URL
uploaded = 0
skipped = 0
errors = 0
for external_id, img_path in sorted(images.items()):
if external_id not in ext_id_to_product_id:
print(f" [SKIP] {Path(img_path).name}: no hay producto con external_id={external_id}")
skipped += 1
continue
product_id = ext_id_to_product_id[external_id]
filename = Path(img_path).name
ext = Path(img_path).suffix.lower()
mime = MIME_TYPES.get(ext, "application/octet-stream")
with open(img_path, "rb") as f:
files = {"image": (filename, f, mime)}
data = {"product": product_id}
response = requests.post(upload_url, headers=headers, files=files, data=data)
if response.status_code == 201:
print(f" [OK] {filename} -> producto {product_id} (id imagen: {response.json()['id']})")
uploaded += 1
else:
print(f" [ERR] {filename} -> producto {product_id}: {response.status_code} {response.text}")
errors += 1
print(f"\nResumen: {uploaded} subidas, {skipped} saltadas, {errors} errores")
if __name__ == "__main__":
main()

View File

@@ -153,6 +153,7 @@ CATALOGUE_BACKGROUND_IMAGES_RGBA = os.environ.get(
_rgba_parts = CATALOGUE_BACKGROUND_IMAGES_RGBA.split(",")
CATALOGUE_BACKGROUND_IMAGES_RGBA = tuple(int(p.strip()) for p in _rgba_parts)
CATALOGUE_MAX_UPLOAD_SIZE = int(os.environ.get("CATALOGUE_MAX_UPLOAD_SIZE", str(5 * 1024 * 1024)))
LOGO_MAX_UPLOAD_SIZE = int(os.environ.get("LOGO_MAX_UPLOAD_SIZE", str(5 * 1024 * 1024)))
# Default primary key field type
# https://docs.djangoproject.com/en/5.0/ref/settings/#default-auto-field
@@ -166,7 +167,7 @@ REST_FRAMEWORK = {
"rest_framework_simplejwt.authentication.JWTAuthentication",
],
"DEFAULT_PERMISSION_CLASSES": [
"rest_framework.permissions.IsAuthenticated",
"rest_framework.permissions.AllowAny",
],
}

View File

@@ -12,7 +12,7 @@ import os
from .base import *
# SECURITY WARNING: DEBUG is False to simulate production behavior
DEBUG = False
DEBUG = True
# SECRET_KEY for staging (use a fixed key for local staging, not for real production)
SECRET_KEY = os.environ.get(
@@ -21,9 +21,9 @@ SECRET_KEY = os.environ.get(
)
# Allow localhost for staging
ALLOWED_HOSTS = os.environ.get("ALLOWED_HOSTS", "localhost,127.0.0.1,0.0.0.0").split(
","
)
ALLOWED_HOSTS = os.environ.get(
"ALLOWED_HOSTS", "localhost,127.0.0.1,0.0.0.0"
).split(",")
# CORS settings for staging - permissive for localhost testing
CORS_ALLOWED_ORIGINS = os.environ.get(
@@ -52,9 +52,11 @@ CORS_EXPOSE_HEADERS = ["Content-Type", "X-CSRFToken"]
DATABASES = {
"default": {
"ENGINE": "django.db.backends.postgresql",
"NAME": os.environ.get("DB_NAME", "tienda_ilusion_staging"),
"USER": os.environ.get("DB_USER", "tienda_ilusion_user"),
"PASSWORD": os.environ.get("DB_PASSWORD", "staging_local_password"),
"NAME": os.environ.get("POSTGRES_DB", "tienda_ilusion_staging"),
"USER": os.environ.get("POSTGRES_USER", "tienda_ilusion_user"),
"PASSWORD": os.environ.get(
"POSTGRES_PASSWORD", "staging_local_password"
),
"HOST": os.environ.get("DB_HOST", "localhost"),
"PORT": os.environ.get("DB_PORT", "5432"),
"CONN_MAX_AGE": 600, # Persistent connections
@@ -117,7 +119,9 @@ WHITENOISE_AUTOREFRESH = (
)
WHITENOISE_USE_FINDERS = False # Use collected static files only
WHITENOISE_MANIFEST_STRICT = False # More permissive in staging
WHITENOISE_MAX_AGE = 600 # 10 minutes cache for staging (allows easier testing)
WHITENOISE_MAX_AGE = (
600 # 10 minutes cache for staging (allows easier testing)
)
# Staging logging - similar to production but more verbose
LOGGING = {
@@ -148,7 +152,7 @@ LOGGING = {
},
"root": {
"handlers": ["console", "file"],
"level": "INFO",
"level": "DEBUG",
},
"loggers": {
"django": {
@@ -163,7 +167,7 @@ LOGGING = {
},
"django.db.backends": {
"handlers": ["console"],
"level": "INFO", # Show SQL queries in staging for debugging
"level": "DEBUG", # Show SQL queries in staging for debugging
"propagate": False,
},
},
@@ -177,7 +181,10 @@ pathlib.Path(logs_dir).mkdir(parents=True, exist_ok=True)
# Admin email notifications
ADMINS = [
("Staging Admin", os.environ.get("ADMIN_EMAIL", "admin@tiendailusion.local")),
(
"Staging Admin",
os.environ.get("ADMIN_EMAIL", "admin@tiendailusion.local"),
),
]
MANAGERS = ADMINS

View File

@@ -10,11 +10,18 @@ from .models.sales import (
)
from .models.products import Product, ProductCategory
from .models.payments import ReconciliationJar
from .models.store_settings import StoreSettings
@admin.register(StoreSettings)
class StoreSettingsAdmin(admin.ModelAdmin):
list_display = ("id", "address", "latitude", "longitude", "updated_at")
@admin.register(Customer)
class CustomerAdmin(admin.ModelAdmin):
list_display = (
"id",
"name",
"email",
"phone",
@@ -23,12 +30,59 @@ class CustomerAdmin(admin.ModelAdmin):
)
search_fields = ("name", "email", "phone")
@admin.register(Sale)
class SaleAdmin(admin.ModelAdmin):
list_display = (
"id",
"date",
"customer",
"payment_method",
"external_id",
"code",
)
search_fields = ("date", "customer__name", "payment_method", "external_id", "code")
list_filter = ("date", "customer__name", "payment_method")
@admin.register(SaleLine)
class SaleLineAdmin(admin.ModelAdmin):
list_display = ("sale", "product", "quantity", "unit_price")
search_fields = ("sale__id", "product__name")
list_filter = ("sale__date", "product__name")
@admin.register(CatalogSale)
class CatalogSaleAdmin(admin.ModelAdmin):
list_display = (
"id",
"date",
"customer_name",
"customer_phone",
"customer_address",
"pickup_method",
"external_id",
"code",
)
search_fields = ("date","customer_name", "customer_phone", "pickup_method","external_id", "code")
list_filter = ("date", "customer_name", "pickup_method")
@admin.register(CatalogSaleLine)
class CatalogSaleLineAdmin(admin.ModelAdmin):
list_display = ("catalog_sale", "product", "quantity", "unit_price")
search_fields = ("catalog_sale__id", "product__name")
list_filter = ("catalog_sale__date", "product__name")
@admin.register(Product)
class ProductAdmin(admin.ModelAdmin):
list_display = ("id", "name", "price", "measuring_unit", "external_id")
search_fields = ("id", "name",)
list_filter = ("name", "id")
@admin.register(ProductCategory)
class ProductCategoryAdmin(admin.ModelAdmin):
list_display = ("id", "name", "alternative_name", "active", "external_id")
search_fields = ("name", "alternative_name", "external_id")
list_filter = ("active",)
admin.site.register(Sale)
admin.site.register(SaleLine)
admin.site.register(CatalogSale)
admin.site.register(CatalogSaleLine)
admin.site.register(Product)
admin.site.register(ProductCategory)
admin.site.register(Payment)
admin.site.register(ReconciliationJar)

View File

@@ -1,11 +1,13 @@
from .catalogue_images import CatalogueImageViewSet
from .products import ProductView, ProductsFromTrytonView
from .categories import ProductCategoryView
from .customers import CustomerView, CustomersFromTrytonView
from .sales import (
SaleView,
CatalogSaleView,
SaleSummary,
CatalogSaleSummary,
PublicSaleSummaryView,
SalesForTrytonView,
SalesToTrytonView,
CatalogSalesToTrytonView,
@@ -18,6 +20,15 @@ from .payments import (
Pagination,
)
from .admin import AdminCodeValidateView
from .store_settings import StoreSettingsView
from .provenance import (
OrganizationView,
SupplierView,
CountryView,
DepartmentView,
MunicipalityView,
SeedGeographyView,
)
__all__ = [
# Catalogue Images
@@ -25,6 +36,8 @@ __all__ = [
# Products
"ProductView",
"ProductsFromTrytonView",
# Product Categories
"ProductCategoryView",
# Customers
"CustomerView",
"CustomersFromTrytonView",
@@ -33,6 +46,7 @@ __all__ = [
"CatalogSaleView",
"SaleSummary",
"CatalogSaleSummary",
"PublicSaleSummaryView",
"SalesForTrytonView",
"SalesToTrytonView",
"CatalogSalesToTrytonView",
@@ -44,4 +58,13 @@ __all__ = [
"Pagination",
# Admin
"AdminCodeValidateView",
# Store Settings
"StoreSettingsView",
# Provenance
"OrganizationView",
"SupplierView",
"CountryView",
"DepartmentView",
"MunicipalityView",
"SeedGeographyView",
]

View File

@@ -1,5 +1,5 @@
from rest_framework import viewsets
from rest_framework.permissions import IsAuthenticated
from rest_framework.permissions import AllowAny, IsAuthenticated
from ..image_service import resize_catalogue_image
from ..models.catalogue_images import CatalogueImage
@@ -14,7 +14,7 @@ class CatalogueImageViewSet(viewsets.ModelViewSet):
def get_permissions(self):
if self.action in ("create", "update", "partial_update", "destroy"):
return [IsAuthenticated(), IsAdministrator()]
return [IsAuthenticated()]
return [AllowAny()]
def perform_create(self, serializer):
instance = serializer.save()

View File

@@ -0,0 +1,85 @@
from rest_framework import viewsets
from rest_framework.response import Response
from rest_framework.decorators import action
from rest_framework.permissions import IsAuthenticated
from ..models.products import ProductCategory
from ..serializers import ProductCategorySerializer
from ..permissions import IsAdministrator, IsNotPublico
from ..services.tryton.categories import CategoryTrytonService
from ..services.tryton.client import get_tryton_client
class ProductCategoryView(viewsets.ModelViewSet):
queryset = ProductCategory.objects.all()
serializer_class = ProductCategorySerializer
def create(self, request):
return Response(status=405)
def destroy(self, request, pk=None):
return Response(status=405)
def get_permissions(self):
if self.action in ("list", "retrieve"):
return [IsNotPublico()]
return [IsAuthenticated(), IsAdministrator()]
def get_queryset(self):
"""
Filters categories by active status for list operations.
Query params for list:
- active=true (default): Only active categories
- active=false: Only inactive categories
- active=all: All categories regardless of status
"""
queryset = ProductCategory.objects.all()
if self.action != "list":
return queryset
active_param = self.request.query_params.get("active", "true")
if active_param.lower() == "all":
return queryset
elif active_param.lower() in ["true", "1", "yes"]:
return queryset.filter(active=True)
elif active_param.lower() in ["false", "0", "no"]:
return queryset.filter(active=False)
else:
return queryset.filter(active=True)
@action(detail=False, methods=["post"])
def import_from_external(self, request):
tryton_client = get_tryton_client()
service = CategoryTrytonService(tryton_client)
return Response(service.import_from_external(), status=200)
@action(detail=False, methods=["post"])
def deactivate(self, request):
return self._change_active(request, active=False)
@action(detail=False, methods=["post"])
def activate(self, request):
return self._change_active(request, active=True)
def _change_active(self, request, active):
"""Activa o desactiva categorías por ids. Ignora ids inexistentes."""
key = "activated" if active else "deactivated"
affected = []
failed = []
for category_id in request.data.get("ids", []):
try:
category = ProductCategory.objects.get(pk=category_id)
except ProductCategory.DoesNotExist:
continue
except Exception as e:
failed.append({"id": category_id, "error": str(e)})
continue
try:
category.active = active
category.save()
affected.append({"id": category.id, "name": category.name})
except Exception as e:
failed.append({"id": category_id, "error": str(e)})
return Response({key: affected, "failed": failed})

View File

@@ -5,7 +5,7 @@ from rest_framework.permissions import IsAuthenticated
from ..models.customers import Customer
from ..serializers import CustomerSerializer
from ..permissions import IsAdministrator
from ..permissions import IsAdministrator, IsNotPublico
from ..services.tryton.customers import CustomerTrytonService
from ..services.tryton.client import get_tryton_client
@@ -13,6 +13,7 @@ from ..services.tryton.client import get_tryton_client
class CustomerView(viewsets.ModelViewSet):
queryset = Customer.objects.all()
serializer_class = CustomerSerializer
permission_classes = [IsNotPublico]
class CustomersFromTrytonView(APIView):

View File

@@ -3,7 +3,7 @@ from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.status import HTTP_400_BAD_REQUEST
from rest_framework.pagination import PageNumberPagination
from rest_framework.permissions import IsAuthenticated
from rest_framework.permissions import AllowAny, IsAuthenticated
from decimal import Decimal
from ..models.sales import Sale
@@ -87,6 +87,8 @@ class ReconciliateJarModelView(viewsets.ModelViewSet):
class PaymentMethodView(APIView):
permission_classes = [AllowAny]
def get(self, request):
serializer = PaymentMethodSerializer(PaymentMethods.choices, many=True)
return Response(serializer.data)

View File

@@ -1,11 +1,11 @@
from rest_framework import viewsets
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated
from rest_framework.permissions import AllowAny, IsAuthenticated
from ..models.products import Product
from ..serializers import ProductSerializer
from ..permissions import IsAdministrator
from ..permissions import IsAdministrator, IsNotPublico
from ..services.tryton.products import ProductTrytonService
from ..services.tryton.client import get_tryton_client
@@ -14,6 +14,11 @@ class ProductView(viewsets.ModelViewSet):
queryset = Product.objects.all()
serializer_class = ProductSerializer
def get_permissions(self):
if self.action in ("list", "retrieve"):
return [AllowAny()]
return [IsNotPublico()]
def get_queryset(self):
"""
Filters products by active status for list operations.

View File

@@ -0,0 +1,122 @@
from rest_framework import viewsets
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated
from ..models.geography import Country, Department, Municipality
from ..models.provenance import Organization, Supplier
from ..serializers import (
CountrySerializer,
DepartmentSerializer,
MunicipalitySerializer,
OrganizationSerializer,
SupplierSerializer,
)
from ..permissions import IsAdministrator, IsNotPublico
class _AdminWriteMixin:
def get_permissions(self):
if self.action in ("list", "retrieve"):
return [IsNotPublico()]
return [IsAuthenticated(), IsAdministrator()]
class OrganizationView(_AdminWriteMixin, viewsets.ModelViewSet):
queryset = Organization.objects.all()
serializer_class = OrganizationSerializer
class SupplierView(_AdminWriteMixin, viewsets.ModelViewSet):
queryset = Supplier.objects.select_related(
"organization", "municipality"
)
serializer_class = SupplierSerializer
class CountryView(_AdminWriteMixin, viewsets.ModelViewSet):
queryset = Country.objects.all()
serializer_class = CountrySerializer
class DepartmentView(_AdminWriteMixin, viewsets.ModelViewSet):
queryset = Department.objects.select_related("country")
serializer_class = DepartmentSerializer
class MunicipalityView(_AdminWriteMixin, viewsets.ModelViewSet):
queryset = Municipality.objects.select_related(
"department", "country"
)
serializer_class = MunicipalitySerializer
class SeedGeographyView(APIView):
permission_classes = [IsAuthenticated, IsAdministrator]
def post(self, request):
country_data = request.data.get("country") or {}
country_name = country_data.get("name")
if not country_name:
return Response(
{"detail": "country.name is required"}, status=400
)
country, _ = Country.objects.get_or_create(
name=country_name,
defaults={"code": country_data.get("code", "")},
)
departments_created = 0
departments = {}
for department_data in request.data.get("departments", []):
department, created = Department.objects.get_or_create(
name=department_data["name"],
defaults={"country": country},
)
departments[department.name] = department
if created:
departments_created += 1
municipalities_created = 0
for municipality_data in request.data.get(
"municipalities", []
):
department_name = municipality_data["department"]
department = departments.get(department_name)
if department is None:
department, created = Department.objects.get_or_create(
name=department_name, defaults={"country": country}
)
departments[department.name] = department
if created:
departments_created += 1
municipality, created = Municipality.objects.get_or_create(
name=municipality_data["name"],
department=department,
defaults={"country": country},
)
if created:
municipalities_created += 1
latitude = municipality_data.get("latitude")
longitude = municipality_data.get("longitude")
if latitude is not None or longitude is not None:
municipality.country = country
if latitude is not None:
municipality.latitude = latitude
if longitude is not None:
municipality.longitude = longitude
municipality.save()
return Response(
{
"country": country.name,
"departments_created": departments_created,
"municipalities_created": municipalities_created,
"total_departments": Department.objects.count(),
"total_municipalities": Municipality.objects.count(),
},
status=200,
)

View File

@@ -1,11 +1,17 @@
from rest_framework import viewsets
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated
from rest_framework.permissions import AllowAny, IsAuthenticated
from rest_framework.decorators import action
import io
import csv
from ..models.sales import Sale, SaleLine, CatalogSale
from ..models.sales import (
Sale,
SaleLine,
CatalogSale,
generate_order_code,
)
from ..models.customers import Customer
from ..models.products import Product
from ..serializers import (
@@ -14,15 +20,32 @@ from ..serializers import (
SaleSummarySerializer,
CatalogSaleSummarySerializer,
)
from ..permissions import IsAdministrator
from ..permissions import IsAdministrator, IsNotPublico
from ..services.tryton.sales import SaleTrytonService
from ..services.tryton.client import get_tryton_client
from ..views import sales_to_tryton_csv
def _fill_codes(Model):
rows = Model.objects.all()
for row in rows:
row.code = generate_order_code()
Model.objects.bulk_update(rows, ["code"])
return len(rows)
class SaleView(viewsets.ModelViewSet):
queryset = Sale.objects.all()
serializer_class = SaleSerializer
permission_classes = [IsNotPublico]
@action(
detail=False,
methods=["post"],
permission_classes=[IsAuthenticated, IsAdministrator],
)
def fill_codes(self, request):
return Response({"sales_updated": _fill_codes(Sale)})
def create(self, request):
data = request.data
@@ -59,21 +82,77 @@ class CatalogSaleView(viewsets.ModelViewSet):
queryset = CatalogSale.objects.all()
serializer_class = CatalogSaleSerializer
@action(
detail=False,
methods=["post"],
permission_classes=[IsAuthenticated, IsAdministrator],
)
def fill_codes(self, request):
return Response({"catalog_sales_updated": _fill_codes(CatalogSale)})
def get_permissions(self):
if self.action == "fill_codes":
return [IsAuthenticated(), IsAdministrator()]
if self.action == "create":
return [IsAuthenticated()]
return [IsNotPublico()]
class SaleSummary(APIView):
permission_classes = [IsNotPublico]
def get(self, request, id):
sale = Sale.objects.get(pk=id)
serializer = SaleSummarySerializer(sale)
serializer = SaleSummarySerializer(
sale, context={"request": request}
)
return Response(serializer.data)
class CatalogSaleSummary(APIView):
permission_classes = [IsAuthenticated]
def get(self, request, id):
catalog_sale = CatalogSale.objects.get(pk=id)
serializer = CatalogSaleSummarySerializer(catalog_sale)
serializer = CatalogSaleSummarySerializer(
catalog_sale, context={"request": request}
)
return Response(serializer.data)
class PublicSaleSummaryView(APIView):
permission_classes = [AllowAny]
def get(self, request, code=None):
code = code or request.query_params.get("code")
if not code:
return Response(
{"detail": "code is required"}, status=400
)
sale = Sale.objects.filter(code=code).first()
if sale:
serializer = SaleSummarySerializer(
sale, context={"request": request}
)
data = serializer.data
data["type"] = "sale"
return Response(data)
catalog_sale = CatalogSale.objects.filter(code=code).first()
if catalog_sale:
serializer = CatalogSaleSummarySerializer(
catalog_sale, context={"request": request}
)
data = serializer.data
data["type"] = "catalog"
return Response(data)
return Response(
{"detail": "order not found"}, status=404
)
class SalesForTrytonView(APIView):
permission_classes = [IsAuthenticated, IsAdministrator]

View File

@@ -0,0 +1,32 @@
from rest_framework import status
from rest_framework.permissions import AllowAny, IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView
from ..models.store_settings import StoreSettings
from ..permissions import IsAdministrator
from ..serializers.store_settings import StoreSettingsSerializer
class StoreSettingsView(APIView):
def get_permissions(self):
if self.request.method == "GET":
return [AllowAny()]
return [IsAuthenticated(), IsAdministrator()]
def get(self, request):
settings = StoreSettings.get_singleton()
return Response(
StoreSettingsSerializer(
settings, context={"request": request}
).data
)
def patch(self, request):
settings = StoreSettings.get_singleton()
serializer = StoreSettingsSerializer(
settings, data=request.data, partial=True, context={"request": request}
)
serializer.is_valid(raise_exception=True)
serializer.save()
return Response(serializer.data, status=status.HTTP_200_OK)

View File

@@ -0,0 +1,24 @@
# Generated by Django 5.0.6 on 2026-08-08 18:24
import don_confiao.models.sales
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('don_confiao', '0049_catalogueimage'),
]
operations = [
migrations.AddField(
model_name='catalogsale',
name='code',
field=models.CharField(default=don_confiao.models.sales.generate_order_code, editable=False, max_length=32),
),
migrations.AddField(
model_name='sale',
name='code',
field=models.CharField(default=don_confiao.models.sales.generate_order_code, editable=False, max_length=32),
),
]

View File

@@ -0,0 +1,27 @@
# Generated by Django 5.0.6 on 2026-08-08 17:33
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('don_confiao', '0049_catalogueimage'),
]
operations = [
migrations.CreateModel(
name='StoreSettings',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('address', models.TextField()),
('latitude', models.FloatField(blank=True, null=True)),
('longitude', models.FloatField(blank=True, null=True)),
('updated_at', models.DateTimeField(auto_now=True)),
],
options={
'verbose_name': 'Configuración de la tienda',
'verbose_name_plural': 'Configuración de la tienda',
},
),
]

View File

@@ -0,0 +1,18 @@
# Generated by Django 5.0.6 on 2026-08-08 19:10
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('don_confiao', '0050_storesettings'),
]
operations = [
migrations.AddField(
model_name='storesettings',
name='logo',
field=models.ImageField(blank=True, null=True, upload_to='logos/'),
),
]

View File

@@ -0,0 +1,14 @@
# Generated by Django 5.0.6 on 2026-08-10 03:51
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
('don_confiao', '0050_catalogsale_code_sale_code'),
('don_confiao', '0051_storesettings_logo'),
]
operations = [
]

View File

@@ -0,0 +1,101 @@
# Generated by Django 5.0.6 on 2026-08-15 21:17
import django.db.models.deletion
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('don_confiao', '0052_merge_20260810_0351'),
]
operations = [
migrations.CreateModel(
name='Country',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('name', models.CharField(max_length=100, unique=True)),
('code', models.CharField(blank=True, max_length=3, null=True, unique=True)),
],
options={
'verbose_name': 'Country',
'verbose_name_plural': 'Countries',
'ordering': ['name'],
},
),
migrations.CreateModel(
name='Organization',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('name', models.CharField(max_length=100, unique=True)),
('description', models.TextField(blank=True, null=True)),
('website', models.CharField(blank=True, max_length=255, null=True)),
('contact_email', models.CharField(blank=True, max_length=255, null=True)),
('contact_phone', models.CharField(blank=True, max_length=100, null=True)),
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
],
options={
'verbose_name': 'Organization',
'verbose_name_plural': 'Organizations',
'ordering': ['name'],
},
),
migrations.CreateModel(
name='Department',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('name', models.CharField(max_length=100, unique=True)),
('country', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='departments', to='don_confiao.country')),
],
options={
'verbose_name': 'Department',
'verbose_name_plural': 'Departments',
'ordering': ['name'],
},
),
migrations.CreateModel(
name='Municipality',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('name', models.CharField(max_length=100)),
('country', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='municipalities', to='don_confiao.country')),
('department', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='municipalities', to='don_confiao.department')),
],
options={
'verbose_name': 'Municipality',
'verbose_name_plural': 'Municipalities',
'ordering': ['name'],
},
),
migrations.CreateModel(
name='Supplier',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('name', models.CharField(max_length=100, unique=True)),
('description', models.TextField(blank=True, null=True)),
('website', models.CharField(blank=True, max_length=255, null=True)),
('contact_email', models.CharField(blank=True, max_length=255, null=True)),
('contact_phone', models.CharField(blank=True, max_length=100, null=True)),
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
('municipality', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='suppliers', to='don_confiao.municipality')),
('organization', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='suppliers', to='don_confiao.organization')),
],
options={
'verbose_name': 'Supplier',
'verbose_name_plural': 'Suppliers',
'ordering': ['name'],
},
),
migrations.AddField(
model_name='product',
name='suppliers',
field=models.ManyToManyField(blank=True, related_name='products', to='don_confiao.supplier'),
),
migrations.AddConstraint(
model_name='municipality',
constraint=models.UniqueConstraint(fields=('name', 'department'), name='unique_municipality_name_per_department'),
),
]

View File

@@ -0,0 +1,23 @@
# Generated by Django 5.0.6 on 2026-08-16 08:29
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('don_confiao', '0053_country_organization_department_municipality_and_more'),
]
operations = [
migrations.AddField(
model_name='municipality',
name='latitude',
field=models.DecimalField(blank=True, decimal_places=7, max_digits=10, null=True),
),
migrations.AddField(
model_name='municipality',
name='longitude',
field=models.DecimalField(blank=True, decimal_places=7, max_digits=10, null=True),
),
]

View File

@@ -0,0 +1,18 @@
# Generated by Django 5.0.6 on 2026-08-22 20:33
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('don_confiao', '0054_municipality_latitude_municipality_longitude'),
]
operations = [
migrations.AddField(
model_name='productcategory',
name='external_id',
field=models.CharField(blank=True, max_length=100, null=True),
),
]

View File

@@ -0,0 +1,28 @@
# Generated by Django 5.0.6 on 2026-10-10 21:26
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('don_confiao', '0055_productcategory_external_id'),
]
operations = [
migrations.AddField(
model_name='productcategory',
name='active',
field=models.BooleanField(default=True),
),
migrations.AddField(
model_name='productcategory',
name='alternative_name',
field=models.CharField(blank=True, max_length=100, null=True),
),
migrations.AddField(
model_name='productcategory',
name='description',
field=models.TextField(blank=True, default=''),
),
]

View File

@@ -0,0 +1,12 @@
from .store_settings import StoreSettings
from .geography import Country, Department, Municipality
from .provenance import Organization, Supplier
__all__ = [
"StoreSettings",
"Country",
"Department",
"Municipality",
"Organization",
"Supplier",
]

View File

@@ -0,0 +1,60 @@
from django.db import models
class Country(models.Model):
name = models.CharField(max_length=100, unique=True)
code = models.CharField(
max_length=3, unique=True, null=True, blank=True
)
class Meta:
verbose_name = "Country"
verbose_name_plural = "Countries"
ordering = ["name"]
def __str__(self):
return self.name
class Department(models.Model):
name = models.CharField(max_length=100, unique=True)
country = models.ForeignKey(
Country, on_delete=models.PROTECT, related_name="departments"
)
class Meta:
verbose_name = "Department"
verbose_name_plural = "Departments"
ordering = ["name"]
def __str__(self):
return self.name
class Municipality(models.Model):
name = models.CharField(max_length=100)
department = models.ForeignKey(
Department, on_delete=models.PROTECT, related_name="municipalities"
)
country = models.ForeignKey(
Country, on_delete=models.PROTECT, related_name="municipalities"
)
latitude = models.DecimalField(
max_digits=10, decimal_places=7, null=True, blank=True
)
longitude = models.DecimalField(
max_digits=10, decimal_places=7, null=True, blank=True
)
class Meta:
verbose_name = "Municipality"
verbose_name_plural = "Municipalities"
ordering = ["name"]
constraints = [
models.UniqueConstraint(
fields=["name", "department"], name="unique_municipality_name_per_department"
)
]
def __str__(self):
return self.name

View File

@@ -8,6 +8,10 @@ class MeasuringUnits(models.TextChoices):
class ProductCategory(models.Model):
name = models.CharField(max_length=100, unique=True)
external_id = models.CharField(max_length=100, null=True, blank=True)
description = models.TextField(blank=True, default="")
alternative_name = models.CharField(max_length=100, null=True, blank=True)
active = models.BooleanField(default=True)
def __str__(self):
return self.name
@@ -26,6 +30,9 @@ class Product(models.Model):
max_length=100, null=True, blank=True
)
categories = models.ManyToManyField(ProductCategory)
suppliers = models.ManyToManyField(
"Supplier", blank=True, related_name="products"
)
external_id = models.CharField(max_length=100, null=True, blank=True)
def __str__(self):
@@ -42,7 +49,15 @@ class Product(models.Model):
"price_list": product.price,
"uom": product.measuring_unit,
"external_id": product.external_id,
"categories": [c.name for c in product.categories.all()],
"categories": [
{
"id": c.id,
"name": c.name,
"alternative_name": c.alternative_name,
"active": c.active,
}
for c in product.categories.all()
],
}
products_list.append(rproduct)
return products_list

View File

@@ -0,0 +1,53 @@
from django.db import models
from .geography import Municipality
class Organization(models.Model):
name = models.CharField(max_length=100, unique=True)
description = models.TextField(null=True, blank=True)
website = models.CharField(max_length=255, null=True, blank=True)
contact_email = models.CharField(max_length=255, null=True, blank=True)
contact_phone = models.CharField(max_length=100, null=True, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
verbose_name = "Organization"
verbose_name_plural = "Organizations"
ordering = ["name"]
def __str__(self):
return self.name
class Supplier(models.Model):
name = models.CharField(max_length=100, unique=True)
description = models.TextField(null=True, blank=True)
organization = models.ForeignKey(
Organization,
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name="suppliers",
)
municipality = models.ForeignKey(
Municipality,
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name="suppliers",
)
website = models.CharField(max_length=255, null=True, blank=True)
contact_email = models.CharField(max_length=255, null=True, blank=True)
contact_phone = models.CharField(max_length=100, null=True, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
verbose_name = "Supplier"
verbose_name_plural = "Suppliers"
ordering = ["name"]
def __str__(self):
return self.name

View File

@@ -4,6 +4,11 @@ from .products import Product
from .payments import PaymentMethods, ReconciliationJar
from django.core.exceptions import ValidationError
from datetime import datetime
import uuid
def generate_order_code():
return uuid.uuid4().hex
class SaleAbstractModel(models.Model):
@@ -11,6 +16,9 @@ class SaleAbstractModel(models.Model):
date = models.DateTimeField("Date")
phone = models.CharField(max_length=13, null=True, blank=True)
description = models.CharField(max_length=255, null=True, blank=True)
code = models.CharField(
max_length=32, default=generate_order_code, editable=False
)
class Meta:
abstract = True

View File

@@ -0,0 +1,21 @@
from django.db import models
class StoreSettings(models.Model):
address = models.TextField()
latitude = models.FloatField(null=True, blank=True)
longitude = models.FloatField(null=True, blank=True)
logo = models.ImageField(upload_to="logos/", null=True, blank=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
verbose_name = "Configuración de la tienda"
verbose_name_plural = "Configuración de la tienda"
def __str__(self):
return f"StoreSettings (id={self.id})"
@classmethod
def get_singleton(cls):
instance, _ = cls.objects.get_or_create(pk=1)
return instance

View File

@@ -1,6 +1,26 @@
from rest_framework.permissions import BasePermission
def _get_user_type(user):
if not user or not user.is_authenticated:
return None
profile = getattr(user, "profile", None)
if profile:
return profile.user_type
return "administrator" if user.is_staff else "user"
class IsAdministrator(BasePermission):
def has_permission(self, request, view):
return request.user and request.user.is_staff
return _get_user_type(request.user) == "administrator"
class IsNotPublico(BasePermission):
def has_permission(self, request, view):
user_type = _get_user_type(request.user)
return user_type is not None and user_type != "publico"
class IsPublico(BasePermission):
def has_permission(self, request, view):
return _get_user_type(request.user) == "publico"

View File

@@ -1,5 +1,10 @@
from .catalogue_images import CatalogueImageSerializer
from .products import ProductSerializer, ListProductSerializer
from .products import (
ProductSerializer,
ListProductSerializer,
ProductCategorySerializer,
ProductCategoryBriefSerializer,
)
from .customers import CustomerSerializer, ListCustomerSerializer
from .sales import (
SaleSerializer,
@@ -16,6 +21,9 @@ from .payments import (
ReconciliationJarSerializer,
PaymentMethodSerializer,
)
from .store_settings import StoreSettingsSerializer
from .geography import CountrySerializer, DepartmentSerializer, MunicipalitySerializer
from .provenance import OrganizationSerializer, SupplierSerializer
__all__ = [
# Catalogue Images
@@ -23,6 +31,8 @@ __all__ = [
# Products
"ProductSerializer",
"ListProductSerializer",
"ProductCategorySerializer",
"ProductCategoryBriefSerializer",
# Customers
"CustomerSerializer",
"ListCustomerSerializer",
@@ -39,4 +49,13 @@ __all__ = [
# Payments
"ReconciliationJarSerializer",
"PaymentMethodSerializer",
# Store Settings
"StoreSettingsSerializer",
# Geography
"CountrySerializer",
"DepartmentSerializer",
"MunicipalitySerializer",
# Provenance
"OrganizationSerializer",
"SupplierSerializer",
]

View File

@@ -0,0 +1,55 @@
from rest_framework import serializers
from ..models.geography import Country, Department, Municipality
class CountrySerializer(serializers.ModelSerializer):
class Meta:
model = Country
fields = ["id", "name", "code"]
class CountryBriefSerializer(serializers.ModelSerializer):
class Meta:
model = Country
fields = ["id", "name", "code"]
class DepartmentBriefSerializer(serializers.ModelSerializer):
class Meta:
model = Department
fields = ["id", "name"]
class MunicipalityBriefSerializer(serializers.ModelSerializer):
class Meta:
model = Municipality
fields = ["id", "name", "latitude", "longitude"]
class DepartmentSerializer(serializers.ModelSerializer):
country_detail = CountryBriefSerializer(source="country", read_only=True)
class Meta:
model = Department
fields = ["id", "name", "country", "country_detail"]
class MunicipalitySerializer(serializers.ModelSerializer):
department_detail = DepartmentBriefSerializer(
source="department", read_only=True
)
country_detail = CountryBriefSerializer(source="country", read_only=True)
class Meta:
model = Municipality
fields = [
"id",
"name",
"department",
"department_detail",
"country",
"country_detail",
"latitude",
"longitude",
]

View File

@@ -1,10 +1,35 @@
from rest_framework import serializers
from ..models.products import Product, ProductCategory
from ..models.provenance import Supplier
class ProductCategorySerializer(serializers.ModelSerializer):
class Meta:
model = ProductCategory
fields = [
"id",
"name",
"external_id",
"description",
"alternative_name",
"active",
]
read_only_fields = ["name", "external_id"]
class ProductCategoryBriefSerializer(serializers.ModelSerializer):
class Meta:
model = ProductCategory
fields = ["id", "name", "alternative_name", "active"]
class ProductSerializer(serializers.ModelSerializer):
catalogue_images = serializers.SerializerMethodField()
suppliers = serializers.PrimaryKeyRelatedField(
many=True, queryset=Supplier.objects.all(), required=False
)
categories = ProductCategoryBriefSerializer(many=True, read_only=True)
class Meta:
model = Product
@@ -15,6 +40,7 @@ class ProductSerializer(serializers.ModelSerializer):
"price",
"measuring_unit",
"categories",
"suppliers",
"external_id",
"catalogue_images",
]

View File

@@ -0,0 +1,79 @@
from rest_framework import serializers
from ..models.provenance import Organization, Supplier
from .geography import (
CountryBriefSerializer,
DepartmentBriefSerializer,
MunicipalityBriefSerializer,
)
class OrganizationSerializer(serializers.ModelSerializer):
class Meta:
model = Organization
fields = [
"id",
"name",
"description",
"website",
"contact_email",
"contact_phone",
"created_at",
"updated_at",
]
class OrganizationBriefSerializer(serializers.ModelSerializer):
class Meta:
model = Organization
fields = [
"id",
"name",
"description",
"website",
"contact_email",
"contact_phone",
]
class SupplierSerializer(serializers.ModelSerializer):
organization_detail = OrganizationBriefSerializer(
source="organization", read_only=True
)
municipality_detail = MunicipalityBriefSerializer(
source="municipality", read_only=True
)
products = serializers.PrimaryKeyRelatedField(
many=True, read_only=True
)
class Meta:
model = Supplier
fields = [
"id",
"name",
"description",
"organization",
"organization_detail",
"municipality",
"municipality_detail",
"website",
"contact_email",
"contact_phone",
"products",
"created_at",
"updated_at",
]
class SupplierBriefSerializer(serializers.ModelSerializer):
class Meta:
model = Supplier
fields = [
"id",
"name",
"description",
"website",
"contact_email",
"contact_phone",
]

View File

@@ -1,5 +1,7 @@
from rest_framework import serializers
from django.urls import reverse
from ..models.products import Product
from ..models.sales import (
Sale,
SaleLine,
@@ -9,6 +11,28 @@ from ..models.sales import (
)
from .products import ListProductSerializer
from .customers import ListCustomerSerializer
from ..services.provenance import build_product_provenance
class PublicSummaryLinkMixin:
def get_link(self, obj):
path = reverse(
"don_confiao:public_purchase_summary", args=[obj.code]
)
request = self.context.get("request")
if request:
return request.build_absolute_uri(path)
return path
class ProvenanceSummaryMixin:
def get_product_provenance(self, obj):
products = Product.objects.filter(
saleline__sale=obj
).distinct()
return build_product_provenance(
products, self.context.get("request")
)
class SaleLineSerializer(serializers.ModelSerializer):
@@ -24,6 +48,7 @@ class SaleSerializer(serializers.ModelSerializer):
model = Sale
fields = [
"id",
"code",
"customer",
"date",
"saleline_set",
@@ -56,6 +81,7 @@ class CatalogSaleSerializer(serializers.ModelSerializer):
model = CatalogSale
fields = [
"id",
"code",
"customer",
"date",
"catalogsaleline_set",
@@ -87,13 +113,26 @@ class SummarySaleLineSerializer(serializers.ModelSerializer):
fields = ["product", "quantity", "unit_price", "description"]
class SaleSummarySerializer(serializers.ModelSerializer):
class SaleSummarySerializer(
PublicSummaryLinkMixin, ProvenanceSummaryMixin, serializers.ModelSerializer
):
customer = ListCustomerSerializer()
lines = SummarySaleLineSerializer(many=True, source="saleline_set")
link = serializers.SerializerMethodField()
product_provenance = serializers.SerializerMethodField()
class Meta:
model = Sale
fields = ["id", "date", "customer", "payment_method", "lines"]
fields = [
"id",
"code",
"date",
"customer",
"payment_method",
"lines",
"link",
"product_provenance",
]
class CatalogSummarySaleLineSerializer(serializers.ModelSerializer):
@@ -104,15 +143,31 @@ class CatalogSummarySaleLineSerializer(serializers.ModelSerializer):
fields = ["product", "quantity", "unit_price", "description"]
class CatalogSaleSummarySerializer(serializers.ModelSerializer):
class CatalogProvenanceSummaryMixin(ProvenanceSummaryMixin):
def get_product_provenance(self, obj):
products = Product.objects.filter(
catalogsaleline__catalog_sale=obj
).distinct()
return build_product_provenance(
products, self.context.get("request")
)
class CatalogSaleSummarySerializer(
PublicSummaryLinkMixin,
CatalogProvenanceSummaryMixin,
serializers.ModelSerializer,
):
customer = ListCustomerSerializer()
lines = CatalogSummarySaleLineSerializer(
many=True, source="catalogsaleline_set"
)
link = serializers.SerializerMethodField()
product_provenance = serializers.SerializerMethodField()
class Meta:
model = CatalogSale
fields = ["id", "date", "customer", "lines"]
fields = ["id", "code", "date", "customer", "lines", "link", "product_provenance"]
class SaleForRenconciliationSerializer(serializers.Serializer):

View File

@@ -0,0 +1,22 @@
from django.conf import settings
from rest_framework import serializers
from ..models.store_settings import StoreSettings
class StoreSettingsSerializer(serializers.ModelSerializer):
class Meta:
model = StoreSettings
fields = ["id", "address", "latitude", "longitude", "logo", "updated_at"]
read_only_fields = ["id", "updated_at"]
def validate_logo(self, value):
if value is None:
return value
max_size = settings.LOGO_MAX_UPLOAD_SIZE
if value.size > max_size:
raise serializers.ValidationError(
f"Logo size exceeds the maximum allowed size of "
f"{max_size // (1024 * 1024)}MB."
)
return value

View File

@@ -0,0 +1,66 @@
from ..serializers.geography import (
CountryBriefSerializer,
DepartmentBriefSerializer,
MunicipalityBriefSerializer,
)
from ..serializers.provenance import (
OrganizationBriefSerializer,
SupplierBriefSerializer,
)
from ..serializers.products import ListProductSerializer
def build_product_provenance(products, request=None):
"""Domain data for provenance charts.
Returns a list of products, each with its suppliers. Each supplier
entry includes the linked organization and the territorial hierarchy
(municipality -> department -> country) derived from the supplier's
municipality.
"""
result = []
for product in products:
suppliers = []
supplier_qs = product.suppliers.all().select_related(
"organization", "municipality__department__country"
)
for supplier in supplier_qs:
municipality = supplier.municipality
entry = {
"supplier": SupplierBriefSerializer(supplier).data,
"organization": (
OrganizationBriefSerializer(supplier.organization).data
if supplier.organization
else None
),
"municipality": (
MunicipalityBriefSerializer(municipality).data
if municipality
else None
),
"department": (
DepartmentBriefSerializer(
municipality.department
).data
if municipality
else None
),
"country": (
CountryBriefSerializer(
municipality.department.country
).data
if municipality
else None
),
}
suppliers.append(entry)
result.append(
{
"product": ListProductSerializer(
product, context={"request": request}
).data,
"suppliers": suppliers,
}
)
return result

View File

@@ -0,0 +1,147 @@
from ...models.products import ProductCategory
def upsert_category(category_data):
"""Crea o actualiza una ProductCategory desde datos de Tryton.
Solo administra name y external_id (fuente Tryton). Preserva los
campos administrados por Don Confiao (description, alternative_name
y active). Devuelve (category, status, changes) donde status es
'created', 'updated' o 'untouched' y changes es la lista de cambios.
"""
external_id = str(category_data.get("id"))
name = category_data.get("name")
category = ProductCategory.objects.filter(external_id=external_id).first()
if category is None and name:
category = ProductCategory.objects.filter(name=name).first()
is_new = category is None
if is_new:
category = ProductCategory()
changes = []
if category.name != name:
changes.append(f"Nombre: {category.name} → {name}")
if category.external_id != external_id:
changes.append(
f"External ID: {category.external_id} → {external_id}"
)
category.name = name
category.external_id = external_id
category.save()
status = "created" if is_new else ("updated" if changes else "untouched")
return category, status, changes
class CategoryTrytonService:
"""Servicio para sincronización de categorías con Tryton"""
def __init__(self, tryton_client):
self.client = tryton_client
def import_from_external(self):
"""Importa desde Tryton las categorías asociadas a productos.
Lee las categorías referenciadas por los productos salables
(sin suponer eliminación de categorías locales). Reporta las
categorías revisadas, creadas, actualizadas, intactas y fallidas.
"""
context = {"company": 1}
product_ids = self._search_salable_products(context)
if not product_ids:
return self._empty_report()
tryton_products = self._get_product_details(product_ids, context)
category_ids = set()
for tryton_product in tryton_products:
category_ids.update(self._extract_category_ids(tryton_product))
if not category_ids:
return self._empty_report()
tryton_categories = self._get_category_details(
sorted(category_ids), context
)
checked = [
{"id": c.get("id"), "name": c.get("name")}
for c in tryton_categories
]
created_categories = []
updated_categories = []
untouched_categories = []
failed_categories = []
for tryton_category in tryton_categories:
try:
category, status, changes = upsert_category(tryton_category)
item = {
"id": category.id,
"name": category.name,
"external_id": category.external_id,
}
if status == "created":
created_categories.append(item)
elif status == "updated":
updated_categories.append(
{**item, "detail": ", ".join(changes)}
)
else:
untouched_categories.append(item)
except Exception as e:
print(
f"Error al importar categoría: {e}"
f"La categoría: {tryton_category}"
)
failed_categories.append(
{
"external_id": tryton_category.get("id"),
"name": tryton_category.get("name"),
"error": f"Error al importar la categoría: {e}",
}
)
return {
"checked_tryton_categories": checked,
"created_categories": created_categories,
"updated_categories": updated_categories,
"untouched_categories": untouched_categories,
"failed_categories": failed_categories,
}
def _search_salable_products(self, context):
method = "model.product.product.search"
params = [
[["salable", "=", True]],
0,
1000,
[["rec_name", "ASC"], ["id", None]],
context,
]
return self.client.call(method, params) or []
def _get_product_details(self, product_ids, context):
tryton_fields = ["id", "name", "template.categories"]
method = "model.product.product.read"
params = (product_ids, tryton_fields, context)
return self.client.call(method, params)
@staticmethod
def _extract_category_ids(tryton_product):
"""Extrae los IDs de categorías de la respuesta RPC de Tryton"""
template = tryton_product.get("template.") or {}
return template.get("categories") or []
def _get_category_details(self, category_ids, context):
method = "model.product.category.read"
params = (category_ids, ["id", "name"], context)
return self.client.call(method, params)
def _empty_report(self):
return {
"checked_tryton_categories": [],
"created_categories": [],
"updated_categories": [],
"untouched_categories": [],
"failed_categories": [],
}

View File

@@ -43,20 +43,28 @@ class TrytonSale:
"shipment_address": self.sale.customer.address_external_id,
"invoice_address": self.sale.customer.address_external_id,
"currency": TRYTON_COP_CURRENCY,
"comment": self.sale.description or "",
"description": "Metodo pago: " + str(self.sale.payment_method or ""),
"comment": self.build_tryton_comment(),
"description": "Metodo pago: "
+ str(self.sale.payment_method or ""),
"party": self.sale.customer.external_id,
"reference": "don_confiao " + str(self.sale.id),
"sale_date": self._format_date(self.sale.date),
"lines": [
[
"create",
[TrytonLineSale(line).to_tryton() for line in self.lines],
[
TrytonLineSale(line).to_tryton()
for line in self.lines
],
]
],
"self_pick_up": True,
}
def build_tryton_comment(self):
"""Construye el comentario para la venta en Tryton"""
return self.sale.description or ""
class TrytonLineSale:
"""Representa una línea de venta para exportación a Tryton"""
@@ -98,20 +106,36 @@ class TrytonCatalogSale:
"shipment_address": self.catalog_sale.customer.address_external_id,
"invoice_address": self.catalog_sale.customer.address_external_id,
"currency": TRYTON_COP_CURRENCY,
"comment": self.catalog_sale.description or "",
"comment": self.build_tryton_comment(),
"description": "Venta de catálogo",
"party": self.catalog_sale.customer.external_id,
"reference": "don_confiao_catalog " + str(self.catalog_sale.id),
"reference": "don_confiao_catalog "
+ str(self.catalog_sale.id),
"sale_date": self._format_date(self.catalog_sale.date),
"lines": [
[
"create",
[TrytonCatalogSaleLine(line).to_tryton() for line in self.lines],
[
TrytonCatalogSaleLine(line).to_tryton()
for line in self.lines
],
]
],
"self_pick_up": True,
}
def build_tryton_comment(self):
"""Construye el comentario para la venta en Tryton"""
comment = f"Cliente: {self.catalog_sale.customer_name or ''}\n"
comment += f"Teléfono: {self.catalog_sale.customer_phone or ''}\n"
comment += (
f"Dirección: {self.catalog_sale.customer_address or ''}\n"
)
comment += f"Método de recogida: {self.catalog_sale.pickup_method or ''}\n"
return comment
class TrytonCatalogSaleLine:
"""Representa una línea de catalog sale para exportación a Tryton"""
@@ -125,8 +149,12 @@ class TrytonCatalogSaleLine:
def to_tryton(self):
return {
"product": self.catalog_sale_line.product.external_id,
"quantity": self._format_decimal(self.catalog_sale_line.quantity),
"quantity": self._format_decimal(
self.catalog_sale_line.quantity
),
"type": "line",
"unit": self.catalog_sale_line.product.unit_external_id,
"unit_price": self._format_decimal(self.catalog_sale_line.unit_price),
"unit_price": self._format_decimal(
self.catalog_sale_line.unit_price
),
}

View File

@@ -15,7 +15,10 @@ class CustomerTrytonService:
party_ids = self.client.call(method, params)
tryton_parties = self._get_party_details(party_ids, context)
checked_tryton_parties = party_ids
checked_tryton_parties = [
{"id": p.get("id"), "name": p.get("name")}
for p in tryton_parties
]
failed_parties = []
updated_customers = []
created_customers = []
@@ -25,15 +28,35 @@ class CustomerTrytonService:
try:
customer = Customer.objects.get(external_id=tryton_party.get("id"))
except Customer.DoesNotExist:
customer = self._create_customer(tryton_party)
created_customers.append(customer.id)
continue
if self._need_update(customer, tryton_party):
self._update_customer(customer, tryton_party)
updated_customers.append(customer.id)
else:
untouched_customers.append(customer.id)
try:
customer = self._create_customer(tryton_party)
created_customers.append(self._build_item(customer))
continue
except Exception as e:
print(
f"Error al importar clientes: {e}El cliente: {tryton_party}"
)
failed_parties.append(
self._build_failure(tryton_party, "crear", e)
)
continue
try:
if self._need_update(customer, tryton_party):
changes = self._describe_changes(customer, tryton_party)
self._update_customer(customer, tryton_party)
updated_customers.append(
{**self._build_item(customer), "detail": changes}
)
else:
untouched_customers.append(self._build_item(customer))
except Exception as e:
print(
f"Error al importar clientes: {e}El cliente: {tryton_party}"
)
failed_parties.append(
self._build_failure(tryton_party, "actualizar", e)
)
return {
"checked_tryton_parties": checked_tryton_parties,
@@ -43,6 +66,41 @@ class CustomerTrytonService:
"untouched_customers": untouched_customers,
}
def _build_item(self, customer):
"""Construye el objeto de un cliente sincronizado"""
return {"id": customer.id, "name": customer.name}
def _describe_changes(self, customer, tryton_party):
"""Describe los cambios detectados entre el cliente local y Tryton"""
changes = []
name = tryton_party.get("name")
if customer.name != name:
changes.append(f"Nombre: {customer.name} → {name}")
if tryton_party.get("addresses") and tryton_party.get("addresses")[0]:
new_address = str(tryton_party.get("addresses")[0])
if customer.address_external_id != new_address:
changes.append(
f"Dirección: {customer.address_external_id} → {new_address}"
)
return ", ".join(changes)
def _build_failure(self, tryton_party, operation, exc):
"""Construye el detalle de un cliente que falló al sincronizar"""
error = self._friendly_error(tryton_party, str(exc))
return {
"external_id": tryton_party.get("id"),
"name": tryton_party.get("name"),
"error": f"Error al {operation} cliente: {error}",
}
def _friendly_error(self, tryton_party, message):
"""Traduce causas conocidas de error a mensajes legibles"""
if "name" in message and (
"not-null constraint" in message or "NOT NULL constraint" in message
):
return "El cliente no tiene nombre en Tryton"
return message
def _get_party_details(self, party_ids, context):
"""Obtiene detalles de clientes desde Tryton"""
tryton_fields = ["id", "name", "addresses"]

View File

@@ -1,4 +1,5 @@
from ...models.products import Product
from ...models.products import Product, ProductCategory
from .categories import upsert_category
class ProductTrytonService:
@@ -20,8 +21,24 @@ class ProductTrytonService:
]
product_ids = self.client.call(method, params)
tryton_products = self._get_product_details(product_ids, context)
checked_tryton_products = product_ids
try:
(
categories_by_tryton_id,
category_report,
) = self._import_categories(tryton_products, context)
created_categories = category_report["created"]
updated_categories = category_report["updated"]
except Exception as e:
print(f"Error al importar categorías: {e}")
categories_by_tryton_id = {}
created_categories = []
updated_categories = []
checked_tryton_products = [
{"id": p.get("id"), "name": p.get("name")}
for p in tryton_products
]
failed_products = []
updated_products = []
created_products = []
@@ -29,24 +46,50 @@ class ProductTrytonService:
for tryton_product in tryton_products:
try:
product = Product.objects.get(external_id=tryton_product.get("id"))
product = Product.objects.get(
external_id=tryton_product.get("id")
)
except Product.DoesNotExist:
try:
product = self._create_product(tryton_product)
created_products.append(product.id)
self._sync_product_categories(
product, tryton_product, categories_by_tryton_id
)
created_products.append(self._build_item(product))
continue
except Exception as e:
print(
f"Error al importar productos: {e}El producto: {tryton_product}"
)
failed_products.append(tryton_product.get("id"))
failed_products.append(
self._build_failure(tryton_product, "crear", e)
)
continue
if self._need_update(product, tryton_product):
self._update_product(product, tryton_product)
updated_products.append(product.id)
else:
untouched_products.append(product.id)
try:
if self._need_update(product, tryton_product):
changes = self._describe_changes(
product, tryton_product
)
self._update_product(product, tryton_product)
self._sync_product_categories(
product, tryton_product, categories_by_tryton_id
)
updated_products.append(
{**self._build_item(product), "detail": changes}
)
else:
self._sync_product_categories(
product, tryton_product, categories_by_tryton_id
)
untouched_products.append(self._build_item(product))
except Exception as e:
print(
f"Error al importar productos: {e}El producto: {tryton_product}"
)
failed_products.append(
self._build_failure(tryton_product, "actualizar", e)
)
return {
"checked_tryton_products": checked_tryton_products,
@@ -54,8 +97,171 @@ class ProductTrytonService:
"updated_products": updated_products,
"created_products": created_products,
"untouched_products": untouched_products,
"created_categories": created_categories,
"updated_categories": updated_categories,
}
def _import_categories(self, tryton_products, context):
"""Importa las categorías de Tryton usadas por los productos
sincronizados. Devuelve un mapa {id_tryton: ProductCategory} y un
reporte {"created": [...], "updated": [...]}
"""
empty_report = {"created": [], "updated": []}
category_ids = {
category_id
for tryton_product in tryton_products
for category_id in self._extract_category_ids(tryton_product)
}
if not category_ids:
return {}, empty_report
categories = {}
report = {"created": [], "updated": []}
tryton_categories = self._get_category_details(
sorted(category_ids), context
)
for tryton_category in tryton_categories:
try:
category, status, changes = upsert_category(tryton_category)
categories[tryton_category.get("id")] = category
item = {
"id": category.id,
"name": category.name,
"external_id": category.external_id,
}
if status == "created":
report["created"].append(item)
elif status == "updated":
report["updated"].append(
{**item, "detail": ", ".join(changes)}
)
except Exception as e:
print(
f"Error al importar categoría: {e}"
f"La categoría: {tryton_category}"
)
return categories, report
@staticmethod
def _extract_category_ids(tryton_product):
"""Extrae los IDs de categorías de la respuesta RPC de Tryton
(los campos con puntos llegan agrupados bajo la clave 'template.')"""
template = tryton_product.get("template.") or {}
return template.get("categories") or []
def _get_category_details(self, category_ids, context):
"""Obtiene los nombres de categorías desde Tryton"""
method = "model.product.category.read"
params = (category_ids, ["id", "name"], context)
return self.client.call(method, params)
def _sync_product_categories(self, product, tryton_product, categories):
"""Vincula al producto las categorías importadas desde Tryton"""
category_ids = self._extract_category_ids(tryton_product)
product.categories.set(
[
categories[category_id]
for category_id in category_ids
if category_id in categories
]
)
def _build_item(self, product):
"""Construye el objeto de un producto sincronizado"""
return {"id": product.id, "name": product.name}
def _describe_changes(self, product, tryton_product):
"""Describe los cambios detectados entre el producto local y Tryton"""
changes = []
name = tryton_product.get("name")
if product.name != name:
changes.append(f"Nombre: {product.name} → {name}")
price = tryton_product.get("list_price")
if product.price != price:
changes.append(
f"Precio: {self._format_price(product.price)} → "
f"{self._format_price(price)}"
)
unit = tryton_product.get("default_uom.")
if unit:
unit_name = unit.get("rec_name")
if product.measuring_unit != unit_name:
changes.append(
f"Unidad: {product.measuring_unit} → {unit_name}"
)
category_changes = self._describe_category_changes(
product, tryton_product
)
if category_changes:
changes.append(category_changes)
return ", ".join(changes)
def _describe_category_changes(self, product, tryton_product):
"""Describe los cambios de categorías entre el producto local y Tryton"""
tryton_ids = set(
str(category_id)
for category_id in self._extract_category_ids(tryton_product)
)
local = {
external_id: name
for external_id, name in product.categories.values_list(
"external_id", "name"
)
if external_id
}
local_ids = set(local)
added = sorted(tryton_ids - local_ids)
removed = sorted(local_ids - tryton_ids)
if not added and not removed:
return ""
parts = []
if added:
names = {}
for external_id in added:
category = (
ProductCategory.objects.filter(
external_id=external_id
)
.values_list("name", flat=True)
.first()
)
names[external_id] = category or external_id
parts.append(", ".join(f"+{names[e]}" for e in added))
if removed:
parts.append(", ".join(f"-{local[e]}" for e in removed))
return f"Categoría: {', '.join(parts)}"
@staticmethod
def _format_price(value):
"""Formatea un precio sin ceros decimales innecesarios"""
return f"{value:.2f}".rstrip("0").rstrip(".")
def _build_failure(self, tryton_product, operation, exc):
"""Construye el detalle de un producto que falló al sincronizar"""
error = self._friendly_error(tryton_product, str(exc))
return {
"external_id": tryton_product.get("id"),
"name": tryton_product.get("name"),
"error": f"Error al {operation} producto: {error}",
}
def _friendly_error(self, tryton_product, message):
"""Traduce causas conocidas de error a mensajes legibles"""
if "price" in message and (
"not-null constraint" in message
or "NOT NULL constraint" in message
):
return (
"El producto no tiene precio en Tryton (list_price nulo)"
)
if "name" in message and (
"duplicate key" in message or "UNIQUE constraint" in message
):
return f"Ya existe un producto con el nombre '{tryton_product.get('name')}'"
return message
def _get_product_details(self, product_ids, context):
"""Obtiene detalles de productos desde Tryton"""
tryton_fields = [
@@ -64,6 +270,7 @@ class ProductTrytonService:
"default_uom.id",
"default_uom.rec_name",
"list_price",
"template.categories",
]
method = "model.product.product.read"
params = (product_ids, tryton_fields, context)
@@ -79,6 +286,17 @@ class ProductTrytonService:
unit = tryton_product.get("default_uom.")
if not product.measuring_unit == unit.get("rec_name"):
return True
tryton_category_ids = set(
str(category_id)
for category_id in self._extract_category_ids(tryton_product)
)
local_category_ids = set(
product.categories.exclude(
external_id__isnull=True
).values_list("external_id", flat=True)
)
if tryton_category_ids != local_category_ids:
return True
return False
def _create_product(self, tryton_product):

View File

@@ -27,10 +27,10 @@ class SaleTrytonService:
external_ids = self.client.call(method, tryton_params)
sale.external_id = external_ids[0]
sale.save()
successful.append(sale.id)
successful.append(self._build_item(sale))
except Exception as e:
print(f"Error al enviar la venta: {e}venta_id: {sale.id}")
failed.append(sale.id)
failed.append(self._build_failure(sale, e))
continue
return {"successful": successful, "failed": failed}
@@ -61,16 +61,28 @@ class SaleTrytonService:
external_ids = self.client.call(method, tryton_params)
catalog_sale.external_id = external_ids[0]
catalog_sale.save()
successful.append(catalog_sale.id)
successful.append(self._build_item(catalog_sale))
except Exception as e:
print(
f"Error al enviar catalog sale: {e}, catalog_sale_id: {catalog_sale.id}"
)
failed.append(catalog_sale.id)
failed.append(self._build_failure(catalog_sale, e))
continue
return {"successful": successful, "failed": failed}
def _build_item(self, sale):
"""Construye el objeto de una venta sincronizada"""
return {"id": sale.id, "code": sale.code}
def _build_failure(self, sale, exc):
"""Construye el detalle de una venta que falló al sincronizar"""
return {
"id": sale.id,
"code": sale.code,
"error": f"Error al enviar la venta: {exc}",
}
def _catalog_sale_to_tryton_params(self, catalog_sale, lines, tryton_context):
"""Convierte catalog sale a parámetros para Tryton"""
sale_tryton = TrytonCatalogSale(catalog_sale, lines)

View File

@@ -0,0 +1,200 @@
from unittest.mock import patch
from django.test import TestCase
from ..models.products import ProductCategory
from ..services.tryton.categories import CategoryTrytonService
class TestCategoryTrytonImport(TestCase):
def _service(self, mock_call):
fake_client = type("FakeClient", (), {"call": mock_call})()
return CategoryTrytonService(fake_client)
def _empty_report(self):
return {
"checked_tryton_categories": [],
"created_categories": [],
"updated_categories": [],
"untouched_categories": [],
"failed_categories": [],
}
@patch("sabatron_tryton_rpc_client.client.Client.call")
def test_import_creates_only_referenced_categories(self, mock_call):
def fake_call(*args, **kwargs):
if args[0] == "model.product.product.search":
return [1, 2]
if args[0] == "model.product.product.read":
return [
{
"id": 1,
"name": "Producto 1",
"template.": {"id": 999, "categories": [6]},
},
{
"id": 2,
"name": "Producto 2",
"template.": {"id": 999, "categories": [7]},
},
]
if args[0] == "model.product.category.read":
return [
{"id": 6, "name": "Alimentos"},
{"id": 7, "name": "Verduras"},
]
raise Exception(f"args no esperados: {args}")
mock_call.side_effect = fake_call
result = self._service(mock_call).import_from_external()
self.assertEqual(
result["checked_tryton_categories"],
[
{"id": 6, "name": "Alimentos"},
{"id": 7, "name": "Verduras"},
],
)
self.assertEqual(
result["created_categories"],
[
{
"id": ProductCategory.objects.get(external_id="6").pk,
"name": "Alimentos",
"external_id": "6",
},
{
"id": ProductCategory.objects.get(external_id="7").pk,
"name": "Verduras",
"external_id": "7",
},
],
)
self.assertEqual(result["updated_categories"], [])
self.assertEqual(result["untouched_categories"], [])
self.assertEqual(result["failed_categories"], [])
self.assertEqual(ProductCategory.objects.count(), 2)
@patch("sabatron_tryton_rpc_client.client.Client.call")
def test_import_preserves_don_confiao_fields_on_update(self, mock_call):
local_category = ProductCategory.objects.create(
name="Granos",
external_id="7",
description="Granos enteros",
alternative_name="Cereales",
active=False,
)
def fake_call(*args, **kwargs):
if args[0] == "model.product.product.search":
return [1]
if args[0] == "model.product.product.read":
return [
{
"id": 1,
"name": "Producto 1",
"template.": {"id": 999, "categories": [7]},
},
]
if args[0] == "model.product.category.read":
return [{"id": 7, "name": "Cereales y Granos"}]
raise Exception(f"args no esperados: {args}")
mock_call.side_effect = fake_call
result = self._service(mock_call).import_from_external()
self.assertEqual(result["created_categories"], [])
self.assertEqual(
result["updated_categories"],
[
{
"id": local_category.pk,
"name": "Cereales y Granos",
"external_id": "7",
"detail": "Nombre: Granos → Cereales y Granos",
}
],
)
refreshed = ProductCategory.objects.get(pk=local_category.pk)
self.assertEqual(refreshed.name, "Cereales y Granos")
self.assertEqual(refreshed.description, "Granos enteros")
self.assertEqual(refreshed.alternative_name, "Cereales")
self.assertFalse(refreshed.active)
@patch("sabatron_tryton_rpc_client.client.Client.call")
def test_import_counts_untouched_categories(self, mock_call):
local_category = ProductCategory.objects.create(
name="Alimentos", external_id="6"
)
def fake_call(*args, **kwargs):
if args[0] == "model.product.product.search":
return [1]
if args[0] == "model.product.product.read":
return [
{
"id": 1,
"name": "Producto 1",
"template.": {"id": 999, "categories": [6]},
},
]
if args[0] == "model.product.category.read":
return [{"id": 6, "name": "Alimentos"}]
raise Exception(f"args no esperados: {args}")
mock_call.side_effect = fake_call
result = self._service(mock_call).import_from_external()
self.assertEqual(result["created_categories"], [])
self.assertEqual(result["updated_categories"], [])
self.assertEqual(
result["untouched_categories"],
[
{
"id": local_category.pk,
"name": "Alimentos",
"external_id": "6",
}
],
)
self.assertEqual(result["failed_categories"], [])
self.assertEqual(ProductCategory.objects.count(), 1)
@patch("sabatron_tryton_rpc_client.client.Client.call")
def test_import_without_products_returns_empty_report(self, mock_call):
def fake_call(*args, **kwargs):
if args[0] == "model.product.product.search":
return []
raise Exception(f"args no esperados: {args}")
mock_call.side_effect = fake_call
result = self._service(mock_call).import_from_external()
self.assertEqual(result, self._empty_report())
self.assertEqual(ProductCategory.objects.count(), 0)
@patch("sabatron_tryton_rpc_client.client.Client.call")
def test_import_failed_category_reported(self, mock_call):
def fake_call(*args, **kwargs):
if args[0] == "model.product.product.search":
return [1]
if args[0] == "model.product.product.read":
return [
{
"id": 1,
"name": "Producto 1",
"template.": {"id": 999, "categories": [6]},
},
]
if args[0] == "model.product.category.read":
return [{"id": 6, "name": None}]
raise Exception(f"args no esperados: {args}")
mock_call.side_effect = fake_call
result = self._service(mock_call).import_from_external()
self.assertEqual(result["created_categories"], [])
self.assertEqual(len(result["failed_categories"]), 1)
failure = result["failed_categories"][0]
self.assertEqual(failure["external_id"], 6)
self.assertEqual(failure["name"], None)
self.assertIn("Error", failure["error"])

View File

@@ -62,11 +62,25 @@ class TestCustomersFromTryton(TestCase, LoginMixin):
content = json.loads(response.content.decode("utf-8"))
expected_response = {
"checked_tryton_parties": [5, 6, 7, 8],
"created_customers": [3, 4],
"untouched_customers": [2],
"checked_tryton_parties": [
{"id": 5, "name": "Carlos"},
{"id": 6, "name": "Cristian"},
{"id": 7, "name": "Ana"},
{"id": 8, "name": "José"},
],
"created_customers": [
{"id": 3, "name": "Ana"},
{"id": 4, "name": "José"},
],
"untouched_customers": [{"id": 2, "name": "Cristian"}],
"failed_parties": [],
"updated_customers": [1],
"updated_customers": [
{
"id": 1,
"name": "Carlos",
"detail": "Nombre: Calos → Carlos, Dirección: None → 303",
}
],
}
self.assertEqual(content, expected_response)
@@ -79,3 +93,115 @@ class TestCustomersFromTryton(TestCase, LoginMixin):
self.assertEqual(updated_customer.external_id, str(5))
self.assertEqual(updated_customer.name, "Carlos")
self.assertIn(updated_customer.address_external_id, str(303))
@patch("sabatron_tryton_rpc_client.client.Client.call")
@patch("sabatron_tryton_rpc_client.client.Client.connect")
def test_create_failure_customer(self, mock_connect, mock_call):
def fake_call(*args, **kwargs):
party_search = "model.party.party.search"
search_args = [
[],
0,
1000,
[["name", "ASC"], ["id", None]],
{"company": 1},
]
if args == (party_search, search_args):
return [5, 9]
party_read = "model.party.party.read"
read_args = (
[5, 9],
["id", "name", "addresses"],
{"company": 1},
)
if args == (party_read, read_args):
return [
{"id": 5, "name": "Carlos", "addresses": [303]},
{"id": 9, "name": None, "addresses": []},
]
raise Exception(
f"Sorry, args non expected on this test: {args}"
)
mock_call.side_effect = fake_call
url = "/don_confiao/api/importar_clientes_de_tryton"
response = self.client.post(url)
self.assertEqual(response.status_code, 200)
content = json.loads(response.content.decode("utf-8"))
self.assertEqual(len(content["failed_parties"]), 1)
failure = content["failed_parties"][0]
self.assertEqual(failure["external_id"], 9)
self.assertEqual(failure["name"], None)
self.assertEqual(
failure["error"],
"Error al crear cliente: El cliente no tiene nombre en Tryton",
)
self.assertEqual(
content["updated_customers"],
[
{
"id": 1,
"name": "Carlos",
"detail": "Nombre: Calos → Carlos, Dirección: None → 303",
}
],
)
@patch("sabatron_tryton_rpc_client.client.Client.call")
@patch("sabatron_tryton_rpc_client.client.Client.connect")
def test_update_failure_customer(self, mock_connect, mock_call):
def fake_call(*args, **kwargs):
party_search = "model.party.party.search"
search_args = [
[],
0,
1000,
[["name", "ASC"], ["id", None]],
{"company": 1},
]
if args == (party_search, search_args):
return [5, 6]
party_read = "model.party.party.read"
read_args = (
[5, 6],
["id", "name", "addresses"],
{"company": 1},
)
if args == (party_read, read_args):
return [
{"id": 5, "name": "Carlos", "addresses": [303]},
{"id": 6, "name": None, "addresses": []},
]
raise Exception(
f"Sorry, args non expected on this test: {args}"
)
mock_call.side_effect = fake_call
url = "/don_confiao/api/importar_clientes_de_tryton"
response = self.client.post(url)
self.assertEqual(response.status_code, 200)
content = json.loads(response.content.decode("utf-8"))
self.assertEqual(len(content["failed_parties"]), 1)
failure = content["failed_parties"][0]
self.assertEqual(failure["external_id"], 6)
self.assertIn("Error al actualizar cliente", failure["error"])
self.assertEqual(
content["updated_customers"],
[
{
"id": 1,
"name": "Carlos",
"detail": "Nombre: Calos → Carlos, Dirección: None → 303",
}
],
)

View File

@@ -0,0 +1,311 @@
import json
from unittest.mock import patch, MagicMock
from django.test import TestCase
from ..models.sales import Sale, SaleLine, CatalogSale, CatalogSaleLine
from ..models.customers import Customer
from ..models.products import Product
from ..services.tryton.client import TrytonSale, TrytonCatalogSale
from .Mixins import LoginMixin
class TestSendSalesToTryton(TestCase, LoginMixin):
def setUp(self):
self.login()
self.product = Product.objects.create(
name="Panela",
price=5000,
measuring_unit="UNIT",
unit_external_id="1",
external_id="1",
)
self.customer = Customer.objects.create(
name="Camilo",
external_id="1",
address_external_id="307",
)
self.sale = Sale.objects.create(
customer=self.customer,
date="2024-09-02",
payment_method="CASH",
description="un comentario",
)
SaleLine.objects.create(
product=self.product,
quantity=2,
unit_price=3000,
sale=self.sale,
)
SaleLine.objects.create(
product=self.product,
quantity=3,
unit_price=5000,
sale=self.sale,
)
self.catalog_sale = CatalogSale.objects.create(
customer=self.customer,
date="2024-09-02",
customer_name="Camilo",
customer_phone="3001234567",
customer_address="Calle 123",
pickup_method="Recogida en tienda",
)
CatalogSaleLine.objects.create(
catalog_sale=self.catalog_sale,
product=self.product,
quantity=2,
unit_price=3000,
)
CatalogSaleLine.objects.create(
catalog_sale=self.catalog_sale,
product=self.product,
quantity=3,
unit_price=5000,
)
@patch("don_confiao.api.sales.get_tryton_client")
def test_send_sales_to_tryton(self, mock_get_tryton_client):
external_id = "23423"
mock_client = MagicMock()
mock_client.call.return_value = [external_id]
mock_get_tryton_client.return_value = mock_client
url = "/don_confiao/api/enviar_ventas_a_tryton"
response = self.client.post(url)
self.assertEqual(response.status_code, 200)
content = json.loads(response.content.decode("utf-8"))
self.assertEqual(
content,
{
"successful": [
{"id": self.sale.id, "code": self.sale.code}
],
"failed": [],
},
)
updated_sale = Sale.objects.get(id=self.sale.id)
self.assertEqual(updated_sale.external_id, external_id)
mock_get_tryton_client.assert_called_once()
mock_client.call.assert_called_once()
call_args = mock_client.call.call_args
method, params = call_args[0]
self.assertEqual(method, "model.sale.sale.create")
tryton_context = params[1]
self.assertEqual(tryton_context, {"company": 1, "shops": [1]})
sale_params = params[0][0]
self.assertEqual(sale_params["company"], 1)
self.assertEqual(sale_params["shipment_address"], "307")
self.assertEqual(sale_params["invoice_address"], "307")
self.assertEqual(sale_params["currency"], 31)
self.assertEqual(sale_params["comment"], "un comentario")
self.assertEqual(sale_params["description"], "Metodo pago: CASH")
self.assertEqual(sale_params["party"], "1")
self.assertEqual(
sale_params["reference"], f"don_confiao {self.sale.id}"
)
self.assertEqual(
sale_params["sale_date"],
{"__class__": "date", "year": 2024, "month": 9, "day": 2},
)
self.assertTrue(sale_params["self_pick_up"])
sale_tryton = TrytonSale(
self.sale, SaleLine.objects.filter(sale=self.sale)
)
self.assertEqual(
sale_tryton.build_tryton_comment(), self.sale.description
)
lines_block = sale_params["lines"]
self.assertEqual(len(lines_block), 1)
self.assertEqual(lines_block[0][0], "create")
lines = lines_block[0][1]
self.assertEqual(len(lines), 2)
self.assertEqual(lines[0]["product"], "1")
self.assertEqual(
lines[0]["quantity"],
{"__class__": "Decimal", "decimal": "2.00"},
)
self.assertEqual(lines[0]["type"], "line")
self.assertEqual(lines[0]["unit"], "1")
self.assertEqual(
lines[0]["unit_price"],
{"__class__": "Decimal", "decimal": "3000.00"},
)
self.assertEqual(lines[1]["product"], "1")
self.assertEqual(
lines[1]["quantity"],
{"__class__": "Decimal", "decimal": "3.00"},
)
self.assertEqual(lines[1]["type"], "line")
self.assertEqual(lines[1]["unit"], "1")
self.assertEqual(
lines[1]["unit_price"],
{"__class__": "Decimal", "decimal": "5000.00"},
)
@patch("don_confiao.api.sales.get_tryton_client")
def test_send_catalog_sales_to_tryton(self, mock_get_tryton_client):
external_id = "23423"
mock_client = MagicMock()
mock_client.call.return_value = [external_id]
mock_get_tryton_client.return_value = mock_client
url = "/don_confiao/api/enviar_catalog_sales_a_tryton"
response = self.client.post(url)
self.assertEqual(response.status_code, 200)
content = json.loads(response.content.decode("utf-8"))
self.assertEqual(
content,
{
"successful": [
{
"id": self.catalog_sale.id,
"code": self.catalog_sale.code,
}
],
"failed": [],
},
)
updated = CatalogSale.objects.get(id=self.catalog_sale.id)
self.assertEqual(updated.external_id, external_id)
mock_get_tryton_client.assert_called_once()
mock_client.call.assert_called_once()
call_args = mock_client.call.call_args
method, params = call_args[0]
self.assertEqual(method, "model.sale.sale.create")
tryton_context = params[1]
self.assertEqual(tryton_context, {"company": 1, "shops": [1]})
sale_params = params[0][0]
self.assertEqual(sale_params["company"], 1)
self.assertEqual(sale_params["shipment_address"], "307")
self.assertEqual(sale_params["invoice_address"], "307")
self.assertEqual(sale_params["currency"], 31)
self.assertEqual(sale_params["description"], "Venta de catálogo")
self.assertEqual(sale_params["party"], "1")
self.assertEqual(
sale_params["reference"],
f"don_confiao_catalog {self.catalog_sale.id}",
)
self.assertEqual(
sale_params["sale_date"],
{"__class__": "date", "year": 2024, "month": 9, "day": 2},
)
self.assertTrue(sale_params["self_pick_up"])
expected_comment = (
"Cliente: Camilo\n"
"Teléfono: 3001234567\n"
"Dirección: Calle 123\n"
"Método de recogida: Recogida en tienda\n"
)
self.assertEqual(sale_params["comment"], expected_comment)
catalog_tryton = TrytonCatalogSale(
self.catalog_sale,
CatalogSaleLine.objects.filter(
catalog_sale=self.catalog_sale
),
)
self.assertEqual(
catalog_tryton.build_tryton_comment(), expected_comment
)
lines_block = sale_params["lines"]
self.assertEqual(len(lines_block), 1)
self.assertEqual(lines_block[0][0], "create")
lines = lines_block[0][1]
self.assertEqual(len(lines), 2)
self.assertEqual(lines[0]["product"], "1")
self.assertEqual(
lines[0]["quantity"],
{"__class__": "Decimal", "decimal": "2.00"},
)
self.assertEqual(lines[0]["type"], "line")
self.assertEqual(lines[0]["unit"], "1")
self.assertEqual(
lines[0]["unit_price"],
{"__class__": "Decimal", "decimal": "3000.00"},
)
self.assertEqual(lines[1]["product"], "1")
self.assertEqual(
lines[1]["quantity"],
{"__class__": "Decimal", "decimal": "3.00"},
)
self.assertEqual(lines[1]["type"], "line")
self.assertEqual(lines[1]["unit"], "1")
self.assertEqual(
lines[1]["unit_price"],
{"__class__": "Decimal", "decimal": "5000.00"},
)
@patch("don_confiao.api.sales.get_tryton_client")
def test_send_sales_to_tryton_failure(self, mock_get_tryton_client):
mock_client = MagicMock()
mock_client.call.side_effect = Exception("error de tryton")
mock_get_tryton_client.return_value = mock_client
url = "/don_confiao/api/enviar_ventas_a_tryton"
response = self.client.post(url)
self.assertEqual(response.status_code, 200)
content = json.loads(response.content.decode("utf-8"))
self.assertEqual(content["successful"], [])
self.assertEqual(len(content["failed"]), 1)
failure = content["failed"][0]
self.assertEqual(failure["id"], self.sale.id)
self.assertEqual(failure["code"], self.sale.code)
self.assertEqual(
failure["error"], "Error al enviar la venta: error de tryton"
)
sale = Sale.objects.get(id=self.sale.id)
self.assertIsNone(sale.external_id)
@patch("don_confiao.api.sales.get_tryton_client")
def test_send_catalog_sales_to_tryton_failure(
self, mock_get_tryton_client
):
mock_client = MagicMock()
mock_client.call.side_effect = Exception("error de tryton")
mock_get_tryton_client.return_value = mock_client
url = "/don_confiao/api/enviar_catalog_sales_a_tryton"
response = self.client.post(url)
self.assertEqual(response.status_code, 200)
content = json.loads(response.content.decode("utf-8"))
self.assertEqual(content["successful"], [])
self.assertEqual(len(content["failed"]), 1)
failure = content["failed"][0]
self.assertEqual(failure["id"], self.catalog_sale.id)
self.assertEqual(failure["code"], self.catalog_sale.code)
self.assertEqual(
failure["error"], "Error al enviar la venta: error de tryton"
)
catalog_sale = CatalogSale.objects.get(id=self.catalog_sale.id)
self.assertIsNone(catalog_sale.external_id)

View File

@@ -0,0 +1,129 @@
from django.db import IntegrityError, transaction
from django.test import TestCase
from ..models.geography import Country, Department, Municipality
class TestCountryModel(TestCase):
def test_create_country(self):
country = Country.objects.create(name="Colombia", code="CO")
self.assertEqual(country.name, "Colombia")
self.assertEqual(country.code, "CO")
self.assertEqual(str(country), "Colombia")
def test_country_name_is_unique(self):
Country.objects.create(name="Colombia", code="CO")
with self.assertRaises(IntegrityError):
with transaction.atomic():
Country.objects.create(name="Colombia", code="CO2")
def test_country_code_is_unique(self):
Country.objects.create(name="Colombia", code="CO")
with self.assertRaises(IntegrityError):
with transaction.atomic():
Country.objects.create(name="Colombia2", code="CO")
class TestDepartmentModel(TestCase):
def setUp(self):
self.country = Country.objects.create(name="Colombia", code="CO")
def test_create_department(self):
department = Department.objects.create(
name="Cundinamarca", country=self.country
)
self.assertEqual(department.country, self.country)
def test_department_requires_country(self):
with self.assertRaises(IntegrityError):
with transaction.atomic():
Department.objects.create(name="Antioquia")
def test_department_name_is_unique(self):
Department.objects.create(
name="Cundinamarca", country=self.country
)
with self.assertRaises(IntegrityError):
with transaction.atomic():
Department.objects.create(
name="Cundinamarca", country=self.country
)
class TestMunicipalityModel(TestCase):
def setUp(self):
self.country = Country.objects.create(name="Colombia", code="CO")
self.department = Department.objects.create(
name="Cundinamarca", country=self.country
)
def test_create_municipality(self):
municipality = Municipality.objects.create(
name="La Mesa",
department=self.department,
country=self.country,
)
self.assertEqual(municipality.department, self.department)
self.assertEqual(municipality.country, self.country)
self.assertEqual(municipality.department.country.name, "Colombia")
self.assertEqual(str(municipality), "La Mesa")
def test_municipality_geolocation_fields(self):
municipality = Municipality.objects.create(
name="La Mesa",
department=self.department,
country=self.country,
latitude=4.63092,
longitude=-74.39152,
)
self.assertEqual(municipality.latitude, 4.63092)
self.assertEqual(municipality.longitude, -74.39152)
def test_municipality_geolocation_optional(self):
municipality = Municipality.objects.create(
name="La Mesa",
department=self.department,
country=self.country,
)
self.assertIsNone(municipality.latitude)
self.assertIsNone(municipality.longitude)
def test_municipality_requires_department_and_country(self):
with self.assertRaises(IntegrityError):
with transaction.atomic():
Municipality.objects.create(name="La Mesa")
def test_municipality_name_is_unique_per_department(self):
Municipality.objects.create(
name="La Mesa",
department=self.department,
country=self.country,
)
with self.assertRaises(IntegrityError):
with transaction.atomic():
Municipality.objects.create(
name="La Mesa",
department=self.department,
country=self.country,
)
def test_municipality_same_name_allowed_in_different_department(
self,
):
department2 = Department.objects.create(
name="Antioquia", country=self.country
)
Municipality.objects.create(
name="La Mesa",
department=self.department,
country=self.country,
)
municipality2 = Municipality.objects.create(
name="La Mesa",
department=department2,
country=self.country,
)
self.assertEqual(
Municipality.objects.filter(name="La Mesa").count(), 2
)
self.assertEqual(municipality2.department, department2)

View File

@@ -0,0 +1,222 @@
from datetime import datetime, timezone
from django.contrib.auth.models import User
from rest_framework import status
from rest_framework.test import APIClient, APITestCase
from ..admin import CatalogSaleAdmin, SaleAdmin
from ..models.customers import Customer
from ..models.products import Product
from ..models.sales import CatalogSale, Sale
from ..serializers import CatalogSaleSerializer, SaleSerializer
from .Mixins import LoginMixin
class TestOrderSummaryPublic(APITestCase, LoginMixin):
def setUp(self):
self.product = Product.objects.create(
name="Panela", price=5000, measuring_unit="UNIT"
)
self.customer = Customer.objects.create(
name="Camilo", external_id="18"
)
def _create_sale(self):
sale = Sale.objects.create(
customer=self.customer,
date=datetime(2024, 9, 2, tzinfo=timezone.utc),
payment_method="CASH",
)
sale.saleline_set.create(
product=self.product, quantity=2, unit_price=3000
)
return sale
def _create_catalog_sale(self):
catalog_sale = CatalogSale.objects.create(
customer=self.customer,
date=datetime(2024, 9, 2, tzinfo=timezone.utc),
customer_name="Camilo",
)
catalog_sale.catalogsaleline_set.create(
product=self.product, quantity=2, unit_price=3000
)
return catalog_sale
def test_sale_has_auto_generated_code(self):
sale = self._create_sale()
self.assertTrue(sale.code)
self.assertEqual(len(sale.code), 32)
int(sale.code, 16)
def test_sale_codes_are_distinct(self):
sale1 = self._create_sale()
sale2 = self._create_sale()
self.assertNotEqual(sale1.code, sale2.code)
def test_catalog_sale_has_auto_generated_code(self):
catalog_sale = self._create_catalog_sale()
self.assertTrue(catalog_sale.code)
self.assertEqual(len(catalog_sale.code), 32)
int(catalog_sale.code, 16)
def test_public_summary_by_code_without_auth(self):
sale = self._create_sale()
response = self.client.get(
f"/don_confiao/resumen_publico/{sale.code}"
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data["code"], sale.code)
self.assertIn("link", data)
self.assertTrue(
data["link"].endswith(
f"/don_confiao/resumen_publico/{sale.code}"
)
)
self.assertEqual(data["type"], "sale")
self.assertIn("lines", data)
self.assertGreaterEqual(len(data["lines"]), 1)
def test_public_summary_invalid_code_returns_404(self):
response = self.client.get(
"/don_confiao/resumen_publico/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
)
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
def test_public_summary_by_query_param(self):
sale = self._create_sale()
response = self.client.get(
"/don_confiao/resumen_publico", {"code": sale.code}
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.json()["code"], sale.code)
def test_public_summary_for_catalog_sale(self):
catalog_sale = self._create_catalog_sale()
response = self.client.get(
f"/don_confiao/resumen_publico/{catalog_sale.code}"
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data["code"], catalog_sale.code)
self.assertEqual(data["type"], "catalog")
def test_public_summary_with_duplicate_codes_does_not_500(self):
sale1 = self._create_sale()
sale2 = self._create_sale()
shared = "a" * 32
Sale.objects.filter(pk__in=[sale1.pk, sale2.pk]).update(code=shared)
response = self.client.get(
f"/don_confiao/resumen_publico/{shared}"
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
def test_existing_authenticated_summaries_include_code_and_link(self):
self.login()
sale = self._create_sale()
catalog_sale = self._create_catalog_sale()
response = self.client.get(
f"/don_confiao/resumen_compra_json/{sale.id}"
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data["code"], sale.code)
self.assertIn("link", data)
response = self.client.get(
f"/don_confiao/resumen_compra_catalogo_json/{catalog_sale.id}"
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data["code"], catalog_sale.code)
self.assertIn("link", data)
def test_fill_codes_sales_endpoint_rewrites_all_codes(self):
self.login()
sale1 = self._create_sale()
sale2 = self._create_sale()
old_codes = {sale1.code, sale2.code}
response = self.client.post("/don_confiao/api/sales/fill_codes/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.json()["sales_updated"], 2)
sale1.refresh_from_db()
sale2.refresh_from_db()
new_codes = {sale1.code, sale2.code}
self.assertNotIn(sale1.code, old_codes)
self.assertEqual(len(new_codes), 2)
def test_fill_codes_catalog_sales_endpoint_rewrites_all_codes(self):
self.login()
catalog_sale = self._create_catalog_sale()
old_code = catalog_sale.code
response = self.client.post(
"/don_confiao/api/catalog_sales/fill_codes/"
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.json()["catalog_sales_updated"], 1)
catalog_sale.refresh_from_db()
self.assertNotEqual(catalog_sale.code, old_code)
def test_fill_codes_endpoints_require_authentication(self):
response = self.client.post("/don_confiao/api/sales/fill_codes/")
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
response = self.client.post(
"/don_confiao/api/catalog_sales/fill_codes/"
)
self.assertEqual(
response.status_code, status.HTTP_401_UNAUTHORIZED
)
def test_fill_codes_endpoints_require_admin(self):
regular_user = User.objects.create_user(
username="regular", password="regularpass"
)
client = APIClient()
client.force_authenticate(user=regular_user)
response = client.post("/don_confiao/api/sales/fill_codes/")
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
response = client.post(
"/don_confiao/api/catalog_sales/fill_codes/"
)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
def test_admin_list_display_includes_code(self):
self.assertIn("code", SaleAdmin.list_display)
self.assertIn("code", CatalogSaleAdmin.list_display)
def test_sale_serializer_includes_code(self):
sale = self._create_sale()
data = SaleSerializer(sale).data
self.assertIn("code", data)
self.assertEqual(data["code"], sale.code)
def test_catalog_sale_serializer_includes_code(self):
catalog_sale = self._create_catalog_sale()
data = CatalogSaleSerializer(catalog_sale).data
self.assertIn("code", data)
self.assertEqual(data["code"], catalog_sale.code)
def test_sales_list_endpoint_includes_code(self):
self.login()
sale = self._create_sale()
response = self.client.get("/don_confiao/api/sales/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
item = next(i for i in response.json() if i["id"] == sale.id)
self.assertEqual(item["code"], sale.code)
def test_catalog_sales_list_endpoint_includes_code(self):
self.login()
catalog_sale = self._create_catalog_sale()
response = self.client.get("/don_confiao/api/catalog_sales/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
item = next(i for i in response.json() if i["id"] == catalog_sale.id)
self.assertEqual(item["code"], catalog_sale.code)

View File

@@ -0,0 +1,102 @@
from django.test import TestCase
from ..models.products import Product, ProductCategory
class TestProductCategoryModel(TestCase):
def test_create_category_with_don_confiao_fields(self):
category = ProductCategory.objects.create(
name="Bebidas",
external_id="8",
description="Bebidas y jugos naturales",
alternative_name="Jugos",
active=False,
)
self.assertEqual(category.name, "Bebidas")
self.assertEqual(category.external_id, "8")
self.assertEqual(category.description, "Bebidas y jugos naturales")
self.assertEqual(category.alternative_name, "Jugos")
self.assertFalse(category.active)
def test_category_active_por_defecto(self):
category = ProductCategory.objects.create(name="Granos")
self.assertTrue(category.active)
self.assertEqual(category.description, "")
self.assertIsNone(category.alternative_name)
class TestProductCategorySerializers(TestCase):
def test_product_category_serializer_devuelve_campos_don_confiao(self):
from ..serializers import ProductCategorySerializer
category = ProductCategory.objects.create(
name="Bebidas",
external_id="8",
description="Bebidas naturales",
alternative_name="Jugos",
active=False,
)
data = ProductCategorySerializer(category).data
self.assertEqual(
data,
{
"id": category.pk,
"name": "Bebidas",
"external_id": "8",
"description": "Bebidas naturales",
"alternative_name": "Jugos",
"active": False,
},
)
def test_product_serializer_incluye_categorias_anidadas(self):
from ..serializers import ProductSerializer
category = ProductCategory.objects.create(
name="Bebidas", external_id="8", alternative_name="Jugos"
)
product = Product.objects.create(
name="Jugo de naranja",
price=2500,
measuring_unit="UNIT",
external_id="10",
)
product.categories.add(category)
data = ProductSerializer(product).data
self.assertEqual(
data["categories"],
[
{
"id": category.pk,
"name": "Bebidas",
"alternative_name": "Jugos",
"active": True,
}
],
)
def test_product_to_list_incluye_categorias_anidadas(self):
category = ProductCategory.objects.create(
name="Bebidas", external_id="8", alternative_name="Jugos"
)
product = Product.objects.create(
name="Jugo de naranja",
price=2500,
measuring_unit="UNIT",
external_id="10",
)
product.categories.add(category)
data = Product.to_list()
self.assertEqual(
data[0]["categories"],
[
{
"id": category.pk,
"name": "Bebidas",
"alternative_name": "Jugos",
"active": True,
}
],
)

View File

@@ -0,0 +1,224 @@
import json
from unittest.mock import patch
from django.contrib.auth.models import User
from django.test import TestCase
from rest_framework.test import APIClient
from rest_framework_simplejwt.tokens import RefreshToken
from ..models.products import ProductCategory
class ProductCategoryApiTest(TestCase):
def setUp(self):
self.user = User.objects.create_superuser(
username="admin",
email="admin@example.com",
password="adminpass",
)
refresh = RefreshToken.for_user(self.user)
self.client = APIClient()
self.client.credentials(
HTTP_AUTHORIZATION=f"Bearer {refresh.access_token}"
)
def login_as_regular_user(self):
user = User.objects.create_user(
username="vendedor", password="vendedorpass"
)
refresh = RefreshToken.for_user(user)
self.client.credentials(
HTTP_AUTHORIZATION=f"Bearer {refresh.access_token}"
)
def test_list_categories_filters_by_active(self):
ProductCategory.objects.create(name="Granos", external_id="6")
ProductCategory.objects.create(
name="Verduras", external_id="7", active=False
)
response = self.client.get(
"/don_confiao/api/product_categories/?active=all"
)
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data), 2)
response = self.client.get(
"/don_confiao/api/product_categories/?active=true"
)
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data), 1)
self.assertEqual(response.data[0]["name"], "Granos")
response = self.client.get(
"/don_confiao/api/product_categories/?active=false"
)
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data), 1)
self.assertEqual(response.data[0]["name"], "Verduras")
def test_patch_updates_don_confiao_fields_only(self):
category = ProductCategory.objects.create(
name="Granos", external_id="6"
)
response = self.client.patch(
f"/don_confiao/api/product_categories/{category.pk}/",
{
"description": "Granos enteros",
"alternative_name": "Cereales",
"active": False,
"name": "Nombre intentado",
"external_id": "999",
},
format="json",
)
self.assertEqual(response.status_code, 200)
refreshed = ProductCategory.objects.get(pk=category.pk)
self.assertEqual(refreshed.description, "Granos enteros")
self.assertEqual(refreshed.alternative_name, "Cereales")
self.assertFalse(refreshed.active)
self.assertEqual(refreshed.name, "Granos")
self.assertEqual(refreshed.external_id, "6")
def test_deactivate_categories(self):
category = ProductCategory.objects.create(
name="Granos", external_id="6"
)
response = self.client.post(
"/don_confiao/api/product_categories/deactivate/",
{"ids": [category.pk]},
format="json",
)
self.assertEqual(response.status_code, 200)
self.assertEqual(
response.data,
{
"deactivated": [{"id": category.pk, "name": "Granos"}],
"failed": [],
},
)
self.assertFalse(
ProductCategory.objects.get(pk=category.pk).active
)
def test_activate_categories(self):
category = ProductCategory.objects.create(
name="Granos", external_id="6", active=False
)
response = self.client.post(
"/don_confiao/api/product_categories/activate/",
{"ids": [category.pk]},
format="json",
)
self.assertEqual(response.status_code, 200)
self.assertEqual(
response.data,
{
"activated": [{"id": category.pk, "name": "Granos"}],
"failed": [],
},
)
self.assertTrue(ProductCategory.objects.get(pk=category.pk).active)
def test_deactivate_ignores_nonexistent_ids(self):
category = ProductCategory.objects.create(
name="Granos", external_id="6"
)
response = self.client.post(
"/don_confiao/api/product_categories/deactivate/",
{"ids": [category.pk, 99999]},
format="json",
)
self.assertEqual(response.status_code, 200)
self.assertEqual(
response.data["deactivated"],
[{"id": category.pk, "name": "Granos"}],
)
self.assertEqual(response.data["failed"], [])
self.assertFalse(
ProductCategory.objects.get(pk=category.pk).active
)
@patch("sabatron_tryton_rpc_client.client.Client.call")
@patch("sabatron_tryton_rpc_client.client.Client.connect")
def test_import_from_external(self, mock_connect, mock_call):
mock_connect.return_value = None
def fake_call(*args, **kwargs):
if args[0] == "model.product.product.search":
return [1]
if args[0] == "model.product.product.read":
return [
{
"id": 1,
"name": "Producto 1",
"template.": {"id": 999, "categories": [6]},
},
]
if args[0] == "model.product.category.read":
return [{"id": 6, "name": "Alimentos"}]
raise Exception(f"args no esperados: {args}")
mock_call.side_effect = fake_call
response = self.client.post(
"/don_confiao/api/product_categories/import_from_external/"
)
self.assertEqual(response.status_code, 200)
content = json.loads(response.content.decode("utf-8"))
self.assertTrue(
set(
[
"checked_tryton_categories",
"created_categories",
"updated_categories",
"untouched_categories",
"failed_categories",
]
).issubset(content.keys())
)
self.assertEqual(
content["created_categories"],
[
{
"id": ProductCategory.objects.get(external_id="6").pk,
"name": "Alimentos",
"external_id": "6",
}
],
)
@patch("sabatron_tryton_rpc_client.client.Client.call")
@patch("sabatron_tryton_rpc_client.client.Client.connect")
def test_mutations_require_administrator(
self, mock_connect, mock_call
):
mock_connect.return_value = None
self.login_as_regular_user()
category = ProductCategory.objects.create(
name="Granos", external_id="6"
)
response = self.client.post(
"/don_confiao/api/product_categories/deactivate/",
{"ids": [category.pk]},
format="json",
)
self.assertEqual(response.status_code, 403)
response = self.client.post(
"/don_confiao/api/product_categories/import_from_external/"
)
self.assertEqual(response.status_code, 403)
response = self.client.patch(
f"/don_confiao/api/product_categories/{category.pk}/",
{"description": "x"},
format="json",
)
self.assertEqual(response.status_code, 403)

View File

@@ -3,7 +3,7 @@ from decimal import Decimal
from unittest.mock import patch
from django.test import TestCase
from ..models.products import Product
from ..models.products import Product, ProductCategory
from .Mixins import LoginMixin
@@ -55,6 +55,7 @@ class TestProductsFromTryton(TestCase, LoginMixin):
"default_uom.id",
"default_uom.rec_name",
"list_price",
"template.categories",
],
{"company": 1},
)
@@ -65,18 +66,21 @@ class TestProductsFromTryton(TestCase, LoginMixin):
"list_price": Decimal("25000"),
"name": "Producto 1",
"default_uom.": {"id": 1, "rec_name": "Unit"},
"template.": {"id": 999, "categories": []},
},
{
"id": 191,
"list_price": Decimal("6000"),
"name": "Panela2",
"default_uom.": {"id": 1, "rec_name": "Unit"},
"template.": {"id": 999, "categories": []},
},
{
"id": 192,
"list_price": Decimal("4500"),
"name": "Papa",
"default_uom.": {"id": 2, "rec_name": "Kilogram"},
"template.": {"id": 999, "categories": []},
},
]
@@ -92,11 +96,23 @@ class TestProductsFromTryton(TestCase, LoginMixin):
content = json.loads(response.content.decode("utf-8"))
expected_response = {
"checked_tryton_products": [190, 191, 192],
"created_products": [3],
"untouched_products": [2],
"checked_tryton_products": [
{"id": 190, "name": "Producto 1"},
{"id": 191, "name": "Panela2"},
{"id": 192, "name": "Papa"},
],
"created_products": [{"id": 3, "name": "Producto 1"}],
"untouched_products": [{"id": 2, "name": "Papa"}],
"failed_products": [],
"updated_products": [1],
"updated_products": [
{
"id": 1,
"name": "Panela2",
"detail": "Nombre: Panela → Panela2, Precio: 5000 → 6000, Unidad: UNIT → Unit",
}
],
"created_categories": [],
"updated_categories": [],
}
self.assertEqual(content, expected_response)
@@ -140,6 +156,7 @@ class TestProductsFromTryton(TestCase, LoginMixin):
"default_uom.id",
"default_uom.rec_name",
"list_price",
"template.categories",
],
{"company": 1},
)
@@ -150,6 +167,7 @@ class TestProductsFromTryton(TestCase, LoginMixin):
"list_price": Decimal("25000"),
"name": self.product.name,
"default_uom.": {"id": 1, "rec_name": "Unit"},
"template.": {"id": 999, "categories": []},
},
]
@@ -165,10 +183,673 @@ class TestProductsFromTryton(TestCase, LoginMixin):
content = json.loads(response.content.decode("utf-8"))
expected_response = {
"checked_tryton_products": [200],
"checked_tryton_products": [{"id": 200, "name": "Panela"}],
"created_products": [],
"untouched_products": [],
"failed_products": [200],
"failed_products": [
{
"external_id": 200,
"name": "Panela",
"error": "Error al crear producto: Ya existe un producto con el nombre 'Panela'",
}
],
"updated_products": [],
"created_categories": [],
"updated_categories": [],
}
self.assertEqual(content, expected_response)
@patch("sabatron_tryton_rpc_client.client.Client.call")
@patch("sabatron_tryton_rpc_client.client.Client.connect")
def test_import_null_price_product(self, mock_connect, mock_call):
mock_connect.return_value = None
def fake_call(*args, **kwargs):
product_search = "model.product.product.search"
search_args = [
[["salable", "=", True]],
0,
1000,
[["rec_name", "ASC"], ["id", None]],
{"company": 1},
]
if args == (product_search, search_args):
return [201]
product_read = "model.product.product.read"
product_args = (
[201],
[
"id",
"name",
"default_uom.id",
"default_uom.rec_name",
"list_price",
"template.categories",
],
{"company": 1},
)
if args == (product_read, product_args):
return [
{
"id": 201,
"list_price": None,
"name": "ENVASES Y EMPAQUES",
"default_uom.": {"id": 1, "rec_name": "Unit"},
"template.": {"id": 999, "categories": []},
},
]
raise Exception(
f"Sorry, args non expected on this test: {args}"
)
mock_call.side_effect = fake_call
url = "/don_confiao/api/importar_productos_de_tryton"
response = self.client.post(url)
self.assertEqual(response.status_code, 200)
content = json.loads(response.content.decode("utf-8"))
self.assertEqual(content["created_products"], [])
self.assertEqual(len(content["failed_products"]), 1)
failure = content["failed_products"][0]
self.assertEqual(failure["external_id"], 201)
self.assertEqual(failure["name"], "ENVASES Y EMPAQUES")
self.assertEqual(
failure["error"],
"Error al crear producto: El producto no tiene precio en Tryton (list_price nulo)",
)
@patch("sabatron_tryton_rpc_client.client.Client.call")
@patch("sabatron_tryton_rpc_client.client.Client.connect")
def test_update_failure_products(self, mock_connect, mock_call):
mock_connect.return_value = None
def fake_call(*args, **kwargs):
product_search = "model.product.product.search"
search_args = [
[["salable", "=", True]],
0,
1000,
[["rec_name", "ASC"], ["id", None]],
{"company": 1},
]
if args == (product_search, search_args):
return [191, 192]
product_read = "model.product.product.read"
product_args = (
[191, 192],
[
"id",
"name",
"default_uom.id",
"default_uom.rec_name",
"list_price",
"template.categories",
],
{"company": 1},
)
if args == (product_read, product_args):
return [
{
"id": 191,
"list_price": Decimal("6000"),
"name": "Panela",
"default_uom.": None,
"template.": {"id": 999, "categories": []},
},
{
"id": 192,
"list_price": Decimal("4500"),
"name": "Papa",
"default_uom.": {"id": 2, "rec_name": "Kilogram"},
"template.": {"id": 999, "categories": []},
},
]
raise Exception(
f"Sorry, args non expected on this test: {args}"
)
mock_call.side_effect = fake_call
url = "/don_confiao/api/importar_productos_de_tryton"
response = self.client.post(url)
self.assertEqual(response.status_code, 200)
content = json.loads(response.content.decode("utf-8"))
self.assertEqual(len(content["failed_products"]), 1)
failure = content["failed_products"][0]
self.assertEqual(failure["external_id"], 191)
self.assertEqual(failure["name"], "Panela")
self.assertIn("Error al actualizar producto", failure["error"])
self.assertEqual(content["untouched_products"], [{"id": 2, "name": "Papa"}])
@patch("sabatron_tryton_rpc_client.client.Client.call")
@patch("sabatron_tryton_rpc_client.client.Client.connect")
def test_import_products_with_categories(self, mock_connect, mock_call):
mock_connect.return_value = None
def fake_call(*args, **kwargs):
product_search = "model.product.product.search"
search_args = [
[["salable", "=", True]],
0,
1000,
[["rec_name", "ASC"], ["id", None]],
{"company": 1},
]
if args == (product_search, search_args):
return [190]
product_read = "model.product.product.read"
product_args = (
[190],
[
"id",
"name",
"default_uom.id",
"default_uom.rec_name",
"list_price",
"template.categories",
],
{"company": 1},
)
if args == (product_read, product_args):
return [
{
"id": 190,
"list_price": Decimal("25000"),
"name": "Producto 1",
"default_uom.": {"id": 1, "rec_name": "Unit"},
"template.": {"id": 999, "categories": [5, 8]},
},
]
category_read = "model.product.category.read"
category_args = (
[5, 8],
["id", "name"],
{"company": 1},
)
if args == (category_read, category_args):
return [
{"id": 5, "name": "Abarrotes"},
{"id": 8, "name": "Bebidas"},
]
raise Exception(
f"Sorry, args non expected on this test: {args}"
)
mock_call.side_effect = fake_call
url = "/don_confiao/api/importar_productos_de_tryton"
response = self.client.post(url)
self.assertEqual(response.status_code, 200)
content = json.loads(response.content.decode("utf-8"))
self.assertEqual(content["failed_products"], [])
self.assertEqual(
content["created_products"], [{"id": 3, "name": "Producto 1"}]
)
self.assertEqual(content["updated_categories"], [])
self.assertEqual(
content["created_categories"],
[
{"id": 1, "name": "Abarrotes", "external_id": "5"},
{"id": 2, "name": "Bebidas", "external_id": "8"},
],
)
created_product = Product.objects.get(name="Producto 1")
self.assertEqual(created_product.external_id, str(190))
categories = created_product.categories.order_by("external_id")
self.assertEqual(
[(c.external_id, c.name) for c in categories],
[("5", "Abarrotes"), ("8", "Bebidas")],
)
self.assertEqual(ProductCategory.objects.count(), 2)
@patch("sabatron_tryton_rpc_client.client.Client.call")
@patch("sabatron_tryton_rpc_client.client.Client.connect")
def test_import_updates_category_name_from_tryton(
self, mock_connect, mock_call
):
mock_connect.return_value = None
local_category = ProductCategory.objects.create(
name="Granos", external_id="7"
)
def fake_call(*args, **kwargs):
product_search = "model.product.product.search"
search_args = [
[["salable", "=", True]],
0,
1000,
[["rec_name", "ASC"], ["id", None]],
{"company": 1},
]
if args == (product_search, search_args):
return [190]
product_read = "model.product.product.read"
product_args = (
[190],
[
"id",
"name",
"default_uom.id",
"default_uom.rec_name",
"list_price",
"template.categories",
],
{"company": 1},
)
if args == (product_read, product_args):
return [
{
"id": 190,
"list_price": Decimal("25000"),
"name": "Producto 1",
"default_uom.": {"id": 1, "rec_name": "Unit"},
"template.": {"id": 999, "categories": [7]},
},
]
category_read = "model.product.category.read"
category_args = (
[7],
["id", "name"],
{"company": 1},
)
if args == (category_read, category_args):
return [
{"id": 7, "name": "Cereales"},
]
raise Exception(
f"Sorry, args non expected on this test: {args}"
)
mock_call.side_effect = fake_call
url = "/don_confiao/api/importar_productos_de_tryton"
response = self.client.post(url)
self.assertEqual(response.status_code, 200)
content = json.loads(response.content.decode("utf-8"))
self.assertEqual(content["created_categories"], [])
self.assertEqual(
content["updated_categories"],
[
{
"id": local_category.pk,
"name": "Cereales",
"external_id": "7",
"detail": "Nombre: Granos → Cereales",
}
],
)
refreshed_category = ProductCategory.objects.get(
pk=local_category.pk
)
self.assertEqual(refreshed_category.name, "Cereales")
self.assertEqual(refreshed_category.external_id, "7")
self.assertEqual(ProductCategory.objects.count(), 1)
created_product = Product.objects.get(name="Producto 1")
self.assertEqual(
list(created_product.categories.all()), [refreshed_category]
)
@patch("sabatron_tryton_rpc_client.client.Client.call")
@patch("sabatron_tryton_rpc_client.client.Client.connect")
def test_import_links_existing_category_by_name(
self, mock_connect, mock_call
):
mock_connect.return_value = None
existing_category = ProductCategory.objects.create(name="Abarrotes")
def fake_call(*args, **kwargs):
product_search = "model.product.product.search"
search_args = [
[["salable", "=", True]],
0,
1000,
[["rec_name", "ASC"], ["id", None]],
{"company": 1},
]
if args == (product_search, search_args):
return [190]
product_read = "model.product.product.read"
product_args = (
[190],
[
"id",
"name",
"default_uom.id",
"default_uom.rec_name",
"list_price",
"template.categories",
],
{"company": 1},
)
if args == (product_read, product_args):
return [
{
"id": 190,
"list_price": Decimal("25000"),
"name": "Producto 1",
"default_uom.": {"id": 1, "rec_name": "Unit"},
"template.": {"id": 999, "categories": [5]},
},
]
category_read = "model.product.category.read"
category_args = (
[5],
["id", "name"],
{"company": 1},
)
if args == (category_read, category_args):
return [
{"id": 5, "name": "Abarrotes"},
]
raise Exception(
f"Sorry, args non expected on this test: {args}"
)
mock_call.side_effect = fake_call
url = "/don_confiao/api/importar_productos_de_tryton"
response = self.client.post(url)
self.assertEqual(response.status_code, 200)
content = json.loads(response.content.decode("utf-8"))
self.assertEqual(content["created_categories"], [])
self.assertEqual(
content["updated_categories"],
[
{
"id": existing_category.pk,
"name": "Abarrotes",
"external_id": "5",
"detail": "External ID: None → 5",
}
],
)
self.assertEqual(
ProductCategory.objects.filter(name="Abarrotes").count(), 1
)
linked_category = ProductCategory.objects.get(
pk=existing_category.pk
)
self.assertEqual(linked_category.external_id, "5")
created_product = Product.objects.get(name="Producto 1")
self.assertEqual(
list(created_product.categories.all()), [linked_category]
)
@patch("sabatron_tryton_rpc_client.client.Client.call")
@patch("sabatron_tryton_rpc_client.client.Client.connect")
def test_import_product_without_categories(
self, mock_connect, mock_call
):
mock_connect.return_value = None
def fake_call(*args, **kwargs):
product_search = "model.product.product.search"
search_args = [
[["salable", "=", True]],
0,
1000,
[["rec_name", "ASC"], ["id", None]],
{"company": 1},
]
if args == (product_search, search_args):
return [190]
product_read = "model.product.product.read"
product_args = (
[190],
[
"id",
"name",
"default_uom.id",
"default_uom.rec_name",
"list_price",
"template.categories",
],
{"company": 1},
)
if args == (product_read, product_args):
return [
{
"id": 190,
"list_price": Decimal("25000"),
"name": "Producto 1",
"default_uom.": {"id": 1, "rec_name": "Unit"},
"template.": {"id": 999, "categories": []},
},
]
raise Exception(
f"Sorry, args non expected on this test: {args}"
)
mock_call.side_effect = fake_call
url = "/don_confiao/api/importar_productos_de_tryton"
response = self.client.post(url)
self.assertEqual(response.status_code, 200)
content = json.loads(response.content.decode("utf-8"))
self.assertEqual(content["failed_products"], [])
self.assertEqual(content["created_categories"], [])
self.assertEqual(content["updated_categories"], [])
created_product = Product.objects.get(name="Producto 1")
self.assertEqual(created_product.categories.count(), 0)
self.assertEqual(ProductCategory.objects.count(), 0)
@patch("sabatron_tryton_rpc_client.client.Client.call")
@patch("sabatron_tryton_rpc_client.client.Client.connect")
def test_import_assigns_categories_marks_product_as_updated(
self, mock_connect, mock_call
):
mock_connect.return_value = None
def fake_call(*args, **kwargs):
product_search = "model.product.product.search"
search_args = [
[["salable", "=", True]],
0,
1000,
[["rec_name", "ASC"], ["id", None]],
{"company": 1},
]
if args == (product_search, search_args):
return [192]
product_read = "model.product.product.read"
product_args = (
[192],
[
"id",
"name",
"default_uom.id",
"default_uom.rec_name",
"list_price",
"template.categories",
],
{"company": 1},
)
if args == (product_read, product_args):
return [
{
"id": 192,
"list_price": Decimal("4500"),
"name": "Papa",
"default_uom.": {"id": 2, "rec_name": "Kilogram"},
"template.": {"id": 999, "categories": [9]},
},
]
category_read = "model.product.category.read"
category_args = (
[9],
["id", "name"],
{"company": 1},
)
if args == (category_read, category_args):
return [
{"id": 9, "name": "Tubérculos"},
]
raise Exception(
f"Sorry, args non expected on this test: {args}"
)
mock_call.side_effect = fake_call
url = "/don_confiao/api/importar_productos_de_tryton"
response = self.client.post(url)
self.assertEqual(response.status_code, 200)
content = json.loads(response.content.decode("utf-8"))
self.assertEqual(content["untouched_products"], [])
self.assertEqual(
content["updated_products"],
[
{
"id": 2,
"name": "Papa",
"detail": "Categoría: +Tubérculos",
}
],
)
updated_product = Product.objects.get(id=2)
categories = updated_product.categories.all()
self.assertEqual(len(categories), 1)
self.assertEqual(categories[0].name, "Tubérculos")
self.assertEqual(categories[0].external_id, "9")
self.assertEqual(content["updated_categories"], [])
self.assertEqual(
content["created_categories"],
[
{
"id": categories[0].id,
"name": "Tubérculos",
"external_id": "9",
}
],
)
@patch("sabatron_tryton_rpc_client.client.Client.call")
@patch("sabatron_tryton_rpc_client.client.Client.connect")
def test_category_change_marks_product_as_updated(
self, mock_connect, mock_call
):
mock_connect.return_value = None
local_abarrotes = ProductCategory.objects.create(
name="Abarrotes", external_id="5"
)
self.product2.categories.add(local_abarrotes)
def fake_call(*args, **kwargs):
product_search = "model.product.product.search"
search_args = [
[["salable", "=", True]],
0,
1000,
[["rec_name", "ASC"], ["id", None]],
{"company": 1},
]
if args == (product_search, search_args):
return [192]
product_read = "model.product.product.read"
product_args = (
[192],
[
"id",
"name",
"default_uom.id",
"default_uom.rec_name",
"list_price",
"template.categories",
],
{"company": 1},
)
if args == (product_read, product_args):
return [
{
"id": 192,
"list_price": Decimal("4500"),
"name": "Papa",
"default_uom.": {"id": 2, "rec_name": "Kilogram"},
"template.": {"id": 999, "categories": [8]},
},
]
category_read = "model.product.category.read"
category_args = (
[8],
["id", "name"],
{"company": 1},
)
if args == (category_read, category_args):
return [
{"id": 8, "name": "Bebidas"},
]
raise Exception(
f"Sorry, args non expected on this test: {args}"
)
mock_call.side_effect = fake_call
url = "/don_confiao/api/importar_productos_de_tryton"
response = self.client.post(url)
self.assertEqual(response.status_code, 200)
content = json.loads(response.content.decode("utf-8"))
self.assertEqual(content["untouched_products"], [])
self.assertEqual(content["failed_products"], [])
self.assertEqual(
content["updated_products"],
[
{
"id": self.product2.pk,
"name": "Papa",
"detail": "Categoría: +Bebidas, -Abarrotes",
}
],
)
self.assertEqual(
content["created_categories"],
[
{
"id": ProductCategory.objects.get(
external_id="8"
).pk,
"name": "Bebidas",
"external_id": "8",
}
],
)
refreshed_product = Product.objects.get(pk=self.product2.pk)
categories = list(refreshed_product.categories.all())
self.assertEqual(
[(c.external_id, c.name) for c in categories],
[("8", "Bebidas")],
)

View File

@@ -0,0 +1,399 @@
from django.contrib.auth.models import User
from rest_framework import status
from rest_framework.test import APIClient, APITestCase
from rest_framework_simplejwt.tokens import RefreshToken
from ..models.geography import Country, Department, Municipality
from ..models.provenance import Organization, Supplier
from ..models.products import Product
from .Mixins import LoginMixin
def _create_user(username, user_type):
user = User.objects.create_user(
username=username, password="password123"
)
user.profile.user_type = user_type
user.profile.save()
return user
def _client_for(user):
refresh = RefreshToken.for_user(user)
client = APIClient()
client.credentials(
HTTP_AUTHORIZATION=f"Bearer {str(refresh.access_token)}"
)
return client
class TestProvenanceAPIPermissions(APITestCase, LoginMixin):
resources = [
"/don_confiao/api/organizations/",
"/don_confiao/api/suppliers/",
"/don_confiao/api/countries/",
"/don_confiao/api/departments/",
"/don_confiao/api/municipalities/",
]
def test_anonymous_get_is_forbidden(self):
for url in self.resources:
response = self.client.get(url)
self.assertEqual(
response.status_code,
status.HTTP_401_UNAUTHORIZED,
url,
)
def test_anonymous_write_is_forbidden(self):
for url in self.resources:
response = self.client.post(
url, {"name": "x"}, format="json"
)
self.assertEqual(
response.status_code,
status.HTTP_401_UNAUTHORIZED,
url,
)
def test_normal_user_can_read(self):
user = _create_user("normal", "user")
client = _client_for(user)
for url in self.resources:
response = client.get(url)
self.assertEqual(
response.status_code, status.HTTP_200_OK, url
)
def test_publico_user_cannot_read(self):
user = _create_user("publico", "publico")
client = _client_for(user)
for url in self.resources:
response = client.get(url)
self.assertEqual(
response.status_code, status.HTTP_403_FORBIDDEN, url
)
def test_normal_user_cannot_write(self):
user = _create_user("normal", "user")
client = _client_for(user)
for url in self.resources:
response = client.post(
url, {"name": "x"}, format="json"
)
self.assertEqual(
response.status_code, status.HTTP_403_FORBIDDEN, url
)
def test_publico_user_cannot_write(self):
user = _create_user("publico", "publico")
client = _client_for(user)
for url in self.resources:
response = client.post(
url, {"name": "x"}, format="json"
)
self.assertEqual(
response.status_code, status.HTTP_403_FORBIDDEN, url
)
class TestProvenanceCRUD(APITestCase, LoginMixin):
def setUp(self):
self.login()
self.country = Country.objects.create(name="Colombia", code="CO")
self.department = Department.objects.create(
name="Cundinamarca", country=self.country
)
self.municipality = Municipality.objects.create(
name="La Mesa",
department=self.department,
country=self.country,
)
self.organization = Organization.objects.create(name="Asociación")
self.supplier = Supplier.objects.create(
name="Proveedor 1",
organization=self.organization,
municipality=self.municipality,
)
def test_admin_can_create_organization(self):
response = self.client.post(
"/don_confiao/api/organizations/",
{"name": "Org 2"},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertEqual(
Organization.objects.filter(name="Org 2").count(), 1
)
def test_admin_can_update_organization(self):
response = self.client.patch(
f"/don_confiao/api/organizations/{self.organization.id}/",
{"description": "Nueva descripción"},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.organization.refresh_from_db()
self.assertEqual(
self.organization.description, "Nueva descripción"
)
def test_admin_can_delete_organization(self):
response = self.client.delete(
f"/don_confiao/api/organizations/{self.organization.id}/"
)
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
self.assertFalse(
Organization.objects.filter(pk=self.organization.id).exists()
)
def test_admin_can_create_supplier(self):
response = self.client.post(
"/don_confiao/api/suppliers/",
{
"name": "Proveedor 2",
"organization": self.organization.id,
"municipality": self.municipality.id,
},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
supplier = Supplier.objects.get(name="Proveedor 2")
self.assertEqual(supplier.organization, self.organization)
self.assertEqual(supplier.municipality, self.municipality)
def test_admin_can_unlink_supplier_organization(self):
response = self.client.patch(
f"/don_confiao/api/suppliers/{self.supplier.id}/",
{"organization": None},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.supplier.refresh_from_db()
self.assertIsNone(self.supplier.organization)
def test_admin_can_unlink_supplier_municipality(self):
response = self.client.patch(
f"/don_confiao/api/suppliers/{self.supplier.id}/",
{"municipality": None},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.supplier.refresh_from_db()
self.assertIsNone(self.supplier.municipality)
def test_admin_can_create_department(self):
response = self.client.post(
"/don_confiao/api/departments/",
{"name": "Antioquia", "country": self.country.id},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertEqual(
Department.objects.filter(name="Antioquia").count(), 1
)
def test_admin_can_create_municipality(self):
response = self.client.post(
"/don_confiao/api/municipalities/",
{
"name": "El Colegio",
"department": self.department.id,
"country": self.country.id,
},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertEqual(
Municipality.objects.filter(name="El Colegio").count(), 1
)
def test_supplier_serializer_includes_details(self):
response = self.client.get(
f"/don_confiao/api/suppliers/{self.supplier.id}/"
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data["organization_detail"]["name"], "Asociación")
self.assertEqual(data["municipality_detail"]["name"], "La Mesa")
self.assertIn("products", data)
def test_department_serializer_includes_country_detail(self):
response = self.client.get(
f"/don_confiao/api/departments/{self.department.id}/"
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data["country"], self.country.id)
self.assertEqual(data["country_detail"]["name"], "Colombia")
self.assertEqual(data["country_detail"]["code"], "CO")
def test_municipality_serializer_includes_fk_details(self):
response = self.client.get(
f"/don_confiao/api/municipalities/{self.municipality.id}/"
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data["department"], self.department.id)
self.assertEqual(data["department_detail"]["name"], "Cundinamarca")
self.assertEqual(data["country"], self.country.id)
self.assertEqual(data["country_detail"]["name"], "Colombia")
self.assertEqual(data["country_detail"]["code"], "CO")
def test_municipality_serializer_includes_geolocation(self):
self.municipality.latitude = 4.63092
self.municipality.longitude = -74.39152
self.municipality.save()
response = self.client.get(
f"/don_confiao/api/municipalities/{self.municipality.id}/"
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(float(data["latitude"]), 4.63092)
self.assertEqual(float(data["longitude"]), -74.39152)
def test_municipality_serializer_geolocation_optional(self):
response = self.client.get(
f"/don_confiao/api/municipalities/{self.municipality.id}/"
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertIsNone(data["latitude"])
self.assertIsNone(data["longitude"])
def test_link_product_to_suppliers_via_product_api(self):
product = Product.objects.create(name="Panela", price=5000)
supplier2 = Supplier.objects.create(name="Proveedor 2")
response = self.client.patch(
f"/don_confiao/api/products/{product.id}/",
{"suppliers": [self.supplier.id, supplier2.id]},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
product.refresh_from_db()
self.assertEqual(
set(product.suppliers.all()), {self.supplier, supplier2}
)
def test_unlink_product_supplier_via_product_api(self):
product = Product.objects.create(name="Panela", price=5000)
product.suppliers.add(self.supplier)
response = self.client.patch(
f"/don_confiao/api/products/{product.id}/",
{"suppliers": []},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
product.refresh_from_db()
self.assertEqual(product.suppliers.count(), 0)
class TestSeedGeography(APITestCase, LoginMixin):
url = "/don_confiao/api/seed_geography"
def _payload(self):
return {
"country": {"name": "Colombia", "code": "CO"},
"departments": [{"name": "Cundinamarca", "code": "CO-CUN"}],
"municipalities": [
{"name": "La Mesa", "department": "Cundinamarca"},
{"name": "El Colegio", "department": "Cundinamarca"},
],
}
def test_seed_requires_authentication(self):
response = self.client.post(
self.url, self._payload(), format="json"
)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
def test_seed_requires_admin(self):
user = _create_user("normal", "user")
client = _client_for(user)
response = client.post(self.url, self._payload(), format="json")
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
def test_seed_creates_geography(self):
self.login()
response = self.client.post(
self.url, self._payload(), format="json"
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data["country"], "Colombia")
self.assertEqual(data["total_departments"], 1)
self.assertEqual(data["total_municipalities"], 2)
def test_seed_is_idempotent(self):
self.login()
self.client.post(self.url, self._payload(), format="json")
response = self.client.post(
self.url, self._payload(), format="json"
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(Country.objects.count(), 1)
self.assertEqual(Department.objects.count(), 1)
self.assertEqual(Municipality.objects.count(), 2)
def test_seed_creates_municipality_with_geolocation(self):
self.login()
payload = self._payload()
payload["municipalities"][0] = {
"name": "La Mesa",
"department": "Cundinamarca",
"latitude": 4.63092,
"longitude": -74.39152,
}
response = self.client.post(self.url, payload, format="json")
self.assertEqual(response.status_code, status.HTTP_200_OK)
municipality = Municipality.objects.get(name="La Mesa")
self.assertEqual(float(municipality.latitude), 4.63092)
self.assertEqual(float(municipality.longitude), -74.39152)
def test_seed_updates_geolocation_on_reseed(self):
self.login()
self.client.post(self.url, self._payload(), format="json")
payload = self._payload()
payload["municipalities"][0] = {
"name": "La Mesa",
"department": "Cundinamarca",
"latitude": 4.63092,
"longitude": -74.39152,
}
response = self.client.post(self.url, payload, format="json")
self.assertEqual(response.status_code, status.HTTP_200_OK)
municipality = Municipality.objects.get(name="La Mesa")
self.assertEqual(float(municipality.latitude), 4.63092)
self.assertEqual(float(municipality.longitude), -74.39152)
def test_seed_requires_country_name(self):
self.login()
response = self.client.post(
self.url, {"departments": []}, format="json"
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
def test_seed_creates_duplicate_municipality_names_in_different_departments(
self,
):
self.login()
payload = {
"country": {"name": "Colombia", "code": "CO"},
"departments": [
{"name": "Cundinamarca", "code": "CO-CUN"},
{"name": "Antioquia", "code": "CO-ANT"},
],
"municipalities": [
{"name": "Bolivar", "department": "Cundinamarca"},
{"name": "Bolivar", "department": "Antioquia"},
],
}
response = self.client.post(self.url, payload, format="json")
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data["total_departments"], 2)
self.assertEqual(data["total_municipalities"], 2)
self.assertEqual(
Municipality.objects.filter(name="Bolivar").count(), 2
)

View File

@@ -0,0 +1,180 @@
from datetime import datetime, timezone
from rest_framework import status
from rest_framework.test import APITestCase
from ..models.customers import Customer
from ..models.geography import Country, Department, Municipality
from ..models.provenance import Organization, Supplier
from ..models.products import Product
from ..models.sales import CatalogSale, Sale
from .Mixins import LoginMixin
class TestProvenanceInSummaries(APITestCase, LoginMixin):
def setUp(self):
self.country = Country.objects.create(name="Colombia", code="CO")
self.department = Department.objects.create(
name="Cundinamarca", country=self.country
)
self.municipality = Municipality.objects.create(
name="La Mesa",
department=self.department,
country=self.country,
)
self.organization = Organization.objects.create(name="Asociación")
self.supplier = Supplier.objects.create(
name="Proveedor 1",
organization=self.organization,
municipality=self.municipality,
)
self.product = Product.objects.create(name="Panela", price=5000)
self.product.suppliers.add(self.supplier)
self.customer = Customer.objects.create(
name="Camilo", external_id="18"
)
def _create_sale(self):
sale = Sale.objects.create(
customer=self.customer,
date=datetime(2024, 9, 2, tzinfo=timezone.utc),
payment_method="CASH",
)
sale.saleline_set.create(
product=self.product, quantity=2, unit_price=3000
)
return sale
def _create_catalog_sale(self):
catalog_sale = CatalogSale.objects.create(
customer=self.customer,
date=datetime(2024, 9, 2, tzinfo=timezone.utc),
customer_name="Camilo",
)
catalog_sale.catalogsaleline_set.create(
product=self.product, quantity=2, unit_price=3000
)
return catalog_sale
def test_public_sale_summary_includes_product_provenance(self):
sale = self._create_sale()
response = self.client.get(
f"/don_confiao/resumen_publico/{sale.code}"
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertIn("product_provenance", data)
self.assertEqual(len(data["product_provenance"]), 1)
entry = data["product_provenance"][0]
self.assertEqual(entry["product"]["name"], "Panela")
self.assertEqual(len(entry["suppliers"]), 1)
supplier = entry["suppliers"][0]
self.assertEqual(supplier["supplier"]["name"], "Proveedor 1")
self.assertEqual(supplier["organization"]["name"], "Asociación")
self.assertEqual(supplier["municipality"]["name"], "La Mesa")
self.assertEqual(supplier["department"]["name"], "Cundinamarca")
self.assertEqual(supplier["country"]["name"], "Colombia")
def test_public_summary_municipality_includes_geolocation(self):
self.municipality.latitude = 4.63092
self.municipality.longitude = -74.39152
self.municipality.save()
sale = self._create_sale()
response = self.client.get(
f"/don_confiao/resumen_publico/{sale.code}"
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
supplier = response.json()["product_provenance"][0]["suppliers"][0]
self.assertEqual(
float(supplier["municipality"]["latitude"]), 4.63092
)
self.assertEqual(
float(supplier["municipality"]["longitude"]), -74.39152
)
def test_public_catalog_summary_includes_product_provenance(self):
catalog_sale = self._create_catalog_sale()
response = self.client.get(
f"/don_confiao/resumen_publico/{catalog_sale.code}"
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertIn("product_provenance", data)
self.assertEqual(
len(data["product_provenance"][0]["suppliers"]), 1
)
def test_authenticated_summaries_include_product_provenance(self):
self.login()
sale = self._create_sale()
response = self.client.get(
f"/don_confiao/resumen_compra_json/{sale.id}"
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn("product_provenance", response.json())
catalog_sale = self._create_catalog_sale()
response = self.client.get(
f"/don_confiao/resumen_compra_catalogo_json/{catalog_sale.id}"
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn("product_provenance", response.json())
def test_summary_existing_fields_are_preserved(self):
sale = self._create_sale()
response = self.client.get(
f"/don_confiao/resumen_publico/{sale.code}"
)
data = response.json()
for field in [
"id",
"code",
"date",
"customer",
"payment_method",
"lines",
"link",
"type",
]:
self.assertIn(field, data)
def test_product_without_suppliers_returns_empty_list(self):
product2 = Product.objects.create(name="Café", price=6000)
sale = self._create_sale()
sale.saleline_set.create(
product=product2, quantity=1, unit_price=6000
)
response = self.client.get(
f"/don_confiao/resumen_publico/{sale.code}"
)
data = response.json()
entry = next(
e
for e in data["product_provenance"]
if e["product"]["name"] == "Café"
)
self.assertEqual(entry["suppliers"], [])
def test_supplier_without_municipality_returns_null_territory(self):
supplier2 = Supplier.objects.create(
name="Proveedor sin ubicación"
)
product2 = Product.objects.create(name="Café", price=6000)
product2.suppliers.add(supplier2)
sale = self._create_sale()
sale.saleline_set.create(
product=product2, quantity=1, unit_price=6000
)
response = self.client.get(
f"/don_confiao/resumen_publico/{sale.code}"
)
data = response.json()
entry = next(
e
for e in data["product_provenance"]
if e["product"]["name"] == "Café"
)
supplier = entry["suppliers"][0]
self.assertIsNone(supplier["municipality"])
self.assertIsNone(supplier["department"])
self.assertIsNone(supplier["country"])

View File

@@ -0,0 +1,126 @@
from django.db import IntegrityError, transaction
from django.test import TestCase
from ..models.geography import Country, Department, Municipality
from ..models.provenance import Organization, Supplier
from ..models.products import Product
class TestOrganizationModel(TestCase):
def test_create_organization(self):
organization = Organization.objects.create(
name="Asociación La Mesa",
description="Cooperativa de campesinos",
)
self.assertEqual(organization.name, "Asociación La Mesa")
self.assertEqual(
organization.description, "Cooperativa de campesinos"
)
self.assertEqual(str(organization), "Asociación La Mesa")
def test_update_organization(self):
organization = Organization.objects.create(name="Asociación")
organization.description = "Nueva descripción"
organization.save()
organization.refresh_from_db()
self.assertEqual(organization.description, "Nueva descripción")
def test_delete_organization(self):
organization = Organization.objects.create(name="Asociación")
organization.delete()
self.assertFalse(
Organization.objects.filter(pk=organization.pk).exists()
)
def test_organization_name_is_unique(self):
Organization.objects.create(name="Asociación")
with self.assertRaises(IntegrityError):
with transaction.atomic():
Organization.objects.create(name="Asociación")
class TestSupplierModel(TestCase):
def setUp(self):
self.country = Country.objects.create(name="Colombia", code="CO")
self.department = Department.objects.create(
name="Cundinamarca", country=self.country
)
self.municipality = Municipality.objects.create(
name="La Mesa",
department=self.department,
country=self.country,
)
self.organization = Organization.objects.create(name="Asociación")
self.product = Product.objects.create(name="Panela", price=5000)
def test_create_supplier(self):
supplier = Supplier.objects.create(
name="Proveedor 1", municipality=self.municipality
)
self.assertEqual(supplier.municipality, self.municipality)
self.assertEqual(
supplier.municipality.department.country.name, "Colombia"
)
def test_supplier_organization_can_be_linked_and_unlinked(self):
supplier = Supplier.objects.create(name="Proveedor 1")
supplier.organization = self.organization
supplier.save()
supplier.refresh_from_db()
self.assertEqual(supplier.organization, self.organization)
supplier.organization = None
supplier.save()
supplier.refresh_from_db()
self.assertIsNone(supplier.organization)
def test_supplier_municipality_can_be_linked_and_unlinked(self):
supplier = Supplier.objects.create(name="Proveedor 1")
supplier.municipality = self.municipality
supplier.save()
supplier.refresh_from_db()
self.assertEqual(supplier.municipality, self.municipality)
supplier.municipality = None
supplier.save()
supplier.refresh_from_db()
self.assertIsNone(supplier.municipality)
def test_supplier_without_organization_or_municipality(self):
supplier = Supplier.objects.create(name="Proveedor 1")
self.assertIsNone(supplier.organization)
self.assertIsNone(supplier.municipality)
def test_supplier_name_is_unique(self):
Supplier.objects.create(name="Proveedor 1")
with self.assertRaises(IntegrityError):
with transaction.atomic():
Supplier.objects.create(name="Proveedor 1")
def test_delete_organization_unlinks_supplier(self):
supplier = Supplier.objects.create(
name="Proveedor 1", organization=self.organization
)
self.organization.delete()
supplier.refresh_from_db()
self.assertIsNone(supplier.organization)
def test_delete_municipality_unlinks_supplier(self):
supplier = Supplier.objects.create(
name="Proveedor 1", municipality=self.municipality
)
self.municipality.delete()
supplier.refresh_from_db()
self.assertIsNone(supplier.municipality)
def test_product_can_be_linked_to_multiple_suppliers(self):
supplier1 = Supplier.objects.create(name="Proveedor 1")
supplier2 = Supplier.objects.create(name="Proveedor 2")
self.product.suppliers.add(supplier1, supplier2)
self.assertEqual(self.product.suppliers.count(), 2)
self.product.suppliers.remove(supplier1)
self.assertEqual(list(self.product.suppliers.all()), [supplier2])
self.assertEqual(set(supplier2.products.all()), {self.product})

View File

@@ -0,0 +1,246 @@
import base64
from django.contrib.auth.models import User
from django.core.files.uploadedfile import SimpleUploadedFile
from django.test import override_settings
from rest_framework import status
from rest_framework.test import APIClient, APITestCase
from rest_framework_simplejwt.tokens import RefreshToken
from ..models.store_settings import StoreSettings
from .Mixins import LoginMixin
URL = "/don_confiao/api/store_settings"
PNG_1X1 = base64.b64decode(
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk"
"+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg=="
)
class TestStoreSettingsModel(APITestCase, LoginMixin):
def test_singleton_returns_single_instance(self):
first = StoreSettings.get_singleton()
second = StoreSettings.get_singleton()
self.assertEqual(first.pk, second.pk)
self.assertEqual(StoreSettings.objects.count(), 1)
def test_singleton_created_without_address(self):
instance = StoreSettings.get_singleton()
self.assertEqual(instance.address, "")
def test_str(self):
instance = StoreSettings.get_singleton()
self.assertEqual(str(instance), f"StoreSettings (id={instance.id})")
class TestStoreSettingsAPIPermissions(APITestCase, LoginMixin):
def setUp(self):
self.url = URL
def test_get_unauthenticated(self):
client = APIClient()
response = client.get(self.url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
def test_patch_unauthenticated(self):
client = APIClient()
response = client.patch(
self.url,
{"address": "Calle falsa 123"},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
def test_patch_non_admin(self):
user = User.objects.create_user(
username="regularuser",
email="regular@example.com",
password="regularpass",
)
refresh = RefreshToken.for_user(user)
client = APIClient()
client.credentials(
HTTP_AUTHORIZATION=f"Bearer {str(refresh.access_token)}"
)
response = client.patch(
self.url,
{"address": "Calle falsa 123"},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
def test_patch_publico(self):
user = User.objects.create_user(
username="publico",
email="publico@example.com",
password="publicopass",
)
user.profile.user_type = "publico"
user.profile.save()
refresh = RefreshToken.for_user(user)
client = APIClient()
client.credentials(
HTTP_AUTHORIZATION=f"Bearer {str(refresh.access_token)}"
)
response = client.patch(
self.url,
{"address": "Calle falsa 123"},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
class TestStoreSettingsAPI(APITestCase, LoginMixin):
def setUp(self):
self.login()
self.url = URL
def test_get_creates_singleton(self):
self.assertEqual(StoreSettings.objects.count(), 0)
response = self.client.get(self.url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(StoreSettings.objects.count(), 1)
data = response.json()
self.assertIn("id", data)
self.assertIn("address", data)
self.assertIn("latitude", data)
self.assertIn("longitude", data)
self.assertIn("logo", data)
self.assertIn("updated_at", data)
def test_patch_updates_settings(self):
self.client.get(self.url)
response = self.client.patch(
self.url,
{
"address": "Carrera 5 # 10-20, Bogotá",
"latitude": 4.6097,
"longitude": -74.0817,
},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data["address"], "Carrera 5 # 10-20, Bogotá")
self.assertEqual(data["latitude"], 4.6097)
self.assertEqual(data["longitude"], -74.0817)
settings = StoreSettings.objects.get(pk=1)
self.assertEqual(settings.address, "Carrera 5 # 10-20, Bogotá")
self.assertEqual(settings.latitude, 4.6097)
self.assertEqual(settings.longitude, -74.0817)
def test_patch_partial(self):
self.client.get(self.url)
response = self.client.patch(
self.url,
{"address": "Solo dirección"},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
settings = StoreSettings.objects.get(pk=1)
self.assertEqual(settings.address, "Solo dirección")
self.assertIsNone(settings.latitude)
def test_patch_reuses_singleton(self):
self.client.get(self.url)
self.client.patch(
self.url,
{"address": "Primera"},
format="json",
)
self.client.patch(
self.url,
{"address": "Segunda"},
format="json",
)
self.assertEqual(StoreSettings.objects.count(), 1)
def test_patch_invalid_coordinates(self):
self.client.get(self.url)
response = self.client.patch(
self.url,
{"latitude": "not-a-number"},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
def test_patch_upload_logo(self):
self.client.get(self.url)
logo = SimpleUploadedFile("logo.png", PNG_1X1, content_type="image/png")
response = self.client.patch(
self.url,
{"logo": logo},
format="multipart",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertIn("/media/logos/", data["logo"])
self.assertTrue(data["logo"].startswith("http"))
settings = StoreSettings.objects.get(pk=1)
self.assertIsNotNone(settings.logo)
self.assertTrue(settings.logo.name.startswith("logos/logo"))
def test_get_returns_absolute_logo_url(self):
self.client.get(self.url)
logo = SimpleUploadedFile("logo.png", PNG_1X1, content_type="image/png")
self.client.patch(self.url, {"logo": logo}, format="multipart")
response = self.client.get(self.url)
data = response.json()
self.assertTrue(data["logo"].startswith("http"))
def test_patch_logo_non_image(self):
self.client.get(self.url)
fake = SimpleUploadedFile("logo.txt", b"not an image", content_type="text/plain")
response = self.client.patch(
self.url,
{"logo": fake},
format="multipart",
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
@override_settings(LOGO_MAX_UPLOAD_SIZE=10)
def test_patch_logo_too_large(self):
self.client.get(self.url)
logo = SimpleUploadedFile("logo.png", PNG_1X1, content_type="image/png")
response = self.client.patch(
self.url,
{"logo": logo},
format="multipart",
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
settings = StoreSettings.objects.get(pk=1)
self.assertFalse(settings.logo)
def test_patch_clear_logo(self):
self.client.get(self.url)
logo = SimpleUploadedFile("logo.png", PNG_1X1, content_type="image/png")
self.client.patch(self.url, {"logo": logo}, format="multipart")
response = self.client.patch(
self.url,
{"logo": None},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertIsNone(data["logo"])
settings = StoreSettings.objects.get(pk=1)
self.assertFalse(settings.logo)
def test_patch_upload_logo_keeps_other_fields(self):
self.client.get(self.url)
logo = SimpleUploadedFile("logo.png", PNG_1X1, content_type="image/png")
response = self.client.patch(
self.url,
{
"address": "Avenida Siempre Viva 742",
"logo": logo,
},
format="multipart",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
settings = StoreSettings.objects.get(pk=1)
self.assertEqual(settings.address, "Avenida Siempre Viva 742")
self.assertIsNotNone(settings.logo)

View File

@@ -8,6 +8,8 @@ from .api import (
# Products
ProductView,
ProductsFromTrytonView,
# Product Categories
ProductCategoryView,
# Customers
CustomerView,
CustomersFromTrytonView,
@@ -16,6 +18,7 @@ from .api import (
CatalogSaleView,
SaleSummary,
CatalogSaleSummary,
PublicSaleSummaryView,
SalesForTrytonView,
SalesToTrytonView,
CatalogSalesToTrytonView,
@@ -26,6 +29,15 @@ from .api import (
SalesForReconciliationView,
# Admin
AdminCodeValidateView,
# Store Settings
StoreSettingsView,
# Provenance
OrganizationView,
SupplierView,
CountryView,
DepartmentView,
MunicipalityView,
SeedGeographyView,
)
app_name = "don_confiao"
@@ -35,6 +47,11 @@ router.register(r"sales", SaleView, basename="sale")
router.register(r"catalog_sales", CatalogSaleView, basename="catalog_sale")
router.register(r"customers", CustomerView, basename="customer")
router.register(r"products", ProductView, basename="product")
router.register(
r"product_categories",
ProductCategoryView,
basename="product_category",
)
router.register(
r"catalogue_images",
CatalogueImageViewSet,
@@ -45,6 +62,11 @@ router.register(
ReconciliateJarModelView,
basename="reconciliate_jar",
)
router.register(r"organizations", OrganizationView, basename="organization")
router.register(r"suppliers", SupplierView, basename="supplier")
router.register(r"countries", CountryView, basename="country")
router.register(r"departments", DepartmentView, basename="department")
router.register(r"municipalities", MunicipalityView, basename="municipality")
urlpatterns = [
path("productos", views.products, name="products"),
@@ -58,6 +80,16 @@ urlpatterns = [
CatalogSaleSummary.as_view(),
name="catalog_purchase_json_summary",
),
path(
"resumen_publico/<str:code>",
PublicSaleSummaryView.as_view(),
name="public_purchase_summary",
),
path(
"resumen_publico",
PublicSaleSummaryView.as_view(),
name="public_purchase_summary_query",
),
path(
"payment_methods/all/select_format",
PaymentMethodView.as_view(),
@@ -95,4 +127,10 @@ urlpatterns = [
AdminCodeValidateView.as_view(),
),
path("api/sales/for_tryton", SalesForTrytonView.as_view()),
path("api/store_settings", StoreSettingsView.as_view()),
path(
"api/seed_geography",
SeedGeographyView.as_view(),
name="seed_geography",
),
]

View File

@@ -1,3 +1,23 @@
from django.contrib import admin
from django.contrib.auth.models import User
from django.contrib.auth.admin import UserAdmin
# Register your models here.
from .models import UserProfile
class UserProfileInline(admin.StackedInline):
model = UserProfile
can_delete = False
class CustomUserAdmin(UserAdmin):
inlines = [UserProfileInline]
def get_inline_instances(self, request, obj=None):
if not obj:
return []
return super().get_inline_instances(request, obj)
admin.site.unregister(User)
admin.site.register(User, CustomUserAdmin)

View File

@@ -3,3 +3,6 @@ from django.apps import AppConfig
class UsersConfig(AppConfig):
name = 'users'
def ready(self):
import users.signals # noqa

View File

@@ -0,0 +1,25 @@
# Generated by Django 5.0.6 on 2026-06-22 17:26
import django.db.models.deletion
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.CreateModel(
name='UserProfile',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('user_type', models.CharField(choices=[('publico', 'Público'), ('user', 'Usuario'), ('administrator', 'Administrador')], default='user', max_length=20)),
('user', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='profile', to=settings.AUTH_USER_MODEL)),
],
),
]

View File

@@ -0,0 +1,24 @@
from django.db import migrations
def backfill_user_profiles(apps, schema_editor):
User = apps.get_model("auth", "User")
UserProfile = apps.get_model("users", "UserProfile")
for user in User.objects.all():
UserProfile.objects.get_or_create(
user=user,
defaults={
"user_type": "administrator" if user.is_staff else "user",
},
)
class Migration(migrations.Migration):
dependencies = [
('users', '0001_initial'),
]
operations = [
migrations.RunPython(backfill_user_profiles, migrations.RunPython.noop),
]

View File

@@ -1,3 +1,19 @@
from django.db import models
from django.contrib.auth.models import User
# Create your models here.
class UserProfile(models.Model):
USER_TYPE_CHOICES = [
("publico", "Público"),
("user", "Usuario"),
("administrator", "Administrador"),
]
user = models.OneToOneField(
User, on_delete=models.CASCADE, related_name="profile"
)
user_type = models.CharField(
max_length=20, choices=USER_TYPE_CHOICES, default="user"
)
def __str__(self):
return f"{self.user.username} - {self.user_type}"

View File

@@ -1,13 +1,25 @@
from django.contrib.auth.models import User
from rest_framework import serializers
from .models import UserProfile
class UserSerializer(serializers.ModelSerializer):
role = serializers.SerializerMethodField()
class Meta:
model = User
fields = ('id', 'username', 'email', 'first_name', 'last_name', 'role')
fields = (
"id",
"username",
"email",
"first_name",
"last_name",
"role",
)
def get_role(self, obj):
return 'administrator' if obj.is_staff else 'user'
try:
return obj.profile.user_type
except UserProfile.DoesNotExist:
return "administrator" if obj.is_staff else "user"

View File

@@ -0,0 +1,12 @@
from django.db.models.signals import post_save
from django.dispatch import receiver
from django.contrib.auth.models import User
from .models import UserProfile
@receiver(post_save, sender=User)
def create_user_profile(sender, instance, created, **kwargs):
if created:
user_type = "administrator" if instance.is_staff else "user"
UserProfile.objects.create(user=instance, user_type=user_type)

View File

@@ -4,6 +4,8 @@ from django.contrib.auth.models import User
from rest_framework.test import APIClient
from rest_framework_simplejwt.tokens import RefreshToken
from .models import UserProfile
class MeEndpointTests(TestCase):
def setUp(self):
@@ -49,7 +51,6 @@ class MeEndpointTests(TestCase):
username='regular',
email='regular@example.com',
password='regularpass',
is_staff=False
)
refresh = RefreshToken.for_user(regular_user)
@@ -87,6 +88,31 @@ class MeEndpointTests(TestCase):
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()['role'], 'administrator')
def test_publico_user_role_is_publico(self):
"""
Verifica que un usuario con user_type='publico' recibe role 'publico'.
"""
publico_user = User.objects.create_user(
username='publico',
email='publico@example.com',
password='publicopass',
)
profile = UserProfile.objects.get(user=publico_user)
profile.user_type = 'publico'
profile.save()
refresh = RefreshToken.for_user(publico_user)
access_token = str(refresh.access_token)
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f'Bearer {access_token}')
url = reverse('current-user')
response = client.get(url)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()['role'], 'publico')
def test_me_endpoint_requires_authentication(self):
"""
Sin token el endpoint debe devolver 401 Unauthorized.