diff --git a/fastapi_startkit/tests/masoniteorm/commands/test_db_seed_command.py b/fastapi_startkit/tests/masoniteorm/commands/test_db_seed_command.py new file mode 100644 index 00000000..ebc40860 --- /dev/null +++ b/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() diff --git a/fastapi_startkit/tests/masoniteorm/commands/test_make_commands.py b/fastapi_startkit/tests/masoniteorm/commands/test_make_commands.py new file mode 100644 index 00000000..6f36fe88 --- /dev/null +++ b/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() diff --git a/fastapi_startkit/tests/masoniteorm/commands/test_migrate_commands.py b/fastapi_startkit/tests/masoniteorm/commands/test_migrate_commands.py index 64f7614e..3aca1a07 100644 --- a/fastapi_startkit/tests/masoniteorm/commands/test_migrate_commands.py +++ b/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())