Merge pull request #3 from i701/feat/topups

feature/topups

api/signals.py

@@ -18,7 +18,12 @@ def assign_device_permissions(sender, instance, created, **kwargs):
         atoll_read_permission = Permission.objects.get(codename="view_atoll")
         island_read_permission = Permission.objects.get(codename="view_island")
         payment_permissions = Permission.objects.filter(content_type__model="payment")
+        topup_permissions = Permission.objects.filter(
+            content_type__model="topup"
+        ).exclude(codename__startswith="delete_")
 
+        for permission in topup_permissions:
+            instance.user_permissions.add(permission)
         for permission in device_permissions:
             instance.user_permissions.add(permission)
         instance.user_permissions.add(atoll_read_permission, island_read_permission)

billing/admin.py

@@ -18,6 +18,24 @@ class PaymentAdmin(admin.ModelAdmin):
     )
 
 
+class TopupAdmin(admin.ModelAdmin):
+    list_display = (
+        "id",
+        "user",
+        "amount",
+        "paid",
+        "paid_at",
+        "created_at",
+        "updated_at",
+    )
+    search_fields = (
+        "user__first_name",
+        "user__last_name",
+        "user__id_card",
+        "user__mobile",
+    )
+
+
 admin.site.register(Payment, PaymentAdmin)
 admin.site.register(BillFormula)
-admin.site.register(Topup)
+admin.site.register(Topup, TopupAdmin)

billing/filters.py

@@ -1,5 +1,6 @@
 import django_filters
-from .models import Payment
+from .models import Payment, Topup
+from django.db.models import Q
 
 
 class PaymentFilter(django_filters.FilterSet):
@@ -16,3 +17,30 @@ class PaymentFilter(django_filters.FilterSet):
     class Meta:
         model = Payment
         fields = "__all__"
+
+
+class TopupFilter(django_filters.FilterSet):
+    amount = django_filters.RangeFilter(field_name="amount")
+    paid = django_filters.BooleanFilter(field_name="paid")
+    user = django_filters.CharFilter(method="filter_user_search")
+    created_at = django_filters.DateFromToRangeFilter(field_name="created_at")
+
+    def filter_user_search(self, queryset, name, value):
+        """
+        Search across multiple user fields: first_name, last_name, id_card, mobile
+        """
+        return queryset.filter(
+            Q(user__first_name__icontains=value)
+            | Q(user__last_name__icontains=value)
+            | Q(user__id_card__icontains=value)
+            | Q(user__mobile__icontains=value)
+        )
+
+    class Meta:
+        model = Topup  # Assuming Topup is a subclass of Payment
+        fields = [
+            "amount",
+            "paid",
+            "user",
+            "created_at",
+        ]

billing/management/commands/seed_topups.py

@@ -0,0 +1,63 @@
+# billing/management/commands/seed_billing.py
+
+import random
+from django.core.management.base import BaseCommand
+from django.utils import timezone
+from faker import Faker
+from billing.models import (
+    Payment,
+)
+from billing.models import Topup
+from api.models import User
+
+
+class Command(BaseCommand):
+    help = "Seeds topup models with dummy data."
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--number",
+            type=int,
+            default=10,
+            help="The number of topups to create.",
+        )
+
+    def handle(self, *args, **options):
+        number = options["number"]
+        fake = Faker()
+
+        users = User.objects.all()
+        if not users.exists():
+            self.stdout.write(
+                self.style.ERROR(
+                    "No users found. Please seed users first (e.g., python manage.py seed_users)."
+                )
+            )
+            return
+
+        self.stdout.write(self.style.NOTICE(f"Seeding {number} topups..."))
+
+        for _ in range(number):
+            random_user = random.choice(users)
+
+            paid_status = fake.boolean(chance_of_getting_true=80)
+            paid_at_date = fake.date_time_this_year() if paid_status else None
+
+            if paid_at_date and timezone.is_naive(paid_at_date):
+                paid_at_date = timezone.make_aware(paid_at_date)
+
+            Topup.objects.create(
+                amount=fake.pydecimal(
+                    left_digits=4,
+                    right_digits=2,
+                    positive=True,
+                    min_value=100.00,
+                    max_value=5000.00,
+                ),
+                paid=paid_status,
+                user=random_user,
+                paid_at=paid_at_date,
+                updated_at=timezone.now(),
+            )
+
+        self.stdout.write(self.style.SUCCESS(f"Successfully seeded {number} topups."))

