Add test coverage for masoniteorm make and seed console commands

fastapi_startkit/tests/masoniteorm/commands/fixtures/databases/seeds/database_seeder.py

@@ -0,0 +1,8 @@
+from fastapi_startkit.masoniteorm.seeds import Seeder
+
+from .recorder import CALLS
+
+
+class DatabaseSeeder(Seeder):
+    async def run(self):
+        CALLS.append(("database", self.connection))

fastapi_startkit/tests/masoniteorm/commands/fixtures/databases/seeds/recorder.py

@@ -0,0 +1,3 @@
+"""Shared call recorder used by the fixture seeder classes below."""
+
+CALLS = []

fastapi_startkit/tests/masoniteorm/commands/fixtures/databases/seeds/sample_seeder.py

@@ -0,0 +1,8 @@
+from fastapi_startkit.masoniteorm.seeds import Seeder
+
+from .recorder import CALLS
+
+
+class SampleSeeder(Seeder):
+    async def run(self):
+        CALLS.append(("sample", self.connection))

fastapi_startkit/tests/masoniteorm/commands/fixtures/databases/seeds/special_seeder.py

@@ -0,0 +1,8 @@
+from fastapi_startkit.masoniteorm.seeds import Seeder
+
+from .recorder import CALLS
+
+
+class SpecialSeeder(Seeder):
+    async def run(self):
+        CALLS.append(("special", self.connection))

fastapi_startkit/tests/masoniteorm/commands/fixtures/databases/seeds/user_table_seeder.py

@@ -0,0 +1,8 @@
+from fastapi_startkit.masoniteorm.seeds import Seeder
+
+from .recorder import CALLS
+
+
+class UserTableSeeder(Seeder):
+    async def run(self):
+        CALLS.append(("user_table", self.connection))

fastapi_startkit/tests/masoniteorm/commands/test_db_seed_command.py

