ci: unify auto-merge across bots

.github/workflows/dependabot-auto-merge.yml → .github/workflows/auto-merge.yml

@@ -1,4 +1,4 @@
-name: 'Linuxfabrik: Dependabot auto-merge'
+name: 'Linuxfabrik: Auto-merge'
 
 on:
   pull_request: {}
@@ -17,7 +17,7 @@ env:
   FROZEN_LOCKFILES: '["pip:/lockfiles/py39"]'
 
 jobs:
-  auto-merge:
+  dependabot:
     runs-on: 'ubuntu-latest'
     if: 'github.actor == ''dependabot[bot]'''
     permissions:
@@ -63,3 +63,24 @@ jobs:
         env:
           GH_TOKEN: '${{ secrets.GITHUB_TOKEN }}'
           PR_URL: '${{ github.event.pull_request.html_url }}'
+
+  # The weekly hook bump carries nothing but `rev:` changes in
+  # .pre-commit-config.yaml, and it arrives in every repository at once.
+  # Merging that by hand is pure overhead, so it goes in as soon as the
+  # required checks pass. The same fallback as above applies, and it is only
+  # safe because the ruleset keeps enforcing those checks server-side.
+  pre-commit-autoupdate:
+    runs-on: 'ubuntu-latest'
+    if: >-
+      github.actor == 'linuxfabrik-automation[bot]'
+      && github.head_ref == 'chore/pre-commit-autoupdate'
+    permissions:
+      contents: 'write'
+      pull-requests: 'write'
+    steps:
+
+      - run: |
+          gh pr merge --auto --squash "$PR_URL" || gh pr merge --squash "$PR_URL"
+        env:
+          GH_TOKEN: '${{ secrets.GITHUB_TOKEN }}'
+          PR_URL: '${{ github.event.pull_request.html_url }}'

.gitignore

@@ -126,4 +126,3 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
-

LICENSE

@@ -21,4 +21,4 @@ OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
 ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
 OTHER DEALINGS IN THE SOFTWARE.
 
-For more information, please refer to <http://unlicense.org/>
+For more information, please refer to <http://unlicense.org/>

huawei_dorado.py

@@ -84,11 +84,7 @@ def _redact(value):
     """
     if isinstance(value, dict):
         return {
-            key: (
-                '******'
-                if str(key).lower() in _REDACTED_FIELDS
-                else _redact(inner)
-            )
+            key: ('******' if str(key).lower() in _REDACTED_FIELDS else _redact(inner))
             for key, inner in value.items()
         }
     if isinstance(value, list):
@@ -1870,6 +1866,7 @@ def get_performance(uuid, data_ids, args):
     385: ('avg_write_io_response_time', 's', 1 / 1_000_000),
 }
 
+
 def get_performance_perfdata(prefix, samples, indicators=None):
     """
     Turn the samples `get_performance()` returned into performance data.

huawei_pacific.py

@@ -63,11 +63,7 @@ def _redact(value):
     """
     if isinstance(value, dict):
         return {
-            key: (
-                '******'
-                if str(key).lower() in _REDACTED_FIELDS
-                else _redact(inner)
-            )
+            key: ('******' if str(key).lower() in _REDACTED_FIELDS else _redact(inner))
             for key, inner in value.items()
         }
     if isinstance(value, list):
@@ -1317,7 +1313,10 @@ def get_performance(object_type, indicators, args, ids=None, window=PERFORMANCE_
     {'1': {'68': '17.0', '69': '45.0'}, '2': {'68': '9.0', '69': '37.0'}}
     """
     end_time = time.now()
-    query = {'object_type': str(object_type), 'indicators': [str(i) for i in indicators]}
+    query = {
+        'object_type': str(object_type),
+        'indicators': [str(i) for i in indicators],
+    }
     if ids is not None:
         query['ids'] = list(ids)
 

ssh.py

@@ -56,6 +56,7 @@ def _with_password(cmd, password):
         return cmd, None
     return ['sshpass', '-e', *cmd], {'SSHPASS': password}
 
