hookmeup

A Git hook to automate your Pipenv and Django workflow
git clone git://git.danielmoch.com/hookmeup.git
Log | Files | Refs | README | LICENSE

commit eab0da457ce4fda1e60af1fb470d626411c97a7c
parent 48f72ac619383012828c5ad1cb4cf87d4269563c
Author: Daniel Moch <daniel@danielmoch.com>
Date:   Tue, 21 Aug 2018 21:44:46 -0400

DjangoMigrator implementation complete

Ready for initial testing

Diffstat:
Mhookmeup/hookmeup.py | 53++++++++++++++++++++++++++++++++++++++++++++++++-----
Mtests/pylintrc | 4+++-
Mtests/test_hookmeup.py | 86+++++++++++++++++++++++++++++++++++++++++++++++++------------------------------
3 files changed, 105 insertions(+), 38 deletions(-)

diff --git a/hookmeup/hookmeup.py b/hookmeup/hookmeup.py @@ -16,19 +16,62 @@ class DjangoMigrator(): Class responsible for parsing, applying, and unapplying Django migrations """ - def __init__(self): - pass + def __init__(self, args): + self.added_migration_apps = [] + self.oldest_deleted = {} + self._migrate_command = ['pipenv', + 'run', + 'python', + 'manage.py', + 'migrate'] + deleted_migrations = {} + stdout = call_checked_subprocess( + ['git', 'diff', '--name-status', args['old'], args['new']] + ) + diff_lines = stdout.splitlines() + for line in diff_lines: + if line.find(os.path.sep + 'migrations' + os.path.sep) >= 0: + file_status = line[0] + file_path = line[1:-1].strip() + file_path_segments = file_path.split(os.path.sep) + migration_name = file_path_segments[-1].replace('.py', '') + app_name = file_path_segments[-3] + if file_status in ['D', 'M']: + if app_name not in deleted_migrations: + deleted_migrations[app_name] = [] + deleted_migrations[app_name].append(migration_name) + if file_status == 'A' \ + and app_name not in self.added_migration_apps: + self.added_migration_apps.append(app_name) + for app_name, migrations_list in deleted_migrations.items(): + migrations_list.sort() + self.oldest_deleted[app_name] = \ + int(migrations_list[0].split('_')[0]) def migrations_changed(self): """ Returns true if there are migrations that need to be applied or unapplied """ - pass + return self.added_migration_apps != [] or \ + self.oldest_deleted != {} def migrate(self): """Apply/unapply any migrations as necessary""" - pass + for app, oldest in self.oldest_deleted.items(): + target_migration = format(oldest - 1, '04d') + if target_migration == '0000': + target_migration = 'zero' + call_checked_subprocess( + self._migrate_command + [app, target_migration], + 'rollback migration for {} failed'.format(app) + ) + + if self.added_migration_apps != []: + call_checked_subprocess( + self._migrate_command + self.added_migration_apps, + 'migration failed' + ) def call_checked_subprocess(arg_list, msg="fatal error"): """Handle return data from a call to a subprocess""" @@ -67,7 +110,7 @@ def pipfile_changed(args): def post_checkout(args): """Run post-checkout hook""" - migrator = DjangoMigrator() + migrator = DjangoMigrator(args) if args['branch_checkout'] == 1: if migrator.migrations_changed(): migrator.migrate() diff --git a/tests/pylintrc b/tests/pylintrc @@ -239,7 +239,9 @@ generated-members=REQUEST, assert_called_once_with, assert_called_once, call_count, - print + print, + call_args_list, + assert_any_call # Tells whether missing members accessed in mixin class should be ignored. A # mixin class is detected if its name ends with "mixin" (case insensitive). diff --git a/tests/test_hookmeup.py b/tests/test_hookmeup.py @@ -7,8 +7,9 @@ import pytest import hookmeup -from hookmeup.hookmeup import HookMeUpError +from hookmeup.hookmeup import HookMeUpError, DjangoMigrator +# pylint: disable=protected-access @pytest.fixture def mock_install(mocker): """Mock low-level API's called by install""" @@ -85,7 +86,9 @@ def test_post_checkout(mocker): """Test nominal post_checkout""" mocker.patch( 'subprocess.check_output', - new=mocker.MagicMock(return_value=b'M Pipfile\n') + new=mocker.MagicMock(return_value=b'M \ + Pipfile\nA \ + app1/migrations/0003_test.py') ) mocker.patch('hookmeup.hookmeup.adjust_pipenv') hookmeup.hookmeup.post_checkout({ @@ -93,7 +96,7 @@ def test_post_checkout(mocker): 'old': 'HEAD^', 'new': 'HEAD' }) - subprocess.check_output.assert_called_once() + assert subprocess.check_output.call_count == 3 hookmeup.hookmeup.adjust_pipenv.assert_called_once() def test_post_checkout_no_changes(mocker): @@ -108,7 +111,7 @@ def test_post_checkout_no_changes(mocker): 'old': 'HEAD^', 'new': 'HEAD' }) - subprocess.check_output.assert_called_once() + assert subprocess.check_output.call_count == 2 assert hookmeup.hookmeup.adjust_pipenv.call_count == 0 def test_adjust_pipenv(mocker): @@ -135,44 +138,63 @@ def test_migrate_up(mocker): """Test a nominal Django migration""" mocker.patch( 'subprocess.check_output', - new=mocker.MagicMock(return_value=b'\ - A app1/migrations/0002_auto.py\n\ - A app2/migrations/0003_test.py\n\ - A other_file.py\n') + new=mocker.MagicMock(return_value=b'A\ + app1/migrations/0002_auto.py\nA\ + app2/migrations/0003_test.py\nA\ + other_file.py\n') + ) + migrator = DjangoMigrator({'old': 'test', 'new': 'test2'}) + assert migrator.migrations_changed() is True + subprocess.check_output.assert_called_once() + mocker.resetall() + migrator.migrate() + subprocess.check_output.assert_called_once_with( + migrator._migrate_command + ['app1', 'app2'] ) def test_migrate_down(mocker): """Test a nominal Django migration downgrade""" mocker.patch( 'subprocess.check_output', - new=mocker.MagicMock(return_value=b'\ - D app1/migrations/0002_auto.py\n\ - D app2/migrations/0003_test.py\n\ - A other_file.py\n') + new=mocker.MagicMock(return_value=b'D \ + app1/migrations/0002_auto.py\nD \ + app2/migrations/0003_test.py\nA \ + other_file.py\n') + ) + migrator = DjangoMigrator({'old': 'test', 'new': 'test2'}) + assert migrator.migrations_changed() is True + subprocess.check_output.assert_called_once() + mocker.resetall() + migrator.migrate() + assert subprocess.check_output.call_count == 2 + subprocess.check_output.assert_any_call( + migrator._migrate_command + ['app1', '0001'] + ) + subprocess.check_output.assert_any_call( + migrator._migrate_command + ['app2', '0002'] ) -def test_squashed_migrate_up(mocker): +def test__migrate_to_zero(mocker): """Test a Django migration upgrade with an intervening squash""" mocker.patch( 'subprocess.check_output', - new=mocker.MagicMock(return_value=b'\ - A app1/migrations/0002_auto.py\n\ - A app2/migrations/0003_test.py\n\ - D app3/migrations/0001_initial.py\n\ - D app3/migrations/0002_auto.py\n\ - A app3/migrations/0001_squashed.py\n\ - A other_file.py\n') + new=mocker.MagicMock(return_value=b'A \ + app1/migrations/0002_auto.py\nA \ + app2/migrations/0003_test.py\nD \ + app3/migrations/0001_initial.py\nD \ + app3/migrations/0002_auto.py\nA \ + app3/migrations/0001_squashed.py\nA \ + other_file.py\n') ) - -def test_squashed_migrate_down(mocker): - """Test a Django migration downgrade with an intervening squash""" - mocker.patch( - 'subprocess.check_output', - new=mocker.MagicMock(return_value=b'\ - A app1/migrations/0002_auto.py\n\ - A app2/migrations/0003_test.py\n\ - A app3/migrations/0001_initial.py\n\ - A app3/migrations/0002_auto.py\n\ - D app3/migrations/0001_squashed.py\n\ - A other_file.py\n') + migrator = DjangoMigrator({'old': 'test', 'new': 'test2'}) + assert migrator.migrations_changed() is True + subprocess.check_output.assert_called_once() + mocker.resetall() + migrator.migrate() + assert subprocess.check_output.call_count == 2 + subprocess.check_output.assert_any_call( + migrator._migrate_command + ['app3', 'zero'] + ) + subprocess.check_output.assert_any_call( + migrator._migrate_command + ['app1', 'app2', 'app3'] )