@@ -0,0 +1,139 @@
+import io
+import unittest
+from contextlib import redirect_stdout
+from unittest import mock
+
+from cleo.testers.command_tester import CommandTester
+
+from fastapi_startkit.masoniteorm.commands.DBSeedCommand import DBSeedCommand
+
+from .fixtures.databases.seeds.recorder import CALLS
+
+FIXTURE_SEED_PATH = "tests.masoniteorm.commands.fixtures.databases.seeds"
+
+
+class TestDBSeedCommand(unittest.TestCase):
+    def setUp(self):
+        from .fixtures.app import create_app
+
+        self.app = create_app()
+        CALLS.clear()
+
+    def tearDown(self):
+        CALLS.clear()
+
+    def _run(self, args=""):
+        tester = CommandTester(DBSeedCommand())
+        tester.execute(args)
+        return tester.io.fetch_output()
+
+    def _run_app(self, args=""):
+        buffer = io.StringIO()
+        with redirect_stdout(buffer):
+            self.app.run("db:seed", args)
+        return buffer.getvalue()
+
+    # -- option/argument resolution, exercised against a mocked Seeder --
+
+    def test_runs_database_seeder_by_default(self):
+        with mock.patch("fastapi_startkit.masoniteorm.seeds.Seeder", autospec=True) as MockSeeder:
+            output = self._run("")
+
+        self.assertIn("Database Seeder seeded!", output)
+        MockSeeder.assert_called_once_with(seed_path="databases/seeds", connection="default")
+        MockSeeder.return_value.run_database_seed.assert_awaited_once_with()
+        MockSeeder.return_value.run_specific_seed.assert_not_awaited()
+
+    def test_seeds_specific_table_from_argument(self):
+        with mock.patch("fastapi_startkit.masoniteorm.seeds.Seeder", autospec=True) as MockSeeder:
+            output = self._run("posts")
+
+        self.assertIn("PostsTableSeeder seeded!", output)
+        MockSeeder.return_value.run_specific_seed.assert_awaited_once_with("posts_table_seeder.PostsTableSeeder")
+
+    def test_class_option_resolves_plain_class_name(self):
+        with mock.patch("fastapi_startkit.masoniteorm.seeds.Seeder", autospec=True) as MockSeeder:
+            output = self._run("--class PostSeeder")
+
+        self.assertIn("PostSeeder seeded!", output)
+        MockSeeder.return_value.run_specific_seed.assert_awaited_once_with("post_seeder.PostSeeder")
+
+    def test_class_option_resolves_table_seeder_suffix(self):
+        with mock.patch("fastapi_startkit.masoniteorm.seeds.Seeder", autospec=True) as MockSeeder:
+            output = self._run("--class PostTableSeeder")
+
+        self.assertIn("PostTableSeeder seeded!", output)
+        MockSeeder.return_value.run_specific_seed.assert_awaited_once_with("post_table_seeder.PostTableSeeder")
+
+    def test_class_option_accepts_dotted_path(self):
+        with mock.patch("fastapi_startkit.masoniteorm.seeds.Seeder", autospec=True) as MockSeeder:
+            output = self._run("--class custom.MySeeder")
+
+        self.assertIn("MySeeder seeded!", output)
+        MockSeeder.return_value.run_specific_seed.assert_awaited_once_with("custom.MySeeder")
+
+    def test_connection_and_directory_options_are_forwarded(self):
+        with mock.patch("fastapi_startkit.masoniteorm.seeds.Seeder", autospec=True) as MockSeeder:
+            self._run("--connection sqlite --directory db/seeds")
+
+        MockSeeder.assert_called_once_with(seed_path="db/seeds", connection="sqlite")
+
+    # -- end-to-end, driven through the registered console app against real fixture seeders --
+
+    def test_runs_database_seeder_by_default_via_app(self):
+        output = self._run_app(f"--directory {FIXTURE_SEED_PATH} --connection sqlite")
+
+        self.assertIn("Database Seeder seeded!", output)
+        self.assertEqual(CALLS, [("database", "sqlite")])
+
+    def test_runs_seeder_for_table_argument_via_app(self):
+        output = self._run_app(f"user --directory {FIXTURE_SEED_PATH} --connection sqlite")
+
+        self.assertIn("UserTableSeeder seeded!", output)
+        self.assertEqual(CALLS, [("user_table", "sqlite")])
+
+    def test_runs_seeder_for_class_option_without_table_suffix_via_app(self):
+        output = self._run_app(f"--directory {FIXTURE_SEED_PATH} --class SampleSeeder --connection sqlite")
+
+        self.assertIn("SampleSeeder seeded!", output)
+        self.assertEqual(CALLS, [("sample", "sqlite")])
+
+    def test_runs_seeder_for_class_option_with_table_suffix_via_app(self):
+        output = self._run_app(f"--directory {FIXTURE_SEED_PATH} --class UserTableSeeder --connection sqlite")
+
+        self.assertIn("UserTableSeeder seeded!", output)
+        self.assertEqual(CALLS, [("user_table", "sqlite")])
+
+    def test_runs_seeder_for_fully_qualified_class_option_via_app(self):
+        output = self._run_app(
+            f"--directory {FIXTURE_SEED_PATH} --class special_seeder.SpecialSeeder --connection sqlite"
+        )
+
+        self.assertIn("SpecialSeeder seeded!", output)
+        self.assertEqual(CALLS, [("special", "sqlite")])
+
+    def test_class_option_takes_precedence_over_table_argument_via_app(self):
+        output = self._run_app(f"user --directory {FIXTURE_SEED_PATH} --class SampleSeeder --connection sqlite")
+
+        self.assertIn("SampleSeeder seeded!", output)
+        self.assertEqual(CALLS, [("sample", "sqlite")])
+
+    def test_uses_default_connection_option_via_app(self):
+        self._run_app(f"--directory {FIXTURE_SEED_PATH}")
+
+        self.assertEqual(CALLS, [("database", "default")])
+
+    # -- error paths: driven directly through CommandTester, since the console
+    # application catches command exceptions instead of propagating them --
+
+    def test_raises_when_seeder_class_cannot_be_found(self):
+        with self.assertRaises(ValueError):
+            self._run(f"--directory {FIXTURE_SEED_PATH} --class does_not_exist.NopeSeeder --connection sqlite")
+
+    def test_raises_when_database_seeder_missing_from_directory(self):
+        with self.assertRaises(ValueError):
+            self._run("--directory tests.masoniteorm.commands.fixtures.databases.migrations --connection sqlite")
+
+
+if __name__ == "__main__":
+    unittest.main()

fastapi_startkit/tests/masoniteorm/commands/test_make_commands.py

@@ -0,0 +1,170 @@
+import os
+import shutil
+import tempfile
+import unittest
+
+from cleo.testers.command_tester import CommandTester
+
+from fastapi_startkit.masoniteorm.commands.MakeMigrationCommand import MakeMigrationCommand
+from fastapi_startkit.masoniteorm.commands.MakeModelCommand import MakeModelCommand
+from fastapi_startkit.masoniteorm.commands.MakeObserverCommand import MakeObserverCommand
+from fastapi_startkit.masoniteorm.commands.MakeSeedCommand import MakeSeedCommand
+
+
+class _TempCwdTestCase(unittest.TestCase):
+    """Base case that runs each test inside an isolated temp working directory.
+
+    The make/generator commands write files relative to ``os.getcwd()``, so the
+    working directory is swapped for a throwaway one to keep the repo clean.
+    """
+
+    def setUp(self):
+        self._original_cwd = os.getcwd()
+        self._tmp_dir = tempfile.mkdtemp()
+        os.chdir(self._tmp_dir)
+
+    def tearDown(self):
+        os.chdir(self._original_cwd)
+        shutil.rmtree(self._tmp_dir, ignore_errors=True)
+
+    def _tester(self, command_class):
+        return CommandTester(command_class())
+
+    def _read_single_file(self, directory):
+        files = [f for f in os.listdir(directory) if f.endswith(".py")]
+        self.assertEqual(len(files), 1, f"expected one generated file in {directory}, got {files}")
+        with open(os.path.join(directory, files[0])) as fp:
+            return files[0], fp.read()
+
+
+class TestMakeMigrationCommand(_TempCwdTestCase):
+    def test_creates_migration_with_inferred_table(self):
+        os.makedirs("databases/migrations")
+        tester = self._tester(MakeMigrationCommand)
+        tester.execute("create_posts_table")
+
+        output = tester.io.fetch_output()
+        self.assertIn("Migration file created:", output)
+
+        file_name, content = self._read_single_file("databases/migrations")
+        self.assertTrue(file_name.endswith("_create_posts_table.py"))
+        self.assertIn("class CreatePostsTable(Migration)", content)
+        self.assertIn('self.schema.create("posts")', content)
+
+    def test_create_option_uses_create_stub(self):
+        os.makedirs("databases/migrations")
+        tester = self._tester(MakeMigrationCommand)
+        tester.execute("add_users --create users")
+
+        _, content = self._read_single_file("databases/migrations")
+        self.assertIn('self.schema.create("users")', content)
+
+    def test_table_option_uses_table_stub(self):
+        os.makedirs("databases/migrations")
+        tester = self._tester(MakeMigrationCommand)
+        tester.execute("modify_users --table users")
+
+        _, content = self._read_single_file("databases/migrations")
+        self.assertIn('self.schema.table("users")', content)
+        self.assertNotIn("self.schema.create", content)
+
+    def test_custom_directory_option(self):
+        os.makedirs("custom/migrations")
+        tester = self._tester(MakeMigrationCommand)
+        tester.execute("create_posts_table --directory custom/migrations")
+
+        output = tester.io.fetch_output()
+        self.assertIn("custom/migrations", output)
+        self.assertTrue(any(f.endswith(".py") for f in os.listdir("custom/migrations")))
+
+
+class TestMakeModelCommand(_TempCwdTestCase):
+    def test_creates_model_file(self):
+        os.makedirs("app")
+        tester = self._tester(MakeModelCommand)
+        tester.execute("Post")
+
+        output = tester.io.fetch_output()
+        self.assertIn("Model created:", output)
+
+        content = open(os.path.join("app", "Post.py")).read()
+        self.assertIn("class Post(Model)", content)
+
+    def test_pep_option_uses_underscore_filename(self):
+        os.makedirs("app")
+        tester = self._tester(MakeModelCommand)
+        tester.execute("BlogPost --pep")
+
+        self.assertTrue(os.path.exists(os.path.join("app", "blog_post.py")))
+
+    def test_reports_when_model_already_exists(self):
+        os.makedirs("app")
+        self._tester(MakeModelCommand).execute("Post")
+
+        tester = self._tester(MakeModelCommand)
+        tester.execute("Post")
+        output = tester.io.fetch_output()
+        self.assertIn('Model "Post" Already Exists', output)
+
+    def test_custom_directory_option(self):
+        os.makedirs("models")
+        tester = self._tester(MakeModelCommand)
+        tester.execute("Post --directory models")
+
+        self.assertTrue(os.path.exists(os.path.join("models", "Post.py")))
+
+
+class TestMakeSeedCommand(_TempCwdTestCase):
+    def test_creates_seed_file(self):
+        os.makedirs("databases/seeds")
+        tester = self._tester(MakeSeedCommand)
+        tester.execute("Post")
+
+        output = tester.io.fetch_output()
+        self.assertIn("Seed file created:", output)
+
+        content = open(os.path.join("databases/seeds", "post_table_seeder.py")).read()
+        self.assertIn("class PostTableSeeder(Seeder)", content)
+
+    def test_reports_when_seed_already_exists(self):
+        os.makedirs("databases/seeds")
+        self._tester(MakeSeedCommand).execute("Post")
+
+        tester = self._tester(MakeSeedCommand)
+        tester.execute("Post")
+        output = tester.io.fetch_output()
+        self.assertIn("already exists.", output)
+
+
+class TestMakeObserverCommand(_TempCwdTestCase):
+    def test_creates_observer_file(self):
+        tester = self._tester(MakeObserverCommand)
+        tester.execute("Post")
+
+        output = tester.io.fetch_output()
+        self.assertIn("Observer created:", output)
+
+        content = open(os.path.join("app/observers", "PostObserver.py")).read()
+        self.assertIn("class PostObserver:", content)
+        self.assertIn("def created(self, post):", content)
+
+    def test_model_option_controls_variable_and_type(self):
+        tester = self._tester(MakeObserverCommand)
+        tester.execute("Audit --model User")
+
+        content = open(os.path.join("app/observers", "AuditObserver.py")).read()
+        self.assertIn("class AuditObserver:", content)
+        self.assertIn("def created(self, user):", content)
+        self.assertIn("User model.", content)
+
+    def test_reports_when_observer_already_exists(self):
+        self._tester(MakeObserverCommand).execute("Post")
+
+        tester = self._tester(MakeObserverCommand)
+        tester.execute("Post")
+        output = tester.io.fetch_output()
+        self.assertIn('Observer "Post" Already Exists', output)
+
+
+if __name__ == "__main__":
+    unittest.main()

fastapi_startkit/tests/masoniteorm/commands/test_migrate_commands.py

@@ -5,6 +5,7 @@
 
 from fastapi_startkit.masoniteorm.commands.DBMigrateCommand import DBMigrateCommand
 from fastapi_startkit.masoniteorm.commands.MigrateFreshCommand import MigrateFreshCommand
+from fastapi_startkit.masoniteorm.commands.MigrateRefreshCommand import MigrateRefreshCommand
 from fastapi_startkit.masoniteorm.commands.MigrateResetCommand import MigrateResetCommand
 from fastapi_startkit.masoniteorm.commands.MigrateRollbackCommand import MigrateRollbackCommand
 from fastapi_startkit.masoniteorm.commands.MigrateStatusCommand import MigrateStatusCommand
@@ -103,6 +104,17 @@ def test_reset_rolls_back_all_migrations(self):
         self.assertIn("Rolled back:", output)
         self.assertIn("create_posts_table", output)
 
+    def test_refresh_resets_and_remigrates(self):
+        asyncio.run(self._migrate())
+
+        cmd = self._make_command(MigrateRefreshCommand)
+        tester = CommandTester(cmd)
+        tester.execute("--connection sqlite")
+        output = tester.io.fetch_output()
+        self.assertIn("Rolled back:", output)
+        self.assertIn("Migrated:", output)
+        self.assertIn("create_posts_table", output)
+
     def test_fresh_drops_all_tables_and_remigrates(self):
         asyncio.run(self._migrate())