billing/migrations/0006_topup_mib_reference.py

@@ -0,0 +1,17 @@
+# Generated by Django 5.2 on 2025-07-03 11:30
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("billing", "0005_alter_payment_options_payment_mib_reference"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="topup",
+            name="mib_reference",
+            field=models.CharField(blank=True, default="", null=True),
+        ),
+    ]

billing/migrations/0007_topup_paid_at.py

@@ -0,0 +1,17 @@
+# Generated by Django 5.2 on 2025-07-03 12:04
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("billing", "0006_topup_mib_reference"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="topup",
+            name="paid_at",
+            field=models.DateTimeField(blank=True, null=True),
+        ),
+    ]

billing/models.py

@@ -51,8 +51,13 @@ class Topup(models.Model):
     amount = models.FloatField()
     user = models.ForeignKey(User, on_delete=models.CASCADE, related_name="topups")
     paid = models.BooleanField(default=False)
+    paid_at = models.DateTimeField(null=True, blank=True)
+    mib_reference = models.CharField(default="", null=True, blank=True)
     created_at = models.DateTimeField(default=timezone.now)
     updated_at = models.DateTimeField(auto_now=True)
 
     def __str__(self):
         return f"Topup for {self.user}"
+
+    class Meta:
+        ordering = ["-created_at"]

billing/serializers.py

@@ -1,19 +1,47 @@
 from rest_framework import serializers
-from .models import Payment
+from .models import Payment, Topup
 from devices.serializers import DeviceSerializer
 
 
 class PaymentSerializer(serializers.ModelSerializer):
     devices = DeviceSerializer(many=True, read_only=True)
 
-    class Meta:
+    class Meta:  # type: ignore
         model = Payment
         fields = "__all__"
 
 
 class UpdatePaymentSerializer(serializers.ModelSerializer):
-    class Meta:
+    class Meta:  # type: ignore
         model = Payment
         fields = [
             "number_of_months",
         ]
+
+
+class TopupSerializer(serializers.ModelSerializer):
+    user = serializers.SerializerMethodField()
+
+    def get_user(self, obj):
+        user = obj.user
+        if user:
+            return {
+                "id": user.id,
+                "name": user.first_name + " " + user.last_name,
+                "id_card": user.id_card,
+                "mobile": user.mobile,
+            }
+        return None
+
+    class Meta:  # type: ignore
+        model = Topup
+        fields = [
+            "id",
+            "amount",
+            "user",
+            "paid",
+            "mib_reference",
+            "created_at",
+            "updated_at",
+        ]
+        read_only_fields = ["id", "created_at", "updated_at"]

billing/tests.py

@@ -1,3 +1,192 @@
 from django.test import TestCase
+from django.urls import reverse
+from rest_framework import status
+from rest_framework.test import APIClient
+from django.contrib.auth import get_user_model
+from .models import Topup
+from .serializers import TopupSerializer
+from django.utils import timezone
+from datetime import timedelta
+from decouple import config
+from unittest import mock
 
