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 48f72ac619383012828c5ad1cb4cf87d4269563c
parent b6d50752c716138bcccb02cef5de687de6d305d4
Author: Daniel Moch <daniel@danielmoch.com>
Date:   Tue, 21 Aug 2018 06:04:24 -0400

Refactor subprocess and error handling

Diffstat:
Mhookmeup/hookmeup.py | 62++++++++++++++++++++++++++++++++++----------------------------
Mtests/test_hookmeup.py | 152+++++++++++++++++++++++--------------------------------------------------------
2 files changed, 77 insertions(+), 137 deletions(-)

diff --git a/hookmeup/hookmeup.py b/hookmeup/hookmeup.py @@ -2,6 +2,7 @@ """hookmeup module.""" import os import subprocess +from subprocess import CalledProcessError class HookMeUpError(Exception): """Errors raised by hookmeup""" @@ -10,35 +11,48 @@ class HookMeUpError(Exception): def __str__(self): return "hookmeup: {}".format(self.args[0]) -def handle_completed_process(completed_process, msg="fatal error"): - """Handle return data from a call to subprocess.run""" - if completed_process.returncode != 0: +class DjangoMigrator(): + """ + Class responsible for parsing, applying, and unapplying Django + migrations + """ + def __init__(self): + pass + + def migrations_changed(self): + """ + Returns true if there are migrations that need to be applied + or unapplied + """ + pass + + def migrate(self): + """Apply/unapply any migrations as necessary""" + pass + +def call_checked_subprocess(arg_list, msg="fatal error"): + """Handle return data from a call to a subprocess""" + try: + return subprocess.check_output(arg_list).decode('utf-8') + except CalledProcessError: raise HookMeUpError(msg) def adjust_pipenv(): """Adjust pipenv to match Pipfile""" print('Adjusting virtualenv to match Pipfile') - completed_process = subprocess.run( + call_checked_subprocess( ['pipenv', 'clean'], - check=True - ) - handle_completed_process( - completed_process, 'Attempt to clean pipenv failed' ) - completed_process = subprocess.run( + call_checked_subprocess( ['pipenv', 'sync', '--dev'], - check=True - ) - handle_completed_process( - completed_process, 'Attempt to sync pipenv failed' ) def pipfile_changed(args): """Test if the Pipfile has changed""" - completed_process = subprocess.run( + stdout = call_checked_subprocess( ['git', 'diff', '--name-status', @@ -46,19 +60,17 @@ def pipfile_changed(args): args['new'], '--', 'Pipfile'], - check=True, - capture_output=True - ) - handle_completed_process( - completed_process, 'Not in a Git repository' ) - return completed_process.stdout.decode('utf-8').startswith('M') + return stdout.startswith('M') def post_checkout(args): """Run post-checkout hook""" + migrator = DjangoMigrator() if args['branch_checkout'] == 1: + if migrator.migrations_changed(): + migrator.migrate() if pipfile_changed(args): adjust_pipenv() @@ -69,19 +81,13 @@ def install(args): "Argument passed to 'install', but expected none" ) - completed_process = subprocess.run( + stdout = call_checked_subprocess( ['git', 'rev-parse', '--git-dir'], - check=True, - capture_output=True - ) - - handle_completed_process( - completed_process, 'Not in a Git repository' ) hook_path = os.path.join( - completed_process.stdout.decode('utf-8').strip(), + stdout.strip(), 'hooks', 'post-checkout' ) diff --git a/tests/test_hookmeup.py b/tests/test_hookmeup.py @@ -1,24 +1,20 @@ # -*- coding: utf-8 -*- -"""Tests for `hookmeup` package.""" +"""Tests for hookmeup package.""" import os import subprocess +from subprocess import CalledProcessError import pytest import hookmeup +from hookmeup.hookmeup import HookMeUpError @pytest.fixture def mock_install(mocker): """Mock low-level API's called by install""" - completed_process = subprocess.CompletedProcess( - args=['git', 'rev-parse', '--git-dir'], - returncode=0, - stdout=b'.git', - stderr=b'' - ) mocker.patch( - 'subprocess.run', - new=mocker.MagicMock(return_value=completed_process) + 'subprocess.check_output', + new=mocker.MagicMock(return_value=b'.git') ) def test_install(mock_install, mocker): @@ -50,22 +46,18 @@ def test_install_existing_hook(mock_install, mocker): def test_install_bad_arg(mocker): """Test install function when arg inappropriately provided""" - with pytest.raises(hookmeup.hookmeup.HookMeUpError): + with pytest.raises(HookMeUpError): hookmeup.hookmeup.install({'oops': 'don\t do this'}) def test_install_outside_repo(mocker): """Test install outside of Git repository""" - completed_process = subprocess.CompletedProcess( - args=['git', 'rev-parse', '--git-dir'], - returncode=128, - stdout=b'', - stderr=b'fatal: not a git repository' - ) mocker.patch( - 'subprocess.run', - new=mocker.MagicMock(return_value=completed_process) + 'subprocess.check_output', + new=mocker.Mock( + side_effect=CalledProcessError(returncode=1, cmd='cmd') + ) ) - with pytest.raises(hookmeup.hookmeup.HookMeUpError): + with pytest.raises(HookMeUpError): hookmeup.hookmeup.install({}) def test_install_already_installed(mock_install, mocker): @@ -85,27 +77,15 @@ def test_install_already_installed(mock_install, mocker): def test_error(): """Test accessing error members""" try: - raise hookmeup.hookmeup.HookMeUpError('test error') - except hookmeup.hookmeup.HookMeUpError as error: + raise HookMeUpError('test error') + except HookMeUpError as error: assert str(error) == 'hookmeup: test error' def test_post_checkout(mocker): """Test nominal post_checkout""" - completed_process = subprocess.CompletedProcess( - args=['git', - 'diff', - '--name-status', - 'HEAD^', - 'HEAD', - '--', - 'Pipfile'], - returncode=0, - stdout=b'M Pipfile\n', - stderr=b'' - ) mocker.patch( - 'subprocess.run', - new=mocker.MagicMock(return_value=completed_process) + 'subprocess.check_output', + new=mocker.MagicMock(return_value=b'M Pipfile\n') ) mocker.patch('hookmeup.hookmeup.adjust_pipenv') hookmeup.hookmeup.post_checkout({ @@ -113,26 +93,14 @@ def test_post_checkout(mocker): 'old': 'HEAD^', 'new': 'HEAD' }) - subprocess.run.assert_called_once() + subprocess.check_output.assert_called_once() hookmeup.hookmeup.adjust_pipenv.assert_called_once() def test_post_checkout_no_changes(mocker): """Test nominal post_checkout""" - completed_process = subprocess.CompletedProcess( - args=['git', - 'diff', - '--name-status', - 'HEAD^', - 'HEAD', - '--', - 'Pipfile'], - returncode=0, - stdout=b'\n', - stderr=b'' - ) mocker.patch( - 'subprocess.run', - new=mocker.MagicMock(return_value=completed_process) + 'subprocess.check_output', + new=mocker.MagicMock(return_value=b'\n') ) mocker.patch('hookmeup.hookmeup.adjust_pipenv') hookmeup.hookmeup.post_checkout({ @@ -140,105 +108,71 @@ def test_post_checkout_no_changes(mocker): 'old': 'HEAD^', 'new': 'HEAD' }) - subprocess.run.assert_called_once() + subprocess.check_output.assert_called_once() assert hookmeup.hookmeup.adjust_pipenv.call_count == 0 def test_adjust_pipenv(mocker): """Test call to adjust_pipenv""" - completed_process = subprocess.CompletedProcess( - args=['pipenv', 'clean'], - returncode=0, - stdout=b'.git', - stderr=b'' - ) mocker.patch( - 'subprocess.run', - new=mocker.MagicMock(return_value=completed_process) + 'subprocess.check_output', + new=mocker.MagicMock(return_value=b'.git\n') ) hookmeup.hookmeup.adjust_pipenv() - assert subprocess.run.call_count == 2 + assert subprocess.check_output.call_count == 2 def test_adjust_pipenv_failure(mocker): """Test adjust_pipenv with failed subprocess call""" - completed_process = subprocess.CompletedProcess( - args=['pipenv', 'clean'], - returncode=1, - stdout=b'.git', - stderr=b'' - ) mocker.patch( - 'subprocess.run', - new=mocker.MagicMock(return_value=completed_process) + 'subprocess.check_output', + new=mocker.Mock( + side_effect=CalledProcessError(returncode=1, cmd='cmd') + ) ) - with pytest.raises(hookmeup.hookmeup.HookMeUpError): + with pytest.raises(HookMeUpError): hookmeup.hookmeup.adjust_pipenv() def test_migrate_up(mocker): """Test a nominal Django migration""" - completed_process = subprocess.CompletedProcess( - args=['git', 'diff', '--name-status'], - returncode=0, - stdout=b'\ + 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', - stderr=b'' - ) - mocker.patch( - 'subprocess.run', - new=mocker.MagicMock(return_value=completed_process) + A other_file.py\n') ) def test_migrate_down(mocker): """Test a nominal Django migration downgrade""" - completed_process = subprocess.CompletedProcess( - args=['git', 'diff', '--name-status'], - returncode=0, - stdout=b'\ + 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', - stderr=b'' - ) - mocker.patch( - 'subprocess.run', - new=mocker.MagicMock(return_value=completed_process) + A other_file.py\n') ) def test_squashed_migrate_up(mocker): """Test a Django migration upgrade with an intervening squash""" - completed_process = subprocess.CompletedProcess( - args=['git', 'diff', '--name-status'], - returncode=0, - stdout=b'\ + 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', - stderr=b'' - ) - mocker.patch( - 'subprocess.run', - new=mocker.MagicMock(return_value=completed_process) + A other_file.py\n') ) def test_squashed_migrate_down(mocker): """Test a Django migration downgrade with an intervening squash""" - completed_process = subprocess.CompletedProcess( - args=['git', 'diff', '--name-status'], - returncode=0, - stdout=b'\ + 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', - stderr=b'' - ) - mocker.patch( - 'subprocess.run', - new=mocker.MagicMock(return_value=completed_process) + A other_file.py\n') )