New upstream version 1.30.2.upstream/1.30.2

1 files changed, 114 insertions, 62 deletions

diff --git a/test/test_results.py b/test/test_results.py
index 6e04e1d..4b1c4c1 100644
--- a/test/test_results.py
+++ b/test/test_results.py
@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 
-# Copyright 2015-2023 Mike Fährmann
+# Copyright 2015-2025 Mike Fährmann
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License version 2 as
@@ -80,9 +80,9 @@ class TestExtractorResults(unittest.TestCase):
     @classmethod
     def tearDownClass(cls):
         if cls._skipped:
-            print("\n\nSkipped tests:")
-            for url, exc in cls._skipped:
-                print('- {} ("{}")'.format(url, exc))
+            sys.stdout.write("\n\nSkipped tests:\n")
+            for url, reason in cls._skipped:
+                sys.stdout.write(f'- {url} ("{reason}")\n')
 
     def assertRange(self, value, range, msg=None):
         if range.step > 1:
@@ -91,6 +91,24 @@ class TestExtractorResults(unittest.TestCase):
             self.assertLessEqual(value, range.stop, msg=msg)
             self.assertGreaterEqual(value, range.start, msg=msg)
 
+    def assertLogEqual(self, expected, output):
+        if isinstance(expected, str):
+            expected = (expected,)
+        self.assertEqual(len(expected), len(output), "#log/count")
+
+        for exp, out in zip(expected, output):
+            level, name, message = out.split(":", 2)
+
+            if isinstance(exp, str):
+                return self.assertEqual(exp, message, "#log")
+
+            self.assertEqual(exp[0].lower(), level.lower(), "#log/level")
+            if len(exp) < 3:
+                self.assertEqual(exp[1], message, "#log/message")
+            else:
+                self.assertEqual(exp[1], name   , "#log/name")
+                self.assertEqual(exp[2], message, "#log/message")
+
     def _run_test(self, result):
         if result.get("#fail"):
             del result["#fail"]
@@ -145,7 +163,11 @@ class TestExtractorResults(unittest.TestCase):
             return
 
         try:
-            tjob.run()
+            if "#log" in result:
+                with self.assertLogs() as log_info:
+                    tjob.run()
+            else:
+                tjob.run()
         except exception.StopExtraction:
             pass
         except exception.HttpError as exc:
@@ -156,6 +178,9 @@ class TestExtractorResults(unittest.TestCase):
                 self.skipTest(exc)
             raise
 