-# Create your tests here.
+User = get_user_model()
+
+REAL_USER_FIRST_NAME = config("REAL_USER_FIRST_NAME", default="josh")
+REAL_USER_LAST_NAME = config("REAL_USER_LAST_NAME", default="mosh")
+
+
+class TopupTests(TestCase):
+    def setUp(self):
+        self.client = APIClient()
+        self.real_user = User.objects.create_user(
+            username="testuser",
+            password="testpassword",
+            first_name=REAL_USER_FIRST_NAME,
+            last_name=REAL_USER_LAST_NAME,
+            acc_no="7770000010629",
+            is_admin=True,
+        )
+        self.user = User.objects.create_user(
+            username="plskillme",
+            password="modewasgayithink",
+            first_name="mode",
+            last_name="hussain",
+            acc_no="1122334455",
+        )
+        self.admin_user = User.objects.create_superuser(
+            username="adminuser",
+            password="adminpassword",
+            email="admin@example.com",
+            first_name="Admin",
+            last_name="User",
+            acc_no="987654321",
+        )
+        self.client.force_authenticate(user=self.real_user)
+
+    def test_create_topup(self):
+        url = reverse("create-list-topups")
+        data = {"amount": 100.00}
+        response = self.client.post(url, data, format="json")
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+        self.assertEqual(Topup.objects.count(), 1)
+        topup = Topup.objects.first()
+        self.assertEqual(getattr(topup, "amount"), 100.00)
+        self.assertEqual(getattr(topup, "user"), self.real_user)
+
+    def test_create_topup_no_amount(self):
+        url = reverse("create-list-topups")
+        data = {}
+        response = self.client.post(url, data, format="json")
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_list_topups(self):
+        Topup.objects.create(amount=50.00, user=self.real_user)
+        Topup.objects.create(amount=75.00, user=self.real_user)
+        url = reverse("create-list-topups")
+        response = self.client.get(url, format="json")
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(len(response.json()["data"]), 2)
+
+    def test_list_topups_admin(self):
+        self.client = APIClient()
+        self.client.force_authenticate(user=self.admin_user)
+        Topup.objects.create(amount=50.00, user=self.real_user)
+        admin_user = User.objects.create_user(
+            username="anotheruser",
+            password="testpassword",
+            first_name="Another",
+            last_name="User",
+            acc_no="1122334455",
+            is_admin=True,
+        )
+        Topup.objects.create(amount=75.00, user=admin_user)
+        url = reverse("create-list-topups")
+        response = self.client.get(url, format="json")
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(len(response.json()["data"]), 2)  # Admin sees all
+
+    def test_list_topups_filtered_by_user(self):
+        other_user = User.objects.create_user(
+            username="otheruser",
+            password="testpassword",
+            first_name="Other",
+            last_name="User",
+            acc_no="5544332211",
+        )
+        Topup.objects.create(amount=50.00, user=self.user)
+        Topup.objects.create(amount=75.00, user=other_user)
+        url = reverse("create-list-topups")
+        response = self.client.get(url, format="json")
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(len(response.json()["data"]), 2)
+
+    @mock.patch("billing.views.localtime")
+    def test_verify_topup_payment(self, mock_localtime):
+        fixed_time = timezone.datetime(
+            2025, 7, 3, 19, 36, tzinfo=timezone.get_current_timezone()
+        )
+        mock_localtime.return_value = fixed_time
+
+        topup = Topup.objects.create(amount=1.5, user=self.real_user)
+        url = reverse("verify-topup-payment", kwargs={"pk": topup.pk})
+        self.client = APIClient()
+
+        self.client.force_authenticate(user=self.real_user)
+        data = {
+            "benefName": f"{REAL_USER_FIRST_NAME} {REAL_USER_LAST_NAME}",
+            "accountNo": "7770000010629",
+            "absAmount": 1.5,
+            "time": fixed_time.strftime("%Y-%m-%d %H:%M"),  # Use the same fixed time
+        }
+        response = self.client.patch(url, data, format="json")
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        topup.refresh_from_db()
+        self.assertEqual(topup.paid, True)
+
+    def test_verify_topup_payment_already_verified(self):
+        topup = Topup.objects.create(amount=100.00, user=self.real_user, paid=True)
+        url = reverse("verify-topup-payment", kwargs={"pk": topup.pk})
+        data = {}
+        response = self.client.patch(url, data, format="json")
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_verify_topup_payment_unauthorized(self):
+        other_user = User.objects.create_user(
+            username="otheruser",
+            password="testpassword",
+            first_name="Other",
+            last_name="User",
+            acc_no="5544332211",
+        )
+        topup = Topup.objects.create(amount=100.00, user=other_user)
+        url = reverse("verify-topup-payment", kwargs={"pk": topup.pk})
+        data = {}
+        response = self.client.patch(url, data, format="json")
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_topup_serializer(self):
+        topup = Topup.objects.create(amount=120.00, user=self.real_user)
+        serializer = TopupSerializer(topup)
+        self.assertEqual(serializer.data["amount"], 120.00)
+        self.assertEqual(serializer.data["user"]["id"], getattr(self.real_user, "id"))
+
+    def test_topup_filter_amount(self):
+        Topup.objects.create(amount=50.00, user=self.real_user)
+        Topup.objects.create(amount=100.00, user=self.real_user)
+        url = reverse("create-list-topups") + "?amount_min=75"
+        response = self.client.get(url, format="json")
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(len(response.json()["data"]), 1)
+        self.assertEqual(response.json()["data"][0]["amount"], 100.00)
+
+    def test_topup_filter_user_search(self):
+        Topup.objects.create(amount=50.00, user=self.real_user)
+        other_user = User.objects.create_user(
+            username="otheruser",
+            password="testpassword",
+            first_name="Other",
+            last_name="User",
+            id_card="12345",
+            mobile="1234567890",
+            acc_no="5544332211",
+        )
+        Topup.objects.create(amount=75.00, user=other_user)
+        url = reverse("create-list-topups") + "?user=Other"
+        response = self.client.get(url, format="json")
+        print(response.json())
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(len(response.json()["data"]), 1)
+        self.assertEqual(response.json()["data"][0]["amount"], 75.00)
+
+    def test_topup_filter_created_at(self):
+        now = timezone.now()
+        Topup.objects.create(
+            amount=50.00, user=self.real_user, created_at=now - timedelta(days=2)
+        )
+        Topup.objects.create(amount=100.00, user=self.real_user, created_at=now)
+        url = reverse("create-list-topups") + f"?created_at_after={now.date()}"
+        response = self.client.get(url, format="json")
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(len(response.json()["data"]), 1)
+        self.assertEqual(response.json()["data"][0]["amount"], 100.00)

