Move CLI to its own module; improve test coverage

drover/__init__.py

@@ -1,5 +1,4 @@
 """drover: a command-line utility to deploy Python packages to Lambda functions"""
-import argparse
 import logging
 import os
 import re
@@ -15,13 +14,12 @@
 import botocore
 import boto3
 import tqdm
-import yaml
-from pydantic import BaseModel, ValidationError
+from pydantic import BaseModel
 
 from drover.io import ArchiveMapping, format_file_size, get_digest, get_relative_file_names, write_archive
 from drover.models import S3BucketFileVersion, S3BucketPath, Settings, Stage
 
-__version__ = '0.7.1.dev2'
+__version__ = '0.7.1'
 _logger = logging.getLogger(__name__)
 
 
@@ -309,57 +307,3 @@ def _get_runtime_library_path(runtime: str) -> Path:
         if python_pattern.match(runtime):
             return Path('python')
         raise NotImplementedError(f'Unsupported runtime: {runtime}')
-
-
-def main():
-    """The main command-line entry point for the Drover interface"""
-    parser = argparse.ArgumentParser(description=__doc__.partition('\n')[0])
-    parser.add_argument('--version', '-V', action='version', version=f'%(prog)s {__version__}')
-    group = parser.add_mutually_exclusive_group()
-    group.add_argument('--verbose', '-v', action='count', default=0, help='increase output verbosity')
-    group.add_argument('--quiet', action='store_true', help='disable output')
-    group = parser.add_mutually_exclusive_group()
-    group.add_argument('--interactive', action='store_true', help='enable interactive output (i.e. for a PTY)')
-    group.add_argument('--non-interactive', action='store_true', help='disable interactive output')
-
-    parser.add_argument('--settings-file', default=Path('drover.yml'), type=Path,
-                        help='Settings file name (default: "drover.yml")')
-    parser.add_argument('--install-path', default=Path(), type=Path,
-                        help='Package install path (e.g. from "pip install -t"; default: working directory)')
-    parser.add_argument('stage', type=str)
-    arguments = parser.parse_args()
-
-    if not arguments.quiet:
-        logging.basicConfig(format='%(message)s', stream=sys.stdout)
-        logging_level = max(1, logging.INFO - (10 * arguments.verbose))
-        _logger.setLevel(logging_level)
-
-    interactive = True if arguments.interactive else False if arguments.non_interactive else sys.__stdin__.isatty()
-
-    settings_file_name = arguments.settings_file
-    install_path: Path = arguments.install_path
-
-    settings: Settings = None
-    try:
-        with open(settings_file_name, 'r') as settings_file:
-            settings = Settings.parse_obj(yaml.safe_load(settings_file))
-    except (ValueError, ValidationError) as e:
-        _logger.error('Settings file is invalid: %s', e)
-        _logger.debug('', exc_info=e)
-        sys.exit(1)
-    except FileNotFoundError as e:
-        _logger.error('Settings file does not exist: %s', e)
-        _logger.debug('', exc_info=e)
-        sys.exit(1)
-
-    try:
-        drover = Drover(settings, arguments.stage, interactive=interactive)
-        drover.update(install_path)
-    except SettingsError as e:
-        _logger.error('Initialization failed: %s', e)
-        _logger.debug('', exc_info=e)
-        sys.exit(1)
-    except UpdateError as e:
-        _logger.error('Update failed: %s', e)
-        _logger.debug('', exc_info=e)
-        sys.exit(1)

drover/cli.py

@@ -0,0 +1,74 @@
+"""Command-line interface functionality for the Drover interface"""
+import argparse
+import logging
+import sys
+from pathlib import Path
+
+import yaml
+from pydantic import ValidationError
+
+from drover import __version__, Drover, SettingsError, UpdateError
+from drover.models import Settings
+
+_logger = logging.getLogger(__name__)
+
+
+def _parse_arguments():
+    parser = argparse.ArgumentParser(description=__doc__.partition('\n')[0])
+    parser.add_argument('--version', '-V', action='version', version=f'%(prog)s {__version__}')
+    group = parser.add_mutually_exclusive_group()
+    group.add_argument('--verbose', '-v', action='count', default=0, help='increase output verbosity')
+    group.add_argument('--quiet', action='store_true', help='disable output')
+    group = parser.add_mutually_exclusive_group()
+    group.add_argument('--interactive', action='store_true', help='enable interactive output (i.e. for a PTY)')
+    group.add_argument('--non-interactive', action='store_true', help='disable interactive output')
+
+    parser.add_argument('--settings-file', default=Path('drover.yml'), type=Path,
+                        help='Settings file name (default: "drover.yml")')
+    parser.add_argument('--install-path', default=Path(), type=Path,
+                        help='Package install path (e.g. from "pip install -t"; default: working directory)')
+    parser.add_argument('stage', type=str)
+    return parser.parse_args()
+
+
+def _parse_settings(settings_file_name: Path) -> Settings:
+    try:
+        with open(settings_file_name, 'r') as settings_file:
+            return Settings.parse_obj(yaml.safe_load(settings_file))
+    except (ValueError, ValidationError) as e:
+        _logger.error('Settings file is invalid: %s', e)
+        _logger.debug('', exc_info=e)
+        sys.exit(1)
+    except FileNotFoundError as e:
+        _logger.error('Settings file does not exist: %s', e)
+        _logger.debug('', exc_info=e)
+        sys.exit(1)
+
+
+def main():
+    """The main command-line entry point for the Drover interface"""
+    arguments = _parse_arguments()
+
+    if not arguments.quiet:
+        logging.basicConfig(format='%(message)s', stream=sys.stdout)
+        logging_level = max(1, logging.INFO - (10 * arguments.verbose))
+        _logger.setLevel(logging_level)
+
+    interactive = True if arguments.interactive else False if arguments.non_interactive else sys.__stdin__.isatty()
+
+    settings_file_name = arguments.settings_file
+    install_path: Path = arguments.install_path
+
+    settings: Settings = _parse_settings(settings_file_name)
+
+    try:
+        drover = Drover(settings, arguments.stage, interactive=interactive)
+        drover.update(install_path)
+    except SettingsError as e:
+        _logger.error('Initialization failed: %s', e)
+        _logger.debug('', exc_info=e)
+        sys.exit(1)
+    except UpdateError as e:
+        _logger.error('Update failed: %s', e)
+        _logger.debug('', exc_info=e)
+        sys.exit(1)

