Merge branch 'develop' into staging

src/hope_api_auth/admin.py

@@ -25,7 +25,6 @@
 Key: {obj.key}
 Grants: {obj.grants}
 Expires: {expire}
-Business Areas: {areas}
 
 Regards
 

src/hope_payment_gateway/apps/fsp/moneygram/client.py

@@ -7,7 +7,7 @@
 import phonenumbers
 import requests
 from phonenumbers import NumberParseException
-from urllib3.connectionpool import HTTPSConnectionPool
+from urllib3.exceptions import PoolError
 
 from hope_payment_gateway.apps.core.models import Singleton
 from hope_payment_gateway.apps.gateway.flows import PaymentRecordFlow
@@ -43,7 +43,7 @@ def get_token(self):
 
         try:
             response = requests.get(url, headers=headers)
-        except HTTPSConnectionPool:
+        except PoolError:
             self.token = None
             self.token_response = None
         else:

src/hope_payment_gateway/apps/fsp/western_union/endpoints/client.py

@@ -26,7 +26,7 @@ def __init__(self, wsdl_filename) -> None:
         self.client.set_ns_prefix("xrsi", "http://www.westernunion.com/schema/xrsi")
 
     def response_context(self, service_name, payload, wsdl_name=None, port=None):
-        response = ""
+        response = dict()
         error = ""
         format = "string"
         try:
@@ -63,7 +63,7 @@ def response_context(self, service_name, payload, wsdl_name=None, port=None):
             code = 400
             logger.exception(exc)
         except Exception as exc:
-            title = f"{exc.message} [{exc.code}]"
+            title = type(exc).__name__
             code = 400
             error = str(exc)
             logger.exception(exc)

src/hope_payment_gateway/apps/fsp/western_union/endpoints/das.py

