tests: fix pylint errors

Signed-off-by: alindima <[email protected]>

Author: alindima

tests/conftest.py

@@ -86,9 +86,9 @@ def test_with_any_microvm(test_microvm_any):
 
 import host_tools.cargo_build as build_tools
 import host_tools.network as net_tools
-import host_tools.proc as proc
-import framework.utils as utils
-import framework.defs as defs
+from host_tools import proc
+from framework import utils
+from framework import defs
 from framework.artifacts import ArtifactCollection
 from framework.microvm import Microvm
 from framework.s3fetcher import MicrovmImageS3Fetcher
@@ -144,7 +144,7 @@ def __init__(self, test_name: str, append=True):
         # Create the root directory, if it doesn't exist.
         self._root_path.mkdir(exist_ok=True)
 
-        self._file = open(self._root_path / test_name, flags)
+        self._file = open(self._root_path / test_name, flags, encoding='utf-8')
 
     def writeln(self, data: str):
         """Write the `data` string to the output file, appending a newline."""

tests/framework/builder.py

@@ -13,7 +13,7 @@
     ArtifactCollection, Artifact, DiskArtifact, Snapshot,
     SnapshotType, NetIfaceConfig
 )
-import framework.utils as utils
+from framework import utils
 import host_tools.logging as log_tools
 import host_tools.network as net_tools
 
@@ -129,7 +129,8 @@ def build(self,
             )
             assert vm.api_session.is_status_no_content(response.status_code)
 
-        with open(config.local_path()) as microvm_config_file:
+        with open(config.local_path(), encoding='utf-8') as \
+                microvm_config_file:
             microvm_config = json.load(microvm_config_file)
 
         response = vm.basic_config(

tests/framework/decorators.py

@@ -15,8 +15,8 @@ class ApiTimeoutException(Exception):
         def __init__(self, duration, method, resource, payload):
             """Compose the error message from the API call components."""
             super().__init__(
-                'API call exceeded maximum duration: {:.2f} ms.\n'
-                'Call: {} {} {}'.format(duration, method, resource, payload)
+                f'API call exceeded maximum duration: {float(duration)} ms.\n'
+                f'Call: {method} {resource} {payload}'
             )
 
     def timed(*args, **kwargs):

tests/framework/gitlint_rules.py

@@ -4,6 +4,9 @@
 
 from gitlint.rules import CommitRule, RuleViolation
 
+# Too few public methods (1/2) (too-few-public-methods)
+# pylint: disable=R0903
+
 
 class SignedOffBy(CommitRule):
     """Make sure that each commit contains a "Signed-off-by" line."""

tests/framework/jailer.py

@@ -7,8 +7,8 @@
 import stat
 from pathlib import Path
 from retry.api import retry_call
-import framework.utils as utils
-import framework.defs as defs
+from framework import utils
+from framework import defs
 from framework.defs import FC_BINARY_NAME
 
 # Default name for the socket used for API calls.

tests/framework/microvm.py

@@ -28,7 +28,7 @@
 import host_tools.memory as mem_tools
 import host_tools.network as net_tools
 
-import framework.utils as utils
+from framework import utils
 from framework.defs import MICROVM_KERNEL_RELPATH, MICROVM_FSFILES_RELPATH, \
     FC_PID_FILE_NAME
 from framework.http import Session
@@ -337,7 +337,7 @@ def pid_in_new_ns(self):
         pid_file_path = f"{self.jailer.chroot_path()}/{FC_PID_FILE_NAME}"
         if os.path.exists(pid_file_path):
             # Read the PID stored inside the file.
-            with open(pid_file_path) as file:
+            with open(pid_file_path, encoding='utf-8') as file:
                 fc_pid = int(file.readline())
 
         return fc_pid
@@ -591,8 +591,8 @@ def spawn(self, create_logger=True,
             self._screen_pid = screen_pid
 
             self.jailer_clone_pid = int(open('/proc/{0}/task/{0}/children'
-                                             .format(screen_pid)
-                                             ).read().strip())
+                                             .format(screen_pid),
+                                             encoding='utf-8').read().strip())
 
             # Configure screen to flush stdout to file.
             flush_cmd = 'screen -S {session} -X colon "logfile flush 0^M"'
