diff --git a/fastapi_startkit/pyproject.toml b/fastapi_startkit/pyproject.toml index d64cae4e..5c090120 100644 --- a/fastapi_startkit/pyproject.toml +++ b/fastapi_startkit/pyproject.toml @@ -90,7 +90,7 @@ omit = [ [tool.coverage.report] show_missing = true skip_covered = false -fail_under = 40 +fail_under = 50 exclude_lines = [ "pragma: no cover", "if TYPE_CHECKING:", diff --git a/fastapi_startkit/src/fastapi_startkit/masoniteorm.backup/tests/factories/test_factories.py b/fastapi_startkit/src/fastapi_startkit/masoniteorm.backup/tests/factories/test_factories.py deleted file mode 100644 index 868264b0..00000000 --- a/fastapi_startkit/src/fastapi_startkit/masoniteorm.backup/tests/factories/test_factories.py +++ /dev/null @@ -1,66 +0,0 @@ -import unittest - -from src.masoniteorm.factories import Factory as factory -from src.masoniteorm.models import Model - - -class User(Model): - pass - - -class AfterCreatedModel(Model): - __dry__ = True - pass - - -class TestFactories(unittest.TestCase): - def setUp(self): - factory.register(User, self.user_factory) - factory.register(User, self.named_user_factory, name="admin") - factory.register(AfterCreatedModel, self.named_user_factory) - factory.after_creating(AfterCreatedModel, self.after_creating) - - def user_factory(self, faker): - return {"id": 1, "name": faker.name()} - - def named_user_factory(self, faker): - return {"id": 1, "name": faker.name(), "admin": 1} - - def after_creating(self, model, faker): - model.after_created = True - - def test_can_make_single(self): - user = factory(User).make({"id": 1, "name": "Joe"}) - - self.assertEqual(user.name, "Joe") - self.assertIsInstance(user, User) - - def test_can_make_several(self): - users = factory(User).make([{"id": 1, "name": "Joe"}, {"id": 2, "name": "Bob"}]) - - self.assertEqual(users.count(), 2) - - def test_can_make_any_number(self): - users = factory(User, 50).make() - - self.assertEqual(users.count(), 50) - - def test_can_make_named_factory(self): - user = factory(User).make(name="admin") - self.assertEqual(user.admin, 1) - - def test_after_creates(self): - user = factory(AfterCreatedModel).create() - self.assertTrue(user.name) - self.assertEqual(user.after_created, True) - - users = factory(AfterCreatedModel).create({"name": "billy"}) - self.assertEqual(users.name, "billy") - self.assertEqual(users.after_created, True) - - users = factory(AfterCreatedModel).make() - self.assertEqual(users.after_created, True) - - users = factory(AfterCreatedModel, 2).make() - for user in users: - self.assertEqual(user.after_created, True) diff --git a/fastapi_startkit/src/fastapi_startkit/masoniteorm.backup/tests/integrations/config/database.py b/fastapi_startkit/src/fastapi_startkit/masoniteorm.backup/tests/integrations/config/database.py deleted file mode 100644 index d963cdae..00000000 --- a/fastapi_startkit/src/fastapi_startkit/masoniteorm.backup/tests/integrations/config/database.py +++ /dev/null @@ -1,119 +0,0 @@ -"""Database Settings""" - -import logging -import os - -from dotenv import load_dotenv - -from fastapi_startkit.masoniteorm.connections.factory import ConnectionFactory - - -load_dotenv(".env") - - -DATABASES = { - "default": "dev", - "mysql": { - "driver": "mysql", - "host": os.getenv("MYSQL_DATABASE_HOST"), - "user": os.getenv("MYSQL_DATABASE_USER"), - "password": os.getenv("MYSQL_DATABASE_PASSWORD"), - "database": os.getenv("MYSQL_DATABASE_DATABASE"), - "port": os.getenv("MYSQL_DATABASE_PORT"), - "prefix": "", - "options": {"charset": "utf8mb4"}, - "log_queries": True, - "propagate": False, - "connection_pooling_enabled": True, - "connection_pooling_max_size": 10, - "connection_pooling_min_size": None, - }, - "t": { - "driver": "sqlite", - "database": "orm.sqlite3", - "log_queries": True, - "foreign_keys": True, - }, - "devprod": { - "driver": "mysql", - "host": os.getenv("MYSQL_DATABASE_HOST"), - "user": os.getenv("MYSQL_DATABASE_USER"), - "password": os.getenv("MYSQL_DATABASE_PASSWORD"), - "database": "DEVPROD", - "port": os.getenv("MYSQL_DATABASE_PORT"), - "prefix": "", - "options": {"charset": "utf8mb4"}, - "log_queries": True, - "propagate": False, - }, - "many": { - "driver": "mysql", - "host": "localhost", - "user": "root", - "password": "", - "database": "replicate", - "port": os.getenv("MYSQL_DATABASE_PORT"), - "options": {"charset": "utf8mb4"}, - "log_queries": True, - "propagate": False, - }, - "postgres": { - "driver": "postgres", - "host": os.getenv("POSTGRES_DATABASE_HOST"), - "user": os.getenv("POSTGRES_DATABASE_USER"), - "password": os.getenv("POSTGRES_DATABASE_PASSWORD"), - "database": os.getenv("POSTGRES_DATABASE_DATABASE"), - "port": os.getenv("POSTGRES_DATABASE_PORT"), - "connection_pooling_enabled": True, - "connection_pooling_max_size": 10, - "connection_pooling_min_size": 2, - "prefix": "", - "log_queries": True, - "propagate": False, - }, - # Example with db_url() - # "postgres": db_url( - # "postgres://user:@localhost:5432/postgres", log_queries=True - # ), - "dev": { - "driver": "sqlite", - "database": ":memory:", - "prefix": "", - "log_queries": True, - }, - # Example with db_url() - # "dev": {**db_url("sqlite://orm.sqlite3"), "prefix": "", "log_queries": True}, - "mssql": { - "driver": "mssql", - "host": os.getenv("MSSQL_DATABASE_HOST"), - "user": os.getenv("MSSQL_DATABASE_USER"), - "password": os.getenv("MSSQL_DATABASE_PASSWORD"), - "database": os.getenv("MSSQL_DATABASE_DATABASE"), - "port": os.getenv("MSSQL_DATABASE_PORT"), - "prefix": "", - "log_queries": True, - "options": { - "trusted_connection": "Yes", - "integrated_security": "sspi", - "instance": "SQLExpress", - "authentication": "ActiveDirectoryPassword", - "driver": "ODBC Driver 17 for SQL Server", - "connection_timeout": 15, - "connection_pooling": False, - "connection_pooling_size": 100, - }, - }, -} - -DB = ConnectionFactory(connection_details=DATABASES) - -logger = logging.getLogger("masoniteorm.connection.queries") -logger.setLevel(logging.DEBUG) - -formatter = logging.Formatter("It executed the query %(query)s") - -stream_handler = logging.StreamHandler() -file_handler = logging.FileHandler("queries.log") - -logger.addHandler(stream_handler) -logger.addHandler(file_handler) diff --git a/fastapi_startkit/src/fastapi_startkit/masoniteorm.backup/tests/integrations/databases/migrations/2026_01_01_000000_create_users_table.py b/fastapi_startkit/src/fastapi_startkit/masoniteorm.backup/tests/integrations/databases/migrations/2026_01_01_000000_create_users_table.py deleted file mode 100644 index 34de112a..00000000 --- a/fastapi_startkit/src/fastapi_startkit/masoniteorm.backup/tests/integrations/databases/migrations/2026_01_01_000000_create_users_table.py +++ /dev/null @@ -1,29 +0,0 @@ -from fastapi_startkit.masoniteorm.migrations import Migration - - -class CreateUsersTable(Migration): - async def up(self): - """ - Run the migrations. - """ - async with await self.schema.create("users") as table: - table.increments("id") - table.string("name") - table.string("username").unique() - table.string("email").unique() - table.string("password") - table.integer("age").nullable() - table.float("balance").nullable() - table.json("metadata").nullable() - table.string("gender").nullable() - table.json("address").nullable() - table.boolean("is_active").default(True) - table.text("bio").nullable() - table.decimal("price", 10, 2).nullable() - table.timestamps() - - async def down(self): - """ - Revert the migrations. - """ - await self.schema.drop("users") diff --git a/fastapi_startkit/src/fastapi_startkit/masoniteorm.backup/tests/integrations/model.py b/fastapi_startkit/src/fastapi_startkit/masoniteorm.backup/tests/integrations/model.py deleted file mode 100644 index e1823ef8..00000000 --- a/fastapi_startkit/src/fastapi_startkit/masoniteorm.backup/tests/integrations/model.py +++ /dev/null @@ -1,57 +0,0 @@ -from dataclasses import dataclass, asdict -from enum import StrEnum - -from fastapi_startkit.masoniteorm import Model - - -@dataclass -class Address: - city: str = None - country: str = None - street: str = None - - def get(self, value): - import json - - if isinstance(value, str): - value = json.loads(value) - - return Address( - city=value.get("city"), - country=value.get("country"), - street=value.get("street"), - ) - - def set(self, value): - import json - - if isinstance(value, Address): - return json.dumps(asdict(value)) - return json.dumps(value) - - -class Gender(StrEnum): - MALE = "male" - FEMALE = "female" - OTHER = "other" - - -class User(Model): - __primary_key__ = "id" - __table__ = "users" - __casts__ = { - "metadata": "json", - } - - id: int - name: str - age: int - balance: float - metadata: dict - is_active: bool - bio: str - price: float - - # created_at: Date - gender: Gender - address: Address diff --git a/fastapi_startkit/src/fastapi_startkit/masoniteorm.backup/tests/integrations/test_model.py b/fastapi_startkit/src/fastapi_startkit/masoniteorm.backup/tests/integrations/test_model.py deleted file mode 100644 index f6c7696e..00000000 --- a/fastapi_startkit/src/fastapi_startkit/masoniteorm.backup/tests/integrations/test_model.py +++ /dev/null @@ -1,70 +0,0 @@ -from fastapi_startkit.masoniteorm.testing import TestCase -from fastapi_startkit.masoniteorm.tests.integrations.model import User, Gender, Address - - -class TestModelCast(TestCase): - migration_directory = "src/fastapi_startkit/masoniteorm/tests/integrations/databases/migrations" - - async def test_database_is_isolated(self): - user = await User.first() - self.assertIsNone(user) - - async def test_first_record_can_be_fetch(self): - await User.create(name="Joe", username="joe", email="joe@test.com", password="password") - - user = await User.first() - self.assertEqual(user.name, "Joe") - self.assertEqual(user.username, "joe") - self.assertEqual(user.email, "joe@test.com") - self.assertEqual(user.password, "password") - - async def test_can_create_with_dict(self): - await User.create( - { - "name": "Jane", - "username": "jane", - "email": "jane@test.com", - "password": "password", - } - ) - - user = await User.first() - self.assertEqual(user.name, "Jane") - - async def test_can_insert_all_types(self): - data = { - "name": "All Types", - "username": "alltypes", - "email": "all@types.com", - "password": "password", - "age": 25, - "balance": 1000.50, - "metadata": {"key": "value", "active": True}, - "is_active": True, - "bio": "A long text for bio...", - "price": 19.99, - "gender": Gender.MALE, - "address": {"city": "New York", "country": "USA", "street": "Broadway"}, - } - await User.create(data) - - user = await User.where("username", "alltypes").first() - self.assertEqual(user.name, "All Types") - self.assertEqual(user.age, 25) - self.assertEqual(user.balance, 1000.50) - - # Test JSON - self.assertEqual(user.metadata, {"key": "value", "active": True}) - - self.assertTrue(user.is_active) - self.assertEqual(user.bio, "A long text for bio...") - self.assertEqual(float(user.price), 19.99) - - # Test Gender - self.assertEqual(user.gender, Gender.MALE) - - # Test Address - self.assertIsInstance(user.address, Address) - self.assertEqual(user.address.city, "New York") - self.assertEqual(user.address.country, "USA") - self.assertEqual(user.address.street, "Broadway") diff --git a/fastapi_startkit/tests/masoniteorm/models/test_model.py b/fastapi_startkit/tests/masoniteorm/models/test_model.py index c3f4870c..88ebdbc6 100644 --- a/fastapi_startkit/tests/masoniteorm/models/test_model.py +++ b/fastapi_startkit/tests/masoniteorm/models/test_model.py @@ -1,56 +1,6 @@ -import pytest - -from fastapi_startkit.carbon import Carbon -from fastapi_startkit.masoniteorm.models.fields import DateTimeField -from fastapi_startkit.masoniteorm.connections.factory import ConnectionFactory -from fastapi_startkit.masoniteorm.connections.manager import DatabaseManager from fastapi_startkit.masoniteorm.models.model import Model - -# --------------------------------------------------------------------------- -# Shared fixtures -# --------------------------------------------------------------------------- - -SQLITE_CONFIG = { - "default": "sqlite", - "connections": { - "sqlite": { - "driver": "sqlite", - "url": "sqlite+aiosqlite:///:memory:", - }, - }, -} - - -@pytest.fixture -def db(): - return DatabaseManager(ConnectionFactory(), SQLITE_CONFIG) - - -@pytest.fixture -def UserModel(db): - class User(Model): - id: int - name: str - email: str - email_verified_at: Carbon = DateTimeField(fmt="%Y-%m-%d %H:%M:%S", tz="UTC") - - User.db_manager = db - return User - - -@pytest.fixture -async def users_table(db): - """Create the user's table and drop it after each test.""" - schema = db.get_schema_builder() - await schema.drop_table_if_exists("users") - async with await schema.on("default").create("users") as table: - table.id() - table.string("name") - table.string("email").unique() - table.timestamp("email_verified_at").nullable() - table.timestamps() - yield schema - await schema.drop_table_if_exists("users") +from tests.masoniteorm.fixtures.model import User +from tests.masoniteorm.sqlite.test_case import TestCase # --------------------------------------------------------------------------- @@ -58,20 +8,16 @@ async def users_table(db): # --------------------------------------------------------------------------- -class TestSchema: - async def test_create_table(self, db, users_table): - schema = db.get_schema_builder() - assert await schema.on("default").has_table("users") +class TestSchema(TestCase): + async def test_create_table(self): + assert await self.schema.on("default").has_table("users") - async def test_drop_table_if_exists(self, db): - schema = db.get_schema_builder() - # table does not exist yet — should not raise - await schema.drop_table_if_exists("nonexistent_table") + async def test_drop_table_if_exists(self): + await self.schema.drop_table_if_exists("nonexistent_table") - async def test_drop_table_removes_table(self, db, users_table): - schema = db.get_schema_builder() - await schema.drop_table("users") - assert not await schema.on("default").has_table("users") + async def test_drop_table_removes_table(self): + await self.schema.drop_table("users") + assert not await self.schema.on("default").has_table("users") # --------------------------------------------------------------------------- @@ -79,62 +25,47 @@ async def test_drop_table_removes_table(self, db, users_table): # --------------------------------------------------------------------------- -class TestModelInsert: - async def test_save_inserts_row(self, UserModel, users_table): - user = UserModel(name="Alex", email="alex@gmail.com") - saved = await user.save() - - assert saved is True +class TestModelInsert(TestCase): + async def test_save_inserts_row(self): + user = User(name="Alex", email="alex@gmail.com") + assert await user.save() is True - async def test_save_sets_exists_flag(self, UserModel, users_table): - user = UserModel(name="Alex", email="alex@gmail.com") + async def test_save_sets_exists_flag(self): + user = User(name="Alex", email="alex@gmail.com") await user.save() assert user._exists is True - async def test_save_sets_was_recently_created(self, UserModel, users_table): - user = UserModel(name="Alex", email="alex@gmail.com") + async def test_save_sets_was_recently_created(self): + user = User(name="Alex", email="alex@gmail.com") await user.save() assert user._was_recently_created is True - async def test_save_populates_primary_key(self, UserModel, users_table): - user = UserModel(name="Alex", email="alex@gmail.com") + async def test_save_populates_primary_key(self): + user = User(name="Alex", email="alex@gmail.com") await user.save() assert user.id is not None assert isinstance(user.id, int) - async def test_save_with_datetime_field(self, UserModel, users_table): - user = UserModel( - name="Alex", - email="alex@gmail.com", - email_verified_at="2026-10-01 12:12:12", - ) - saved = await user.save() - - assert saved is True - assert user.email_verified_at.format("YYYY-MM-DD HH:mm:ss") == "2026-10-01 12:12:12" - # --------------------------------------------------------------------------- # UPDATE (Model.save on an existing record) # --------------------------------------------------------------------------- -class TestModelUpdate: - async def test_save_updates_dirty_attribute(self, UserModel, users_table): - user = UserModel(name="Alex", email="alex@gmail.com") +class TestModelUpdate(TestCase): + async def test_save_updates_dirty_attribute(self): + user = User(name="Alex", email="alex@gmail.com") await user.save() user.name = "Ram" - saved = await user.save() - - assert saved is True + assert await user.save() is True assert user.name == "Ram" - async def test_save_update_clears_dirty(self, UserModel, users_table): - user = UserModel(name="Alex", email="alex@gmail.com") + async def test_save_update_clears_dirty(self): + user = User(name="Alex", email="alex@gmail.com") await user.save() user.name = "Ram" @@ -142,17 +73,14 @@ async def test_save_update_clears_dirty(self, UserModel, users_table): assert not user.is_dirty() - async def test_save_noop_when_not_dirty(self, UserModel, users_table): - user = UserModel(name="Alex", email="alex@gmail.com") + async def test_save_noop_when_not_dirty(self): + user = User(name="Alex", email="alex@gmail.com") await user.save() - # Nothing changed — save should return True without hitting the DB - saved = await user.save() + assert await user.save() is True - assert saved is True - - async def test_save_update_increments_id_stays_same(self, UserModel, users_table): - user = UserModel(name="Alex", email="alex@gmail.com") + async def test_save_update_keeps_same_id(self): + user = User(name="Alex", email="alex@gmail.com") await user.save() original_id = user.id @@ -160,3 +88,37 @@ async def test_save_update_increments_id_stays_same(self, UserModel, users_table await user.save() assert user.id == original_id + + async def test_update_iterating_all_persists_changes(self): + await User.create({"name": "Alice", "email": "alice@test.com"}) + await User.create({"name": "Bob", "email": "bob@test.com"}) + + for user in await User.all(): + await user.update({"name": "Updated"}) + + assert all(u.name == "Updated" for u in await User.all()) + + +# --------------------------------------------------------------------------- +# Fillable guard +# --------------------------------------------------------------------------- + + +class TestFillable(TestCase): + async def test_annotated_fields_are_in_fillable(self): + class Post(Model): + __table__ = "posts" + title: str + body: str + + assert "title" in Post.__fillable__ + assert "body" in Post.__fillable__ + + async def test_framework_fields_excluded_from_fillable(self): + class Post(Model): + __table__ = "posts" + title: str + + assert "db_manager" not in Post.__fillable__ + assert "created_at" not in Post.__fillable__ + assert "updated_at" not in Post.__fillable__ diff --git a/fastapi_startkit/src/fastapi_startkit/masoniteorm.backup/tests/integrations/config/__init__.py b/fastapi_startkit/tests/masoniteorm/mysql/__init__.py similarity index 100% rename from fastapi_startkit/src/fastapi_startkit/masoniteorm.backup/tests/integrations/config/__init__.py rename to fastapi_startkit/tests/masoniteorm/mysql/__init__.py diff --git a/fastapi_startkit/tests/masoniteorm/mysql/builder/__init__.py b/fastapi_startkit/tests/masoniteorm/mysql/builder/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/fastapi_startkit/tests/masoniteorm/mysql/builder/test_mysql_query_builder.py b/fastapi_startkit/tests/masoniteorm/mysql/builder/test_mysql_query_builder.py new file mode 100644 index 00000000..619e8f70 --- /dev/null +++ b/fastapi_startkit/tests/masoniteorm/mysql/builder/test_mysql_query_builder.py @@ -0,0 +1,288 @@ +from unittest.mock import AsyncMock + +from fastapi_startkit.masoniteorm import Model + +from ..test_case import TestCase +from ..fixtures.db import DB +from ...fixtures.model import User + + +class Payment(Model): + __table__ = "payments" + __connection__ = "mysql" + + +class TestMySQLQueryBuilder(TestCase): + # --- SELECT / WHERE --- + + async def test_where(self): + sql = User.query().where("name", "Joe").to_sql() + self.assertEqual(sql, "SELECT * FROM `users` WHERE `users`.`name` = 'Joe'") + + async def test_where_with_operator(self): + sql = User.query().where("age", ">", "20").to_sql() + self.assertEqual(sql, "SELECT * FROM `users` WHERE `users`.`age` > '20'") + + async def test_where_lt(self): + sql = User.query().where("age", "<", "20").to_sql() + self.assertEqual(sql, "SELECT * FROM `users` WHERE `users`.`age` < '20'") + + async def test_where_lte(self): + sql = User.query().where("age", "<=", "20").to_sql() + self.assertEqual(sql, "SELECT * FROM `users` WHERE `users`.`age` <= '20'") + + async def test_where_gte(self): + sql = User.query().where("age", ">=", "20").to_sql() + self.assertEqual(sql, "SELECT * FROM `users` WHERE `users`.`age` >= '20'") + + async def test_where_ne(self): + sql = User.query().where("age", "!=", "20").to_sql() + self.assertEqual(sql, "SELECT * FROM `users` WHERE `users`.`age` != '20'") + + async def test_or_where(self): + sql = User.query().where("age", "20").or_where("age", "<", 20).to_sql() + self.assertEqual( + sql, + "SELECT * FROM `users` WHERE `users`.`age` = '20' OR `users`.`age` < '20'", + ) + + async def test_where_null(self): + sql = User.query().where_null("name").to_sql() + self.assertEqual(sql, "SELECT * FROM `users` WHERE `users`.`name` IS NULL") + + async def test_where_not_null(self): + sql = User.query().where_not_null("name").to_sql() + self.assertEqual(sql, "SELECT * FROM `users` WHERE `users`.`name` IS NOT NULL") + + async def test_where_in(self): + sql = User.query().where_in("id", [1, 2, 3]).to_sql() + self.assertEqual(sql, "SELECT * FROM `users` WHERE `users`.`id` IN ('1','2','3')") + + async def test_where_not_in(self): + sql = User.query().where_not_in("id", [1, 2, 3]).to_sql() + self.assertEqual(sql, "SELECT * FROM `users` WHERE `users`.`id` NOT IN ('1','2','3')") + + async def test_where_like(self): + sql = User.query().where("age", "like", "%name%").to_sql() + self.assertEqual(sql, "SELECT * FROM `users` WHERE `users`.`age` LIKE '%name%'") + + async def test_where_not_like(self): + sql = User.query().where("age", "not like", "%name%").to_sql() + self.assertEqual(sql, "SELECT * FROM `users` WHERE `users`.`age` NOT LIKE '%name%'") + + async def test_where_column(self): + sql = User.query().where_column("name", "username").to_sql() + self.assertEqual(sql, "SELECT * FROM `users` WHERE name = username") + + async def test_between(self): + sql = User.query().between("id", 2, 5).to_sql() + self.assertEqual(sql, "SELECT * FROM `users` WHERE `users`.`id` BETWEEN '2' AND '5'") + + async def test_not_between(self): + sql = User.query().not_between("id", 2, 5).to_sql() + self.assertEqual(sql, "SELECT * FROM `users` WHERE `users`.`id` NOT BETWEEN '2' AND '5'") + + async def test_when_true_applies_condition(self): + sql = User.query().when(True, lambda q: q.where("is_admin", 1)).to_sql() + self.assertEqual(sql, "SELECT * FROM `users` WHERE `users`.`is_admin` = '1'") + + async def test_when_false_skips_condition(self): + sql = User.query().when(False, lambda q: q.where("is_admin", 1)).to_sql() + self.assertEqual(sql, "SELECT * FROM `users`") + + # --- SELECT columns --- + + async def test_select_columns(self): + sql = User.query().select("name", "email").to_sql() + self.assertEqual(sql, "SELECT `users`.`name`, `users`.`email` FROM `users`") + + # --- LIMIT / OFFSET --- + + async def test_limit(self): + sql = User.query().limit(5).to_sql() + self.assertEqual(sql, "SELECT * FROM `users` LIMIT 5") + + async def test_offset(self): + sql = User.query().limit(10).offset(5).to_sql() + self.assertEqual(sql, "SELECT * FROM `users` LIMIT 10 OFFSET 5") + + # --- ORDER BY --- + + async def test_order_by_asc(self): + sql = User.query().order_by("email", "asc").to_sql() + self.assertEqual(sql, "SELECT * FROM `users` ORDER BY `email` ASC") + + async def test_order_by_desc(self): + sql = User.query().order_by("email", "desc").to_sql() + self.assertEqual(sql, "SELECT * FROM `users` ORDER BY `email` DESC") + + async def test_latest(self): + sql = User.query().latest("email").to_sql() + self.assertEqual(sql, "SELECT * FROM `users` ORDER BY `email` DESC") + + async def test_oldest(self): + sql = User.query().oldest("email").to_sql() + self.assertEqual(sql, "SELECT * FROM `users` ORDER BY `email` ASC") + + # --- JOINS --- + + async def test_join(self): + sql = User.query().join("profiles", "users.id", "=", "profiles.user_id").to_sql() + self.assertEqual( + sql, + "SELECT * FROM `users` INNER JOIN `profiles` ON `users`.`id` = `profiles`.`user_id`", + ) + + async def test_left_join(self): + sql = User.query().left_join("profiles", "users.id", "=", "profiles.user_id").to_sql() + self.assertEqual( + sql, + "SELECT * FROM `users` LEFT JOIN `profiles` ON `users`.`id` = `profiles`.`user_id`", + ) + + async def test_right_join(self): + sql = User.query().right_join("profiles", "users.id", "=", "profiles.user_id").to_sql() + self.assertEqual( + sql, + "SELECT * FROM `users` RIGHT JOIN `profiles` ON `users`.`id` = `profiles`.`user_id`", + ) + + # --- AGGREGATES (mock select_one to capture generated SQL) --- + + async def test_count(self): + mock_select_one = AsyncMock(return_value={"id": 5}) + DB.connection("mysql").select_one = mock_select_one + + await User.query().count("id") + + sql, bindings = mock_select_one.call_args[0] + self.assertEqual(sql, "SELECT COUNT(`users`.`id`) AS id FROM `users`") + self.assertEqual(list(bindings), []) + + async def test_sum(self): + mock_select_one = AsyncMock(return_value={"age": 100}) + DB.connection("mysql").select_one = mock_select_one + + await User.query().sum("age") + + sql, bindings = mock_select_one.call_args[0] + self.assertEqual(sql, "SELECT SUM(`users`.`age`) AS age FROM `users`") + self.assertEqual(list(bindings), []) + + async def test_max(self): + mock_select_one = AsyncMock(return_value={"age": 99}) + DB.connection("mysql").select_one = mock_select_one + + await User.query().max("age") + + sql, bindings = mock_select_one.call_args[0] + self.assertEqual(sql, "SELECT MAX(`users`.`age`) AS age FROM `users`") + self.assertEqual(list(bindings), []) + + async def test_min(self): + mock_select_one = AsyncMock(return_value={"age": 1}) + DB.connection("mysql").select_one = mock_select_one + + await User.query().min("age") + + sql, bindings = mock_select_one.call_args[0] + self.assertEqual(sql, "SELECT MIN(`users`.`age`) AS age FROM `users`") + self.assertEqual(list(bindings), []) + + async def test_avg(self): + mock_select_one = AsyncMock(return_value={"age": 30}) + DB.connection("mysql").select_one = mock_select_one + + await User.query().avg("age") + + sql, bindings = mock_select_one.call_args[0] + self.assertEqual(sql, "SELECT AVG(`users`.`age`) AS age FROM `users`") + self.assertEqual(list(bindings), []) + + # --- GROUP BY / HAVING --- + + async def test_group_by(self): + mock_select_one = AsyncMock(return_value={"salary": 500}) + DB.connection("mysql").select_one = mock_select_one + + await Payment.query().select("user_id").group_by("user_id").min("salary") + + sql, bindings = mock_select_one.call_args[0] + self.assertEqual( + sql, + "SELECT `payments`.`user_id`, MIN(`payments`.`salary`) AS salary FROM `payments` GROUP BY `payments`.`user_id`", + ) + self.assertEqual(list(bindings), []) + + async def test_having(self): + mock_select_one = AsyncMock(return_value={"salary": 2000}) + DB.connection("mysql").select_one = mock_select_one + + await Payment.query().select("user_id").group_by("user_id").having("salary", ">=", "1000").avg("salary") + + sql, bindings = mock_select_one.call_args[0] + self.assertEqual( + sql, + "SELECT `payments`.`user_id`, AVG(`payments`.`salary`) AS salary FROM `payments` GROUP BY `payments`.`user_id` HAVING `payments`.`salary` >= '1000'", + ) + self.assertEqual(list(bindings), []) + + # --- DML: INSERT / UPDATE / DELETE --- + + async def test_insert_sql(self): + mock_insert = AsyncMock(return_value=1) + DB.connection("mysql").insert = mock_insert + + await User.query().insert({"name": "Joe"}) + + mock_insert.assert_called_once() + sql, bindings = mock_insert.call_args[0] + self.assertEqual(sql, "INSERT INTO `users` (`name`) VALUES (?)") + self.assertEqual(list(bindings), ["Joe"]) + + async def test_bulk_insert_sql(self): + mock_insert = AsyncMock(return_value=3) + DB.connection("mysql").insert = mock_insert + + await User.query().insert([{"name": "Joe"}, {"name": "Bill"}, {"name": "John"}]) + + mock_insert.assert_called_once() + sql, bindings = mock_insert.call_args[0] + self.assertEqual(sql, "INSERT INTO `users` (`name`) VALUES (?), (?), (?)") + self.assertEqual(list(bindings), ["Joe", "Bill", "John"]) + + async def test_update_sql(self): + mock_update = AsyncMock(return_value=1) + DB.connection("mysql").update = mock_update + + await User.query().where("id", 1).update({"name": "Joe", "email": "joe@test.com"}) + + mock_update.assert_called_once() + sql, bindings = mock_update.call_args[0] + self.assertEqual( + sql, + "UPDATE `users` SET `users`.`name` = ?, `users`.`email` = ? WHERE `users`.`id` = ?", + ) + self.assertEqual(list(bindings), ["Joe", "joe@test.com", 1]) + + async def test_delete_by_where(self): + mock_delete = AsyncMock(return_value=1) + DB.connection("mysql").delete = mock_delete + + await User.query().where("id", 1).delete() + + mock_delete.assert_called_once() + sql, bindings = mock_delete.call_args[0] + self.assertEqual(sql, "DELETE FROM `users` WHERE `users`.`id` = ?") + self.assertEqual(list(bindings), [1]) + + async def test_delete_where_in(self): + mock_delete = AsyncMock(return_value=3) + DB.connection("mysql").delete = mock_delete + + await User.query().where_in("id", [1, 2, 3]).delete() + + mock_delete.assert_called_once() + sql, bindings = mock_delete.call_args[0] + self.assertEqual(sql, "DELETE FROM `users` WHERE `users`.`id` IN (?, ?, ?)") + self.assertEqual(list(bindings), [1, 2, 3]) diff --git a/fastapi_startkit/tests/masoniteorm/mysql/fixtures/__init__.py b/fastapi_startkit/tests/masoniteorm/mysql/fixtures/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/fastapi_startkit/tests/masoniteorm/mysql/fixtures/db.py b/fastapi_startkit/tests/masoniteorm/mysql/fixtures/db.py new file mode 100644 index 00000000..20c7ebc6 --- /dev/null +++ b/fastapi_startkit/tests/masoniteorm/mysql/fixtures/db.py @@ -0,0 +1,18 @@ +from fastapi_startkit.masoniteorm.connections.factory import ConnectionFactory +from fastapi_startkit.masoniteorm.connections.manager import DatabaseManager +from fastapi_startkit.masoniteorm.models.model import Model + +DB = DatabaseManager( + ConnectionFactory(), + { + "default": "mysql", + "connections": { + "mysql": { + "driver": "mysql", + "url": "mysql+aiomysql://root:@localhost:3306/test", + }, + }, + }, +) + +Model.db_manager = DB diff --git a/fastapi_startkit/tests/masoniteorm/mysql/test_case.py b/fastapi_startkit/tests/masoniteorm/mysql/test_case.py new file mode 100644 index 00000000..0bc47d20 --- /dev/null +++ b/fastapi_startkit/tests/masoniteorm/mysql/test_case.py @@ -0,0 +1,15 @@ +from unittest import IsolatedAsyncioTestCase + +from fastapi_startkit.masoniteorm import Model + +from .fixtures.db import DB + + +class TestCase(IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self._prev_db_manager = Model.db_manager + Model.db_manager = DB + + async def asyncTearDown(self): + await DB.clear() + Model.db_manager = self._prev_db_manager