enable strict check for arguments (#183)

* enable strict check for arguments

* remove arguments in input_data

Author: njzjz

dpdispatcher/dp_cloud_server_context.py

@@ -251,25 +251,7 @@ def machine_subfields(cls) -> List[Argument]:
             Argument("email", str, optional=False, doc="Email"),
             Argument("password", str, optional=False, doc="Password"),
             Argument("program_id", int, optional=False, doc="Program ID"),
-            Argument("input_data", dict, [
-                Argument("job_name", str, optional=True, doc="Job name"),
-                Argument("image_name", str, optional=True, doc="Name of the image which run the job, optional "
-                                                               "when platform is not ali/oss."),
-                Argument("disk_size", str, optional=True, doc="disk size (GB), optional "
-                                                              "when platform is not ali/oss."),
-                Argument("scass_type", str, optional=False, doc="machine configuration."),
-                Argument("platform", str, optional=False, doc="Job run in which platform."),
-                Argument("log_file", str, optional=True, doc="location of log file."),
-                Argument('checkpoint_files', [str, list], optional=True, doc="location of checkpoint files when "
-                                                                                  "it is list type. record file "
-                                                                                  "changes when it is string value "
-                                                                                  "'sync_files'"),
-                Argument('checkpoint_time', int, optional=True, default=15, doc='interval of checkpoint data been '
-                                                                                'stored minimum 15.'),
-                Argument('backward_files', list, optional=True, doc='which files to be uploaded to remote '
-                                                                         'resources. Upload all the files when it is '
-                                                                         'None or empty.')
-            ], optional=False, doc="Configuration of job"),
+            Argument("input_data", dict, optional=False, doc="Configuration of job"),
         ], doc=doc_remote_profile)]
 
 

dpdispatcher/machine.py

@@ -120,6 +120,11 @@ def load_from_dict(cls, machine_dict):
         except KeyError as e:
             dlog.error(f"KeyError:batch_type; cls.subclasses_dict:{cls.subclasses_dict}; batch_type:{batch_type};")
             raise e
+        # check dict
+        base = cls.arginfo()
+        machine_dict = base.normalize_value(machine_dict, trim_pattern="_*")
+        base.check_value(machine_dict, strict=True)
+
         context = BaseContext.load_from_dict(machine_dict)
         machine = machine_class(context=context)
         return machine

dpdispatcher/submission.py

@@ -439,6 +439,15 @@ def get_hash(self):
     def load_from_json(cls, json_file):
         with open(json_file, 'r') as f:
             task_dict = json.load(f)
+        return cls.load_from_dict(task_dict)
+
+    @classmethod
+    def load_from_dict(cls, task_dict: dict) -> "Task":
+        # check dict
+        base = cls.arginfo()
+        task_dict = base.normalize_value(task_dict, trim_pattern="_*")
+        base.check_value(task_dict, strict=True)
+
         task = cls.deserialize(task_dict=task_dict)
         return task
 
@@ -793,10 +802,15 @@ def load_from_json(cls, json_file):
 
     @classmethod
     def load_from_dict(cls, resources_dict):
+        # check dict
+        base = cls.arginfo(detail_kwargs=False)
+        resources_dict = base.normalize_value(resources_dict, trim_pattern="_*")
+        base.check_value(resources_dict, strict=True)
+
         return cls.deserialize(resources_dict=resources_dict)
 
     @staticmethod
-    def arginfo():
+    def arginfo(detail_kwargs=True):
         doc_number_node = 'The number of node need for each `job`'
         doc_cpu_per_node = 'cpu numbers of each node assigned to each job.'
         doc_gpu_per_node = 'gpu numbers of each node assigned to each job.'
@@ -837,14 +851,20 @@ def arginfo():
             Argument("wait_time", [int, float], optional=True, doc=doc_wait_time, default=0)
         ]
 
-        batch_variant = Variant(
-            "batch_type",
-            [machine.resources_arginfo() for machine in set(Machine.subclasses_dict.values())],
-            optional=False,
-            doc='The batch job system type loaded from machine/batch_type.',
-        )
+        if detail_kwargs:
+            batch_variant = Variant(
+                "batch_type",
+                [machine.resources_arginfo() for machine in set(Machine.subclasses_dict.values())],
+                optional=False,
+                doc='The batch job system type loaded from machine/batch_type.',
+            )
 