@@ -883,7 +883,7 @@ def start_console_logger(self, log_fifo):
         """
         def monitor_fd(microvm, path):
             try:
-                fd = open(path, "r")
+                fd = open(path, "r", encoding='utf-8')
                 while True:
                     try:
                         if microvm().logging_thread.stopped():
@@ -936,7 +936,8 @@ def open(self):
             # serial already opened
             return
 
-        screen_log_fd = os.open(self._vm.screen_log, os.O_RDONLY)
+        screen_log_fd = os.open(self._vm.screen_log,
+                                os.O_RDONLY)
         self._poller = select.poll()
         self._poller.register(screen_log_fd,
                               select.POLLIN | select.POLLHUP)

tests/framework/report.py

@@ -52,7 +52,7 @@ def test_net_device():
 import subprocess
 from collections import namedtuple
 from pathlib import Path
-from . import mpsing   # pylint: disable=relative-beyond-top-level
+from framework import mpsing
 
 # Try to see if we're in a git repo. If yes, set the commit ID
 COMMIT_ID = ""
@@ -168,8 +168,8 @@ def parse_data(test):
 
             # Create a dict with default values
             found_data = {
-                key: Report.doc_items[key].value
-                for key in Report.doc_items}
+                key: value.value
+                for (key, value) in Report.doc_items.items()}
             found_data["name"] = test.nodeid
 
             # Handle None docstrings
@@ -205,7 +205,7 @@ def parse_data(test):
                         f"{crt_doc_item.one_of}, not {item_value}")
 
                 # Check if the item was found twice
-                if crt_item in docstring_items.keys():
+                if crt_item in docstring_items:
                     raise ValueError(f"Item {crt_item} specified twice.")
 
                 docstring_items[crt_item] = item_value
@@ -267,7 +267,8 @@ def write_report(self):
         self._data_loc.mkdir(exist_ok=True, parents=True)
 
         # Dump the JSON file
-        with open(self._data_loc / Report.FNAME_JSON, "w") as json_file:
+        with open(self._data_loc / Report.FNAME_JSON, "w", encoding='utf-8') \
+                as json_file:
             total_duration = 0
             test_items = []
             for test_name in sorted(self._collected_items):

tests/framework/scheduler.py

@@ -33,9 +33,7 @@
 from _pytest.mark import Expression, MarkMatcher
 from _pytest.main import ExitCode
 
-from . import defs  # pylint: disable=relative-beyond-top-level
-from . import mpsing  # pylint: disable=relative-beyond-top-level
-from . import report as treport  # pylint: disable=relative-beyond-top-level
+from framework import defs, mpsing, report as treport
 
 
 class PytestScheduler(mpsing.MultiprocessSingleton):
@@ -318,7 +316,7 @@ def _worker_main(self, items, startup_delay=0):
         for item, nextitem in zip(
                 self.session.items,
                 self.session.items[1:] + [None]
-                ):
+        ):
             item.ihook.pytest_runtest_protocol(item=item, nextitem=nextitem)
 
     @mpsing.ipcmethod

tests/framework/stats/__init__.py

@@ -137,7 +137,7 @@
 # The baseline provider is a requirement for the `DictProvider`.
 class RandintBaselineProvider(BaselineProvider):
     def __init__(self, exercise_id, env_id):
-        super().__init__(DictQuery(dict()))
+        super().__init__(DictQuery({}))
         if "baselines" in CONFIG:
             super().__init__(DictQuery(CONFIG["baselines"][exercise_id]))
         self._tag = "{}/" + env_id + "/{}"
@@ -184,9 +184,9 @@ def baseline(ms_name: str, st_name: str, exercise_id: str):
     }
 
 def measurements(exercise_id: str):
-    ms_list = list()
+    ms_list  = []
     for ms_name in CONFIG["measurements"][exercise_id]:
-        st_list = list()
+        st_list  = []
         unit = CONFIG["measurements"][exercise_id][ms_name]["unit"]
         st_defs = CONFIG["measurements"][exercise_id][ms_name]["statistics"]
         for st_def in st_defs:

tests/framework/stats/consumer.py

@@ -37,15 +37,15 @@ def __init__(self,
         """Initialize a consumer."""
         self._iteration = 0
         self._results = defaultdict()  # Aggregated results.
-        self._custom = dict() if not custom else custom
+        self._custom = {} if not custom else custom
         self._metadata_provider = metadata_provider
 
-        self._measurements_defs = dict()
+        self._measurements_defs = {}
         if metadata_provider:
             self._measurements_defs = metadata_provider.measurements
 
         # Final statistics.
-        self._statistics = dict()
+        self._statistics = {}
 
         self._failure_aggregator = ProcessingException()
 
@@ -57,24 +57,24 @@ def consume_data(self, ms_name: str, value: Number):
         """Aggregate measurement."""
         results = self._results.get(ms_name)
         if not results:
-            self._results[ms_name] = dict()
-            self._results[ms_name][self.DATA_KEY] = list()
+            self._results[ms_name] = {}
+            self._results[ms_name][self.DATA_KEY] = []
         self._results[ms_name][self.DATA_KEY].append(value)
 
     def consume_stat(self, st_name: str, ms_name: str, value: Number):
         """Aggregate statistics."""
         results = self._results.get(ms_name)
         if not results:
-            self._results[ms_name] = dict()
+            self._results[ms_name] = {}
         self._results[ms_name][st_name] = value
 
     def consume_custom(self, name: str, value: Any):
         """Aggregate custom information."""
         if not self._custom.get(self._iteration):
-            self._custom[self._iteration] = dict()
+            self._custom[self._iteration] = {}
 
         if not self._custom[self._iteration].get(name):
-            self._custom[self._iteration][name] = list()
+            self._custom[self._iteration][name] = []
 
         self._custom[self._iteration][name].append(value)
 
@@ -120,7 +120,7 @@ def process(self, fail_fast=False) -> (dict, dict):
                         self._statistics[ms_name][st_def.name] = {
                             "value": st_def.func(self._results[ms_name][
                                                  self.DATA_KEY])
-                        }
+                    }
                 else:
                     self._statistics[ms_name][st_def.name] = {
                         "value": self._results[ms_name][st_def.name]
@@ -131,7 +131,7 @@ def process(self, fail_fast=False) -> (dict, dict):
                     self._statistics[ms_name][st_def.name][
                         "pass_criteria"] = {
                             pass_criteria.name: pass_criteria.baseline
-                        }
+                    }
                     res = self._statistics[ms_name][st_def.name]["value"]
                     try:
                         pass_criteria.check(res)

tests/framework/stats/metadata.py

@@ -113,7 +113,7 @@ def __init__(self,
         """
         super().__init__(baseline_provider)
 
