Add unit tests for product service

backend/python/django_app/adapters/http/hello_handler.py

@@ -0,0 +1,9 @@
+from django.http import JsonResponse
+from django_app.core.services.hello_service import HelloService
+
+service = HelloService()
+
+def hello_world(request):
+    name = request.GET.get('name')
+    message = service.greet(name)
+    return JsonResponse({"message": message})

backend/python/django_app/adapters/http/product_category_handler.py

@@ -0,0 +1,47 @@
+import json
+from django.http import JsonResponse
+from django.views.decorators.csrf import csrf_exempt
+from django_app.core.services.product_category_service import ProductCategoryService
+from django_app.adapters.repositories.mongo_product_category_repository import MongoProductCategoryRepository
+from dataclasses import asdict
+
+category_repository = MongoProductCategoryRepository()
+service = ProductCategoryService(category_repository)
+
+@csrf_exempt
+def categories(request):
+    if request.method == 'GET':
+        all_cats = service.get_all_categories()
+        return JsonResponse([asdict(c) for c in all_cats], safe=False, status=200)
+
+    elif request.method == 'POST':
+        try:
+            data = json.loads(request.body)
+            category = service.create_category(data)
+            return JsonResponse(asdict(category), status=201)
+        except ValueError as e:
+            return JsonResponse({'errors': e.args[0]}, status=400)
+
+@csrf_exempt
+def category_detail(request, category_id):
+    if request.method == 'GET':
+        category = service.get_category(category_id)
+        if not category:
+            return JsonResponse({'error': 'Category not found'}, status=404)
+        return JsonResponse(asdict(category), status=200)
+
+    elif request.method == 'PUT':
+        try:
+            data = json.loads(request.body)
+            category = service.update_category(category_id, data)
+            if not category:
+                return JsonResponse({'error': 'Category not found'}, status=404)
+            return JsonResponse(asdict(category), status=200)
+        except ValueError as e:
+            return JsonResponse({'errors': e.args[0]}, status=400)
+
+    elif request.method == 'DELETE':
+        deleted = service.delete_category(category_id)
+        if not deleted:
+            return JsonResponse({'error': 'Category not found'}, status=404)
+        return JsonResponse({}, status=204)

backend/python/django_app/adapters/http/product_handler.py

@@ -0,0 +1,95 @@
+import json
+from django.http import JsonResponse
+from django.views.decorators.csrf import csrf_exempt
+from django_app.adapters.repositories.mongo_product_repository import MongoProductRepository
+from django_app.core.services.product_service import ProductService
+from django_app.adapters.repositories.in_memory_product_repository import InMemoryProductRepository
+from dataclasses import asdict
+
+# Wire up once (in real apps this is done via dependency injection)
+# repository = InMemoryProductRepository()
+repository = MongoProductRepository()
+service = ProductService(repository)
+
+@csrf_exempt
+def products(request):
+    if request.method == 'GET':
+        all_products = service.get_all_products()
+        return JsonResponse([asdict(p) for p in all_products], safe=False, status=200)
+
+    elif request.method == 'POST':
+        try:
+            data = json.loads(request.body)
+            product = service.create_product(data)
+            return JsonResponse(asdict(product), status=201)
+        except ValueError as e:
+            return JsonResponse({'errors': e.args[0]}, status=400)
+
+@csrf_exempt
+def product_detail(request, product_id):
+    if request.method == 'GET':
+        product = service.get_product(product_id)
+        if not product:
+            return JsonResponse({'error': 'Product not found'}, status=404)
+        return JsonResponse(asdict(product), status=200)
+
+    elif request.method == 'PUT':
+        try:
+            data = json.loads(request.body)
+            product = service.update_product(product_id, data)
+            if not product:
+                return JsonResponse({'error': 'Product not found'}, status=404)
+            return JsonResponse(asdict(product), status=200)
+        except ValueError as e:
+            return JsonResponse({'errors': e.args[0]}, status=400)
+
+    elif request.method == 'DELETE':
+        deleted = service.delete_product(product_id)
+        if not deleted:
+            return JsonResponse({'error': 'Product not found'}, status=404)
+        return JsonResponse({}, status=204)
+
+@csrf_exempt
+def products_by_category(request, category_id):
+    """GET /categories/<id>/products/"""
+    if request.method == 'GET':
+        products = service.get_products_by_category(category_id)
+        return JsonResponse([asdict(p) for p in products], safe=False, status=200)
+
+@csrf_exempt
+def product_category_assign(request, product_id):
+    """POST /products/<id>/category/  — assign category
+       DELETE /products/<id>/category/ — remove category"""
+    if request.method == 'POST':
+        try:
+            data = json.loads(request.body)
+            product = service.assign_category(product_id, data['category_id'])
+            if not product:
+                return JsonResponse({'error': 'Product not found'}, status=404)
+            return JsonResponse(asdict(product), status=200)
+        except ValueError as e:
+            return JsonResponse({'errors': e.args[0]}, status=400)
+
+    elif request.method == 'DELETE':
+        product = service.remove_category(product_id)
+        if not product:
+            return JsonResponse({'error': 'Product not found'}, status=404)
+        return JsonResponse(asdict(product), status=200)
+
+@csrf_exempt
+def bulk_create_products(request):
+    """POST /products/bulk/ — accepts CSV file"""
+    if request.method == 'POST':
+        try:
+            csv_file = request.FILES.get('file')
+            if not csv_file:
+                return JsonResponse({'error': 'No file provided'}, status=400)
+            csv_content = csv_file.read().decode('utf-8')
+            products = service.bulk_create_from_csv(csv_content)
+            return JsonResponse(
+                {'created': len(products),
+                 'products': [asdict(p) for p in products]},
+                status=201
+            )
+        except ValueError as e:
+            return JsonResponse({'errors': e.args[0]}, status=400)