billing/urls.py

@@ -6,6 +6,8 @@ from .views import (
     PaymentDetailAPIView,
     UpdatePaymentAPIView,
     DeletePaymentView,
+    ListCreateTopupView,
+    VerifyTopupPaymentAPIView,
 )
 
 urlpatterns = [
@@ -24,4 +26,12 @@ urlpatterns = [
     path(
         "payment/<str:pk>/verify/", VerifyPaymentView.as_view(), name="verify-payment"
     ),
+    # Topups
+    path("topup/", ListCreateTopupView.as_view(), name="create-list-topups"),
+    # path("topup/<str:pk>/", TopupDetailAPIView.as_view(), name="retrieve-topup"),
+    path(
+        "topup/<str:pk>/verify/",
+        VerifyTopupPaymentAPIView.as_view(),
+        name="verify-topup-payment",
+    ),
 ]

billing/views.py

@@ -15,9 +15,9 @@ from api.tasks import add_new_devices_to_omada
 from apibase.env import BASE_DIR, env
 import logging
 
-from .models import Device, Payment
+from .models import Device, Payment, Topup
-from .serializers import PaymentSerializer, UpdatePaymentSerializer
+from .serializers import PaymentSerializer, UpdatePaymentSerializer, TopupSerializer
-from .filters import PaymentFilter
+from .filters import PaymentFilter, TopupFilter
 
 env.read_env(os.path.join(BASE_DIR, ".env"))
 
@@ -259,3 +259,99 @@ class DeletePaymentView(StaffEditorPermissionMixin, generics.DestroyAPIView):
         devices = instance.devices.all()
         devices.update(is_active=False, expiry_date=None, has_a_pending_payment=False)
         return super().delete(request, *args, **kwargs)
+
+
+class ListCreateTopupView(StaffEditorPermissionMixin, generics.ListCreateAPIView):
+    queryset = Topup.objects.all()
+    serializer_class = TopupSerializer
+    filter_backends = [DjangoFilterBackend]
+    filterset_fields = "__all__"
+    filterset_class = TopupFilter
+
+    def create(self, request, *args, **kwargs):
+        data = request.data
+        user = request.user
+        amount = data.get("amount")
+        if not amount:
+            return Response(
+                {"message": "amount is required."},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        topup = Topup.objects.create(amount=amount, user=user)
+        serializer = TopupSerializer(topup)
+        return Response(serializer.data, status=status.HTTP_201_CREATED)
+
+    def get_queryset(self):
+        queryset = super().get_queryset()
+        if getattr(self.request.user, "is_admin") or self.request.user.is_superuser:
+            return queryset
+        return queryset.filter(user=self.request.user)
+
+
+class VerifyTopupPaymentAPIView(StaffEditorPermissionMixin, generics.UpdateAPIView):
+    queryset = Topup.objects.all()
+    serializer_class = TopupSerializer
+    lookup_field = "pk"
+
+    def verify_transfer_topup(self, data, topup):
+        if not PAYMENT_BASE_URL:
+            raise ValueError(
+                "PAYMENT_BASE_URL is not set. Please set it in your environment variables."
+            )
+        logger.info(data)
+        response = requests.post(
+            f"{PAYMENT_BASE_URL}/verify-payment",
+            json=data,
+            headers={"Content-Type": "application/json"},
+        )
+        try:
+            response.raise_for_status()
+        except requests.exceptions.HTTPError as e:
+            logger.error(f"HTTPError: {e}")
+            return False  # Or handle the error as appropriate
+        mib_resp = response.json()
+        print(mib_resp)
+        if not response.json().get("success"):
+            return mib_resp["success"]
+        else:
+            topup.paid = True
+            # topup.paid_at = timezone.now()  # Assuming Topup model has paid_at field
+            topup.mib_reference = mib_resp["transaction"]["ref"] or ""
+            topup.paid_at = timezone.now()
+            topup.save()
+            return True
+
+    def update(self, request, *args, **kwargs):
+        topup_instance = self.get_object()
+        user = request.user
+
+        if topup_instance.paid:
+            return Response(
+                {"message": "Payment has already been verified."},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        if topup_instance.user != user and not user.is_superuser:
+            return Response(
+                {"message": "You are not allowed to pay for this topup."},
+                status=status.HTTP_403_FORBIDDEN,
+            )
+        data = {
+            "benefName": f"{user.first_name} {user.last_name}",  # type: ignore
+            "accountNo": user.acc_no,  # type: ignore
+            "absAmount": topup_instance.amount,
+            "time": localtime(
+                topup_instance.created_at + timedelta(minutes=5)
+            ).strftime("%Y-%m-%d %H:%M"),
+        }
+        print("payment payload in view ->", data)
+        topup_status = self.verify_transfer_topup(data, topup_instance)
+        if topup_status:
+            return Response(
+                {"message": "Topup payment verified successfully."},
+                status=status.HTTP_200_OK,
+            )
+        else:
+            return Response(
+                {"message": "Topup payment verification failed."},
+                status=status.HTTP_400_BAD_REQUEST,
+            )

justfile

@@ -4,3 +4,17 @@ default:
 dev:
   python manage.py runserver
   python manage.py procrastinate worker
+
+migrate:
+  python manage.py migrate
+make-migrations:
+  python manage.py makemigrations
+
+seed-topups:
+  python manage.py seed_topups --number=50
+seed-payments:
+  python manage.py seed_payments --number=50
+
+# TESTS
+test-billing:
+  python manage.py test billing