-        self._measurements = dict()
+        self._measurements = {}
         for ms_name in measurements:
             assert DictProvider.UNIT_KEY in measurements[ms_name], \
                 f"'{DictProvider.UNIT_KEY}' field is required for '" \
@@ -125,7 +125,7 @@ def __init__(self,
             unit = measurements[ms_name][DictProvider.UNIT_KEY]
             st_defs = measurements[ms_name][DictProvider.STATISTICS_KEY]
 
-            st_list = list()
+            st_list = []
             for st_def in st_defs:
                 # Mandatory.
                 func_cls_name = st_def.get("function")

tests/framework/stats/types.py

@@ -41,7 +41,7 @@ def create_measurement(cls,
         else:
             pass_criteria = defaultdict(None, pass_criteria)
 
-        stats = list()
+        stats = []
         for func in st_functions:
             stats.append(
                 StatisticDef(

tests/framework/utils.py

@@ -55,7 +55,7 @@ def set_cpu_affinity(pid: int, cpulist: list) -> list:
     def get_cpu_percent(pid: int) -> float:
         """Return the instant process CPU utilization percent."""
         _, stdout, _ = run_cmd(GET_CPU_LOAD.format(pid))
-        cpu_percentages = dict()
+        cpu_percentages = {}
 
         # Take all except the last line
         lines = stdout.strip().split(sep="\n")
@@ -72,7 +72,7 @@ def get_cpu_percent(pid: int) -> float:
             # Handles `fc_vcpu 0` case as well.
             thread_name = info[11] + (" " + info[12] if info_len > 12 else "")
             if thread_name not in cpu_percentages:
-                cpu_percentages[thread_name] = dict()
+                cpu_percentages[thread_name] = {}
             cpu_percentages[thread_name][task_id] = cpu_percent
 
         return cpu_percentages
@@ -89,14 +89,14 @@ class CpuMap:
     starting from 0.
     """
 
-    arr = list()
+    arr = []
 
-    def __new__(cls, x):
+    def __new__(cls, cpu):
         """Instantiate the class field."""
-        assert CpuMap.len() > x
+        assert CpuMap.len() > cpu
         if not CpuMap.arr:
             CpuMap.arr = CpuMap._cpus()
-        return CpuMap.arr[x]
+        return CpuMap.arr[cpu]
 
     @staticmethod
     def len():
@@ -193,8 +193,8 @@ def with_arg(self, flag, value=""):
     def build(self):
         """Build the command."""
         cmd = self._bin_path + " "
-        for flag in self._args:
-            cmd += flag + " " + "{}".format(self._args[flag]) + " "
+        for (flag, value) in self._args.items():
+            cmd += f"{flag} {value} "
         return cmd
 
 
@@ -240,9 +240,9 @@ class DictQuery:
     1
     """
 
-    def __init__(self, d: dict):
+    def __init__(self, inner: dict):
         """Initialize the dict query."""
-        self._inner = d
+        self._inner = inner
 
     def get(self, keys_path: str, default=None):
         """Retrieve value corresponding to the key path."""
@@ -270,7 +270,7 @@ class ExceptionAggregator(Exception):
     def __init__(self, add_newline=False):
         """Initialize the exception aggregator."""
         super().__init__()
-        self.failures = list()
+        self.failures = []
 
         # If `add_newline` is True then the failures will start one row below,
         # in the logs. This is useful for having the failures starting on an
@@ -545,19 +545,19 @@ def get_cpu_percent(pid: int, iterations: int, omit: int) -> dict:
     """
     assert iterations > 0
     time.sleep(omit)
-    cpu_percentages = dict()
+    cpu_percentages = {}
     for _ in range(iterations):
         current_cpu_percentages = ProcessManager.get_cpu_percent(pid)
         assert len(current_cpu_percentages) > 0
 
-        for thread_name in current_cpu_percentages:
+        for (thread_name, task_ids) in current_cpu_percentages.items():
             if not cpu_percentages.get(thread_name):
-                cpu_percentages[thread_name] = dict()
-            for task_id in current_cpu_percentages[thread_name]:
+                cpu_percentages[thread_name] = {}
+            for task_id in task_ids:
                 if not cpu_percentages[thread_name].get(task_id):
-                    cpu_percentages[thread_name][task_id] = list()
+                    cpu_percentages[thread_name][task_id] = []
                 cpu_percentages[thread_name][task_id].append(
-                    current_cpu_percentages[thread_name][task_id])
+                    task_ids[task_id])
         time.sleep(1)  # 1 second granularity.
     return cpu_percentages