drover/io.py

@@ -54,7 +54,7 @@ def get_digest(source_file_names: Sequence[Path], block_size: int = 8192) -> str
     for source_file_name in sorted(full):
         if package_record_pattern.search(str(source_file_name)):
             package_parent_path = source_file_name.parent.parent
-            with open(source_file_name, 'r') as record:
+            with open(source_file_name, 'r', buffering=block_size) as record:
                 reader = csv.reader(record, delimiter=',', quotechar='"', lineterminator=os.linesep)
                 for item in reader:
                     item_name, item_hash, _other = item[:3]
@@ -64,7 +64,7 @@ def get_digest(source_file_names: Sequence[Path], block_size: int = 8192) -> str
                         done.add(source_file_name)
     remaining = full - done
     for source_file_name in sorted(remaining):
-        with open(source_file_name, 'rb') as source_file:
+        with open(source_file_name, 'rb', buffering=block_size) as source_file:
             if egg_information_pattern.search(str(source_file_name)):
                 # Ensure deterministic field order from PKG-INFO files
                 # See: https://www.python.org/dev/peps/pep-0314/#including-metadata-in-packages

setup.py

@@ -67,7 +67,7 @@ def run(self):
     license='BSD',
     packages=setuptools.find_packages(exclude=['tests*']),
     entry_points={
-        'console_scripts': ['drover=drover:main'],
+        'console_scripts': ['drover=drover.cli:main'],
     },
     python_requires='>=3.6',
     install_requires=[

tests/test_cli.py

@@ -0,0 +1,38 @@
+# pylint: disable=protected-access
+from dataclasses import dataclass
+from pathlib import Path
+from unittest import TestCase
+from unittest.mock import patch
+
+import drover
+import drover.cli
+
+from tests.test_drover import get_basic_settings
+
+
+@dataclass
+class MockArguments:
+    stage: str
+    quiet: bool
+    interactive: bool = True
+    non_interactive: bool = False
+    settings_file: Path = Path('drover.yml')
+    install_path: Path = Path('.')
+
+
+class TestDroverCLI(TestCase):
+    def test_basic_run(self):
+        expected_stage_name = 'stage'
+        expected_settings = get_basic_settings(expected_stage_name)
+        expected_interactive = True
+        mock_arguments = MockArguments(
+            stage='stage',
+            quiet=True,
+            interactive=expected_interactive, non_interactive=not expected_interactive)
+        with patch.object(drover.cli, '_parse_arguments', return_value=mock_arguments) as mock_parse_arguments, \
+             patch.object(drover.cli, '_parse_settings', return_value=expected_settings) as mock_parse_settings, \
+             patch.object(drover.cli, 'Drover', autospec=drover.Drover) as mock_drover:
+            drover.cli.main()
+            mock_drover.assert_called_with(expected_settings, expected_stage_name, interactive=expected_interactive)
+            mock_parse_arguments.assert_called()
+            mock_parse_settings.assert_called_with(mock_arguments.settings_file)

tests/test_drover.py

@@ -4,29 +4,48 @@
 from unittest.mock import patch, MagicMock
 
 import boto3
+import pytest
 
-from drover import Drover
+from drover import Drover, SettingsError
 from drover.models import Settings, Stage
 
 
 def get_supported_compatible_runtime():
     return 'python3.8'
 
 
+def get_basic_settings(expected_stage_name: str) -> Settings:
+    expected_stage = Stage(
+        region_name='region_name',
+        function_name='function_name',
+        compatible_runtime=get_supported_compatible_runtime(),
+        function_file_patterns=[
+            '^function.*'
+        ])
+    expected_settings = Settings(
+        stages={
+            expected_stage_name: expected_stage
+        })
+    return expected_settings
+
+
 class TestDrover(TestCase):
+    def test_init_with_valid_settings_and_invalid_stage_name(self):
+        expected_invalid_stage_name = 'stage-invalid'
+        expected_settings = get_basic_settings('stage')
+        expected_compatible_runtime_library_path = Path('path')
+
+        mock_boto3_client = MagicMock()
+
+        with patch.object(Drover, '_get_runtime_library_path', return_value=expected_compatible_runtime_library_path), \
+             patch.object(boto3, 'client', return_value=mock_boto3_client):
+            with pytest.raises(SettingsError, match=r'^Invalid stage name.*'):
+                Drover(expected_settings, expected_invalid_stage_name, interactive=False)
+
     def test_init_with_valid_settings_and_stage(self):
         expected_stage_name = 'stage'
-        expected_stage = Stage(
-            region_name='region_name',
-            function_name='function_name',
-            compatible_runtime=get_supported_compatible_runtime(),
-            function_file_patterns=[
-                '^function.*'
-            ])
-        expected_settings = Settings(
-            stages={
-                expected_stage_name: expected_stage
-            })
+        expected_settings = get_basic_settings(expected_stage_name)
+        expected_stage = expected_settings.stages[expected_stage_name]
         expected_interactive = False
         expected_requirements_layer_name = 'function_name-requirements'
         expected_compatible_runtime_library_path = Path('path')