alot/utils/argparse: Add a new argparse validators infrastructure

This adds a new argparse.Action class validates input using a new
keyword argument that takes a validator function. This will allow us to
replace the use the type keyword as a validator, which is both more
correct, and frees up the type keyword to do what it's actually meant to
do, convert the input from one type to another.

It also adds 3 new validator functions that will be enabled in the next
commit. One that checks for a required file, one that checks for an
optional directory, and one that looks for a required file, fifo, or
block special device (/dev/null).

2 files changed, 241 insertions, 0 deletions

diff --git a/alot/utils/argparse.py b/alot/utils/argparse.py
index 4922c247..adacdd6e 100644
--- a/alot/utils/argparse.py
+++ b/alot/utils/argparse.py
@@ -20,13 +20,22 @@
 from __future__ import absolute_import
 
 import argparse
+import collections
+import functools
 import itertools
+import os
+import stat
 
 
 _TRUEISH = ['true', 'yes', 'on', '1', 't', 'y']
 _FALSISH = ['false', 'no', 'off', '0', 'f', 'n']
 
 
+class ValidationFailed(Exception):
+    """Exception raised when Validation fails in a ValidatedStoreAction."""
+    pass
+
+
 def _boolean(string):
     string = string.lower()
     if string in _FALSISH:
@@ -38,6 +47,56 @@ def _boolean(string):
             ', '.join(itertools.chain(iter(_TRUEISH), iter(_FALSISH)))))
 
 
+def _path_factory(check):
+    """Create a function that checks paths."""
+
+    @functools.wraps(check)
+    def validator(paths):
+        if isinstance(paths, basestring):
+            check(paths)
+        elif isinstance(paths, collections.Sequence):
+            for path in paths:
+                check(path)
+        else:
+            raise Exception('expected either basestr or sequenc of basstr')
+
+    return validator
+
+
+@_path_factory
+def require_file(path):
+    """Validator that asserts that a file exists.
+
+    This fails if there is nothing at the given path.
+    """
+    if not os.path.isfile(path):
+        raise ValidationFailed('{} is not a valid file.'.format(path))
+
+
+@_path_factory
+def optional_file_like(path):
+    """Validator that ensures that if a file exists it regular, a fifo, or a
+    character device. The file is not required to exist.
+
+    This includes character special devices like /dev/null.
+    """
+    if (os.path.exists(path) and not (os.path.isfile(path) or
+                                      stat.S_ISFIFO(os.stat(path).st_mode) or
+                                      stat.S_ISCHR(os.stat(path).st_mode))):
+        raise ValidationFailed(
+            '{} is not a valid file, character device, or fifo.'.format(path))
+
+
+@_path_factory
+def require_dir(path):
+    """Validator that asserts that a directory exists.
+
+    This fails if there is nothing at the given path.
+    """
+    if not os.path.isdir(path):
+        raise ValidationFailed('{} is not a valid directory.'.format(path))
+
+
 class BooleanAction(argparse.Action):
     """Argparse action that can be used to store boolean values."""
     def __init__(self, *args, **kwargs):
@@ -47,3 +106,31 @@ class BooleanAction(argparse.Action):
 
     def __call__(self, parser, namespace, values, option_string=None):
         setattr(namespace, self.dest, values)