@@ -106,18 +106,19 @@ def das_delivery_services(destination_country, destination_currency, create_corr
     client = WesternUnionClient("DAS_Service_H2HService.wsdl")
     response = client.response_context("DAS_Service", payload, "DAS_Service_H2H", f"SOAP_HTTP_Port_{wu_env}")
 
-    context = response["content"]["MTML"]["REPLY"]["DATA_CONTEXT"]["RECORDSET"]
-    if create_corridors and context:
-        for ds in context["GETDELIVERYSERVICES"]:
-            if ds["SVC_CODE"] == "800":
-                Corridor.objects.get_or_create(
-                    destination_country=destination_country,
-                    destination_currency=destination_currency,
-                    defaults={
-                        "description": f"{destination_country}: {destination_currency}",
-                        "template_code": ds["TEMPLT"],
-                    },
-                )
+    if "content" in response and "MTML" in response["content"]:
+        context = response["content"]["MTML"]["REPLY"]["DATA_CONTEXT"]["RECORDSET"]
+        if create_corridors and context:
+            for ds in context["GETDELIVERYSERVICES"]:
+                if ds["SVC_CODE"] == "800":
+                    Corridor.objects.get_or_create(
+                        destination_country=destination_country,
+                        destination_currency=destination_currency,
+                        defaults={
+                            "description": f"{destination_country}: {destination_currency}",
+                            "template_code": ds["TEMPLT"],
+                        },
+                    )
 
     return response
 
@@ -139,7 +140,7 @@ def das_delivery_option_template(destination_country, destination_currency, temp
     client = WesternUnionClient("DAS_Service_H2HService.wsdl")
     context = client.response_context("DAS_Service", payload, "DAS_Service_H2H", f"SOAP_HTTP_Port_{wu_env}")
 
-    if "content" in context:
+    if "content" in context and "MTML" in context["content"]:
         rows = context["content"]["MTML"]["REPLY"]["DATA_CONTEXT"]["RECORDSET"]
         template = {}
         structure = []
@@ -170,7 +171,7 @@ def das_delivery_option_template(destination_country, destination_currency, temp
                             else:
                                 base[structure[-1]] = [base[structure[-1]], code]
                             if service_provider_code:
-                                sp, created = ServiceProviderCode.objects.get_or_create(
+                                ServiceProviderCode.objects.get_or_create(
                                     code=code,
                                     description=description,
                                     country=destination_country,
@@ -193,4 +194,4 @@ def das_delivery_option_template(destination_country, destination_currency, temp
                                 service_provider_code = False
                 obj.template = template
                 obj.save()
-        return context
+    return context

tests/conftest.py

@@ -2,6 +2,7 @@
 import tempfile
 
 import pytest
+import responses
 from factories import (
     APITokenFactory,
     CorridorFactory,
@@ -31,6 +32,12 @@ def use_override_settings(settings):
     settings.WESTERN_UNION_BASE_URL = "https://wugateway2pi.westernunion.com/"
 
 
+@pytest.fixture()
+def mocked_responses():
+    with responses.RequestsMock(assert_all_requests_are_fired=False) as rsps:
+        yield rsps
+
+
 @pytest.fixture()
 def user(request, db):
     return UserFactory()

tests/factories/__init__.py

@@ -0,0 +1,33 @@
+import importlib
+import pkgutil
+from pathlib import Path
+
+from factory.django import DjangoModelFactory
+from pytest_factoryboy import register
+
+from .base import AutoRegisterModelFactory, TAutoRegisterModelFactory, factories_registry
+from .payment import *  # noqa
+from .social import SocialAuthUserFactory  # noqa
+from .user import GroupFactory, SuperUserFactory, SystemFactory, User, UserFactory  # noqa
+
+for _, name, _ in pkgutil.iter_modules([str(Path(__file__).parent)]):
+    importlib.import_module(f".{name}", __package__)
+
+
+django_model_factories = {factory._meta.model: factory for factory in DjangoModelFactory.__subclasses__()}
+
+
+def get_factory_for_model(
+    _model,
+) -> type[TAutoRegisterModelFactory] | type[DjangoModelFactory]:
+    class Meta:
+        model = _model
+
+    bases = (AutoRegisterModelFactory,)
+    if _model in factories_registry:
+        return factories_registry[_model]  # noqa
+
+    if _model in django_model_factories:
+        return django_model_factories[_model]
+
+    return register(type(f"{_model._meta.model_name}AutoCreatedFactory", bases, {"Meta": Meta}))  # noqa

tests/factories/base.py

@@ -0,0 +1,19 @@
+import typing
+
+import factory
+from factory.base import FactoryMetaClass
+
+TAutoRegisterModelFactory = typing.TypeVar("TAutoRegisterModelFactory", bound="AutoRegisterModelFactory")
+
+factories_registry: dict[str, TAutoRegisterModelFactory] = {}
+
+
+class AutoRegisterFactoryMetaClass(FactoryMetaClass):
+    def __new__(mcs, class_name, bases, attrs):
+        new_class = super().__new__(mcs, class_name, bases, attrs)
+        factories_registry[new_class._meta.model] = new_class
+        return new_class
+
+
+class AutoRegisterModelFactory(factory.django.DjangoModelFactory, metaclass=AutoRegisterFactoryMetaClass):
+    pass

tests/factories/contenttypes.py

@@ -0,0 +1,12 @@
+from django.contrib.contenttypes.models import ContentType
+
+from .base import AutoRegisterModelFactory
+
+
+class ContentTypeFactory(AutoRegisterModelFactory):
+    app_label = "auth"
+    model = "user"
+
+    class Meta:
+        model = ContentType
+        django_get_or_create = ("app_label", "model")

tests/factories/django_auth.py

@@ -0,0 +1,32 @@
+from django.contrib.auth.models import Group, Permission
+
+import factory
+
+from .base import AutoRegisterModelFactory
+from .contenttypes import ContentTypeFactory
+
+
+class PermissionFactory(AutoRegisterModelFactory):
+    content_type = factory.SubFactory(ContentTypeFactory)
+
+    class Meta:
+        model = Permission
+
+
+# class GroupFactory(AutoRegisterModelFactory):
+#     name = factory.Sequence(lambda n: "group %s" % n)
+#
+#     class Meta:
+#         model = Group
+#         django_get_or_create = ("name",)
+#
+#     @factory.post_generation
+#     def permissions(self, create, extracted, **kwargs):
+#         if not create:
+#             # Simple build, do nothing.
+#             return
+#
+#         if extracted:
+#             # A list of groups were passed in, use them
+#             for perm in extracted:
+#                 self.permissions.add(perm)

tests/factories/django_celery_beat.py

@@ -0,0 +1,31 @@
+from django.utils import timezone
+
+from django_celery_beat.models import SOLAR_SCHEDULES, ClockedSchedule, IntervalSchedule, SolarSchedule
+from factory.fuzzy import FuzzyChoice
+
+from .base import AutoRegisterModelFactory
+
+
+class IntervalScheduleFactory(AutoRegisterModelFactory):
+    every = 1
+    period = IntervalSchedule.HOURS
+
+    class Meta:
+        model = IntervalSchedule
+
+
+class SolarScheduleFactory(AutoRegisterModelFactory):
+    event = FuzzyChoice([x[0] for x in SOLAR_SCHEDULES])
+
+    latitude = 10.1
+    longitude = 10.1
+
+    class Meta:
+        model = SolarSchedule
+
+
+class ClockedScheduleFactory(AutoRegisterModelFactory):
+    clocked_time = timezone.now()
+
+    class Meta:
+        model = ClockedSchedule