feat/49-provenance Implementa trazabilidad de proveedores #49 #50

Merged
mono merged 6 commits from feat/49-provenance into main 2026-08-16 22:17:35 -05:00
23 changed files with 2841 additions and 7 deletions

2
.gitignore vendored
View File

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

View File

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

File diff suppressed because it is too large Load Diff

188
scripts/seed_geography.py Normal file
View File

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

View File

@@ -20,6 +20,14 @@ from .payments import (
)
from .admin import AdminCodeValidateView
from .store_settings import StoreSettingsView
from .provenance import (
OrganizationView,
SupplierView,
CountryView,
DepartmentView,
MunicipalityView,
SeedGeographyView,
)
__all__ = [
# Catalogue Images
@@ -49,4 +57,11 @@ __all__ = [
"AdminCodeValidateView",
# Store Settings
"StoreSettingsView",
# Provenance
"OrganizationView",
"SupplierView",
"CountryView",
"DepartmentView",
"MunicipalityView",
"SeedGeographyView",
]

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@@ -26,6 +26,9 @@ class Product(models.Model):
max_length=100, null=True, blank=True
)
categories = models.ManyToManyField(ProductCategory)
suppliers = models.ManyToManyField(
"Supplier", blank=True, related_name="products"
)
external_id = models.CharField(max_length=100, null=True, blank=True)
def __str__(self):

View File

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

View File

@@ -17,6 +17,8 @@ from .payments import (
PaymentMethodSerializer,
)
from .store_settings import StoreSettingsSerializer
from .geography import CountrySerializer, DepartmentSerializer, MunicipalitySerializer
from .provenance import OrganizationSerializer, SupplierSerializer
__all__ = [
# Catalogue Images
@@ -42,4 +44,11 @@ __all__ = [
"PaymentMethodSerializer",
# Store Settings
"StoreSettingsSerializer",
# Geography
"CountrySerializer",
"DepartmentSerializer",
"MunicipalitySerializer",
# Provenance
"OrganizationSerializer",
"SupplierSerializer",
]

View File

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

View File

@@ -1,10 +1,14 @@
from rest_framework import serializers
from ..models.products import Product, ProductCategory
from ..models.provenance import Supplier
class ProductSerializer(serializers.ModelSerializer):
catalogue_images = serializers.SerializerMethodField()
suppliers = serializers.PrimaryKeyRelatedField(
many=True, queryset=Supplier.objects.all(), required=False
)
class Meta:
model = Product
@@ -15,6 +19,7 @@ class ProductSerializer(serializers.ModelSerializer):
"price",
"measuring_unit",
"categories",
"suppliers",
"external_id",
"catalogue_images",
]

View File

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

View File

@@ -1,6 +1,7 @@
from rest_framework import serializers
from django.urls import reverse
from ..models.products import Product
from ..models.sales import (
Sale,
SaleLine,
@@ -10,6 +11,7 @@ from ..models.sales import (
)
from .products import ListProductSerializer
from .customers import ListCustomerSerializer
from ..services.provenance import build_product_provenance
class PublicSummaryLinkMixin:
@@ -23,6 +25,16 @@ class PublicSummaryLinkMixin:
return path
class ProvenanceSummaryMixin:
def get_product_provenance(self, obj):
products = Product.objects.filter(
saleline__sale=obj
).distinct()
return build_product_provenance(
products, self.context.get("request")
)
class SaleLineSerializer(serializers.ModelSerializer):
class Meta:
model = SaleLine
@@ -101,14 +113,26 @@ class SummarySaleLineSerializer(serializers.ModelSerializer):
fields = ["product", "quantity", "unit_price", "description"]
class SaleSummarySerializer(PublicSummaryLinkMixin, serializers.ModelSerializer):
class SaleSummarySerializer(
PublicSummaryLinkMixin, ProvenanceSummaryMixin, serializers.ModelSerializer
):
customer = ListCustomerSerializer()
lines = SummarySaleLineSerializer(many=True, source="saleline_set")
link = serializers.SerializerMethodField()
product_provenance = serializers.SerializerMethodField()
class Meta:
model = Sale
fields = ["id", "code", "date", "customer", "payment_method", "lines", "link"]
fields = [
"id",
"code",
"date",
"customer",
"payment_method",
"lines",
"link",
"product_provenance",
]
class CatalogSummarySaleLineSerializer(serializers.ModelSerializer):
@@ -119,18 +143,31 @@ class CatalogSummarySaleLineSerializer(serializers.ModelSerializer):
fields = ["product", "quantity", "unit_price", "description"]
class CatalogProvenanceSummaryMixin(ProvenanceSummaryMixin):
def get_product_provenance(self, obj):
products = Product.objects.filter(
catalogsaleline__catalog_sale=obj
).distinct()
return build_product_provenance(
products, self.context.get("request")
)
class CatalogSaleSummarySerializer(
PublicSummaryLinkMixin, serializers.ModelSerializer
PublicSummaryLinkMixin,
CatalogProvenanceSummaryMixin,
serializers.ModelSerializer,
):
customer = ListCustomerSerializer()
lines = CatalogSummarySaleLineSerializer(
many=True, source="catalogsaleline_set"
)
link = serializers.SerializerMethodField()
product_provenance = serializers.SerializerMethodField()
class Meta:
model = CatalogSale
fields = ["id", "code", "date", "customer", "lines", "link"]
fields = ["id", "code", "date", "customer", "lines", "link", "product_provenance"]
class SaleForRenconciliationSerializer(serializers.Serializer):

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@@ -29,6 +29,13 @@ from .api import (
AdminCodeValidateView,
# Store Settings
StoreSettingsView,
# Provenance
OrganizationView,
SupplierView,
CountryView,
DepartmentView,
MunicipalityView,
SeedGeographyView,
)
app_name = "don_confiao"
@@ -48,6 +55,11 @@ router.register(
ReconciliateJarModelView,
basename="reconciliate_jar",
)
router.register(r"organizations", OrganizationView, basename="organization")
router.register(r"suppliers", SupplierView, basename="supplier")
router.register(r"countries", CountryView, basename="country")
router.register(r"departments", DepartmentView, basename="department")
router.register(r"municipalities", MunicipalityView, basename="municipality")
urlpatterns = [
path("productos", views.products, name="products"),
@@ -109,4 +121,9 @@ urlpatterns = [
),
path("api/sales/for_tryton", SalesForTrytonView.as_view()),
path("api/store_settings", StoreSettingsView.as_view()),
path(
"api/seed_geography",
SeedGeographyView.as_view(),
name="seed_geography",
),
]