fix codecheck bug

Signed-off-by: deveco_test <liguangjie1@huawei.com>

plugins/ohos/src/ohos/drivers/drivers.py

@@ -24,6 +24,7 @@ import shutil
 import zipfile
 import tempfile
 import stat
+from collections import namedtuple
 from dataclasses import dataclass
 
 from xdevice import ParamError
@@ -1505,7 +1506,7 @@ class RemoteDexRunner:
             self.config.device.execute_shell_command(
                 command, timeout=self.config.timeout,
                 receiver=handler, retry=0)
-        except ConnectionResetError as _:
+        except ConnectionResetError as _:  # pylint:disable=undefined-variable
             if len(listener) == 1 and isinstance(listener[0],
                                                  CollectingTestListener):
                 LOG.info("Try subprocess ")
@@ -2390,8 +2391,7 @@ class JSUnitTestDriver(IDriver):
                                       self.config.resource_path,
                                       self.config.testcases_path)
 
-        package, ability_name, runner, testcase_timeout = \
-            self._get_driver_config(json_config)
+        driver_config = self._get_driver_config(json_config)
         # bms not check release type
         self.config.device.execute_shell_command("bm set -d enable")
         # turn auto rotation off
@@ -2402,7 +2402,8 @@ class JSUnitTestDriver(IDriver):
         # execute test case
         command = "aa start -p %s -n %s " \
                   "-s unittest %s -s rawLog true -s timeout %s" \
-                  % (package, ability_name, runner, testcase_timeout)
+                  % (driver_config.package, driver_config.ability_name,
+                     driver_config.runner, driver_config.testcase_timeout)
         result_value = self.config.device.execute_shell_command(
             command, timeout=self.timeout)
         if self.xml_output == "true":
