Reorganize tests in triggers and utils apps

temba/utils/tests/test_checks.py

@@ -0,0 +1,19 @@
+from django.test import override_settings
+
+from temba.tests import TembaTest
+from temba.utils.checks import storage
+
+
+class SystemChecksTest(TembaTest):
+    def test_storage(self):
+        self.assertEqual(len(storage(None)), 0)
+
+        with override_settings(STORAGES={"default": {"BACKEND": "x"}, "staticfiles": {"BACKEND": "x"}}):
+            self.assertEqual(storage(None)[0].msg, "Missing 'archives' storage config.")
+            self.assertEqual(storage(None)[1].msg, "Missing 'public' storage config.")
+
+        with override_settings(STORAGE_URL=None):
+            self.assertEqual(storage(None)[0].msg, "No storage URL set.")
+
+        with override_settings(STORAGE_URL="http://example.com/uploads/"):
+            self.assertEqual(storage(None)[0].msg, "Storage URL shouldn't end with trailing slash.")

temba/utils/tests/test_compose.py

@@ -0,0 +1,7 @@
+from temba.tests import TembaTest
+from temba.utils.compose import compose_serialize
+
+
+class ComposeTest(TembaTest):
+    def test_empty_compose(self):
+        self.assertEqual(compose_serialize(), {})

temba/utils/tests/test_countries.py

@@ -0,0 +1,10 @@
+from temba.tests import TembaTest
+from temba.utils import countries
+
+
+class CountriesTest(TembaTest):
+    def test_from_tel(self):
+        self.assertIsNone(countries.from_tel(""))
+        self.assertIsNone(countries.from_tel("123"))
+        self.assertEqual("EC", countries.from_tel("+593979123456"))
+        self.assertEqual("US", countries.from_tel("+1 213 621 0002"))

temba/utils/tests/test_dates.py

@@ -0,0 +1,34 @@
+import datetime
+from datetime import date, timezone as tzone
+from zoneinfo import ZoneInfo
+
+from temba.tests import TembaTest
+from temba.utils.dates import date_range, datetime_to_str, datetime_to_timestamp, timestamp_to_datetime
+
+
+class DatesTest(TembaTest):
+    def test_datetime_to_timestamp(self):
+        d1 = datetime.datetime(2014, 1, 2, 3, 4, 5, microsecond=123_456, tzinfo=tzone.utc)
+        self.assertEqual(datetime_to_timestamp(d1), 1_388_631_845_123_456)  # from http://unixtimestamp.50x.eu
+        self.assertEqual(timestamp_to_datetime(1_388_631_845_123_456), d1)
+
+        tz = ZoneInfo("Africa/Kigali")
+        d2 = datetime.datetime(2014, 1, 2, 3, 4, 5, microsecond=123_456).replace(tzinfo=tz)
+        self.assertEqual(datetime_to_timestamp(d2), 1_388_624_645_123_456)
+        self.assertEqual(timestamp_to_datetime(1_388_624_645_123_456), d2.astimezone(tzone.utc))
+
+    def test_datetime_to_str(self):
+        tz = ZoneInfo("Africa/Kigali")
+        d2 = datetime.datetime(2014, 1, 2, 3, 4, 5, 6).replace(tzinfo=tz)
+
+        self.assertIsNone(datetime_to_str(None, "%Y-%m-%d %H:%M", tz=tz))
+        self.assertEqual(datetime_to_str(d2, "%Y-%m-%d %H:%M", tz=tz), "2014-01-02 03:04")
+        self.assertEqual(datetime_to_str(d2, "%Y/%m/%d %H:%M", tz=tzone.utc), "2014/01/02 01:04")
+        self.assertEqual(datetime_to_str(date(2023, 8, 16), "%Y/%m/%d %H:%M", tz=tzone.utc), "2023/08/16 00:00")
+
+    def test_date_range(self):
+        self.assertEqual(
+            [date(2015, 1, 29), date(2015, 1, 30), date(2015, 1, 31), date(2015, 2, 1)],
+            list(date_range(date(2015, 1, 29), date(2015, 2, 2))),
+        )
+        self.assertEqual([], list(date_range(date(2015, 1, 29), date(2015, 1, 29))))

temba/utils/tests/test_fields.py