+
 def build_options(
     configfile=None,
     identity=None,

tests/args/unit-test/run

@@ -120,7 +120,9 @@ class TestLoadSecret(unittest.TestCase):
         self.assertEqual(args.load_secret(self._write('linuxfabrik\n')), 'linuxfabrik')
 
     def test_strips_a_windows_line_ending(self):
-        self.assertEqual(args.load_secret(self._write('linuxfabrik\r\n')), 'linuxfabrik')
+        self.assertEqual(
+            args.load_secret(self._write('linuxfabrik\r\n')), 'linuxfabrik'
+        )
 
     def test_only_the_first_line_is_the_secret(self):
         self.assertEqual(
@@ -130,7 +132,9 @@ class TestLoadSecret(unittest.TestCase):
 
     def test_surrounding_spaces_are_part_of_the_secret(self):
         # Stripping them would make a valid password fail with no way to tell why.
-        self.assertEqual(args.load_secret(self._write('  linuxfabrik  \n')), '  linuxfabrik  ')
+        self.assertEqual(
+            args.load_secret(self._write('  linuxfabrik  \n')), '  linuxfabrik  '
+        )
 
     def test_empty_file_aborts(self):
         # An empty secret would be sent to the remote end as an empty password, which

tests/disk/unit-test/run

@@ -242,9 +242,7 @@ class TestFileHelpers(unittest.TestCase):
     def test_walk_directory_relative_paths_keep_their_subdirectory(self):
         disk.mkdir(os.path.join(self.dir, 'sub'))
         disk.write_file(os.path.join(self.dir, 'sub', 'y.txt'), 'y')
-        self.assertEqual(
-            disk.walk_directory(os.path.join(self.dir, 'sub')), ['y.txt']
-        )
+        self.assertEqual(disk.walk_directory(os.path.join(self.dir, 'sub')), ['y.txt'])
         self.assertEqual(disk.walk_directory(self.dir), [os.path.join('sub', 'y.txt')])
 
     def test_walk_directory_relative_paths_tolerate_a_trailing_separator(self):
@@ -257,9 +255,7 @@ class TestFileHelpers(unittest.TestCase):
         root = os.path.join(self.dir, 'wp')
         disk.mkdir(os.path.join(root, 'a', 'wp'))
         disk.write_file(os.path.join(root, 'a', 'wp', 'b.txt'), 'b')
-        self.assertEqual(
-            disk.walk_directory(root), [os.path.join('a', 'wp', 'b.txt')]
-        )
+        self.assertEqual(disk.walk_directory(root), [os.path.join('a', 'wp', 'b.txt')])
 
     def test_walk_directory_absolute_paths(self):
         disk.write_file(self.file, 'x')
@@ -346,8 +342,10 @@ class TestHostHelpers(unittest.TestCase):
 class TestGetOwningPackage(unittest.TestCase):
     def test_rpm_answer_is_the_package_name(self):
         answer = (True, ('icinga-director-php-1.11.9-1.fc43.noarch\n', '', 0))
-        with mock.patch.object(disk.shell, 'which', side_effect=lambda n: n == 'rpm'), \
-                mock.patch.object(disk.shell, 'shell_exec', return_value=answer) as m:
+        with (
+            mock.patch.object(disk.shell, 'which', side_effect=lambda n: n == 'rpm'),
+            mock.patch.object(disk.shell, 'shell_exec', return_value=answer) as m,
+        ):
             self.assertEqual(
                 disk.get_package('/usr/share/icingaweb2/modules/director'),
                 'icinga-director-php-1.11.9-1.fc43.noarch',
@@ -358,9 +356,14 @@ class TestGetOwningPackage(unittest.TestCase):
         )
 
     def test_dpkg_answer_is_split_off_its_path(self):
-        answer = (True, ('icinga-director: /usr/share/icingaweb2/modules/director\n', '', 0))
-        with mock.patch.object(disk.shell, 'which', side_effect=lambda n: n == 'dpkg'), \
-                mock.patch.object(disk.shell, 'shell_exec', return_value=answer):
+        answer = (
+            True,
+            ('icinga-director: /usr/share/icingaweb2/modules/director\n', '', 0),
+        )
+        with (
+            mock.patch.object(disk.shell, 'which', side_effect=lambda n: n == 'dpkg'),
+            mock.patch.object(disk.shell, 'shell_exec', return_value=answer),
+        ):
             self.assertEqual(
                 disk.get_package('/usr/share/icingaweb2/modules/director'),
                 'icinga-director',
@@ -369,8 +372,10 @@ class TestGetOwningPackage(unittest.TestCase):
     def test_unowned_path_yields_no_package(self):
         # both package managers answer an unowned path with a non-zero return code
         answer = (True, ('file /opt/thing is not owned by any package\n', '', 1))
-        with mock.patch.object(disk.shell, 'which', side_effect=lambda n: n == 'rpm'), \
-                mock.patch.object(disk.shell, 'shell_exec', return_value=answer):
+        with (
+            mock.patch.object(disk.shell, 'which', side_effect=lambda n: n == 'rpm'),
+            mock.patch.object(disk.shell, 'shell_exec', return_value=answer),
+        ):
             self.assertEqual(disk.get_package('/opt/thing'), '')
 
     def test_host_without_a_package_manager_yields_no_package(self):

tests/huawei_dorado/unit-test/run

@@ -371,7 +371,9 @@ class TestRunningStatusState(unittest.TestCase):
     def test_same_code_differs_per_object(self):
         # Charging is the resting state of a backup power module and nonsense
         # anywhere else, which is the whole reason ok_codes is a parameter.
-        self.assertEqual(huawei.get_running_status_state(48, (1, 2, 27, 48, 49)), STATE_OK)
+        self.assertEqual(
+            huawei.get_running_status_state(48, (1, 2, 27, 48, 49)), STATE_OK
+        )
         self.assertEqual(huawei.get_running_status_state(48, (1, 27)), STATE_WARN)
 
     def test_failed_codes_are_critical(self):
@@ -427,10 +429,14 @@ class TestHypermetroDomainRunningStatusState(unittest.TestCase):
         # Read through the shared mapper code 1 would be 'Normal' and come out
         # OK; on a domain it means the mirror is still recovering.
         self.assertEqual(huawei.get_running_status_state(1, (1,)), STATE_OK)
-        self.assertEqual(huawei.get_hypermetro_domain_running_status_state(1), STATE_WARN)
+        self.assertEqual(
+            huawei.get_hypermetro_domain_running_status_state(1), STATE_WARN
+        )
 
     def test_missing_value_warns(self):
-        self.assertEqual(huawei.get_hypermetro_domain_running_status_state(None), STATE_WARN)
+        self.assertEqual(
+            huawei.get_hypermetro_domain_running_status_state(None), STATE_WARN
+        )
 
 
 # ---------------------------------------------------------------------------
@@ -488,7 +494,9 @@ class TestGetUuid(unittest.TestCase):
 
 class TestAssertOk(unittest.TestCase):
     def test_success_returns(self):
-        self.assertIsNone(huawei.assert_ok({'error': {'code': 0}, 'data': []}, 'the fans'))
+        self.assertIsNone(
+            huawei.assert_ok({'error': {'code': 0}, 'data': []}, 'the fans')
+        )
 
     def test_success_as_a_string_returns(self):
         # Some endpoints report the code as a string. Treating that as an error would take
@@ -501,7 +509,10 @@ class TestAssertOk(unittest.TestCase):
         self.assertEqual(ctx.exception.code, STATE_UNKNOWN)
 
     def test_error_text_names_the_query_and_the_appliance_reason(self):
-        with self.assertRaises(SystemExit), contextlib.redirect_stdout(io.StringIO()) as out:
+        with (
+            self.assertRaises(SystemExit),
+            contextlib.redirect_stdout(io.StringIO()) as out,
+        ):
             huawei.assert_ok(
                 {
                     'error': {
@@ -647,7 +658,9 @@ class TestField(unittest.TestCase):
     def test_matches_a_different_case(self):
         # The sfp object is documented in camelCase and answered in upper case on the
         # same page of the REST Interface Reference.
-        self.assertEqual(huawei.field({'RXPOWER': '[12]'}, 'rxPowerReal', 'RXPOWER'), '[12]')
+        self.assertEqual(
+            huawei.field({'RXPOWER': '[12]'}, 'rxPowerReal', 'RXPOWER'), '[12]'
+        )
         self.assertEqual(huawei.field({'healthStatus': '1'}, 'HEALTHSTATUS'), '1')
 
     def test_exact_match_wins_over_a_case_insensitive_one(self):
@@ -1250,7 +1263,10 @@ class TestGetAllData(HuaweiHttpTestBase):
         self.assertNotIn('ID', str(result.get('data', '')))
 
     def test_single_object_response_is_not_paged(self):
-        self.fetch_json.return_value = (True, {'data': {'ID': '0'}, 'error': {'code': 0}})
+        self.fetch_json.return_value = (
+            True,
+            {'data': {'ID': '0'}, 'error': {'code': 0}},
+        )
         result, truncated = huawei.get_all_data('system/', self.args)
         self.assertEqual(result['data'], {'ID': '0'})
         self.assertFalse(truncated)

tests/huawei_pacific/unit-test/run

@@ -358,7 +358,9 @@ class TestComponentStatusState(unittest.TestCase):
     def test_value_is_normalised(self):
         # A firmware that capitalises the value differently must not turn a
         # healthy fan into a warning.
-        self.assertEqual(huawei_pacific.get_component_status_state(' Normal '), STATE_OK)
+        self.assertEqual(
+            huawei_pacific.get_component_status_state(' Normal '), STATE_OK
+        )
         self.assertEqual(huawei_pacific.get_component_status_state('FAULT'), STATE_CRIT)
 
     def test_absent_is_ok(self):
@@ -444,17 +446,13 @@ class TestPoolStatusState(unittest.TestCase):
     def test_faulty_and_stopped_are_critical(self):
         for code in (1, 3, 4):
             with self.subTest(code=code):
-                self.assertEqual(
-                    huawei_pacific.get_pool_status_state(code), STATE_CRIT
-                )
+                self.assertEqual(huawei_pacific.get_pool_status_state(code), STATE_CRIT)
 
     def test_recoverable_states_warn(self):
         # The pool still serves reads, and the last three work themselves out.
         for code in (2, 5, 7, 8):
             with self.subTest(code=code):
-                self.assertEqual(
-                    huawei_pacific.get_pool_status_state(code), STATE_WARN
-                )
+                self.assertEqual(huawei_pacific.get_pool_status_state(code), STATE_WARN)
 
     def test_unmapped_code_warns(self):
         self.assertEqual(huawei_pacific.get_pool_status_state(6), STATE_WARN)
@@ -470,7 +468,9 @@ class TestAssertOk(unittest.TestCase):
         )
 
     def test_success_as_a_string_returns(self):
-        self.assertIsNone(huawei_pacific.assert_ok({'result': {'code': '0'}}, 'the fans'))
+        self.assertIsNone(
+            huawei_pacific.assert_ok({'result': {'code': '0'}}, 'the fans')
+        )
 
     def test_bare_result_of_the_older_endpoints_returns(self):
         # The endpoints below /dsware/service/ send the code bare instead of wrapped in an
@@ -489,7 +489,10 @@ class TestAssertOk(unittest.TestCase):
             huawei_pacific.assert_ok({'result': 1}, 'the disks')
 
     def test_error_text_names_the_query_and_the_appliance_reason(self):
-        with self.assertRaises(SystemExit), contextlib.redirect_stdout(io.StringIO()) as out:
+        with (
+            self.assertRaises(SystemExit),
+            contextlib.redirect_stdout(io.StringIO()) as out,
+        ):
             huawei_pacific.assert_ok(
                 {
                     'result': {
@@ -1083,7 +1086,11 @@ class TestGetPerformance(PacificHttpTestBase):
     def test_posts_the_documented_body(self):
         self.respond([])
         huawei_pacific.get_performance(
-            huawei_pacific.PERFORMANCE_OBJECT_NODE, (68,), self.args, ids=(1,), window=60
+            huawei_pacific.PERFORMANCE_OBJECT_NODE,
+            (68,),
+            self.args,
+            ids=(1,),
+            window=60,
         )
         payload = self.sent_kwargs()['data']
         self.assertEqual(payload['begin_time'], 1_000_000 - 60)

tests/lftest/unit-test/stderr/case

@@ -1 +1 @@
-err text
+err text

tests/lftest/unit-test/stdout/case

@@ -1 +1 @@
-hello world
+hello world

tests/lftest/unit-test/stdout/case-second-pass

@@ -1 +1 @@
-second pass
+second pass

tests/txt/unit-test/run

@@ -415,7 +415,9 @@ class TestStripAnsi(unittest.TestCase):
         self.assertEqual(txt.strip_ansi('\x1b]0;window title\x07body'), 'body')
 
     def test_osc_terminated_by_string_terminator(self):
-        self.assertEqual(txt.strip_ansi('\x1b]8;;https://example.com\x1b\\link'), 'link')
+        self.assertEqual(
+            txt.strip_ansi('\x1b]8;;https://example.com\x1b\\link'), 'link'
+        )
 
     def test_multiple_sequences_in_one_line(self):
         self.assertEqual(

tests/url/unit-test/run

@@ -1040,9 +1040,7 @@ class TestGetLatestVersionFromGithub(unittest.TestCase):
         self.assertFalse(res)
 
     def test_empty_dict_returns_false(self):
-        with mock.patch.object(
-            url, 'fetch_json', return_value=github_answer(200, {})
-        ):
+        with mock.patch.object(url, 'fetch_json', return_value=github_answer(200, {})):
             ok, res = url.get_latest_version_from_github('u', 'r')
         self.assertTrue(ok)
         self.assertFalse(res)
@@ -1101,7 +1099,9 @@ class TestGetLatestTagFromGithub(unittest.TestCase):
             'fetch_json',
             return_value=github_answer(200, list(GITHUB_TAGS_FIXTURE)),
         ) as m:
-            ok, res = url.get_latest_tag_from_github('Icinga', 'icingaweb2-theme-company')
+            ok, res = url.get_latest_tag_from_github(
+                'Icinga', 'icingaweb2-theme-company'
+            )
         self.assertTrue(ok)
         self.assertEqual(res, 'v1.0.0')
         self.assertEqual(