1 Commits

Author SHA1 Message Date
a786f307fd chore: update db variable enviroment. 2026-10-24 15:32:48 -05:00
16 changed files with 39 additions and 1063 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

@@ -137,49 +137,3 @@ post importar_productos_de_tryton
#+begin_src json #+begin_src json
[] []
#+end_src #+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

@@ -77,12 +77,6 @@ class ProductAdmin(admin.ModelAdmin):
search_fields = ("id", "name",) search_fields = ("id", "name",)
list_filter = ("name", "id") list_filter = ("name", "id")
admin.site.register(ProductCategory)
@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(Payment) admin.site.register(Payment)
admin.site.register(ReconciliationJar) admin.site.register(ReconciliationJar)

View File

@@ -1,6 +1,5 @@
from .catalogue_images import CatalogueImageViewSet from .catalogue_images import CatalogueImageViewSet
from .products import ProductView, ProductsFromTrytonView from .products import ProductView, ProductsFromTrytonView
from .categories import ProductCategoryView
from .customers import CustomerView, CustomersFromTrytonView from .customers import CustomerView, CustomersFromTrytonView
from .sales import ( from .sales import (
SaleView, SaleView,
@@ -36,8 +35,6 @@ __all__ = [
# Products # Products
"ProductView", "ProductView",
"ProductsFromTrytonView", "ProductsFromTrytonView",
# Product Categories
"ProductCategoryView",
# Customers # Customers
"CustomerView", "CustomerView",
"CustomersFromTrytonView", "CustomersFromTrytonView",

View File

@@ -1,85 +0,0 @@
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

@@ -1,28 +0,0 @@
# 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

@@ -9,9 +9,6 @@ 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) 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): def __str__(self):
return self.name return self.name
@@ -49,15 +46,7 @@ class Product(models.Model):
"price_list": product.price, "price_list": product.price,
"uom": product.measuring_unit, "uom": product.measuring_unit,
"external_id": product.external_id, "external_id": product.external_id,
"categories": [ "categories": [c.name for c in product.categories.all()],
{
"id": c.id,
"name": c.name,
"alternative_name": c.alternative_name,
"active": c.active,
}
for c in product.categories.all()
],
} }
products_list.append(rproduct) products_list.append(rproduct)
return products_list return products_list

View File