@@ -0,0 +1,70 @@
+from django import forms
+from django.forms import ValidationError
+from django.test import TestCase
+
+from temba.utils.fields import ExternalURLField, NameValidator
+
+
+class TestFields(TestCase):
+    def test_name_validator(self):
+        cases = (
+            (" ", "Cannot begin or end with whitespace."),
+            (" hello", "Cannot begin or end with whitespace."),
+            ("hello\t", "Cannot begin or end with whitespace."),
+            ('hello "', 'Cannot contain the character: "'),
+            ("hello \\", "Cannot contain the character: \\"),
+            ("hello \0 world", "Cannot contain null characters."),
+            ("x" * 65, "Cannot be longer than 64 characters."),
+            ("hello world", None),
+            ("x" * 64, None),
+        )
+
+        validator = NameValidator(64)
+
+        for tc in cases:
+            if tc[1]:
+                with self.assertRaises(ValidationError) as cm:
+                    validator(tc[0])
+
+                self.assertEqual(tc[1], cm.exception.messages[0])
+            else:
+                try:
+                    validator(tc[0])
+                except Exception:
+                    self.fail(f"unexpected validation error for '{tc[0]}'")
+
+        self.assertEqual(NameValidator(64), validator)
+        self.assertNotEqual(NameValidator(32), validator)
+
+    def test_external_url_field(self):
+        class Form(forms.Form):
+            url = ExternalURLField()
+
+        cases = (
+            ("//[", ["Enter a valid URL."]),
+            ("ftp://google.com", ["Must use HTTP or HTTPS."]),
+            ("google.com", ["Enter a valid URL."]),
+            ("http://localhost/foo", ["Cannot be a local or private host."]),
+            ("http://localhost:80/foo", ["Cannot be a local or private host."]),
+            ("http://127.0.00.1/foo", ["Cannot be a local or private host."]),  # loop back
+            ("http://192.168.0.0/foo", ["Cannot be a local or private host."]),  # private
+            ("http://255.255.255.255", ["Cannot be a local or private host."]),  # multicast
+            ("http://169.254.169.254/latest", ["Cannot be a local or private host."]),  # link local
+            ("http://::1:80/foo", ["Unable to resolve host."]),  # no ipv6 addresses for now
+            ("http://google.com/foo", []),
+            ("http://google.com:8000/foo", []),
+            ("HTTP://google.com:8000/foo", []),
+            ("HTTP://8.8.8.8/foo", []),
+        )
+
+        for tc in cases:
+            form = Form({"url": tc[0]})
+            is_valid = form.is_valid()
+
+            if tc[1]:
+                self.assertFalse(is_valid, f"form.is_valid() unexpectedly true for '{tc[0]}'")
+                self.assertEqual({"url": tc[1]}, form.errors, f"validation errors mismatch for '{tc[0]}'")
+
+            else:
+                self.assertTrue(is_valid, f"form.is_valid() unexpectedly false for '{tc[0]}'")
+                self.assertEqual({}, form.errors)

temba/utils/tests/test_json.py

@@ -0,0 +1,41 @@
+from collections import OrderedDict
+from datetime import datetime, timezone as tzone
+from decimal import Decimal
+
+from django.utils import timezone
+
+from temba.tests import TembaTest
+from temba.utils import json
+
+
+class EncoderTest(TembaTest):
+    def test_encode_decode(self):
+        # create a time that has a set millisecond
+        now = timezone.now().replace(microsecond=1000)
+
+        # our dictionary to encode
+        source = dict(name="Date Test", age=Decimal("10"), now=now)
+
+        # encode it
+        encoded = json.dumps(source)
+
+        self.assertEqual(
+            json.loads(encoded), {"name": "Date Test", "age": Decimal("10"), "now": json.encode_datetime(now)}
+        )
+
+        # try it with a microsecond of 0 instead
+        source["now"] = timezone.now().replace(microsecond=0)
+
+        # encode it
+        encoded = json.dumps(source)
+
+        # test that we throw with unknown types
+        with self.assertRaises(TypeError):
+            json.dumps(dict(foo=Exception("invalid")))
+
+    def test_json(self):
+        self.assertEqual(OrderedDict({"one": 1, "two": Decimal("0.2")}), json.loads('{"one": 1, "two": 0.2}'))
+        self.assertEqual(
+            '{"dt": "2018-08-27T20:41:28.123Z"}',
+            json.dumps({"dt": datetime(2018, 8, 27, 20, 41, 28, 123000, tzinfo=tzone.utc)}),
+        )

