refactor: replace type() with isinstance() in intelmqdump.py #2636

intelmq/bin/intelmqctl.py

@@ -1115,7 +1115,7 @@ def upgrade_conf(self, previous=None, dry_run=None, function=None,
                 result['traceback'] = traceback.format_exc()
                 result['success'] = False
             else:
-                if type(retval) is str:
+                if isinstance(retval, str):
                     self._logger.error('Upgrade %r failed: %s', function, retval)
                     result['message'] = retval
                     result['success'] = False
@@ -1207,7 +1207,7 @@ def upgrade_conf(self, previous=None, dry_run=None, function=None,
                         result['traceback'] = traceback.format_exc()
                         result['success'] = False
                     else:
-                        if type(retval) is str:
+                        if isinstance(retval, str):
                             self._logger.error('%s: Upgrade failed: %s', docstring, retval)
                             result['message'] = retval
                             result['success'] = False

intelmq/bin/intelmqdump.py

@@ -131,7 +131,7 @@ def save_file(handle, content):
 def load_meta(dump):
     retval = []
     for key, value in dump.items():
-        if type(value['traceback']) is not list:
+        if not isinstance(value['traceback'], list):
             error = value['traceback'].splitlines()[-1]
         else:
             error = value['traceback'][-1].strip()
@@ -410,7 +410,7 @@ def main(argv=None):
                                     len(value['message']['raw']) > args.truncate):
                                 value['message']['raw'] = value['message'][
                                     'raw'][:args.truncate] + '...[truncated]'
-                    if type(value['traceback']) is not list:
+                    if not isinstance(value['traceback'], list):
                         value['traceback'] = value['traceback'].splitlines()
                     pprint.pprint(value)
             elif answer[0] == 'e':

intelmq/bots/experts/filter/expert.py

@@ -82,21 +82,21 @@ def process(self):
             except ValueError:
                 self.logger.error("Could not parse time.source %s.", event.get('time.source'))
             else:
-                if type(self.not_after) is datetime and event_time > self.not_after:
+                if isinstance(self.not_after, datetime) and event_time > self.not_after:
                     self.acknowledge_message()
                     self.logger.debug("Filtered out event with time.source %s.", event.get('time.source'))
                     return
-                if type(self.not_before) is datetime and event_time < self.not_before:
+                if isinstance(self.not_before, datetime) and event_time < self.not_before:
                     self.acknowledge_message()
                     self.logger.debug("Filtered out event with time.source %r.", event.get('time.source'))
                     return
 
                 now = datetime.now(tz=timezone.utc)
-                if type(self.not_after) is timedelta and event_time > (now - self.not_after):
+                if isinstance(self.not_after, timedelta) and event_time > (now - self.not_after):
                     self.acknowledge_message()
                     self.logger.debug("Filtered out event with time.source %r.", event.get('time.source'))
                     return
-                if type(self.not_before) is timedelta and event_time < (now - self.not_before):
+                if isinstance(self.not_before, timedelta) and event_time < (now - self.not_before):
                     self.acknowledge_message()
                     self.logger.debug("Filtered out event with time.source %r.", event.get('time.source'))
                     return

intelmq/bots/experts/format_field/expert.py

@@ -18,7 +18,7 @@ class FormatFieldExpertBot(ExpertBot):
     split_column    = None
 
     def init(self):
-        if type(self.strip_columns) is str:
+        if isinstance(self.strip_columns, str):
             self.strip_columns = [column.strip() for column in self.strip_columns.split(",")]
 
     def process(self):

intelmq/bots/experts/rdap/expert.py

@@ -46,15 +46,15 @@ def init(self):
 
         # get bootstrapped servers
         for service in self.rdap_bootstrapped_servers:
-            if type(self.rdap_bootstrapped_servers[service]) is str:
+            if isinstance(self.rdap_bootstrapped_servers[service], str):
                 self.__rdap_directory[service] = {"url": self.rdap_bootstrapped_servers[service]}
-            elif type(self.rdap_bootstrapped_servers) is dict:
+            elif isinstance(self.rdap_bootstrapped_servers, dict):
                 self.__rdap_directory[service] = self.rdap_bootstrapped_servers[service]
 
     def parse_entities(self, vcardArray) -> list:
         vcard = []
         for vcardentry in vcardArray:
-            if type(vcardentry) is str:
+            if isinstance(vcardentry, str):
                 continue
 
             for vcarddata in vcardentry:

intelmq/bots/parsers/generic/parser_csv.py

@@ -50,7 +50,7 @@ class GenericCsvParserBot(ParserBot):
 
     def init(self):
         # convert columns to an array
-        if type(self.columns) is str:
+        if isinstance(self.columns, str):
             self.columns = [column.strip() for column in self.columns.split(",")]
 
         if self.type_translation and isinstance(self.type_translation, str):  # not-empty string

intelmq/bots/parsers/html_table/parser.py

@@ -55,11 +55,11 @@ def init(self):
             raise MissingDependencyError("beautifulsoup4")
 
         # convert columns to an array
-        if type(self.columns) is str:
+        if isinstance(self.columns, str):
             self.columns = [column.strip() for column in self.columns.split(",")]
         if self.ignore_values is None:
             self.ignore_values = len(self.columns) * ['']
-        if type(self.ignore_values) is str:
+        if isinstance(self.ignore_values, str):
             self.ignore_values = [value.strip() for value in self.ignore_values.split(",")]
 
         if len(self.columns) != len(self.ignore_values):

intelmq/lib/bot.py

@@ -532,8 +532,7 @@ def __reset_total_path_stats(self):
         """Initially set destination paths to 0 to reset them in stats cache"""
         if not self.destination_queues:
             return
-        queues_type = type(self.destination_queues)
-        if queues_type is dict:
+        if isinstance(self.destination_queues, dict):
             for path in self.destination_queues.keys():
                 self.__message_counter["path_total"][path] = 0
         else:
@@ -1244,7 +1243,7 @@ def process(self):
                 value = self.parse_line(line, report)
                 if value is None:
                     continue
-                elif type(value) is list or isinstance(value, types.GeneratorType):
+                elif isinstance(value, list) or isinstance(value, types.GeneratorType):
                     # filter out None
                     events: list[libmessage.Event] = list(filter(bool, value))
                 else:

intelmq/lib/harmonization.py

@@ -101,7 +101,7 @@ def is_valid(value: str, sanitize: bool = False) -> bool:
         if not GenericType.is_valid(value):
             return False
 
-        if type(value) is not str:
+        if not isinstance(value, str):
             return False
 
         if len(value) == 0:

intelmq/lib/pipeline.py

@@ -116,10 +116,9 @@ def set_queues(self, queues: Optional[str], queues_type: str):
             self.internal_queue = None if queues is None else f'{queues}-internal'
 
         elif queues_type == "destination":
-            type_ = type(queues)
-            if type_ is list:
+            if isinstance(queues, list):
                 q = {"_default": queues}
-            elif type_ is str:
+            elif isinstance(queues, str):
                 q = {"_default": queues.split()}
             elif isinstance(queues, dict):
                 q = queues

intelmq/lib/test.py

@@ -171,7 +171,7 @@ def setUpClass(cls):
                                          'time.observation': '2016-01-01T00:00:00+00:00'}
         elif cls.bot_type != 'collector' and cls.default_input_message == '':
             cls.default_input_message = {'__type': 'Event'}