backend/python/django_app/adapters/repositories/in_memory_product_repository.py

@@ -0,0 +1,22 @@
+from django_app.core.models.product import Product
+from typing import Optional
+
+class InMemoryProductRepository:
+    def __init__(self):
+        self._store = {}  # our "database"
+
+    def save(self, product: Product) -> Product:
+        self._store[product.id] = product
+        return product
+
+    def find_by_id(self, product_id: str) -> Optional[Product]:
+        return self._store.get(product_id)
+
+    def find_all(self) -> list[Product]:
+        return list(self._store.values())
+
+    def delete(self, product_id: str) -> bool:
+        if product_id in self._store:
+            del self._store[product_id]
+            return True
+        return False

backend/python/django_app/adapters/repositories/mongo_models/product_category_document.py

@@ -0,0 +1,7 @@
+from mongoengine import Document, StringField
+
+class ProductCategoryDocument(Document):
+    title = StringField(required=True, max_length=100)
+    description = StringField(max_length=500)
+
+    meta = {'collection': 'product_categories'}

backend/python/django_app/adapters/repositories/mongo_models/product_document.py

@@ -0,0 +1,14 @@
+from mongoengine import Document, StringField, FloatField, IntField, ReferenceField
+from django_app.adapters.repositories.mongo_models.product_category_document import ProductCategoryDocument
+
+class ProductDocument(Document):
+    name = StringField(required=True, max_length=200)
+    description = StringField(max_length=1000)
+    category = ReferenceField(ProductCategoryDocument, null=True)
+    price = FloatField(required=True, min_value=0)
+    brand = StringField(required=True, max_length=100)
+    quantity = IntField(required=True, min_value=0)
+
+    meta = {
+        'collection': 'products'
+    }

backend/python/django_app/adapters/repositories/mongo_product_category_repository.py

@@ -0,0 +1,59 @@
+# adapters/repositories/mongo_product_category_repository.py
+from bson import ObjectId
+from bson.errors import InvalidId
+from django_app.core.models.product_category import ProductCategory
+from django_app.adapters.repositories.mongo_models.product_category_document import ProductCategoryDocument
+from typing import Optional
+
+class MongoProductCategoryRepository:
+
+    def _to_domain(self, doc: ProductCategoryDocument) -> ProductCategory:
+        return ProductCategory(
+            id=str(doc.id),
+            title=doc.title,
+            description=doc.description or '',
+        )
+
+    def _to_object_id(self, category_id: str):
+        try:
+            return ObjectId(category_id)
+        except (InvalidId, TypeError):
+            return None
+
+    def save(self, category: ProductCategory) -> ProductCategory:
+        if category.id:
+            oid = self._to_object_id(category.id)
+            doc = ProductCategoryDocument.objects(id=oid).first()
+            if doc:
+                doc.title = category.title
+                doc.description = category.description
+                doc.save()
+                return self._to_domain(doc)
+
+        doc = ProductCategoryDocument(
+            title=category.title,
+            description=category.description,
+        )
+        doc.save()
+        return self._to_domain(doc)
+
+    def find_by_id(self, category_id: str) -> Optional[ProductCategory]:
+        oid = self._to_object_id(category_id)
+        if not oid:
+            return None
+        doc = ProductCategoryDocument.objects(id=oid).first()
+        return self._to_domain(doc) if doc else None
+
+    def find_all(self) -> list[ProductCategory]:
+        return [self._to_domain(doc) for doc in ProductCategoryDocument.objects.all()]
+
+    def delete(self, category_id: str) -> bool:
+        oid = self._to_object_id(category_id)
+        if not oid:
+            return False
+        doc = ProductCategoryDocument.objects(id=oid).first()
+        if not doc:
+            return False
+        doc.delete()
+        return True
+