temba/utils/tests/test_languages.py

@@ -0,0 +1,67 @@
+from django.test import override_settings
+
+from temba.tests import TembaTest
+from temba.utils import languages
+
+
+class LanguagesTest(TembaTest):
+    def test_get_name(self):
+        with override_settings(NON_ISO6391_LANGUAGES={"acx", "frc", "kir"}):
+            languages.reload()
+            self.assertEqual("French", languages.get_name("fra"))
+            self.assertEqual("Arabic (Omani, ISO-639-3)", languages.get_name("acx"))  # name is overridden
+            self.assertEqual("Cajun French", languages.get_name("frc"))  # non ISO-639-1 lang explicitly included
+            self.assertEqual("Kyrgyz", languages.get_name("kir"))
+            self.assertEqual("Oromifa", languages.get_name("orm"))
+
+            self.assertEqual("", languages.get_name("cpi"))  # not in our allowed languages
+            self.assertEqual("", languages.get_name("xyz"))
+
+            # should strip off anything after an open paren or semicolon
+            self.assertEqual("Haitian", languages.get_name("hat"))
+
+        languages.reload()
+
+    def test_search_by_name(self):
+        # check that search returns results and in the proper order
+        self.assertEqual(
+            [
+                {"value": "afr", "name": "Afrikaans"},
+                {"value": "fra", "name": "French"},
+                {"value": "fry", "name": "Western Frisian"},
+            ],
+            languages.search_by_name("Fr"),
+        )
+
+        # usually only return ISO-639-1 languages but can add inclusions in settings
+        with override_settings(NON_ISO6391_LANGUAGES={"afr", "afb", "acx", "frc"}):
+            languages.reload()
+
+            # order is based on name rather than code
+            self.assertEqual(
+                [
+                    {"value": "afr", "name": "Afrikaans"},
+                    {"value": "frc", "name": "Cajun French"},
+                    {"value": "fra", "name": "French"},
+                    {"value": "fry", "name": "Western Frisian"},
+                ],
+                languages.search_by_name("Fr"),
+            )
+
+            # searching and ordering uses overridden names
+            self.assertEqual(
+                [
+                    {"value": "ara", "name": "Arabic"},
+                    {"value": "afb", "name": "Arabic (Gulf, ISO-639-3)"},
+                    {"value": "acx", "name": "Arabic (Omani, ISO-639-3)"},
+                ],
+                languages.search_by_name("Arabic"),
+            )
+
+        languages.reload()
+
+    def alpha2_to_alpha3(self):
+        self.assertEqual("eng", languages.alpha2_to_alpha3("en"))
+        self.assertEqual("eng", languages.alpha2_to_alpha3("en-us"))
+        self.assertEqual("spa", languages.alpha2_to_alpha3("es"))
+        self.assertIsNone(languages.alpha2_to_alpha3("xx"))

temba/utils/tests/test_matchers.py

@@ -0,0 +1,35 @@
+from temba.tests import TembaTest, matchers
+
+
+class MatchersTest(TembaTest):
+    def test_string(self):
+        self.assertEqual("abc", matchers.String())
+        self.assertEqual("", matchers.String())
+        self.assertNotEqual(None, matchers.String())
+        self.assertNotEqual(123, matchers.String())
+
+        self.assertEqual("abc", matchers.String(pattern=r"\w{3}$"))
+        self.assertNotEqual("ab", matchers.String(pattern=r"\w{3}$"))
+        self.assertNotEqual("abcd", matchers.String(pattern=r"\w{3}$"))
+
+    def test_isodate(self):
+        self.assertEqual("2013-02-01T07:08:09.100000+04:30", matchers.ISODate())
+        self.assertEqual("2018-02-21T20:34:07.198537686Z", matchers.ISODate())
+        self.assertEqual("2018-02-21T20:34:07.19853768Z", matchers.ISODate())
+        self.assertEqual("2018-02-21T20:34:07.198Z", matchers.ISODate())
+        self.assertEqual("2018-02-21T20:34:07Z", matchers.ISODate())
+        self.assertEqual("2013-02-01T07:08:09.100000Z", matchers.ISODate())
+        self.assertNotEqual(None, matchers.ISODate())
+        self.assertNotEqual("abc", matchers.ISODate())
+
+    def test_uuid4string(self):
+        self.assertEqual("85ECBE45-E2DF-4785-8FC8-16FA941E0A79", matchers.UUID4String())
+        self.assertEqual("85ecbe45-e2df-4785-8fc8-16fa941e0a79", matchers.UUID4String())
+        self.assertNotEqual(None, matchers.UUID4String())
+        self.assertNotEqual("abc", matchers.UUID4String())
+
+    def test_dict(self):
+        self.assertEqual({}, matchers.Dict())
+        self.assertEqual({"a": "b"}, matchers.Dict())
+        self.assertNotEqual(None, matchers.Dict())
+        self.assertNotEqual([], matchers.Dict())

