support dispatcher executor (#83)

* support dispatcher executor

* add UTs

Co-authored-by: Han Wang <wang_han@iapcm.ac.cn>

dpgen2/utils/step_config.py

@@ -5,6 +5,7 @@
 )
 from dpgen2.constants import default_image
 from dflow.plugins.lebesgue import LebesgueExecutor
+from dflow.plugins.dispatcher import DispatcherExecutor
 
 def lebesgue_extra_args():
     # It is not possible to strictly check the keys in this section....
@@ -26,10 +27,15 @@ def lebesgue_executor_args():
         Argument("extra", dict, lebesgue_extra_args(), optional = True, doc = doc_extra),
     ]
 
+def dispatcher_args():
+    """free style dispatcher args"""
+    return []
+
 def variant_executor():
     doc = f'The type of the executor.'
     return Variant("type", [
         Argument("lebesgue_v2", dict, lebesgue_executor_args()),
+        Argument("dispatcher", dict, dispatcher_args()),
     ], doc = doc)
 
 def template_conf_args():
@@ -94,6 +100,8 @@ def init_executor(
     etype = executor_dict.pop('type')
     if etype == "lebesgue_v2":
         return LebesgueExecutor(**executor_dict)
+    if etype == "dispatcher":
+        return DispatcherExecutor(**executor_dict)
     else:
         raise RuntimeError('unknown executor type', etype)    
 

tests/utils/test_step_config.py

@@ -85,7 +85,7 @@ def test_init_executor(self):
         }
         odict = normalize(idict)
         ret = init_executor(odict.pop('executor'))
-        self.assertEqual(type(ret), dflow.plugins.lebesgue.LebesgueExecutor)
+        self.assertTrue(isinstance(ret, dflow.plugins.lebesgue.LebesgueExecutor))
 
 
     def test_init_executor_notype(self):
@@ -97,3 +97,16 @@ def test_init_executor_notype(self):
         odict = normalize(idict)
         ret = init_executor(odict.pop('executor'))
         self.assertEqual(ret, None)
+
+
+    def test_init_executor_dispatcher(self):
+        idict = {
+            "executor":{
+                "type" : "dispatcher",
+                "username" : "foo",
+            },
+        }
+        odict = normalize(idict)
+        self.assertEqual(odict['executor'], idict['executor'])
+        ret = init_executor(odict.pop('executor'))
+        self.assertTrue(isinstance(ret, dflow.plugins.dispatcher.DispatcherExecutor))