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) # PostgreSQL Database Configuration (staging local)
DB_NAME=tienda_ilusion_staging DB_NAME=tienda_ilusion_staging
DB_USER=tienda_ilusion_user
DB_PASSWORD=staging_local_password
DB_HOST=postgres
DB_PORT=5432 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 Configuration (console backend para staging)
EMAIL_HOST=smtp.gmail.com EMAIL_HOST=smtp.gmail.com

View File

@@ -21,6 +21,7 @@ class StoreSettingsAdmin(admin.ModelAdmin):
@admin.register(Customer) @admin.register(Customer)
class CustomerAdmin(admin.ModelAdmin): class CustomerAdmin(admin.ModelAdmin):
list_display = ( list_display = (
"id",
"name", "name",
"email", "email",
"phone", "phone",
@@ -72,7 +73,7 @@ class CatalogSaleLineAdmin(admin.ModelAdmin):
@admin.register(Product) @admin.register(Product)
class ProductAdmin(admin.ModelAdmin): 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",) search_fields = ("id", "name",)
list_filter = ("name", "id") 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): class ProductCategory(models.Model):
name = models.CharField(max_length=100, unique=True) name = models.CharField(max_length=100, unique=True)
external_id = models.CharField(max_length=100, null=True, blank=True)
def __str__(self): def __str__(self):
return self.name return self.name

View File

@@ -15,7 +15,10 @@ class CustomerTrytonService:
party_ids = self.client.call(method, params) party_ids = self.client.call(method, params)
tryton_parties = self._get_party_details(party_ids, context) 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 = [] failed_parties = []
updated_customers = [] updated_customers = []
created_customers = [] created_customers = []
@@ -25,15 +28,35 @@ class CustomerTrytonService:
try: try:
customer = Customer.objects.get(external_id=tryton_party.get("id")) customer = Customer.objects.get(external_id=tryton_party.get("id"))
except Customer.DoesNotExist: except Customer.DoesNotExist:
try:
customer = self._create_customer(tryton_party) customer = self._create_customer(tryton_party)
created_customers.append(customer.id) 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 continue
try:
if self._need_update(customer, tryton_party): if self._need_update(customer, tryton_party):
changes = self._describe_changes(customer, tryton_party)
self._update_customer(customer, tryton_party) self._update_customer(customer, tryton_party)
updated_customers.append(customer.id) updated_customers.append(
{**self._build_item(customer), "detail": changes}
)
else: else:
untouched_customers.append(customer.id) 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 { return {
"checked_tryton_parties": checked_tryton_parties, "checked_tryton_parties": checked_tryton_parties,
@@ -43,6 +66,41 @@ class CustomerTrytonService:
"untouched_customers": untouched_customers, "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): def _get_party_details(self, party_ids, context):
"""Obtiene detalles de clientes desde Tryton""" """Obtiene detalles de clientes desde Tryton"""
tryton_fields = ["id", "name", "addresses"] tryton_fields = ["id", "name", "addresses"]

View File

@@ -1,4 +1,4 @@
from ...models.products import Product from ...models.products import Product, ProductCategory
class ProductTrytonService: class ProductTrytonService:
@@ -21,7 +21,23 @@ class ProductTrytonService:
product_ids = self.client.call(method, params) product_ids = self.client.call(method, params)
tryton_products = self._get_product_details(product_ids, context) 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 = [] failed_products = []
updated_products = [] updated_products = []
created_products = [] created_products = []
@@ -29,24 +45,50 @@ class ProductTrytonService:
for tryton_product in tryton_products: for tryton_product in tryton_products:
try: try:
product = Product.objects.get(external_id=tryton_product.get("id")) product = Product.objects.get(
external_id=tryton_product.get("id")
)
except Product.DoesNotExist: except Product.DoesNotExist:
try: try:
product = self._create_product(tryton_product) 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 continue
except Exception as e: except Exception as e:
print( print(
f"Error al importar productos: {e}El producto: {tryton_product}" 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 continue
try:
if self._need_update(product, tryton_product): if self._need_update(product, tryton_product):
changes = self._describe_changes(
product, tryton_product
)
self._update_product(product, tryton_product) self._update_product(product, tryton_product)
updated_products.append(product.id) self._sync_product_categories(
product, tryton_product, categories_by_tryton_id
)
updated_products.append(
{**self._build_item(product), "detail": changes}
)
else: else:
untouched_products.append(product.id) 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 { return {
"checked_tryton_products": checked_tryton_products, "checked_tryton_products": checked_tryton_products,
@@ -54,8 +96,155 @@ class ProductTrytonService:
"updated_products": updated_products, "updated_products": updated_products,
"created_products": created_products, "created_products": created_products,
"untouched_products": untouched_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): def _get_product_details(self, product_ids, context):
"""Obtiene detalles de productos desde Tryton""" """Obtiene detalles de productos desde Tryton"""
tryton_fields = [ tryton_fields = [
@@ -64,6 +253,7 @@ class ProductTrytonService:
"default_uom.id", "default_uom.id",
"default_uom.rec_name", "default_uom.rec_name",
"list_price", "list_price",
"template.categories",
] ]
method = "model.product.product.read" method = "model.product.product.read"
params = (product_ids, tryton_fields, context) params = (product_ids, tryton_fields, context)

View File

@@ -27,10 +27,10 @@ class SaleTrytonService:
external_ids = self.client.call(method, tryton_params) external_ids = self.client.call(method, tryton_params)
sale.external_id = external_ids[0] sale.external_id = external_ids[0]
sale.save() sale.save()
successful.append(sale.id) successful.append(self._build_item(sale))
except Exception as e: except Exception as e:
print(f"Error al enviar la venta: {e}venta_id: {sale.id}") print(f"Error al enviar la venta: {e}venta_id: {sale.id}")
failed.append(sale.id) failed.append(self._build_failure(sale, e))
continue continue
return {"successful": successful, "failed": failed} return {"successful": successful, "failed": failed}
@@ -61,16 +61,28 @@ class SaleTrytonService:
external_ids = self.client.call(method, tryton_params) external_ids = self.client.call(method, tryton_params)
catalog_sale.external_id = external_ids[0] catalog_sale.external_id = external_ids[0]
catalog_sale.save() catalog_sale.save()
successful.append(catalog_sale.id) successful.append(self._build_item(catalog_sale))
except Exception as e: except Exception as e:
print( print(
f"Error al enviar catalog sale: {e}, catalog_sale_id: {catalog_sale.id}" 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 continue
return {"successful": successful, "failed": failed} 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): def _catalog_sale_to_tryton_params(self, catalog_sale, lines, tryton_context):
"""Convierte catalog sale a parámetros para Tryton""" """Convierte catalog sale a parámetros para Tryton"""
sale_tryton = TrytonCatalogSale(catalog_sale, lines) sale_tryton = TrytonCatalogSale(catalog_sale, lines)

View File

@@ -62,11 +62,25 @@ class TestCustomersFromTryton(TestCase, LoginMixin):
content = json.loads(response.content.decode("utf-8")) content = json.loads(response.content.decode("utf-8"))
expected_response = { expected_response = {
"checked_tryton_parties": [5, 6, 7, 8], "checked_tryton_parties": [
"created_customers": [3, 4], {"id": 5, "name": "Carlos"},
"untouched_customers": [2], {"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": [], "failed_parties": [],
"updated_customers": [1], "updated_customers": [
{
"id": 1,
"name": "Carlos",
"detail": "Nombre: Calos → Carlos, Dirección: None → 303",
}
],
} }
self.assertEqual(content, expected_response) 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.external_id, str(5))
self.assertEqual(updated_customer.name, "Carlos") self.assertEqual(updated_customer.name, "Carlos")
self.assertIn(updated_customer.address_external_id, str(303)) 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) self.assertEqual(response.status_code, 200)
content = json.loads(response.content.decode("utf-8")) content = json.loads(response.content.decode("utf-8"))
self.assertEqual( 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) 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")) content = json.loads(response.content.decode("utf-8"))
self.assertEqual( self.assertEqual(
content, 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) updated = CatalogSale.objects.get(id=self.catalog_sale.id)
@@ -247,3 +261,51 @@ class TestSendSalesToTryton(TestCase, LoginMixin):
lines[1]["unit_price"], lines[1]["unit_price"],
{"__class__": "Decimal", "decimal": "5000.00"}, {"__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 unittest.mock import patch
from django.test import TestCase from django.test import TestCase
from ..models.products import Product from ..models.products import Product, ProductCategory
from .Mixins import LoginMixin from .Mixins import LoginMixin
@@ -55,6 +55,7 @@ class TestProductsFromTryton(TestCase, LoginMixin):
"default_uom.id", "default_uom.id",
"default_uom.rec_name", "default_uom.rec_name",
"list_price", "list_price",
"template.categories",
], ],
{"company": 1}, {"company": 1},
) )
@@ -65,18 +66,21 @@ class TestProductsFromTryton(TestCase, LoginMixin):
"list_price": Decimal("25000"), "list_price": Decimal("25000"),
"name": "Producto 1", "name": "Producto 1",
"default_uom.": {"id": 1, "rec_name": "Unit"}, "default_uom.": {"id": 1, "rec_name": "Unit"},
"template.": {"id": 999, "categories": []},
}, },
{ {
"id": 191, "id": 191,
"list_price": Decimal("6000"), "list_price": Decimal("6000"),
"name": "Panela2", "name": "Panela2",
"default_uom.": {"id": 1, "rec_name": "Unit"}, "default_uom.": {"id": 1, "rec_name": "Unit"},
"template.": {"id": 999, "categories": []},
}, },
{ {
"id": 192, "id": 192,
"list_price": Decimal("4500"), "list_price": Decimal("4500"),
"name": "Papa", "name": "Papa",
"default_uom.": {"id": 2, "rec_name": "Kilogram"}, "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")) content = json.loads(response.content.decode("utf-8"))
expected_response = { expected_response = {
"checked_tryton_products": [190, 191, 192], "checked_tryton_products": [
"created_products": [3], {"id": 190, "name": "Producto 1"},
"untouched_products": [2], {"id": 191, "name": "Panela2"},
{"id": 192, "name": "Papa"},
],
"created_products": [{"id": 3, "name": "Producto 1"}],
"untouched_products": [{"id": 2, "name": "Papa"}],
"failed_products": [], "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) self.assertEqual(content, expected_response)
@@ -140,6 +156,7 @@ class TestProductsFromTryton(TestCase, LoginMixin):
"default_uom.id", "default_uom.id",
"default_uom.rec_name", "default_uom.rec_name",
"list_price", "list_price",
"template.categories",
], ],
{"company": 1}, {"company": 1},
) )
@@ -150,6 +167,7 @@ class TestProductsFromTryton(TestCase, LoginMixin):
"list_price": Decimal("25000"), "list_price": Decimal("25000"),
"name": self.product.name, "name": self.product.name,
"default_uom.": {"id": 1, "rec_name": "Unit"}, "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")) content = json.loads(response.content.decode("utf-8"))
expected_response = { expected_response = {
"checked_tryton_products": [200], "checked_tryton_products": [{"id": 200, "name": "Panela"}],
"created_products": [], "created_products": [],
"untouched_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": [], "updated_products": [],
"created_categories": [],
"updated_categories": [],
} }
self.assertEqual(content, expected_response) 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",
}
],
)