feat(#49): add product provenance (suppliers, organizations, geography)

- Country, Department, Municipality models (municipality unique per department)
- Organization and Supplier models, Supplier links to organization and municipality
- Product.suppliers M2M to link products to one or more suppliers
- Authenticated CRUD APIs for organizations, suppliers, countries, departments and municipalities (write only for administrators)
- product_provenance domain data embedded in sale/catalog summaries (public and authenticated)
- seed_geography endpoint (admin, idempotent) and scripts/seed_geography.py using pycountry + DANE municipalities CSV
- TDD tests for models, API permissions/CRUD and summary provenance data
This commit is contained in:
2026-08-15 16:32:20 -05:00
parent b088794716
commit 29d154e140
18 changed files with 1321 additions and 5 deletions

View File

@@ -0,0 +1,109 @@
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_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,325 @@
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_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_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,163 @@
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_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})