7 Commits

Author SHA1 Message Date
a786f307fd chore: update db variable enviroment. 2026-10-24 15:32:48 -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
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
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
d7132733fc Update Admin list view 2026-08-09 20:54:16 -05:00
10 changed files with 1082 additions and 42 deletions

View File

@@ -24,10 +24,10 @@ CSRF_TRUSTED_ORIGINS=http://localhost:8000,http://127.0.0.1:8000
# PostgreSQL Database Configuration (staging local)
DB_NAME=tienda_ilusion_staging
DB_USER=tienda_ilusion_user
DB_PASSWORD=staging_local_password
DB_HOST=postgres
DB_PORT=5432
POSTGRES_DB=tienda_ilusion_staging
POSTGRES_USER=tienda_ilusion_user
POSTGRES_PASSWORD=tienda_ilusion_password
# Email Configuration (console backend para staging)
EMAIL_HOST=smtp.gmail.com

View File

@@ -21,6 +21,7 @@ class StoreSettingsAdmin(admin.ModelAdmin):
@admin.register(Customer)
class CustomerAdmin(admin.ModelAdmin):
list_display = (
"id",
"name",
"email",
"phone",
@@ -72,7 +73,7 @@ class CatalogSaleLineAdmin(admin.ModelAdmin):
@admin.register(Product)
class ProductAdmin(admin.ModelAdmin):
list_display = ("name", "price", "measuring_unit", "external_id")
list_display = ("id", "name", "price", "measuring_unit", "external_id")
search_fields = ("id", "name",)
list_filter = ("name", "id")

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

@@ -8,6 +8,7 @@ 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)
def __str__(self):
return self.name

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,4 @@
from ...models.products import Product
from ...models.products import Product, ProductCategory
class ProductTrytonService:
@@ -20,8 +20,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 +45,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 +96,155 @@ 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:
external_id = str(tryton_category.get("id"))
name = tryton_category.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()
old_name = category.name
old_external_id = category.external_id
category.name = name
category.external_id = external_id
category.save()
categories[tryton_category.get("id")] = category
item = {
"id": category.id,
"name": category.name,
"external_id": category.external_id,
}
if is_new:
report["created"].append(item)
continue
changes = []
if old_name != category.name:
changes.append(f"Nombre: {old_name} → {category.name}")
if old_external_id != category.external_id:
changes.append(
"External ID: "
f"{old_external_id} → {category.external_id}"
)
if changes:
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}"
)
return ", ".join(changes)
@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 +253,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)

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

@@ -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

@@ -79,7 +79,13 @@ class TestSendSalesToTryton(TestCase, LoginMixin):
self.assertEqual(response.status_code, 200)
content = json.loads(response.content.decode("utf-8"))
self.assertEqual(
content, {"successful": [self.sale.id], "failed": []}
content,
{
"successful": [
{"id": self.sale.id, "code": self.sale.code}
],
"failed": [],
},
)
updated_sale = Sale.objects.get(id=self.sale.id)
@@ -165,7 +171,15 @@ class TestSendSalesToTryton(TestCase, LoginMixin):
content = json.loads(response.content.decode("utf-8"))
self.assertEqual(
content,
{"successful": [self.catalog_sale.id], "failed": []},
{
"successful": [
{
"id": self.catalog_sale.id,
"code": self.catalog_sale.code,
}
],
"failed": [],
},
)
updated = CatalogSale.objects.get(id=self.catalog_sale.id)
@@ -247,3 +261,51 @@ class TestSendSalesToTryton(TestCase, LoginMixin):
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

@@ -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,564 @@ 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_to_untouched_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 [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"], [{"id": 2, "name": "Papa"}]
)
untouched_product = Product.objects.get(id=2)
categories = untouched_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",
}
],
)