Implement declarative argument definitions and logic to bundle values.

Expand the arguments module with the notions of a defined grouping of arguments
(ArgumentBundleDefinition) and bundles of particular arguments (ArgumentBundle).

Use this as the mechanism by which payloads are checked for validity. As part
of declaring a bundle definition the expected positional arguments are declared
so implement basic length checks that catch missing positional arguments which
were required. Each argument itself is also tested aganst an optional validity
function which can be specified at the point of definition.

Author: albu-diku

mig/shared/arguments.py

@@ -1,12 +1,88 @@
 import argparse
+from collections import OrderedDict
 
 from mig.shared.compat import PY2
 
 _EMPTY_DICT = {}
+_EMPTY_LIST = {}
 _NO_DEFAULT = object()
 _POSITIONAL_MARKER = '__args__'
 
 
+class ArgumentBundleDefinition:
+    def __init__(self, name, positional=_EMPTY_LIST):
+        self._definition_name = name
+        self._expected_positions = 0
+        self._item_checks = []
+        self._item_names = []
+
+        if positional is not _EMPTY_LIST:
+            self._define_positional(positional)
+
+    @property
+    def _fields(self):
+        return self._item_names
+
+    @property
+    def _validators(self):
+        return self._item_checks
+
+    def __call__(self, *args):
+        return self._extract_and_bundle(args, extract_by='position')
+
+    def _define_positional(self, positional):
+        for flag, name, validator_fn in positional:
+            assert flag is None
+            self._item_names.append(name)
+            self._item_checks.append(validator_fn)
+        self._expected_positions = len(positional)
+
+    def _extract_and_bundle(self, args, extract_by=None):
+        if extract_by == 'position':
+            actual_positions = len(args)
+            if actual_positions < self._expected_positions:
+                raise ValueError('Error: too few arguments given (expected %d got %d)' % (
+                    self._expected_positions, actual_positions))
+            keys_to_bundle = list(range(actual_positions))
+        elif extract_by == 'name':
+            keys_to_bundle = self._item_names
+        elif extract_by == 'short':
+            keys_to_bundle = self._item_short
+        else:
+            raise RuntimeError()
+
+        return ArgumentBundle.from_args(self, args, keys_to_bundle)
+
+    def ensure_bundle(self, bundle_or_args):
+        assert isinstance(self, ArgumentBundleDefinition)
+
+        bundle_definition = self
+
+        if isinstance(bundle_or_args, ArgumentBundle):
+            assert bundle_or_args.name == bundle_definition._definition_name
+            return bundle_or_args
+        else:
+            return bundle_definition(*bundle_or_args)
+
+
+class ArgumentBundle(OrderedDict):
+    def __init__(self, definition, dictionary):
+        super(ArgumentBundle, self).__init__(dictionary)
+        self._definition = definition
+
+    @property
+    def name(self):
+        return self._definition._definition_name
+
+    def __iter__(self):
+        return iter(self.values())
+
+    @classmethod
+    def from_args(cls, definition, args, keys):
+        dictionary = {key: args[key] for key in keys}
+        return cls(definition, dictionary)
+
+
 class GetoptCompatNamespace(argparse.Namespace):
     """Small glue abstraction to provide an object that when iterated yields
     tuples of cli-like options and their corresponding values thus emulating

tests/test_mig_shared_arguments.py

@@ -3,7 +3,46 @@
 
 from tests.support import MigTestCase, testmain
 
-from mig.shared.arguments import parse_getopt_args
+from mig.shared.arguments import parse_getopt_args, \
+    ArgumentBundle, ArgumentBundleDefinition
+
+
+def _is_not_none(value):
+    return value is not None
+
+
+class TestMigSharedArguments__bundles(MigTestCase):
+    ThingsBundle = ArgumentBundleDefinition('Things', [
+        (None, 'some_thing', _is_not_none),
+        (None, 'other_thing', _is_not_none),
+    ])
+
+    def assertBundleOfKind(self, value, bundle_kind=None):
+        assert isinstance(bundle_kind, str) and bundle_kind
+        self.assertIsInstance(value, ArgumentBundle, "value is not an argument bundle")
+        self.assertEqual(value.name, bundle_kind, "expected %s bundle, got %s" % (bundle_kind, value.name))
+
+    def test_bundling_arguments_produces_a_bundle(self):
+        bundle = self.ThingsBundle('abc', 'def')
+
+        self.assertBundleOfKind(bundle, bundle_kind='Things')
+
+    def test_raises_on_missing_positional_arguments(self):
+        with self.assertRaises(ValueError) as raised:
+            self.ThingsBundle(['a'])
+        self.assertEqual(str(raised.exception), 'Error: too few arguments given (expected 2 got 1)')
+
+    def test_ensuring_an_existing_bundle_returns_it_unchanged(self):
+        existing_bundle = self.ThingsBundle('abc', 'def')
+
+        bundle = self.ThingsBundle.ensure_bundle(existing_bundle)
+
+        self.assertIs(bundle, existing_bundle)
+
+    def test_ensuring_an_list_of_arguments_returns_a_bundle(self):
+        bundle = self.ThingsBundle.ensure_bundle(['abc', 'def'])
+
+        self.assertBundleOfKind(bundle, bundle_kind='Things')
 
 
 class TestMigSharedArguments__getopt(MigTestCase):