+        if "#log" in result:
+            self.assertLogEqual(result["#log"], log_info.output)
+
         if result.get("#archive", True):
             self.assertEqual(
                 len(set(tjob.archive_list)),
@@ -220,13 +245,15 @@ class TestExtractorResults(unittest.TestCase):
                 for url, pat in zip(tjob.url_list, pattern):
                     self.assertRegex(url, pat, msg="#pattern")
 
-        if "#urls" in result:
-            expected = result["#urls"]
+        if "#results" in result:
+            expected = result["#results"]
             if isinstance(expected, str):
-                self.assertTrue(tjob.url_list, msg="#urls")
-                self.assertEqual(tjob.url_list[0], expected, msg="#urls")
+                self.assertTrue(tjob.url_list, msg="#results")
+                self.assertEqual(
+                    tjob.url_list[0], expected, msg="#results")
             else:
-                self.assertSequenceEqual(tjob.url_list, expected, msg="#urls")
+                self.assertSequenceEqual(
+                    tjob.url_list, expected, msg="#results")
 
         metadata = {k: v for k, v in result.items() if k[0] != "#"}
         if metadata:
@@ -235,56 +262,74 @@ class TestExtractorResults(unittest.TestCase):
 
     def _test_kwdict(self, kwdict, tests, parent=None):
         for key, test in tests.items():
+
             if key.startswith("?"):
                 key = key[1:]
                 if key not in kwdict:
                     continue
 
+            if key.endswith("[*]"):
+                key = key[:-3]
+                subtest = True
+            else:
+                subtest = False
+
             path = "{}.{}".format(parent, key) if parent else key
+
             if key.startswith("!"):
                 self.assertNotIn(key[1:], kwdict, msg=path)
                 continue
+
             self.assertIn(key, kwdict, msg=path)
             value = kwdict[key]
 
-            if isinstance(test, dict):
-                self._test_kwdict(value, test, path)
-            elif isinstance(test, type):
-                self.assertIsInstance(value, test, msg=path)
-            elif isinstance(test, range):
-                self.assertRange(value, test, msg=path)
-            elif isinstance(test, set):
-                try:
-                    self.assertIn(value, test, msg=path)
-                except AssertionError:
-                    self.assertIn(type(value), test, msg=path)
-            elif isinstance(test, list):
-                subtest = False
-                for idx, item in enumerate(test):
-                    if isinstance(item, dict):
-                        subtest = True
-                        subpath = "{}[{}]".format(path, idx)
-                        self._test_kwdict(value[idx], item, subpath)
-                if not subtest:
-                    self.assertEqual(test, value, msg=path)
-            elif isinstance(test, str):
-                if test.startswith("re:"):
-                    self.assertRegex(value, test[3:], msg=path)
-                elif test.startswith("dt:"):
-                    self.assertIsInstance(value, datetime.datetime, msg=path)
-                    self.assertEqual(test[3:], str(value), msg=path)
-                elif test.startswith("type:"):
-                    self.assertEqual(test[5:], type(value).__name__, msg=path)
-                elif test.startswith("len:"):
-                    cls, _, length = test[4:].rpartition(":")
-                    if cls:
-                        self.assertEqual(
-                            cls, type(value).__name__, msg=path + "/type")
-                    self.assertEqual(int(length), len(value), msg=path)
-                else:
-                    self.assertEqual(test, value, msg=path)
+            if subtest:
+                self.assertNotIsInstance(value, str, msg=path)
+                for idx, item in enumerate(value):
+                    subpath = "{}[{}]".format(path, idx)
+                    self._test_kwdict_value(item, test, subpath)
             else:
+                self._test_kwdict_value(value, test, path)
+
+    def _test_kwdict_value(self, value, test, path):
+        if isinstance(test, dict):
+            self._test_kwdict(value, test, path)
+        elif isinstance(test, type):
+            self.assertIsInstance(value, test, msg=path)
+        elif isinstance(test, range):
+            self.assertRange(value, test, msg=path)
+        elif isinstance(test, set):
+            try:
+                self.assertIn(value, test, msg=path)
+            except AssertionError:
+                self.assertIn(type(value), test, msg=path)
+        elif isinstance(test, list):
+            subtest = False
+            for idx, item in enumerate(test):
+                if isinstance(item, dict):
+                    subtest = True
+                    subpath = "{}[{}]".format(path, idx)
+                    self._test_kwdict(value[idx], item, subpath)
+            if not subtest:
                 self.assertEqual(test, value, msg=path)
+        elif isinstance(test, str):
+            if test.startswith("re:"):
+                self.assertRegex(value, test[3:], msg=path)
+            elif test.startswith("dt:"):
+                self.assertIsInstance(value, datetime.datetime, msg=path)
+                self.assertEqual(test[3:], str(value), msg=path)
+            elif test.startswith("type:"):
+                self.assertEqual(test[5:], type(value).__name__, msg=path)
+            elif test.startswith("len:"):
+                cls, _, length = test[4:].rpartition(":")
+                if cls:
+                    self.assertEqual(
+                        cls, type(value).__name__, msg=path + "/type")
+                self.assertEqual(int(length), len(value), msg=path)
+            else:
+                self.assertEqual(test, value, msg=path)
+        else:
+            self.assertEqual(test, value, msg=path)
 
 
 class ResultJob(job.DownloadJob):
@@ -402,27 +447,31 @@ class TestPathfmt():
 
 class TestFormatter(formatter.StringFormatter):
 
-    @staticmethod
-    def _noop(_):
-        return ""
-
     def _apply_simple(self, key, fmt):
         if key == "extension" or "_parse_optional." in repr(fmt):
-            return self._noop
-
-        def wrap(obj):
-            return fmt(obj[key])
+            def wrap(obj):
+                try:
+                    return fmt(obj[key])
+                except KeyError:
+                    return ""
+        else:
+            def wrap(obj):
+                return fmt(obj[key])
         return wrap
 
     def _apply(self, key, funcs, fmt):
         if key == "extension" or "_parse_optional." in repr(fmt):
-            return self._noop
-
-        def wrap(obj):
-            obj = obj[key]
-            for func in funcs:
-                obj = func(obj)
-            return fmt(obj)
+            def wrap(obj):
+                obj = obj[key] if key in obj else ""
+                for func in funcs:
+                    obj = func(obj)
+                return fmt(obj)
+        else:
+            def wrap(obj):
+                obj = obj[key]
+                for func in funcs:
+                    obj = func(obj)
+                return fmt(obj)
         return wrap
 
 
@@ -457,7 +506,10 @@ def generate_tests():
     """Dynamically generate extractor unittests"""
     def _generate_method(result):
         def test(self):
-            print("\n" + result["#url"])
+            sys.stdout.write(f"\n{result['#url']}\n")
+            if "#comment" in result:
+                sys.stdout.write(f"# {result['#comment']}\n")
+
             try:
                 self._run_test(result)
             except KeyboardInterrupt as exc: