Add test coverage for masoniteorm make and seed console commands

fastapi_startkit/tests/masoniteorm/commands/test_db_seed_command.py

@@ -0,0 +1,97 @@
+import unittest
+from unittest import mock
+
+from cleo.testers.command_tester import CommandTester
+
+from fastapi_startkit.masoniteorm.commands.DBSeedCommand import DBSeedCommand
+
+
+class FakeSeeder:
+    """Records constructor args and awaited methods, mocking DB side effects."""
+
+    instances = []
+
+    def __init__(self, seed_path="databases/seeds", connection=None):
+        self.seed_path = seed_path
+        self.connection = connection
+        self.calls = []
+        FakeSeeder.instances.append(self)
+
+    async def run_database_seed(self):
+        self.calls.append(("run_database_seed", None))
+
+    async def run_specific_seed(self, seed):
+        self.calls.append(("run_specific_seed", seed))
+
+
+class TestDBSeedCommand(unittest.TestCase):
+    def setUp(self):
+        FakeSeeder.instances = []
+        patcher = mock.patch(
+            "fastapi_startkit.masoniteorm.seeds.Seeder",
+            FakeSeeder,
+        )
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+    def _run(self, args=""):
+        tester = CommandTester(DBSeedCommand())
+        tester.execute(args)
+        return tester.io.fetch_output()
+
+    def test_runs_database_seeder_by_default(self):
+        output = self._run("")
+
+        self.assertIn("Database Seeder seeded!", output)
+        seeder = FakeSeeder.instances[-1]
+        self.assertEqual(seeder.calls, [("run_database_seed", None)])
+        self.assertEqual(seeder.seed_path, "databases/seeds")
+        self.assertEqual(seeder.connection, "default")
+
+    def test_seeds_specific_table_from_argument(self):
+        output = self._run("posts")
+
+        self.assertIn("PostsTableSeeder seeded!", output)
+        seeder = FakeSeeder.instances[-1]
+        self.assertEqual(
+            seeder.calls,
+            [("run_specific_seed", "posts_table_seeder.PostsTableSeeder")],
+        )
+
+    def test_class_option_resolves_plain_class_name(self):
+        output = self._run("--class PostSeeder")
+
+        self.assertIn("PostSeeder seeded!", output)
+        seeder = FakeSeeder.instances[-1]
+        self.assertEqual(
+            seeder.calls,
+            [("run_specific_seed", "post_seeder.PostSeeder")],
+        )
+
+    def test_class_option_resolves_table_seeder_suffix(self):
+        output = self._run("--class PostTableSeeder")
+
+        self.assertIn("PostTableSeeder seeded!", output)
+        seeder = FakeSeeder.instances[-1]
+        self.assertEqual(
+            seeder.calls,
+            [("run_specific_seed", "post_table_seeder.PostTableSeeder")],
+        )
+
+    def test_class_option_accepts_dotted_path(self):
+        output = self._run("--class custom.MySeeder")
+
+        self.assertIn("MySeeder seeded!", output)
+        seeder = FakeSeeder.instances[-1]
+        self.assertEqual(seeder.calls, [("run_specific_seed", "custom.MySeeder")])
+
+    def test_connection_and_directory_options_are_forwarded(self):
+        self._run("--connection sqlite --directory db/seeds")
+
+        seeder = FakeSeeder.instances[-1]
+        self.assertEqual(seeder.seed_path, "db/seeds")
+        self.assertEqual(seeder.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())