+
+
+class ValidatedStoreAction(argparse.Action):
+    """An action that allows a validation function to be specificied.
+
+    The validator keyword must be a function taking exactly one argument, that
+    argument is a list of strings or the type specified by the type argument.
+    It must raise ValidationFailed with a message when validation fails.
+    """
+
+    def __init__(self, option_strings, dest=None, nargs=None, default=None,
+                 required=False, type=None, metavar=None, help=None,
+                 validator=None):
+        super(ValidatedStoreAction, self).__init__(
+            option_strings=option_strings, dest=dest, nargs=nargs,
+            default=default, required=required, metavar=metavar, type=type,
+            help=help)
+
+        self.validator = validator
+
+    def __call__(self, parser, namespace, values, option_string=None):
+        if self.validator:
+            try:
+                self.validator(values)
+            except ValidationFailed as e:
+                raise argparse.ArgumentError(self, str(e))
+
+        setattr(namespace, self.dest, values)
diff --git a/tests/utils/argparse_test.py b/tests/utils/argparse_test.py
new file mode 100644
index 00000000..b76e96ec
--- /dev/null
+++ b/tests/utils/argparse_test.py
@@ -0,0 +1,154 @@
+# encoding=utf-8
+# Copyright © 2017 Dylan Baker
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Tests for alot.utils.argparse"""
+
+from __future__ import absolute_import
+
+import argparse
+import contextlib
+import os
+import shutil
+import tempfile
+import unittest
+
+from alot.utils import argparse as cargparse
+
+
+class TestValidatedStore(unittest.TestCase):
+    """Tests for the ValidatedStore action class."""
+
+    def _argparse(self, args):
+        """Create an argparse instance with a validator."""
+
+        def validator(args):
+            if args == 'fail':
+                raise cargparse.ValidationFailed
+
+        parser = argparse.ArgumentParser()
+        parser.add_argument(
+            'foo',
+            action=cargparse.ValidatedStoreAction,
+            validator=validator)
+        return parser.parse_args(args)
+
+    def test_validates(self):
+        # Arparse will raise a SystemExit (calls sys.exit) rather than letting
+        # the exception cause the program to close.
+        with self.assertRaises(SystemExit):
+            self._argparse(['fail'])
+
+
+@contextlib.contextmanager
+def temporary_directory(suffix='', prefix='', dir=None):
+    """Python3 interface implementation.
+
+    Python3 provides a class that can be used as a context manager, which
+    creates a temporary directory and removes it when the context manager
+    exits. This function emulates enough of the interface of
+    TemporaryDirectory, for this module to use, and is designed as a drop in
+    replacement that can be replaced after the python3 port.
+
+    The only user visible difference is that this does not implement the
+    cleanup method that TemporaryDirectory does.
+    """
+    directory = tempfile.mkdtemp(suffix=suffix, prefix=prefix, dir=dir)
+    yield directory
+    shutil.rmtree(directory)
+
+
+class TestRequireFile(unittest.TestCase):
+    """Tests for the require_file validator."""
+
+    def test_doesnt_exist(self):
+        with temporary_directory() as d:
+            with self.assertRaises(cargparse.ValidationFailed):
+                cargparse.require_file(os.path.join(d, 'doesnt-exist'))
+
+    def test_dir(self):
+        with temporary_directory() as d:
+            with self.assertRaises(cargparse.ValidationFailed):
+                cargparse.require_file(d)
+
+    def test_file(self):
+        with tempfile.NamedTemporaryFile() as f:
+            cargparse.require_file(f.name)
+
+    def test_char_special(self):
+        with self.assertRaises(cargparse.ValidationFailed):
+            cargparse.require_file('/dev/null')
+
+    def test_fifo(self):
+        with temporary_directory() as d:
+            path = os.path.join(d, 'fifo')
+            os.mkfifo(path)
+            with self.assertRaises(cargparse.ValidationFailed):
+                cargparse.require_file(path)
+
+
+class TestRequireDir(unittest.TestCase):
+    """Tests for the require_dir validator."""
+
+    def test_doesnt_exist(self):
+        with temporary_directory() as d:
+            with self.assertRaises(cargparse.ValidationFailed):
+                cargparse.require_dir(os.path.join(d, 'doesnt-exist'))
+
+    def test_dir(self):
+        with temporary_directory() as d:
+            cargparse.require_dir(d)
+
+    def test_file(self):
+        with tempfile.NamedTemporaryFile() as f:
+            with self.assertRaises(cargparse.ValidationFailed):
+                cargparse.require_dir(f.name)
+
+    def test_char_special(self):
+        with self.assertRaises(cargparse.ValidationFailed):
+            cargparse.require_dir('/dev/null')
+
+    def test_fifo(self):
+        with temporary_directory() as d:
+            path = os.path.join(d, 'fifo')
+            os.mkfifo(path)
+            with self.assertRaises(cargparse.ValidationFailed):
+                cargparse.require_dir(path)
+
+
+class TestOptionalFileLike(unittest.TestCase):
+    """Tests for the optional_file_like validator."""
+
+    def test_doesnt_exist(self):
+        with temporary_directory() as d:
+            cargparse.optional_file_like(os.path.join(d, 'doesnt-exist'))
+
+    def test_dir(self):
+        with temporary_directory() as d:
+            with self.assertRaises(cargparse.ValidationFailed):
+                cargparse.optional_file_like(d)
+
+    def test_file(self):
+        with tempfile.NamedTemporaryFile() as f:
+            cargparse.optional_file_like(f.name)
+
+    def test_char_special(self):
+        cargparse.optional_file_like('/dev/null')
+
+    def test_fifo(self):
+        with temporary_directory() as d:
+            path = os.path.join(d, 'fifo')
+            os.mkfifo(path)
+            cargparse.optional_file_like(path)