Initial commitv3.3.4

5 files changed, 287 insertions, 0 deletions

diff --git a/tests/models/custom_types/__init__.py b/tests/models/custom_types/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/tests/models/custom_types/__init__.py
diff --git a/tests/models/custom_types/test_aware_datetime.py b/tests/models/custom_types/test_aware_datetime.py
new file mode 100644
index 0000000..a23413c
--- /dev/null
+++ b/tests/models/custom_types/test_aware_datetime.py
@@ -0,0 +1,82 @@
+import logging
+from datetime import datetime, timezone
+from typing import Optional
+
+import pytest
+import pytz
+from pydantic import BaseModel, ValidationError, Field
+
+from generalresearch.models.custom_types import AwareDatetimeISO
+
+logger = logging.getLogger()
+
+
+class AwareDatetimeISOModel(BaseModel):
+    dt_optional: Optional[AwareDatetimeISO] = Field(default=None)
+    dt: AwareDatetimeISO
+
+
+class TestAwareDatetimeISO:
+    def test_str(self):
+        dt = "2023-10-10T01:01:01.0Z"
+        t = AwareDatetimeISOModel(dt=dt, dt_optional=dt)
+        AwareDatetimeISOModel.model_validate_json(t.model_dump_json())
+
+        t = AwareDatetimeISOModel(dt=dt, dt_optional=None)
+        AwareDatetimeISOModel.model_validate_json(t.model_dump_json())
+
+    def test_dt(self):
+        dt = datetime(2023, 10, 10, 1, 1, 1, tzinfo=timezone.utc)
+        t = AwareDatetimeISOModel(dt=dt, dt_optional=dt)
+        AwareDatetimeISOModel.model_validate_json(t.model_dump_json())
+
+        t = AwareDatetimeISOModel(dt=dt, dt_optional=None)
+        AwareDatetimeISOModel.model_validate_json(t.model_dump_json())
+
+        dt = datetime(2023, 10, 10, 1, 1, 1, microsecond=123, tzinfo=timezone.utc)
+        t = AwareDatetimeISOModel(dt=dt, dt_optional=dt)
+        AwareDatetimeISOModel.model_validate_json(t.model_dump_json())
+
+        t = AwareDatetimeISOModel(dt=dt, dt_optional=None)
+        AwareDatetimeISOModel.model_validate_json(t.model_dump_json())
+
+    def test_no_tz(self):
+        dt = datetime(2023, 10, 10, 1, 1, 1)
+
+        with pytest.raises(expected_exception=ValidationError):
+            AwareDatetimeISOModel(dt=dt, dt_optional=None)
+
+        dt = "2023-10-10T01:01:01.0"
+        with pytest.raises(expected_exception=ValidationError):
+            AwareDatetimeISOModel(dt=dt, dt_optional=None)
+
+    def test_non_utc_tz(self):
+        dt = datetime(
+            year=2023,
+            month=10,
+            day=10,
+            hour=1,
+            second=1,
+            minute=1,
+            tzinfo=pytz.timezone("US/Central"),
+        )
+
+        with pytest.raises(expected_exception=ValidationError):
+            AwareDatetimeISOModel(dt=dt, dt_optional=dt)
+
+    def test_invalid_format(self):
+        dt = "2023-10-10T01:01:01Z"
+        with pytest.raises(expected_exception=ValidationError):
+            AwareDatetimeISOModel(dt=dt, dt_optional=dt)
+
+        dt = "2023-10-10T01:01:01"
+        with pytest.raises(expected_exception=ValidationError):
+            AwareDatetimeISOModel(dt=dt, dt_optional=dt)
+        dt = "2023-10-10"
+        with pytest.raises(expected_exception=ValidationError):
+            AwareDatetimeISOModel(dt=dt, dt_optional=dt)
+
+    def test_required(self):
+        dt = "2023-10-10T01:01:01.0Z"
+        with pytest.raises(expected_exception=ValidationError):
+            AwareDatetimeISOModel(dt=None, dt_optional=dt)
diff --git a/tests/models/custom_types/test_dsn.py b/tests/models/custom_types/test_dsn.py
new file mode 100644
index 0000000..b37f2c4
--- /dev/null
+++ b/tests/models/custom_types/test_dsn.py
@@ -0,0 +1,112 @@
+from typing import Optional
+from uuid import uuid4
+
+import pytest
+from pydantic import BaseModel, ValidationError, Field
+from pydantic import MySQLDsn
+from pydantic_core import Url
+
+from generalresearch.models.custom_types import DaskDsn, SentryDsn
+
+
+# --- Test Pydantic Models ---
+
+
+class SettingsModel(BaseModel):
+    dask: Optional["DaskDsn"] = Field(default=None)
+    sentry: Optional["SentryDsn"] = Field(default=None)
+    db: Optional["MySQLDsn"] = Field(default=None)
+
+
+# --- Pytest themselves ---
+
+
+class TestDaskDsn:
+
+    def test_base(self):
+        from dask.distributed import Client
+
+        m = SettingsModel(dask="tcp://dask-scheduler.internal")
+
+        assert m.dask.scheme == "tcp"
+        assert m.dask.host == "dask-scheduler.internal"
+        assert m.dask.port == 8786
+
+        with pytest.raises(expected_exception=TypeError) as cm:
+            Client(m.dask)
+        assert "Scheduler address must be a string or a Cluster instance" in str(
+            cm.value
+        )
+
+        # todo: this requires vpn connection. maybe do this part with a localhost dsn
+        # client = Client(str(m.dask))
+        # self.assertIsInstance(client, Client)
+
+    def test_str(self):
+        m = SettingsModel(dask="tcp://dask-scheduler.internal")
+        assert isinstance(m.dask, Url)
+        assert "tcp://dask-scheduler.internal:8786" == str(m.dask)
+
+    def test_auth(self):
+        with pytest.raises(expected_exception=ValidationError) as cm:
+            SettingsModel(dask="tcp://test:password@dask-scheduler.internal")
+        assert "User & Password are not supported" in str(cm.value)
+
+        with pytest.raises(expected_exception=ValidationError) as cm:
+            SettingsModel(dask="tcp://test:@dask-scheduler.internal")
+        assert "User & Password are not supported" in str(cm.value)
+
+        with pytest.raises(expected_exception=ValidationError) as cm:
+            SettingsModel(dask="tcp://:password@dask-scheduler.internal")
+        assert "User & Password are not supported" in str(cm.value)
+
+    def test_invalid_schema(self):
+        with pytest.raises(expected_exception=ValidationError) as cm:
+            SettingsModel(dask="dask-scheduler.internal")
+        assert "relative URL without a base" in str(cm.value)
+
+        # I look forward to the day we use infiniband interfaces
+        with pytest.raises(expected_exception=ValidationError) as cm:
+            SettingsModel(dask="ucx://dask-scheduler.internal")
+        assert "URL scheme should be 'tcp'" in str(cm.value)
+
+    def test_port(self):
+        m = SettingsModel(dask="tcp://dask-scheduler.internal")
+        assert m.dask.port == 8786
+
+
+class TestSentryDsn:
+    def test_base(self):
+        m = SettingsModel(
+            sentry=f"https://{uuid4().hex}@12345.ingest.us.sentry.io/9876543"
+        )
+
+        assert m.sentry.scheme == "https"
+        assert m.sentry.host == "12345.ingest.us.sentry.io"
+        assert m.sentry.port == 443
+
+    def test_str(self):
+        test_url: str = f"https://{uuid4().hex}@12345.ingest.us.sentry.io/9876543"
+        m = SettingsModel(sentry=test_url)
+        assert isinstance(m.sentry, Url)
+        assert test_url == str(m.sentry)
+
+    def test_auth(self):
+        with pytest.raises(expected_exception=ValidationError) as cm:
+            SettingsModel(
+                sentry="https://0123456789abc:password@12345.ingest.us.sentry.io/9876543"
+            )
+        assert "Sentry password is not supported" in str(cm.value)
+
+        with pytest.raises(expected_exception=ValidationError) as cm:
+            SettingsModel(sentry="https://test:@12345.ingest.us.sentry.io/9876543")
+        assert "Sentry user key seems bad" in str(cm.value)
+
+        with pytest.raises(expected_exception=ValidationError) as cm:
+            SettingsModel(sentry="https://:password@12345.ingest.us.sentry.io/9876543")
+        assert "Sentry URL requires a user key" in str(cm.value)
+
+    def test_port(self):
+        test_url: str = f"https://{uuid4().hex}@12345.ingest.us.sentry.io/9876543"
+        m = SettingsModel(sentry=test_url)
+        assert m.sentry.port == 443
diff --git a/tests/models/custom_types/test_therest.py b/tests/models/custom_types/test_therest.py
new file mode 100644
index 0000000..13e9bae
--- /dev/null
+++ b/tests/models/custom_types/test_therest.py
@@ -0,0 +1,42 @@
+import json
+from uuid import UUID
+
+import pytest
+from pydantic import TypeAdapter, ValidationError
+
+
+class TestAll:
+
+    def test_comma_sep_str(self):
+        from generalresearch.models.custom_types import AlphaNumStrSet
+
+        t = TypeAdapter(AlphaNumStrSet)
+        assert {"a", "b", "c"} == t.validate_python(["a", "b", "c"])
+        assert '"a,b,c"' == t.dump_json({"c", "b", "a"}).decode()
+        assert '""' == t.dump_json(set()).decode()
+        assert {"a", "b", "c"} == t.validate_json('"c,b,a"')
+        assert set() == t.validate_json('""')
+
+        with pytest.raises(ValidationError):
+            t.validate_python({"", "b", "a"})
+
+        with pytest.raises(ValidationError):
+            t.validate_python({""})
+
+        with pytest.raises(ValidationError):
+            t.validate_json('",b,a"')
+
+    def test_UUIDStrCoerce(self):
+        from generalresearch.models.custom_types import UUIDStrCoerce
+
+        t = TypeAdapter(UUIDStrCoerce)
+        uuid_str = "18e70590176e49c693b07682f3c112be"
+        assert uuid_str == t.validate_python("18e70590-176e-49c6-93b0-7682f3c112be")
+        assert uuid_str == t.validate_python(
+            UUID("18e70590-176e-49c6-93b0-7682f3c112be")
+        )
+        assert (
+            json.dumps(uuid_str)
+            == t.dump_json("18e70590176e49c693b07682f3c112be").decode()
+        )
+        assert uuid_str == t.validate_json('"18e70590-176e-49c6-93b0-7682f3c112be"')
diff --git a/tests/models/custom_types/test_uuid_str.py b/tests/models/custom_types/test_uuid_str.py
new file mode 100644
index 0000000..91af9ae
--- /dev/null
+++ b/tests/models/custom_types/test_uuid_str.py
@@ -0,0 +1,51 @@
+from typing import Optional
+from uuid import uuid4
+
+import pytest
+from pydantic import BaseModel, ValidationError, Field
+
+from generalresearch.models.custom_types import UUIDStr
+
+
+class UUIDStrModel(BaseModel):
+    uuid_optional: Optional[UUIDStr] = Field(default_factory=lambda: uuid4().hex)
+    uuid: UUIDStr
+
+
+class TestUUIDStr:
+    def test_str(self):
+        v = "58889cd67f9f4c699b25437112dce638"
+
+        t = UUIDStrModel(uuid=v, uuid_optional=v)
+        UUIDStrModel.model_validate_json(t.model_dump_json())
+
+        t = UUIDStrModel(uuid=v, uuid_optional=None)
+        t2 = UUIDStrModel.model_validate_json(t.model_dump_json())
+
+        assert t2.uuid_optional is None
+        assert t2.uuid == v
+
+    def test_uuid(self):
+        v = uuid4()
+
+        with pytest.raises(ValidationError) as cm:
+            UUIDStrModel(uuid=v, uuid_optional=None)
+        assert "Input should be a valid string" in str(cm.value)
+
+        with pytest.raises(ValidationError) as cm:
+            UUIDStrModel(uuid="58889cd67f9f4c699b25437112dce638", uuid_optional=v)
+        assert "Input should be a valid string" in str(cm.value)
+
+    def test_invalid_format(self):
+        v = "x"
+        with pytest.raises(ValidationError):
+            UUIDStrModel(uuid=v, uuid_optional=None)
+
+        with pytest.raises(ValidationError):
+            UUIDStrModel(uuid="58889cd67f9f4c699b25437112dce638", uuid_optional=v)
+
+    def test_required(self):
+        v = "58889cd67f9f4c699b25437112dce638"
+
+        with pytest.raises(ValidationError):
+            UUIDStrModel(uuid=None, uuid_optional=v)