@@ -2410,7 +2411,7 @@ class JSUnitTestDriver(IDriver):
             if report_name:
                 self.config.target_test_path = "/%s/%s/%s/%s/%s/" \
                                                % ("sdcard", "Android",
-                                                  "data", package, "cache")
+                                                  "data", driver_config.package, "cache")
                 result = ResultManager(report_name,
                                        self.config.report_path,
                                        self.config.device,
@@ -2446,7 +2447,8 @@ class JSUnitTestDriver(IDriver):
         if not package:
             raise ParamError("Can't find package in config file.",
                              error_no="03201")
-        return package, ability_name, runner, testcase_timeout
+        DriverConfig = namedtuple('DriverConfig', 'package ability_name runner testcase_timeout')
+        return DriverConfig(package, ability_name, runner, testcase_timeout)
 
     def run_js_outer(self, request):
         try:

plugins/ohos/src/ohos/environment/device.py

@@ -80,7 +80,7 @@ def perform_device_action(func):
             except ReportException as error:
                 self.log.exception("Generate report error!", exc_info=False)
                 exception = error
-            except (ConnectionResetError, ConnectionRefusedError) as error:
+            except (ConnectionResetError, ConnectionRefusedError) as error:  # pylint:disable=undefined-variable
                 self.log.error("error type: %s, error: %s" %
                                (error.__class__.__name__, error))
                 cmd = "hdc_std target boot"
@@ -613,7 +613,7 @@ class Device(IDevice):
             try:
                 from devicetest.controllers.openharmony import OpenHarmony
                 OpenHarmony.install_harmony_rpc(self)
-            except (ModuleNotFoundError, ImportError) as error:
+            except (ModuleNotFoundError, ImportError) as error:  # pylint:disable=undefined-variable
                 self.log.debug(str(error))
                 self.log.error('please check devicetest extension module is exist.')
                 raise Exception(ErrorMessage.Error_01437.Topic)

src/xdevice/__init__.py

@@ -50,6 +50,7 @@ from _core.constants import ProductForm
 from _core.constants import TestType
 from _core.constants import CKit
 from _core.constants import ConfigConst
+from _core.constants import ReportConst
 from _core.constants import ModeType
 from _core.constants import TestExecType
 from _core.constants import ListenerType
@@ -156,6 +157,7 @@ __all__ = [
     "TestType",
     "CKit",
     "ConfigConst",
+    "ReportConst",
     "ModeType",
     "TestExecType",
     "ListenerType",

src/xdevice/_core/command/console.py

@@ -23,10 +23,12 @@ import signal
 import sys
 import threading
 import copy
+from collections import namedtuple
 
 from _core.config.config_manager import UserConfigManager
 from _core.constants import SchedulerType
 from _core.constants import ConfigConst
+from _core.constants import ReportConst
 from _core.constants import ModeType
 from _core.constants import ToolCommandType
 from _core.environment.manager_env import EnvironmentManager
@@ -48,11 +50,12 @@ LOG = platform_logger("Console")
 try:
     if platform.system() != 'Windows':
         import readline
-except (ModuleNotFoundError, ImportError):
+except (ModuleNotFoundError, ImportError):  # pylint:disable=undefined-variable
     LOG.warning("Readline module is not exist.")
 
 MAX_VISIBLE_LENGTH = 49
 MAX_RESERVED_LENGTH = 46
+Argument = namedtuple('Argument', 'options unparsed valid_param parser')
 
 
 class Console(object):
@@ -323,7 +326,7 @@ class Console(object):
             valid_param = False
             parser.print_help()
             LOG.warning("Parameter parsing system exit exception.")
-        return options, unparsed, valid_param, parser
+        return Argument(options, unparsed, valid_param, parser)
 
     @classmethod
     def _params_pre_processing(cls, para_list):
@@ -383,16 +386,15 @@ class Console(object):
             Scheduler.command_queue.append(args)
             LOG.info("Input command: {}".format(args))
             para_list = args.split()
-            (options, _, valid_param, parser) = self.argument_parser(
-                para_list)
-            if options is None or not valid_param:
+            argument = self.argument_parser( para_list)
+            if argument.options is None or not argument.valid_param:
                 LOG.warning("Options is None.")
                 return None
-            if options.action == ToolCommandType.toolcmd_key_run and \
-                    options.retry:
-                options = self._get_retry_options(options, parser)
-                if options.dry_run:
-                    history_report_path = getattr(options,
+            if argument.options.action == ToolCommandType.toolcmd_key_run and \
+                    argument.options.retry:
+                argument.options = self._get_retry_options(argument.options, argument.parser)
+                if argument.options.dry_run:
+                    history_report_path = getattr(argument.options,
                                                   "history_report_path", "")
                     self._list_retry_case(history_report_path)
                     return
@@ -401,12 +403,12 @@ class Console(object):
                 SuiteReporter.clear_failed_case_list()
                 SuiteReporter.clear_report_result()
 
-            command = options.action
+            command = argument.options.action
             if command == "":
                 LOG.info("Command is empty.")
                 return
 
-            self._process_command(command, options, para_list, parser)
+            self._process_command(command, argument.options, para_list, argument.parser)
         except (ParamError, ValueError, TypeError, SyntaxError,
                 AttributeError) as exception:
             error_no = getattr(exception, "error_no", "00000")
@@ -456,18 +458,18 @@ class Console(object):
                 split_list = split_list[:pos] + split_list[pos+2:]
                 history_command = " ".join(split_list)
 
-        (options, _, _, _) = self.argument_parser(history_command.split())
-        options.dry_run = is_dry_run
-        setattr(options, "history_report_path", history_report_path)
+        argument = self.argument_parser(history_command.split())
+        argument.options.dry_run = is_dry_run
+        setattr(argument.options, "history_report_path", history_report_path)
         # modify history_command -rp param and -sn param
         for option_tuple in self._get_to_be_replaced_option(parser):
             history_command = self._replace_history_option(
-                history_command, (input_options, options), option_tuple)
+                history_command, (input_options, argument.options), option_tuple)
 
         # add history command to Scheduler.command_queue
         LOG.info("Retry command: %s", history_command)
         Scheduler.command_queue[-1] = history_command
-        return options
+        return argument.options
 
     @classmethod
     def _process_command_help(cls, parser, para_list):
@@ -576,8 +578,9 @@ class Console(object):
         if not params:
             raise ParamError("no retry case exists")
         session_id, command, report_path, failed_list = \
-            params[0], params[1], params[2], \
-            [(module, failed) for module, case_list in params[3].items()
+            params[ReportConst.session_id], params[ReportConst.command], \
+            params[ReportConst.report_path], \
+            [(module, failed) for module, case_list in params[ReportConst.unsuccessful_params].items()
              for failed in case_list]
         if Scheduler.mode == ModeType.decc:
             from xdevice import SuiteReporter
@@ -661,7 +664,7 @@ class Console(object):
                     options.session else "'%s' has no command executed" % \
                                          options.session
                 raise ParamError(error_msg)
-            history_command, history_report_path = params[1], params[2]
+            history_command, history_report_path = params[ReportConst.command], params[ReportConst.report_path]
         else:
             history_command, history_report_path = "", ""
             for command_tuple in Scheduler.command_queue[:-1]:

src/xdevice/_core/constants.py

@@ -279,6 +279,15 @@ class ConfigConst(object):
     device_log = "device_log"
 
 
+@dataclass
+class ReportConst(object):
+    session_id = "session_id"
+    command = "command"
+    report_path = "report_path"
+    unsuccessful_params = "unsuccessful_params"
+    data_reports = "data_reports"
+
+
 class FilePermission(object):
     mode_777 = 0o777
     mode_755 = 0o755

src/xdevice/_core/executor/concurrent.py

@@ -26,6 +26,7 @@ from concurrent.futures import wait
 
 from _core.constants import ModeType
 from _core.constants import ConfigConst
+from _core.constants import ReportConst
 from _core.executor.request import Request
 from _core.logger import platform_logger
 from _core.plugin import Config
@@ -269,11 +270,11 @@ class DriversThread(threading.Thread):
                 for i in Handler.DAV.case_id_list:
                     failed_list.append(i + "#" + i)
             else:
-                failed_list = params[3].get(module_name, [])
+                failed_list = params[ReportConst.unsuccessful_params].get(module_name, [])
         except:
-            failed_list = params[3].get(module_name, [])
+            failed_list = params[ReportConst.unsuccessful_params].get(module_name, [])
         if not failed_list:
-            failed_list = params[3].get(str(module_name).split(".")[0], [])
+            failed_list = params[ReportConst.unsuccessful_params].get(str(module_name).split(".")[0], [])
         unpassed_test_params.extend(failed_list)
         LOG.debug("Get unpassed test params %s", unpassed_test_params)
         return unpassed_test_params
@@ -432,7 +433,7 @@ class DriversThread(threading.Thread):
             from _core.report.result_reporter import ResultReporter
             params = ResultReporter.get_task_info_params(history_report_path)
             if params:
-                report_data_dict = dict(params[4])
+                report_data_dict = dict(params[ReportConst.report_path])
                 if execute_result_name in report_data_dict.keys():
                     return report_data_dict.get(execute_result_name)
                 elif execute_result_name.split(".")[0] in \

src/xdevice/_core/executor/scheduler.py

@@ -57,6 +57,7 @@ from _core.constants import DeviceLabelType
 from _core.constants import SchedulerType
 from _core.constants import ListenerType
 from _core.constants import ConfigConst
+from _core.constants import ReportConst
 from _core.constants import HostDrivenTestType
 from _core.executor.concurrent import DriversThread
 from _core.executor.concurrent import QueueMonitorThread
@@ -1149,10 +1150,10 @@ class Scheduler(object):
             history_report_path = \
                 getattr(task.config, ConfigConst.history_report_path, "")
             params = ResultReporter.get_task_info_params(history_report_path)
-            if params and params[3]:
-                if dict(params[3]).get(module_name, []):
+            if params and params[ReportConst.unsuccessful_params]:
+                if dict(params[ReportConst.unsuccessful_params]).get(module_name, []):
                     failed_flag = True
-                elif dict(params[3]).get(str(module_name).split(".")[0], []):
+                elif dict(params[ReportConst.unsuccessful_params]).get(str(module_name).split(".")[0], []):
                     failed_flag = True
         return failed_flag
 

src/xdevice/_core/logger.py

@@ -400,7 +400,7 @@ class EncryptFileHandler(RotatingFileHandler):
             stream = getattr(self, "stream", self._open())
             stream.write(msg)
             self.flush()
-        except RecursionError as _:
+        except RecursionError as _:  # pylint:disable=undefined-variable
             raise
 
     def _encrypt_valid(self):

src/xdevice/_core/report/result_reporter.py

@@ -620,8 +620,7 @@ class ResultReporter(IReporter):
             LOG.error("%s error!", ReportConstant.task_info_record)
             return ()
 
-        return result["session_id"], result["command"], result["report_path"],\
-               result["unsuccessful_params"], result["data_reports"]
+        return result
 
     @classmethod
     def set_summary_report_result(cls, summary_data_path, result_xml):

src/xdevice/_core/utils.py

@@ -95,7 +95,7 @@ def stop_standing_subprocess(process):
         signal_value = signal.SIGINT if sys_type == "Windows" \
             else signal.SIGTERM
         os.kill(process.pid, signal_value)
-    except (PermissionError, AttributeError, FileNotFoundError,
+    except (PermissionError, AttributeError, FileNotFoundError,  # pylint:disable=undefined-variable
             SystemError) as error:
         LOG.error("Stop standing subprocess error '%s'" % error)
 
@@ -203,7 +203,7 @@ def exec_cmd(cmd, timeout=5 * 60, error_print=True, join_result=False, redirect=
         else:
             return err if err else out
 
-    except (TimeoutError, KeyboardInterrupt, AttributeError, ValueError,
+    except (TimeoutError, KeyboardInterrupt, AttributeError, ValueError,  # pylint:disable=undefined-variable
             EOFError, IOError) as _:
         sys_type = platform.system()
         if sys_type == "Linux" or sys_type == "Darwin":