-        resources_format = Argument("resources", dict, resources_args, [batch_variant])
+            resources_format = Argument("resources", dict, resources_args, [batch_variant])
+        else:
+            resources_args.append(
+                Argument("kwargs", dict, optional=True, doc="Vary by different machines.")
+            )
+            resources_format = Argument("resources", dict, resources_args)
         return resources_format
 
 

setup.py

@@ -19,7 +19,7 @@
 with open(path.join(NAME, '_date.py'), 'w') as fp :
     fp.write('date = \'%s\'' % today)
 
-install_requires=['paramiko', 'dargs>=0.2.6', 'requests', 'tqdm']
+install_requires=['paramiko', 'dargs>=0.2.9', 'requests', 'tqdm']
 
 setuptools.setup(
     name=NAME,

tests/test_argcheck.py

@@ -0,0 +1,81 @@
+import os
+import sys
+import unittest
+
+sys.path.insert(0, os.path.abspath(
+    os.path.join(os.path.dirname(__file__), '..')))
+__package__ = 'tests'
+
+from .context import setUpModule
+from .context import Task, Resources, Machine
+
+class TestJob(unittest.TestCase):
+
+    def test_machine_argcheck(self):
+        norm_dict = Machine.load_from_dict({
+            "batch_type": "slurm",
+            "context_type": "local",
+            "local_root": "./",
+            "remote_root": "/some/path",
+        }).serialize()
+        expected_dict = {
+            'batch_type': 'Slurm',
+            'context_type': 'LocalContext',
+            'local_root': './',
+            'remote_root': '/some/path',
+            'remote_profile': {},
+        }
+        self.assertDictEqual(norm_dict, expected_dict)
+
+    def test_resources_argcheck(self):
+        norm_dict = Resources.load_from_dict({
+            "number_node": 1,
+            "cpu_per_node": 2,
+            "gpu_per_node": 0,
+            "queue_name": "haha",
+            "group_size": 1,
+            "envs": {
+                "aa": "bb",
+            },
+            "kwargs": {
+                "cc": True,
+            }
+        }).serialize()
+        expected_dict = {'cpu_per_node': 2,
+                         'custom_flags': [],
+                         'envs': {'aa': 'bb'},
+                         'gpu_per_node': 0,
+                         'group_size': 1,
+                         'kwargs': {
+                             "cc": True,
+                         },
+                         'module_list': [],
+                         'module_purge': False,
+                         'module_unload_list': [],
+                         'number_node': 1,
+                         'para_deg': 1,
+                         'queue_name': 'haha',
+                         'source_list': [],
+                         'strategy': {'if_cuda_multi_devices': False, 'ratio_unfinished': 0.0},
+                         'wait_time': 0,
+                         }
+        self.assertDictEqual(norm_dict, expected_dict)
+
+    def test_task_argcheck(self):
+        norm_dict = Task.load_from_dict({
+            "command": "ls",
+            "task_work_path": "./",
+            "forward_files": [],
+            "backward_files": [],
+            "outlog": "out",
+            "errlog": "err",
+        }).serialize()
+        expected_dict = {
+            "command": "ls",
+            "task_work_path": "./",
+            "forward_files": [],
+            "backward_files": [],
+            "outlog": "out",
+            "errlog": "err",
+        }
+        self.assertDictEqual(norm_dict, expected_dict)

tests/test_class_machine_dispatch.py

@@ -16,7 +16,7 @@
 from .context import dargs
 from .context import DistributedShell, HDFSContext
 from .sample_class import SampleClass
-from dargs.dargs import ArgumentKeyError
+from dargs.dargs import ArgumentValueError
 
 class TestMachineDispatch(unittest.TestCase):
     def setUp(self):
@@ -136,7 +136,7 @@ def test_context_err(self):
             'context_type': 'foo'
         }
         # with self.assertRaises(KeyError):
-        with self.assertRaises(KeyError):
+        with self.assertRaises(ArgumentValueError):
             Machine.load_from_dict(machine_dict=machine_dict)
 
     def test_pbs(self):