-        if type(cls.default_input_message) is dict:
+        if isinstance(cls.default_input_message, dict):
             cls.default_input_message = \
                 utils.decode(json.dumps(cls.default_input_message))
 
@@ -268,9 +268,9 @@ def prepare_source_queue(self):
                 self.input_message = [self.input_message]
             self.input_queue = []
             for msg in self.input_message:
-                if type(msg) is dict:
+                if isinstance(msg, dict):
                     self.input_queue.append(json.dumps(msg))
-                elif issubclass(type(msg), message.Message):
+                elif isinstance(msg, message.Message):
                     self.input_queue.append(msg.serialize())
                 else:
                     self.input_queue.append(msg)

intelmq/lib/upgrades.py

@@ -202,7 +202,7 @@ def v100_dev7_modify_syntax(configuration, harmonization, dry_run, **kwargs):
         if bot["module"] == "intelmq.bots.experts.modify.expert":
             if "configuration_path" in bot["parameters"]:
                 config = load_configuration(bot["parameters"]["configuration_path"])
-                if type(config) is dict:
+                if isinstance(config, dict):
                     new_config = modify_expert_convert_config(config)
                     if len(config) != len(new_config):
                         return 'Error converting modify expert syntax. Different size of configurations. Please report this.'
@@ -529,7 +529,7 @@ def v221_feed_changes(configuration, harmonization, dry_run, **kwargs):
                 continue
             columns = bot["parameters"]["columns"]
             # convert columns to an array
-            if type(columns) is str:
+            if isinstance(columns, str):
                 columns = [column.strip() for column in columns.split(",")]
             if columns == ULRHAUS_OLD:
                 changed = True

intelmq/lib/utils.py

@@ -199,8 +199,8 @@ def flatten_queues(queues: Union[list, dict]) -> Iterator[str]:
     Returns:
         flattened_queues: queues without dictionaries as values, just lists with the values
     """
-    return (item for sublist in (queues.values() if type(queues) is dict else queues) for item in
-            (sublist if type(sublist) is list else [sublist]))
+    return (item for sublist in (queues.values() if isinstance(queues, dict) else queues) for item in
+            (sublist if isinstance(sublist, list) else [sublist]))
 
 
 def load_configuration(configuration_filepath: str) -> dict:
@@ -395,7 +395,7 @@ def log(name: str, log_path: Union[str, bool] = intelmq.DEFAULT_LOGGING_PATH,
         handler.setLevel(log_level)
         handler.setFormatter(logging.Formatter(LOG_FORMAT))
     elif syslog:
-        if type(syslog) is tuple or type(syslog) is list:
+        if isinstance(syslog, (tuple, list)):
             handler = logging.handlers.SysLogHandler(address=tuple(syslog))
         else:
             handler = logging.handlers.SysLogHandler(address=syslog)

intelmq/tests/lib/test_message.py

@@ -90,14 +90,12 @@ def add_event_examples(self, event):
     def test_report_type(self):
         """ Test if MessageFactory returns a Report. """
         report = self.new_report()
-        self.assertEqual(type(report),
-                         message.Report)
+        self.assertIsInstance(report, message.Report)
 
     def test_event_type(self):
         """ Test if MessageFactory returns a Event. """
         event = self.new_event()
-        self.assertEqual(type(event),
-                         message.Event)
+        self.assertIsInstance(event, message.Event)
 
     def test_report_init_auto(self):
         """ Test if serialize does pass auto=True """
@@ -172,7 +170,6 @@ def test_report_invalid_key(self):
             report.add('invalid', 0)
         self.assertIn('not allowed', cm.exception.args[0])
 
-
     def test_report_add_raw(self):
         """ Test if report can add raw value. """
         report = self.new_report(auto=True)