From a276617e870467fbf666f727e6bc615642dd5466 Mon Sep 17 00:00:00 2001 From: tarsil Date: Wed, 2 Aug 2023 17:17:12 +0100 Subject: [PATCH] Add tests to unique constraints --- tests/uniques/test_unique.py | 55 +++++++++++ tests/uniques/test_unique_constraint.py | 121 ++++++++++++++++++++++++ tests/uniques/test_unique_together.py | 117 +++++++++++++++++++++++ 3 files changed, 293 insertions(+) create mode 100644 tests/uniques/test_unique.py create mode 100644 tests/uniques/test_unique_constraint.py create mode 100644 tests/uniques/test_unique_together.py diff --git a/tests/uniques/test_unique.py b/tests/uniques/test_unique.py new file mode 100644 index 00000000..09ce6402 --- /dev/null +++ b/tests/uniques/test_unique.py @@ -0,0 +1,55 @@ +import datetime +from enum import Enum + +import pytest +from asyncpg.exceptions import UniqueViolationError +from tests.settings import DATABASE_URL + +import edgy +from edgy.testclient import DatabaseTestClient as Database + +pytestmark = pytest.mark.anyio + +database = Database(DATABASE_URL) +models = edgy.Registry(database=database) + + +def time(): + return datetime.datetime.now().time() + + +class StatusEnum(Enum): + DRAFT = "Draft" + RELEASED = "Released" + + +class BaseModel(edgy.Model): + class Meta: + registry = models + + +class User(BaseModel): + name = edgy.CharField(max_length=255, unique=True) + email = edgy.CharField(max_length=60) + + +@pytest.fixture(autouse=True, scope="module") +async def create_test_database(): + await models.create_all() + yield + await models.drop_all() + + +@pytest.fixture(autouse=True) +async def rollback_transactions(): + with database.force_rollback(): + async with database: + yield + + +@pytest.mark.skipif(database.url.dialect == "mysql", reason="Not supported on MySQL") +async def test_unique(): + await User.query.create(name="Tiago", email="test@example.com") + + with pytest.raises(UniqueViolationError): + await User.query.create(name="Tiago", email="test2@example.come") diff --git a/tests/uniques/test_unique_constraint.py b/tests/uniques/test_unique_constraint.py new file mode 100644 index 00000000..93982343 --- /dev/null +++ b/tests/uniques/test_unique_constraint.py @@ -0,0 +1,121 @@ +import datetime +from enum import Enum + +import pytest +from asyncpg.exceptions import UniqueViolationError +from tests.settings import DATABASE_URL + +import edgy +from edgy.core.db.datastructures import UniqueConstraint +from edgy.testclient import DatabaseTestClient as Database + +pytestmark = pytest.mark.anyio + +database = Database(DATABASE_URL) +models = edgy.Registry(database=database) + + +def time(): + return datetime.datetime.now().time() + + +class StatusEnum(Enum): + DRAFT = "Draft" + RELEASED = "Released" + + +class BaseModel(edgy.Model): + class Meta: + registry = models + + +class User(BaseModel): + name = edgy.CharField(max_length=255) + email = edgy.CharField(max_length=60) + + class Meta: + unique_together = [UniqueConstraint(fields=["name", "email"])] + + +class HubUser(BaseModel): + name = edgy.CharField(max_length=255) + email = edgy.CharField(max_length=60, null=True) + age = edgy.IntegerField(minimum=18, null=True) + + class Meta: + unique_together = [ + UniqueConstraint(fields=["name", "email"]), + ("email", "age"), + ] + + +class Product(BaseModel): + name = edgy.CharField(max_length=255) + sku = edgy.CharField(max_length=255) + + class Meta: + unique_together = [UniqueConstraint(fields=["name"]), UniqueConstraint(fields=["sku"])] + + +class NewProduct(BaseModel): + name = edgy.CharField(max_length=255) + sku = edgy.CharField(max_length=255) + + class Meta: + unique_together = [UniqueConstraint(fields=["name"]), "sku"] + + +@pytest.fixture(autouse=True, scope="module") +async def create_test_database(): + await models.create_all() + yield + await models.drop_all() + + +@pytest.fixture(autouse=True) +async def rollback_transactions(): + with database.force_rollback(): + async with database: + yield + + +@pytest.mark.skipif(database.url.dialect == "mysql", reason="Not supported on MySQL") +async def test_unique_together(): + await User.query.create(name="Test", email="test@example.com") + await User.query.create(name="Test", email="test2@example.come") + + with pytest.raises(UniqueViolationError): + await User.query.create(name="Test", email="test@example.com") + + +@pytest.mark.skipif(database.url.dialect == "mysql", reason="Not supported on MySQL") +async def test_unique_together_multiple(): + await HubUser.query.create(name="Test", email="test@example.com") + await HubUser.query.create(name="Test", email="test2@example.come") + + with pytest.raises(UniqueViolationError): + await HubUser.query.create(name="Test", email="test@example.com") + + +@pytest.mark.skipif(database.url.dialect == "mysql", reason="Not supported on MySQL") +async def test_unique_together_multiple_name_age(): + await HubUser.query.create(name="NewTest", email="test@example.com", age=18) + + with pytest.raises(UniqueViolationError): + await HubUser.query.create(name="Test", email="test@example.com", age=18) + + +@pytest.mark.skipif(database.url.dialect == "mysql", reason="Not supported on MySQL") +async def test_unique_together_multiple_single_string(): + await Product.query.create(name="android", sku="12345") + + with pytest.raises(UniqueViolationError): + await Product.query.create(name="android", sku="12345") + + +@pytest.mark.skipif(database.url.dialect == "mysql", reason="Not supported on MySQL") +async def test_unique_together_multiple_single_string_two(): + await Product.query.create(name="android", sku="12345") + + with pytest.raises(UniqueViolationError): + await Product.query.create(name="iphone", sku="12345") diff --git a/tests/uniques/test_unique_together.py b/tests/uniques/test_unique_together.py new file mode 100644 index 00000000..139d7e04 --- /dev/null +++ b/tests/uniques/test_unique_together.py @@ -0,0 +1,117 @@ +import datetime +from enum import Enum + +import pytest +from asyncpg.exceptions import UniqueViolationError +from tests.settings import DATABASE_URL + +import edgy +from edgy.testclient import DatabaseTestClient as Database + +pytestmark = pytest.mark.anyio + +database = Database(DATABASE_URL) +models = edgy.Registry(database=database) + + +def time(): + return datetime.datetime.now().time() + + +class StatusEnum(Enum): + DRAFT = "Draft" + RELEASED = "Released" + + +class BaseModel(edgy.Model): + class Meta: + registry = models + + +class User(BaseModel): + name = edgy.CharField(max_length=255) + email = edgy.CharField(max_length=60) + + class Meta: + unique_together = [("name", "email")] + + +class HubUser(BaseModel): + name = edgy.CharField(max_length=255) + email = edgy.CharField(max_length=60, null=True) + age = edgy.IntegerField(minimum=18, null=True) + + class Meta: + unique_together = [("name", "email"), ("email", "age")] + + +class Product(BaseModel): + name = edgy.CharField(max_length=255) + sku = edgy.CharField(max_length=255) + + class Meta: + unique_together = ["name", "sku"] + + +class NewProduct(BaseModel): + name = edgy.CharField(max_length=255) + sku = edgy.CharField(max_length=255) + + class Meta: + unique_together = ["name", "sku"] + + +@pytest.fixture(autouse=True, scope="module") +async def create_test_database(): + await models.create_all() + yield + await models.drop_all() + + +@pytest.fixture(autouse=True) +async def rollback_transactions(): + with database.force_rollback(): + async with database: + yield + + +@pytest.mark.skipif(database.url.dialect == "mysql", reason="Not supported on MySQL") +async def test_unique_together(): + await User.query.create(name="Test", email="test@example.com") + await User.query.create(name="Test", email="test2@example.come") + + with pytest.raises(UniqueViolationError): + await User.query.create(name="Test", email="test@example.com") + + +@pytest.mark.skipif(database.url.dialect == "mysql", reason="Not supported on MySQL") +async def test_unique_together_multiple(): + await HubUser.query.create(name="Test", email="test@example.com") + await HubUser.query.create(name="Test", email="test2@example.come") + + with pytest.raises(UniqueViolationError): + await HubUser.query.create(name="Test", email="test@example.com") + + +@pytest.mark.skipif(database.url.dialect == "mysql", reason="Not supported on MySQL") +async def test_unique_together_multiple_name_age(): + await HubUser.query.create(name="NewTest", email="test@example.com", age=18) + + with pytest.raises(UniqueViolationError): + await HubUser.query.create(name="Test", email="test@example.com", age=18) + + +@pytest.mark.skipif(database.url.dialect == "mysql", reason="Not supported on MySQL") +async def test_unique_together_multiple_single_string(): + await Product.query.create(name="android", sku="12345") + + with pytest.raises(UniqueViolationError): + await Product.query.create(name="android", sku="12345") + + +@pytest.mark.skipif(database.url.dialect == "mysql", reason="Not supported on MySQL") +async def test_unique_together_multiple_single_string_two(): + await Product.query.create(name="android", sku="12345") + + with pytest.raises(UniqueViolationError): + await Product.query.create(name="iphone", sku="12345")