temba/utils/tests/test_middleware.py

@@ -0,0 +1,88 @@
+from django.test import override_settings
+from django.urls import reverse
+
+from temba.orgs.models import OrgRole
+from temba.tests import TembaTest, override_brand
+
+
+class MiddlewareTest(TembaTest):
+    def test_org(self):
+        index_url = reverse("public.public_index")
+
+        response = self.client.get(index_url)
+        self.assertFalse(response.has_header("X-Temba-Org"))
+
+        # if a user has a single org, that becomes the current org
+        self.login(self.admin)
+
+        response = self.client.get(index_url)
+        self.assertEqual(str(self.org.id), response["X-Temba-Org"])
+
+        # if not, org isn't set
+        self.org2.add_user(self.admin, OrgRole.ADMINISTRATOR)
+
+        response = self.client.get(index_url)
+        self.assertFalse(response.has_header("X-Temba-Org"))
+
+        # org will be read from session if set
+        s = self.client.session
+        s.update({"org_id": self.org.id})
+        s.save()
+
+        response = self.client.get(index_url)
+        self.assertEqual(str(self.org.id), response["X-Temba-Org"])
+
+        # org can be sent as a header too and we check it matches
+        response = self.client.post(reverse("flows.flow_create"), {}, headers={"X-Temba-Org": str(self.org.id)})
+        self.assertEqual(200, response.status_code)
+
+        response = self.client.post(reverse("flows.flow_create"), {}, headers={"X-Temba-Org": str(self.org2.id)})
+        self.assertEqual(403, response.status_code)
+
+        self.login(self.customer_support)
+
+        # our staff user doesn't have a default org
+        response = self.client.get(index_url)
+        self.assertFalse(response.has_header("X-Temba-Org"))
+
+        # but they can specify an org to service as a header
+        response = self.client.get(index_url, headers={"X-Temba-Service-Org": str(self.org.id)})
+        self.assertEqual(response["X-Temba-Org"], str(self.org.id))
+
+        response = self.client.get(index_url)
+        self.assertFalse(response.has_header("X-Temba-Org"))
+
+        self.login(self.editor)
+
+        response = self.client.get(index_url)
+        self.assertEqual(response["X-Temba-Org"], str(self.org.id))
+
+        # non-staff can't specify a different org from there own
+        response = self.client.get(index_url, headers={"X-Temba-Service-Org": str(self.org2.id)})
+        self.assertNotEqual(response["X-Temba-Org"], str(self.org2.id))
+
+    def test_redirect(self):
+        self.assertNotRedirect(self.client.get(reverse("public.public_index")), None)
+
+        # now set our brand to redirect
+        with override_brand(redirect="/redirect"):
+            self.assertRedirect(self.client.get(reverse("public.public_index")), "/redirect")
+
+    def test_language(self):
+        def assert_text(text: str):
+            self.assertContains(self.client.get(reverse("orgs.login")), text)
+
+        # default is English
+        assert_text("Sign In")
+
+        # can be overridden in Django settings
+        with override_settings(DEFAULT_LANGUAGE="es"):
+            assert_text("Ingresar")
+
+        # if we have an authenticated user, their setting takes priority
+        self.login(self.admin)
+
+        self.admin.settings.language = "fr"
+        self.admin.settings.save(update_fields=("language",))
+
+        assert_text("Se connecter")