fix more ruff rule exclusions; ignore BLE001 intentionally

pyproject.toml

@@ -41,15 +41,11 @@ ignore = [
     "DTZ",     # flake8-datetimez (except DTZ003) - we only run in UTC so unnecessary
     "ISC004",  # implicit-string-concatenation-in-collection-literal - too many false positives
     "RUF015",  # unnecessary-iterable-allocation-for-first-element - makes code unnecessarily complex
+    "BLE001",  # blind-except - the majority of matches are intentional (health checks, best-effort tasks)
     # The following rules fail currently, and should probably eventually be addressed
     # They are in most-error-occuring order, descending
     "UP031",   # printf-string-formatting
     "RUF012",  # mutable-class-default
-    "SIM115",  # open-file-with-context-handler
-    "BLE001",  # blind-except
-    "RUF059",  # unused-unpacked-variable
-    "SIM102",  # collapsible-if
-    "FLY002",  # static-join-to-f-string
 ]
 extend-select = [
     "B",  # flake8-bugbear

src/olympia/abuse/management/commands/fake_cinder_webhook.py

@@ -26,7 +26,8 @@ def handle(self, *args, **options):
         if settings.ENV != 'local':
             raise CommandError('Only works in local environments')
         try:
-            body = open(options['payload_filename'], 'rb').read()
+            with open(options['payload_filename'], 'rb') as payload_file:
+                body = payload_file.read()
         except FileNotFoundError as exc:
             raise CommandError(
                 'Cannot find payload file. Try using --payload=<path>.'

src/olympia/abuse/tests/test_cinder.py

@@ -1626,7 +1626,7 @@ def test_post_queue_move_no_versions_to_flag(self):
         assert ActivityLog.objects.count() == 0
 
     def test_post_queue_move_with_multiple_reports_including_one_with_no_versions(self):
-        cinder_instance, cinder_job, listed_version, unlisted_version = (
+        cinder_instance, cinder_job, listed_version, _unlisted_version = (
             self._setup_post_queue_move_test()
         )
         other_version = version_factory(

src/olympia/access/acl.py

@@ -7,9 +7,10 @@ def match_rules(rules, app, action):
     """
     for rule in rules.split(','):
         rule_app, rule_action = rule.split(':')
-        if rule_app == '*' or rule_app == app:
-            if rule_action == '*' or rule_action == action or action == '%':
-                return True
+        if (rule_app == '*' or rule_app == app) and (
+            rule_action == '*' or rule_action == action or action == '%'
+        ):
+            return True
     return False
 
 

src/olympia/accounts/views.py

@@ -393,11 +393,10 @@ def get(self, request, user, identity, next_path, token_data):
             # on, we extract that information from the next_path if present
             # and set locale/app on the prefixer instance that reverse() will
             # use automatically.
-            if next_path:
-                if prefixer := get_url_prefix():
-                    splitted = prefixer.split_path(next_path)
-                    prefixer.locale = splitted[0]
-                    prefixer.app = splitted[1]
+            if next_path and (prefixer := get_url_prefix()):
+                splitted = prefixer.split_path(next_path)
+                prefixer.locale = splitted[0]
+                prefixer.app = splitted[1]
             edit_page = reverse('users.edit')
             if next_path:
                 next_path = f'{edit_page}?to={quote_plus(next_path)}'

src/olympia/activity/models.py

@@ -139,7 +139,7 @@ def transfer(self, new_addon):
             # arguments is a structure:
             # ``arguments = [{'addons.addon':12}, {'addons.addon':1}, ... ]``
             arguments = json.loads(self.activity_log._arguments)
-        except Exception:
+        except (TypeError, json.JSONDecodeError):
             log.info(
                 'unserializing data from addon_log failed: %s' % self.activity_log.id
             )
@@ -614,7 +614,7 @@ def handle_renames(value):
                 # `arguments_data` will be a list of dicts like:
                 # `[{'addons.addon':12}, {'addons.addon':1}, ... ]`
                 activity.arguments_data = json.loads(activity._arguments)
-            except Exception as e:
+            except (TypeError, json.JSONDecodeError) as e:
                 log.info('unserializing data from activity_log failed: %s', activity.id)
                 log.info(e)
                 activity.arguments_data = []

src/olympia/addons/indexers.py

@@ -643,9 +643,8 @@ def extract_document(cls, obj):
 
         data['colors'] = None
         # Extract dominant colors from static themes.
-        if obj.type == amo.ADDON_STATICTHEME:
-            if obj.current_previews:
-                data['colors'] = obj.current_previews[0].colors
+        if obj.type == amo.ADDON_STATICTHEME and obj.current_previews:
+            data['colors'] = obj.current_previews[0].colors
 
         data['app'] = [app.id for app in obj.compatible_apps]
         # We can use all_categories because the indexing code goes through the

src/olympia/addons/serializers.py

@@ -1270,12 +1270,13 @@ def run_validation(self, data=serializers.empty):
     def validate_slug(self, value):
         slug_validator(value)
 
-        if not self.instance or value != self.instance.slug:
-            # DeniedSlug.blocked checks for all numeric slugs as well as being denied.
-            if DeniedSlug.blocked(value):
-                raise exceptions.ValidationError(
-                    gettext('This slug cannot be used. Please choose another.')
-                )
+        # DeniedSlug.blocked checks for all numeric slugs as well as being denied.
+        if (not self.instance or value != self.instance.slug) and DeniedSlug.blocked(
+            value
+        ):
+            raise exceptions.ValidationError(
+                gettext('This slug cannot be used. Please choose another.')
+            )
 
         return value
 

src/olympia/addons/tasks.py

@@ -141,20 +141,21 @@ def restore_all_addon_media_from_backup(id, **kwargs):
     if disabled_addon_content:
         log.info('Found some disable content to restore for addon %s', addon.pk)
         if backup_storage_enabled():
-            if disabled_addon_content.icon_backup_name:
-                if icon_contents := download_file_contents_from_backup_storage(
+            if disabled_addon_content.icon_backup_name and (
+                icon_contents := download_file_contents_from_backup_storage(
                     disabled_addon_content.icon_backup_name
-                ):
-                    icon_path = addon.get_icon_path('original')
-                    log.info('Restoring icon %s for addon %s', icon_path, addon.pk)
-                    with storage.open(icon_path, 'wb') as original_file:
-                        original_file.write(icon_contents)
-                    resize_icon.delay(
-                        icon_path,
-                        addon.pk,
-                        amo.ADDON_ICON_SIZES,
-                        set_modified_on=addon.serializable_reference(),
-                    )
+                )
+            ):
+                icon_path = addon.get_icon_path('original')
+                log.info('Restoring icon %s for addon %s', icon_path, addon.pk)
+                with storage.open(icon_path, 'wb') as original_file:
+                    original_file.write(icon_contents)
+                resize_icon.delay(
+                    icon_path,
+                    addon.pk,
+                    amo.ADDON_ICON_SIZES,
+                    set_modified_on=addon.serializable_reference(),
+                )
             for deleted_preview in disabled_addon_content.deletedpreviewfile_set.all():
                 preview = deleted_preview.preview
                 if preview_contents := download_file_contents_from_backup_storage(

src/olympia/addons/tests/test_decorators.py

@@ -105,7 +105,7 @@ def test_slug_isdigit(self):
         addon.update(slug=str(addon.id))
         r = self.view(self.request, addon.slug)
         assert r == mock.sentinel.OK
-        request, addon_ = self.func.call_args[0]
+        addon_ = self.func.call_args[0][1]
         assert addon_ == addon
 
     @mock.patch(
@@ -167,7 +167,7 @@ def test_unlisted_addon_owner(self):
         """Addon owners have access."""
         self.change_channel_for_addon(self.addon, amo.CHANNEL_UNLISTED)
         assert self.view(self.request, self.addon.slug) == mock.sentinel.OK
-        request, addon = self.func.call_args[0]
+        addon = self.func.call_args[0][1]
         assert addon == self.addon
 
     @mock.patch(
@@ -180,5 +180,5 @@ def test_unlisted_addon_unlisted_admin(self):
         """Unlisted addon reviewers have access."""
         self.change_channel_for_addon(self.addon, amo.CHANNEL_UNLISTED)
         assert self.view(self.request, self.addon.slug) == mock.sentinel.OK
-        request, addon = self.func.call_args[0]
+        addon = self.func.call_args[0][1]
         assert addon == self.addon

src/olympia/addons/tests/test_tasks.py

@@ -541,7 +541,7 @@ def _uploader(self, resize_size, final_size):
         img = get_image_path('mozilla.png')
         original_size = (339, 128)
 
-        src = tempfile.NamedTemporaryFile(
+        src = tempfile.NamedTemporaryFile(  # noqa: SIM115 (temp file used across the test (delete=False))
             mode='r+b', suffix='.png', delete=False, dir=settings.TMP_PATH
         )
 

src/olympia/addons/tests/test_views.py

@@ -3363,7 +3363,7 @@ def _submit_source(self, filepath, error=False):
         raise NotImplementedError
 
     def _generate_source_tar(self, suffix='.tar.gz', data=b't' * (2**21), mode=None):
-        source = tempfile.NamedTemporaryFile(suffix=suffix, dir=settings.TMP_PATH)
+        source = tempfile.NamedTemporaryFile(suffix=suffix, dir=settings.TMP_PATH)  # noqa: SIM115 (temp file returned to caller)
         if mode is None:
             mode = 'w:bz2' if suffix.endswith('.tar.bz2') else 'w:gz'
         with tarfile.open(fileobj=source, mode=mode) as tar_file:
@@ -3377,7 +3377,7 @@ def _generate_source_tar(self, suffix='.tar.gz', data=b't' * (2**21), mode=None)
     def _generate_source_zip(
         self, suffix='.zip', data='z' * (2**21), compression=zipfile.ZIP_DEFLATED
     ):
-        source = tempfile.NamedTemporaryFile(suffix=suffix, dir=settings.TMP_PATH)
+        source = tempfile.NamedTemporaryFile(suffix=suffix, dir=settings.TMP_PATH)  # noqa: SIM115 (temp file returned to caller)
         with zipfile.ZipFile(source, 'w', compression=compression) as zip_file:
             zip_file.writestr('foo', data)
         source.seek(0)
@@ -4323,11 +4323,12 @@ def test_compatibility_with_appversion_locked_from_manifest(self):
 
     def _submit_source(self, filepath, error=False):
         _, filename = os.path.split(filepath)
-        src = SimpleUploadedFile(
-            filename,
-            open(filepath, 'rb').read(),
-            content_type=mimetypes.guess_type(filename)[0],
-        )
+        with open(filepath, 'rb') as source_file:
+            src = SimpleUploadedFile(
+                filename,
+                source_file.read(),
+                content_type=mimetypes.guess_type(filename)[0],
+            )
         response = self.client.post(
             self.url, data={**self.minimal_data, 'source': src}, format='multipart'
         )
@@ -4801,11 +4802,12 @@ def test_delete_source_formdata(self):
 
     def _submit_source(self, filepath, error=False):
         _, filename = os.path.split(filepath)
-        src = SimpleUploadedFile(
-            filename,
-            open(filepath, 'rb').read(),
-            content_type=mimetypes.guess_type(filename)[0],
-        )
+        with open(filepath, 'rb') as source_file:
+            src = SimpleUploadedFile(
+                filename,
+                source_file.read(),
+                content_type=mimetypes.guess_type(filename)[0],
+            )
         response = self.client.patch(self.url, data={'source': src}, format='multipart')
         if not error:
             assert response.status_code == 200, response.content
@@ -4882,7 +4884,7 @@ def test_submit_source_pending_rejection_triggers_needs_human_review(self):
             pending_rejection_by=user_factory(),
             pending_content_rejection=False,
         )
-        response, self.version = self._submit_source(new_source)
+        _response, self.version = self._submit_source(new_source)
         self.addon.reload()
         assert self.version.source
         assert self.version.needshumanreview_set.filter(is_active=True).exists()
@@ -7242,7 +7244,7 @@ def test_exclude_addons(self):
 
         # Exclude addon2 and addon3 by slug.
         data = self.perform_search(
-            self.url, {'exclude_addons': ','.join((addon2.slug, addon3.slug))}
+            self.url, {'exclude_addons': f'{addon2.slug},{addon3.slug}'}
         )
 
         assert len(data['results']) == 1

src/olympia/amo/models.py

@@ -583,7 +583,7 @@ def delete_preview_files(cls, sender, instance, **kw):
             try:
                 log.info(f'Removing filename: {filename} for preview: {instance.pk}')
                 storage.delete(filename)
-            except Exception as e:
+            except OSError as e:
                 log.error(f'Error deleting preview file ({filename}): {e}')
 
 

src/olympia/amo/reverse.py

@@ -54,7 +54,7 @@ def resolve(path, urlconf=None):
     """Wraps django's resolve to remove the locale and app from the path."""
     from olympia.amo.urlresolvers import Prefixer
 
-    _lang, application, path_fragment = Prefixer.split_path(path)
+    _lang, _application, path_fragment = Prefixer.split_path(path)
     return django_resolve(f'/{path_fragment}', urlconf)
 
 

src/olympia/amo/tests/test_amo_utils.py

@@ -26,7 +26,7 @@
 
 def test_slug_validator():
     assert slug_validator(u.lower()) is None
-    assert slug_validator('-'.join([u.lower(), u.lower()])) is None
+    assert slug_validator(f'{u.lower()}-{u.lower()}') is None
     pytest.raises(ValidationError, slug_validator, '234.add')
     pytest.raises(ValidationError, slug_validator, 'a a a')
     pytest.raises(ValidationError, slug_validator, 'tags/')
@@ -38,8 +38,8 @@ def test_slug_validator():
         ('xx x  - "#$@ x', 'xx-x-x'),
         ('Bän...g (bang)', 'bäng-bang'),
         (u, u.lower()),
-        ('-'.join([u, u]), '-'.join([u, u]).lower()),
-        (' - '.join([u, u]), '-'.join([u, u]).lower()),
+        (f'{u}-{u}', f'{u}-{u}'.lower()),
+        (f'{u} - {u}', f'{u}-{u}'.lower()),
         ('    a ', 'a'),
         ('tags/', 'tags'),
         ('holy_wars', 'holy_wars'),

src/olympia/amo/tests/test_commands.py

@@ -783,25 +783,15 @@ def _test_full_run_typical_response(self):
         expected_below = (
             'The following locales are below threshold of 80% or completely '
             'absent in one of our projects in Pontoon:\n- '
-            + '\n- '.join(
-                (
-                    'Norwegian (Nynorsk) [nn-NO]',
-                    'Portuguese (Brazilian) [pt-BR]',
-                    'Romanian [ro]',
-                )
-            )
+            'Norwegian (Nynorsk) [nn-NO]\n- '
+            'Portuguese (Brazilian) [pt-BR]\n- '
+            'Romanian [ro]'
         )
         assert expected_below in mail.outbox[0].body
 
         expected_above = (
             'The following locales are above threshold and not yet enabled:\n- '
-            + '\n- '.join(
-                (
-                    'Bulgarian [bg]',
-                    'Danish [da]',
-                    'Indonesian [id]',
-                )
-            )
+            + 'Bulgarian [bg]\n- Danish [da]\n- Indonesian [id]'
         )
         assert expected_above in mail.outbox[0].body
 

src/olympia/amo/tests/test_helpers.py

@@ -318,7 +318,8 @@ def get_image_path(name):
 
 
 def get_uploaded_file(name):
-    data = open(get_image_path(name), mode='rb').read()
+    with open(get_image_path(name), mode='rb') as f:
+        data = f.read()
     return SimpleUploadedFile(name, data, content_type=mimetypes.guess_type(name)[0])
 
 
@@ -328,21 +329,20 @@ def get_addon_file(name):
 
 class TestAnimatedImages(TestCase):
     def test_animated_images(self):
-        img = ImageCheck(open(get_image_path('animated.png'), mode='rb'))
-        assert img.is_animated()
-        img = ImageCheck(open(get_image_path('non-animated.png'), mode='rb'))
-        assert not img.is_animated()
-
-        img = ImageCheck(open(get_image_path('animated.gif'), mode='rb'))
-        assert img.is_animated()
-        img = ImageCheck(open(get_image_path('non-animated.gif'), mode='rb'))
-        assert not img.is_animated()
+        with open(get_image_path('animated.png'), mode='rb') as f:
+            assert ImageCheck(f).is_animated()
+        with open(get_image_path('non-animated.png'), mode='rb') as f:
+            assert not ImageCheck(f).is_animated()
+        with open(get_image_path('animated.gif'), mode='rb') as f:
+            assert ImageCheck(f).is_animated()
+        with open(get_image_path('non-animated.gif'), mode='rb') as f:
+            assert not ImageCheck(f).is_animated()
 
     def test_junk(self):
-        img = ImageCheck(open(__file__, 'rb'))
-        assert not img.is_image()
-        img = ImageCheck(open(get_image_path('non-animated.gif'), mode='rb'))
-        assert img.is_image()
+        with open(__file__, 'rb') as f:
+            assert not ImageCheck(f).is_image()
+        with open(get_image_path('non-animated.gif'), mode='rb') as f:
+            assert ImageCheck(f).is_image()
 
 
 def test_jinja_trans_monkeypatch():

src/olympia/amo/tests/test_monitor.py

@@ -48,7 +48,7 @@ def test_libraries(self):
 
     @pytest.mark.requires_elasticsearch
     def test_elastic(self):
-        status, elastic_result = monitors.elastic()
+        status, _elastic_result = monitors.elastic()
         assert status == ''
 
     @patch('olympia.amo.monitors.get_es', side_effect=Exception('Connection error'))
@@ -68,7 +68,7 @@ def test_elastic_status_red(self):
     @patch('os.path.exists')
     @patch('os.access')
     def test_path(self, mock_exists, mock_access):
-        status, path_result = monitors.path()
+        status, _path_result = monitors.path()
         assert status == ''
 
     @override_settings(TMP_PATH='foo')
@@ -88,7 +88,7 @@ def test_rabbitmq(self, mock_connection):
     def test_signer(self):
         responses.add_passthru(settings.AUTOGRAPH_CONFIG['server_url'])
 
-        status, signer_result = monitors.signer()
+        status, _signer_result = monitors.signer()
         assert status == ''
 
     def test_database(self):