backend/python/django_app/adapters/repositories/mongo_product_repository.py

@@ -0,0 +1,97 @@
+from django_app.core.models.product import Product
+from django_app.adapters.repositories.mongo_models.product_document import ProductDocument
+from typing import Optional
+from bson import ObjectId
+from bson.errors import InvalidId
+
+class MongoProductRepository:
+
+    def _to_domain(self, doc: ProductDocument) -> Product:
+        """Convert a DB document → core domain model"""
+        # print('MongoProductRepository._to_domain doc.category:', doc.category)
+
+        category_title = ''
+        if doc.category:
+            category_title = getattr(doc.category, 'title', '')
+
+        return Product(
+            id=str(doc.id),
+            name=doc.name,
+            description=doc.description or '',
+            category=category_title,
+            price=doc.price,
+            brand=doc.brand or '',
+            quantity=doc.quantity,
+            category_id=str(doc.category.id) if doc.category else None,
+        )
+
+    def _to_object_id(self, product_id: str):
+        """Safely convert string → ObjectId, return None if invalid"""
+        try:
+            return ObjectId(product_id)
+        except (InvalidId, TypeError):
+            return None
+
+    def save(self, product: Product) -> Product:
+        from django_app.adapters.repositories.mongo_models.product_category_document import ProductCategoryDocument
+        from bson import ObjectId
+
+        category_doc = None
+        if product.category_id:
+            category_doc = ProductCategoryDocument.objects(
+                id=ObjectId(product.category_id)
+            ).first()
+
+        if product.id:
+            oid = self._to_object_id(product.id)
+            doc = ProductDocument.objects(id=oid).first()
+            if doc:
+                doc.name = product.name
+                doc.description = product.description
+                # doc.category = product.category
+                doc.price = product.price
+                doc.brand = product.brand
+                doc.quantity = product.quantity
+                doc.category = category_doc
+                doc.save()
+                return self._to_domain(doc)
+
+        # Create new
+        doc = ProductDocument(
+            name=product.name,
+            description=product.description,
+            # category=product.category,
+            price=product.price,
+            brand=product.brand,
+            quantity=product.quantity,
+            category=category_doc,
+        )
+        doc.save()
+        return self._to_domain(doc)
+
+    def find_by_category(self, category_id: str) -> list[Product]:
+        from django_app.adapters.repositories.mongo_models.product_category_document import ProductCategoryDocument
+        oid = self._to_object_id(category_id)
+        if not oid:
+            return []
+        category_doc = ProductCategoryDocument.objects(id=oid).first()
+        if not category_doc:
+            return []
+        docs = ProductDocument.objects(category=category_doc)
+        return [self._to_domain(doc) for doc in docs]
+
+    def find_by_id(self, product_id: str) -> Optional[Product]:
+        doc = ProductDocument.objects(id=product_id).first()
+        if not doc:
+            return None
+        return self._to_domain(doc)
+
+    def find_all(self) -> list[Product]:
+        return [self._to_domain(doc) for doc in ProductDocument.objects.all()]
+
+    def delete(self, product_id: str) -> bool:
+        doc = ProductDocument.objects(id=product_id).first()
+        if not doc:
+            return False
+        doc.delete()
+        return True

backend/python/django_app/core/models/product.py

@@ -0,0 +1,14 @@
+# import uuid
+from dataclasses import dataclass, field
+from typing import Optional
+
+@dataclass
+class Product:
+    name: str
+    description: str
+    price: float
+    brand: str
+    quantity: int
+    category: Optional[str] = None
+    id: Optional[str] = None
+    category_id: Optional[str] = None

backend/python/django_app/core/models/product_category.py

@@ -0,0 +1,8 @@
+from dataclasses import dataclass
+from typing import Optional
+
+@dataclass
+class ProductCategory:
+    title: str
+    description: str
+    id: Optional[str] = None

backend/python/django_app/core/services/hello_service.py

@@ -0,0 +1,6 @@
+
+class HelloService:
+    def greet(self, name: str = None) -> str:
+        if name:
+            return f"Hello, {name}"
+        return "Hello World"