@@ -1,10 +1,5 @@
from .catalogue_images import CatalogueImageSerializer from .catalogue_images import CatalogueImageSerializer
from .products import ( from .products import ProductSerializer, ListProductSerializer
ProductSerializer,
ListProductSerializer,
ProductCategorySerializer,
ProductCategoryBriefSerializer,
)
from .customers import CustomerSerializer, ListCustomerSerializer from .customers import CustomerSerializer, ListCustomerSerializer
from .sales import ( from .sales import (
SaleSerializer, SaleSerializer,
@@ -31,8 +26,6 @@ __all__ = [
# Products # Products
"ProductSerializer", "ProductSerializer",
"ListProductSerializer", "ListProductSerializer",
"ProductCategorySerializer",
"ProductCategoryBriefSerializer",
# Customers # Customers
"CustomerSerializer", "CustomerSerializer",
"ListCustomerSerializer", "ListCustomerSerializer",

View File

@@ -4,32 +4,11 @@ from ..models.products import Product, ProductCategory
from ..models.provenance import Supplier 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): class ProductSerializer(serializers.ModelSerializer):
catalogue_images = serializers.SerializerMethodField() catalogue_images = serializers.SerializerMethodField()
suppliers = serializers.PrimaryKeyRelatedField( suppliers = serializers.PrimaryKeyRelatedField(
many=True, queryset=Supplier.objects.all(), required=False many=True, queryset=Supplier.objects.all(), required=False
) )
categories = ProductCategoryBriefSerializer(many=True, read_only=True)
class Meta: class Meta:
model = Product model = Product

View File

@@ -1,147 +0,0 @@
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

@@ -1,5 +1,4 @@
from ...models.products import Product, ProductCategory from ...models.products import Product, ProductCategory
from .categories import upsert_category
class ProductTrytonService: class ProductTrytonService:
@@ -122,16 +121,42 @@ class ProductTrytonService:
) )
for tryton_category in tryton_categories: for tryton_category in tryton_categories:
try: try:
category, status, changes = upsert_category(tryton_category) 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 categories[tryton_category.get("id")] = category
item = { item = {
"id": category.id, "id": category.id,
"name": category.name, "name": category.name,
"external_id": category.external_id, "external_id": category.external_id,
} }
if status == "created": if is_new:
report["created"].append(item) report["created"].append(item)
elif status == "updated": 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( report["updated"].append(
{**item, "detail": ", ".join(changes)} {**item, "detail": ", ".join(changes)}
) )
@@ -189,50 +214,8 @@ class ProductTrytonService:
changes.append( changes.append(
f"Unidad: {product.measuring_unit} → {unit_name}" 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) 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 @staticmethod
def _format_price(value): def _format_price(value):
"""Formatea un precio sin ceros decimales innecesarios""" """Formatea un precio sin ceros decimales innecesarios"""
@@ -286,17 +269,6 @@ class ProductTrytonService:
unit = tryton_product.get("default_uom.") unit = tryton_product.get("default_uom.")
if not product.measuring_unit == unit.get("rec_name"): if not product.measuring_unit == unit.get("rec_name"):
return True 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 return False
def _create_product(self, tryton_product): def _create_product(self, tryton_product):

View File

@@ -1,200 +0,0 @@
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

@@ -1,102 +0,0 @@
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

@@ -1,224 +0,0 @@
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

@@ -661,7 +661,7 @@ class TestProductsFromTryton(TestCase, LoginMixin):
@patch("sabatron_tryton_rpc_client.client.Client.call") @patch("sabatron_tryton_rpc_client.client.Client.call")
@patch("sabatron_tryton_rpc_client.client.Client.connect") @patch("sabatron_tryton_rpc_client.client.Client.connect")
def test_import_assigns_categories_marks_product_as_updated( def test_import_assigns_categories_to_untouched_products(
self, mock_connect, mock_call self, mock_connect, mock_call
): ):
mock_connect.return_value = None mock_connect.return_value = None
@@ -724,20 +724,12 @@ class TestProductsFromTryton(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(content["untouched_products"], [])
self.assertEqual( self.assertEqual(
content["updated_products"], content["untouched_products"], [{"id": 2, "name": "Papa"}]
[
{
"id": 2,
"name": "Papa",
"detail": "Categoría: +Tubérculos",
}
],
) )
updated_product = Product.objects.get(id=2) untouched_product = Product.objects.get(id=2)
categories = updated_product.categories.all() categories = untouched_product.categories.all()
self.assertEqual(len(categories), 1) self.assertEqual(len(categories), 1)
self.assertEqual(categories[0].name, "Tubérculos") self.assertEqual(categories[0].name, "Tubérculos")
self.assertEqual(categories[0].external_id, "9") self.assertEqual(categories[0].external_id, "9")
@@ -752,104 +744,3 @@ class TestProductsFromTryton(TestCase, LoginMixin):
} }
], ],
) )
@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

@@ -8,8 +8,6 @@ from .api import (
# Products # Products
ProductView, ProductView,
ProductsFromTrytonView, ProductsFromTrytonView,
# Product Categories
ProductCategoryView,
# Customers # Customers
CustomerView, CustomerView,
CustomersFromTrytonView, CustomersFromTrytonView,
@@ -47,11 +45,6 @@ router.register(r"sales", SaleView, basename="sale")
router.register(r"catalog_sales", CatalogSaleView, basename="catalog_sale") router.register(r"catalog_sales", CatalogSaleView, basename="catalog_sale")
router.register(r"customers", CustomerView, basename="customer") router.register(r"customers", CustomerView, basename="customer")
router.register(r"products", ProductView, basename="product") router.register(r"products", ProductView, basename="product")
router.register(
r"product_categories",
ProductCategoryView,
basename="product_category",
)
router.register( router.register(
r"catalogue_images", r"catalogue_images",
CatalogueImageViewSet, CatalogueImageViewSet,