partial work on apply_argspec function

koji/tasks.py

@@ -95,6 +95,59 @@ def safe_rmtree(path, unmount=False, strict=True):
     return 0
 
 
+
+SIGNATURES = {
+    # like getargspec -- args, varargs, keywords, defaults
+    'sleep' : (['n'], None, None, None),
+    'fork' : (['n', 'm'], None, None, None),
+}
+
+
+def apply_argspec(argspec, args, kwargs=None):
+    """Apply an argspec to the given args and return a dictionary"""
+    if kwargs is None:
+        kwargs = {}
+    f_args, f_varargs, f_varkw, f_defaults = argspec
+    print argspec, args, kwargs
+    data = dict(zip(f_args, args))
+    if len(args) > len(f_args):
+        if not f_varargs:
+            raise koji.ParameterError, "Too many args"
+        data[f_varargs] = tuple(args[len(f_args):])
+    elif f_varargs:
+        data[f_varargs] = ()
+    if f_varkw:
+        data[f_varkw] = {}
+    for arg in kwargs:
+        if arg in data:
+            raise koji.ParameterError, "duplicate keyword argument %r" \
+                    % arg
+        if arg in f_args:
+            data[arg] = kwargs[arg]
+        elif not f_varkw:
+            raise koji.ParameterError, "Unexpected keyword argument %r" \
+                % (arg)
+        else:
+            data[f_varkw][arg] = kwargs[arg]
+    if f_defaults:
+        for arg, val in f_defaults:
+            data.setdefault(arg, val)
+    for n, arg in enumerate(f_args):
+        if arg not in data:
+            # missing arg
+            raise koji.ParameterError, "missing required argument %r (#%i)" \
+                    % (arg, n)
+    return data
+
+
+def parse_request(method, request):
+    """Parse a task request into """
+
+    argspec = SIGNATURES.get(method)
+    if not argspec:
+        raise koji.GenericError, "Unknown method"
+
+
 class ServerExit(Exception):
     """Raised to shutdown the server"""
     pass

tests/test_argspec.py

@@ -0,0 +1,33 @@
+#!/usr/bin/python
+
+"""Test argspec functions"""
+
+import koji.tasks
+import unittest
+
+class ArgspecCase(unittest.TestCase):
+    """Main test case container"""
+
+    def test_apply_argspec(self):
+        """Test the parse_NVR method"""
+
+        # Single param
+        argspec = (['n'], None, None, None)
+        self.assertRaises(koji.ParameterError, koji.tasks.apply_argspec,
+            argspec, (), None )
+        self.assertRaises(koji.ParameterError, koji.tasks.apply_argspec,
+            argspec, (1,2), None )
+        ret = koji.tasks.apply_argspec(argspec, (1,), None)
+        self.assertEqual(ret, {'n':1})
+        ret = koji.tasks.apply_argspec(argspec, (), {'n':1})
+        self.assertEqual(ret, {'n':1})
+        self.assertRaises(koji.ParameterError, koji.tasks.apply_argspec,
+            argspec, (1,), {'n':1} )
+        self.assertRaises(koji.ParameterError, koji.tasks.apply_argspec,
+            argspec, (1,), {'m':1} )
+        self.assertRaises(koji.ParameterError, koji.tasks.apply_argspec,
+            argspec, (), {'m':1} )
+
+
+if __name__ == '__main__':
+    unittest.main()