diff --git a/airflow-core/tests/unit/cli/commands/test_api_server_command.py b/airflow-core/tests/unit/cli/commands/test_api_server_command.py
index 93c2c43e27b39..4d6e3f62d5b61 100644
--- a/airflow-core/tests/unit/cli/commands/test_api_server_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_api_server_command.py
@@ -302,7 +302,7 @@ def test_run_command_daemon(
                 ]
         else:
             assert mock_daemon.mock_calls == []
-            mock_setup_locations.mock_calls == []
+            assert mock_setup_locations.mock_calls == []
             mock_pid_file.assert_not_called()
             mock_open.assert_not_called()
 
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index eb763635ddf32..ff1222c1222ee 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -3743,7 +3743,7 @@ def test_dagrun_timeout_fails_run_and_update_next_dagrun(self, dag_maker):
 
         dr = dag_maker.create_dagrun(start_date=timezone.utcnow() - datetime.timedelta(days=1))
         # check that next_dagrun is dr.logical_date
-        dag_maker.dag_model.next_dagrun == dr.logical_date
+        assert dag_maker.dag_model.next_dagrun == dr.logical_date
         scheduler_job = Job()
         self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec])
 
@@ -4600,7 +4600,7 @@ def test_verify_integrity_if_dag_changed(self, dag_maker):
         dr = drs[0]
 
         self.job_runner._schedule_dag_run(dag_run=dr, session=session)
-        len(self.job_runner.scheduler_dag_bag.get_dag_for_run(dr, session).tasks) == 1
+        assert len(self.job_runner.scheduler_dag_bag.get_dag_for_run(dr, session).tasks) == 1
         dag_version_1 = DagVersion.get_latest_version(dr.dag_id, session=session)
         assert dr.dag_versions[-1].id == dag_version_1.id
 
@@ -5799,7 +5799,7 @@ def test_more_runs_are_not_created_when_max_active_runs_is_reached(self, dag_mak
         dag_models = query.all()
         self.job_runner._create_dag_runs(dag_models, session)
         dr = session.scalars(select(DagRun)).one()
-        dr.state == DagRunState.QUEUED
+        assert dr.state == DagRunState.QUEUED
         assert session.scalar(select(func.count()).select_from(DagRun)) == 1
         assert dag_maker.dag_model.next_dagrun_create_after == DEFAULT_DATE + timedelta(days=2)
         assert dag_maker.dag_model.next_dagrun == DEFAULT_DATE + timedelta(days=1)
diff --git a/airflow-core/tests/unit/models/test_dag.py b/airflow-core/tests/unit/models/test_dag.py
index b34ab12dd4aef..f9751684e35cf 100644
--- a/airflow-core/tests/unit/models/test_dag.py
+++ b/airflow-core/tests/unit/models/test_dag.py
@@ -3567,7 +3567,7 @@ def test_get_flat_relative_ids_with_setup(self):
         # now, we know that t1 is the teardown for s1, so now we know that s1 will be "torn down"
         # by the time w4 runs, so we now know that w4 no longer requires s1, so when we clear w4,
         # s1 will not also be cleared
-        self.cleared_downstream(w4) == {w4}
+        assert self.cleared_downstream(w4) == {w4}
         assert set(w1.get_upstreams_only_setups_and_teardowns()) == {s1, t1}
         assert self.cleared_downstream(w1) == {s1, w1, w2, w3, t1, w4}
         assert self.cleared_upstream(w1) == {s1, w1, t1}
diff --git a/airflow-core/tests/unit/serialization/test_dag_serialization.py b/airflow-core/tests/unit/serialization/test_dag_serialization.py
index 0a40f67314e92..8d8cb113c2ea3 100644
--- a/airflow-core/tests/unit/serialization/test_dag_serialization.py
+++ b/airflow-core/tests/unit/serialization/test_dag_serialization.py
@@ -935,9 +935,9 @@ def validate_deserialized_task(
         if isinstance(task, MappedOperator):
             # MappedOperator.operator_class now stores only minimal type information
             # for memory efficiency (task_type and _operator_name).
-            serialized_task.operator_class["task_type"] == type(task).__name__
+            assert serialized_task.operator_class["task_type"] == task.operator_class.__name__
             if isinstance(serialized_task.operator_class, DecoratedOperator):
-                serialized_task.operator_class["_operator_name"] == task._operator_name
+                assert serialized_task.operator_class["_operator_name"] == task._operator_name
 
             # Serialization cleans up default values in partial_kwargs, this
             # adds them back to both sides.
diff --git a/chart/tests/helm_tests/airflow_aux/test_pod_template_file.py b/chart/tests/helm_tests/airflow_aux/test_pod_template_file.py
index 579235cf7bfe8..bbad368fcb398 100644
--- a/chart/tests/helm_tests/airflow_aux/test_pod_template_file.py
+++ b/chart/tests/helm_tests/airflow_aux/test_pod_template_file.py
@@ -1600,7 +1600,7 @@ def test_airflow_local_settings_kerberos_sidecar(self, workers_values):
             show_only=["templates/pod-template-file.yaml"],
             chart_dir=self.temp_chart_dir,
         )
-        jmespath.search("spec.containers[1].name", docs[0]) == "worker-kerberos"
+        assert jmespath.search("spec.containers[1].name", docs[0]) == "worker-kerberos"
 
         assert {
             "name": "config",
diff --git a/providers/amazon/tests/unit/amazon/aws/executors/batch/test_batch_executor.py b/providers/amazon/tests/unit/amazon/aws/executors/batch/test_batch_executor.py
index 00af3ca330d27..00c3d622dcb5e 100644
--- a/providers/amazon/tests/unit/amazon/aws/executors/batch/test_batch_executor.py
+++ b/providers/amazon/tests/unit/amazon/aws/executors/batch/test_batch_executor.py
@@ -397,7 +397,7 @@ def test_attempt_all_jobs_when_some_jobs_fail(self, _, mock_executor):
             submit_job_args["containerOverrides"]["command"] = airflow_commands[i]
             assert mock_executor.batch.submit_job.call_args_list[i].kwargs == submit_job_args
         assert len(mock_executor.pending_jobs) == 1
-        mock_executor.pending_jobs[0].command == airflow_cmd1
+        assert mock_executor.pending_jobs[0].command == airflow_cmd1
         assert len(mock_executor.active_workers.get_all_jobs()) == 1
 
         # Add more tasks to pending_jobs. This simulates tasks being scheduled by Airflow
@@ -417,7 +417,7 @@ def test_attempt_all_jobs_when_some_jobs_fail(self, _, mock_executor):
             submit_job_args["containerOverrides"]["command"] = airflow_commands[i]
             assert mock_executor.batch.submit_job.call_args_list[i].kwargs == submit_job_args
         assert len(mock_executor.pending_jobs) == 1
-        mock_executor.pending_jobs[0].command == airflow_cmd1
+        assert mock_executor.pending_jobs[0].command == airflow_cmd1
         assert len(mock_executor.active_workers.get_all_jobs()) == 3
 
         airflow_commands.append(airflow_cmd1)
diff --git a/providers/amazon/tests/unit/amazon/aws/operators/test_bedrock.py b/providers/amazon/tests/unit/amazon/aws/operators/test_bedrock.py
index 0fb0eba8f081e..7c6b1531fa491 100644
--- a/providers/amazon/tests/unit/amazon/aws/operators/test_bedrock.py
+++ b/providers/amazon/tests/unit/amazon/aws/operators/test_bedrock.py
@@ -174,13 +174,20 @@ def test_customize_model_wait_combinations(
     @mock.patch.object(BedrockHook, "get_waiter")
     def test_ensure_unique_job_name(self, _, side_effect, ensure_unique_name, mock_conn, bedrock_hook):
         mock_conn.create_model_customization_job.side_effect = side_effect
-        expected_call_count = len(side_effect) if ensure_unique_name else 1
+        self.operator.ensure_unique_job_name = ensure_unique_name
         self.operator.wait_for_completion = False
+        expected_call_count = len(side_effect) if ensure_unique_name else 1
+
+        if not ensure_unique_name and any(isinstance(e, ClientError) for e in side_effect):
+            with pytest.raises(ClientError):
+                self.operator.execute({})
+            assert mock_conn.create_model_customization_job.call_count == expected_call_count
+            return
 
         response = self.operator.execute({})
 
         assert response == self.CUSTOMIZE_JOB_ARN
-        mock_conn.create_model_customization_job.call_count == expected_call_count
+        assert mock_conn.create_model_customization_job.call_count == expected_call_count
         bedrock_hook.get_waiter.assert_not_called()
         self.operator.defer.assert_not_called()
 
diff --git a/providers/amazon/tests/unit/amazon/aws/operators/test_emr_serverless.py b/providers/amazon/tests/unit/amazon/aws/operators/test_emr_serverless.py
index c851114e665ce..75fcf1bce3d24 100644
--- a/providers/amazon/tests/unit/amazon/aws/operators/test_emr_serverless.py
+++ b/providers/amazon/tests/unit/amazon/aws/operators/test_emr_serverless.py
@@ -78,11 +78,6 @@ def test_execute_successfully_with_wait_for_completion(self, mock_conn, mock_wai
             "applicationId": application_id,
             "ResponseMetadata": {"HTTPStatusCode": 200},
         }
-        mock_conn.get_application.side_effect = [
-            {"application": {"state": "CREATED"}},
-            {"application": {"state": "STARTED"}},
-        ]
-
         operator = EmrServerlessCreateApplicationOperator(
             task_id=task_id,
             release_label=release_label,
@@ -111,7 +106,6 @@ def test_execute_successfully_with_wait_for_completion(self, mock_conn, mock_wai
 
         mock_conn.start_application.assert_called_once_with(applicationId=application_id)
         assert id == application_id
-        mock_conn.get_application.call_count == 2
 
     @mock.patch.object(EmrServerlessHook, "get_waiter")
     @mock.patch.object(EmrServerlessHook, "conn")
@@ -234,7 +228,7 @@ def test_failed_create_application(self, mock_conn, mock_get_waiter):
             type=job_type,
             **config,
         )
-        mock_conn.create_application.call_count == 2
+        assert mock_conn.create_application.call_count == 2
 
     @mock.patch.object(EmrServerlessHook, "get_waiter")
     @mock.patch.object(EmrServerlessHook, "conn")
@@ -823,7 +817,7 @@ def test_start_job_run_fail_on_wait_for_completion(self, mock_conn, mock_get_wai
         assert "Serverless Job failed:" in str(ex_message.value)
         default_name = operator.name
 
-        mock_conn.get_application.call_count == 2
+        assert mock_conn.get_application.call_count == 1
         mock_conn.start_job_run.assert_called_once_with(
             clientToken=client_request_token,
             applicationId=application_id,
diff --git a/providers/amazon/tests/unit/amazon/aws/operators/test_sagemaker_processing.py b/providers/amazon/tests/unit/amazon/aws/operators/test_sagemaker_processing.py
index df21010a8ee7e..a18058c745e42 100644
--- a/providers/amazon/tests/unit/amazon/aws/operators/test_sagemaker_processing.py
+++ b/providers/amazon/tests/unit/amazon/aws/operators/test_sagemaker_processing.py
@@ -149,7 +149,7 @@ def test_integer_fields_with_stopping_condition(self, _, __, ___, mock_desc):
                 (key3,) = key3_raw
                 assert sagemaker.config[key1][key2][key3] == int(sagemaker.config[key1][key2][key3])
             else:
-                sagemaker.config[key1][key2] == int(sagemaker.config[key1][key2])
+                assert sagemaker.config[key1][key2] == int(sagemaker.config[key1][key2])
 
     @mock.patch.object(SageMakerHook, "describe_processing_job")
     @mock.patch.object(SageMakerHook, "count_processing_jobs_by_name", return_value=0)
diff --git a/providers/amazon/tests/unit/amazon/aws/operators/test_sagemaker_transform.py b/providers/amazon/tests/unit/amazon/aws/operators/test_sagemaker_transform.py
index c91dbedb4d358..40c86115c48e8 100644
--- a/providers/amazon/tests/unit/amazon/aws/operators/test_sagemaker_transform.py
+++ b/providers/amazon/tests/unit/amazon/aws/operators/test_sagemaker_transform.py
@@ -103,7 +103,7 @@ def test_integer_fields(self, _, mock_create_transform, __, ___, mock_desc):
                 (key3,) = key3_org
                 assert self.sagemaker.config[key1][key2][key3] == int(self.sagemaker.config[key1][key2][key3])
             else:
-                self.sagemaker.config[key1][key2] == int(self.sagemaker.config[key1][key2])
+                assert self.sagemaker.config[key1][key2] == int(self.sagemaker.config[key1][key2])
 
     @mock.patch.object(SageMakerHook, "describe_transform_job")
     @mock.patch.object(SageMakerHook, "create_model")
diff --git a/providers/amazon/tests/unit/amazon/aws/utils/test_waiter_with_logging.py b/providers/amazon/tests/unit/amazon/aws/utils/test_waiter_with_logging.py
index 6636f4cdb3a7a..716af7fbdd1af 100644
--- a/providers/amazon/tests/unit/amazon/aws/utils/test_waiter_with_logging.py
+++ b/providers/amazon/tests/unit/amazon/aws/utils/test_waiter_with_logging.py
@@ -269,7 +269,7 @@ def test_wait_with_list_response(self, mock_sleep):
                 "MaxAttempts": 1,
             },
         )
-        mock_waiter.wait.call_count == 3
+        assert mock_waiter.wait.call_count == 3
         mock_sleep.assert_called_with(123)
 
     @mock.patch("time.sleep")
diff --git a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py
index 354fdbf05c900..9269c4debd05f 100644
--- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py
+++ b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py
@@ -1639,7 +1639,7 @@ def test_revoke_task(self, mock_kube_dynamic_client, dag_maker, create_dummy_dag
         assert not executor.has_task(task_instance=ti)
         executor.kube_scheduler.patch_pod_revoked.assert_called_once()
         executor.kube_scheduler.delete_pod.assert_called_once()
-        mock_kube_client.patch_namespaced_pod.calls[0] == []
+        mock_kube_client.patch_namespaced_pod.assert_not_called()
         assert executor.running == set()
 
     @pytest.mark.parametrize(
diff --git a/providers/google/tests/unit/google/cloud/hooks/test_bigquery.py b/providers/google/tests/unit/google/cloud/hooks/test_bigquery.py
index e750b42084cff..ddacebbdbe18a 100644
--- a/providers/google/tests/unit/google/cloud/hooks/test_bigquery.py
+++ b/providers/google/tests/unit/google/cloud/hooks/test_bigquery.py
@@ -1303,7 +1303,7 @@ def test_rowcount(self, mock_get_client):
     def test_fetchone(self, mock_next, mock_get_client):
         bq_cursor = self.hook.get_cursor()
         result = bq_cursor.fetchone()
-        mock_next.call_count == 1
+        assert mock_next.call_count == 1
         assert mock_next.return_value == result
 
     @mock.patch("airflow.providers.google.cloud.hooks.bigquery.BigQueryHook.get_client")
diff --git a/providers/google/tests/unit/google/cloud/transfers/test_sql_to_gcs.py b/providers/google/tests/unit/google/cloud/transfers/test_sql_to_gcs.py
index b252ae198d6a7..6c069b8333570 100644
--- a/providers/google/tests/unit/google/cloud/transfers/test_sql_to_gcs.py
+++ b/providers/google/tests/unit/google/cloud/transfers/test_sql_to_gcs.py
@@ -49,13 +49,15 @@
     ["102", "business", "2017-05-24"],
     ["103", "non-profit", "2018-10-01"],
 ]
-OUTPUT_DATA = json.dumps(
+EXPECTED_JSON_ROW = json.dumps(
     {
         "column_a": "convert_type_return_value",
         "column_b": "convert_type_return_value",
         "column_c": "convert_type_return_value",
-    }
-).encode("utf-8")
+    },
+    sort_keys=True,
+    ensure_ascii=False,
+)
 SCHEMA_FILE = "schema_file.json"
 APP_JSON = "application/json"
 
@@ -163,6 +165,7 @@ def test_exec(self, mock_convert_type, mock_query, mock_upload, mock_writer, moc
         mock_file.flush.reset_mock()
         mock_upload.reset_mock()
         mock_file.close.reset_mock()
+        mock_file.write.reset_mock()
         cursor_mock.reset_mock()
 
         cursor_mock.__iter__ = Mock(return_value=iter(INPUT_DATA))
@@ -183,14 +186,8 @@ def test_exec(self, mock_convert_type, mock_query, mock_upload, mock_writer, moc
         }
 
         mock_query.assert_called_once()
-        mock_file.write.call_args_list == [
-            mock.call(OUTPUT_DATA),
-            mock.call(b"\n"),
-            mock.call(OUTPUT_DATA),
-            mock.call(b"\n"),
-            mock.call(OUTPUT_DATA),
-            mock.call(b"\n"),
-        ]
+        written = "".join(c.args[0] for c in mock_file.write.call_args_list)
+        assert written == (EXPECTED_JSON_ROW + "\n") * 3
         mock_upload.assert_called_once_with(
             BUCKET, FILENAME.format(0), TMP_FILE_NAME, mime_type=APP_JSON, gzip=False, metadata=None
         )
@@ -227,14 +224,8 @@ def test_exec(self, mock_convert_type, mock_query, mock_upload, mock_writer, moc
         }
 
         mock_query.assert_called_once()
-        mock_file.write.call_args_list == [
-            mock.call(OUTPUT_DATA),
-            mock.call(b"\n"),
-            mock.call(OUTPUT_DATA),
-            mock.call(b"\n"),
-            mock.call(OUTPUT_DATA),
-            mock.call(b"\n"),
-        ]
+        written = "".join(c.args[0] for c in mock_file.write.call_args_list)
+        assert written == (EXPECTED_JSON_ROW + "\n") * 3
 
         mock_file.flush.assert_called_once()
         mock_upload.assert_called_once_with(
diff --git a/pyproject.toml b/pyproject.toml
index 128bb11e71428..0cef71e43f57b 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -601,6 +601,7 @@ extend-select = [
     "G", # flake8-logging-format rules
     "LOG", # flake8-logging rules, most of them autofixable
     "PT", # flake8-pytest-style rules
+    "B015", # Useless comparison: bare `a == b` is a no-op; prepend `assert` or remove
     "TID25", # flake8-tidy-imports rules
     "E", # pycodestyle rules
     "W", # pycodestyle rules
diff --git a/shared/observability/tests/observability/metrics/test_otel_logger.py b/shared/observability/tests/observability/metrics/test_otel_logger.py
index f4c4cd369bf23..21d27789def09 100644
--- a/shared/observability/tests/observability/metrics/test_otel_logger.py
+++ b/shared/observability/tests/observability/metrics/test_otel_logger.py
@@ -190,7 +190,7 @@ def test_decr_with_rate_limit_works(self, mock_random, name):
         assert mock_random.call_count == 2
         # add() is called once in the initial stats.incr and once for the decr that passed the rate check.
         self.map[full_name(name)].add.assert_has_calls(expected_calls)
-        self.map[full_name(name)].add.call_count == 2
+        assert self.map[full_name(name)].add.call_count == 2
 
     def test_gauge_new_metric(self, name):
         self.stats.gauge(name, value=1)
@@ -205,7 +205,7 @@ def test_gauge_new_metric_with_tags(self, name):
         self.stats.gauge(name, value=1, tags=tags)
 
         self.meter.get_meter().create_gauge.assert_called_once_with(name=full_name(name))
-        self.map[key].attributes == tags
+        assert self.map[key].attributes == tags
 
     def test_gauge_existing_metric(self, name):
         self.stats.gauge(name, value=1)
diff --git a/task-sdk/tests/task_sdk/definitions/test_taskgroup.py b/task-sdk/tests/task_sdk/definitions/test_taskgroup.py
index ed03e6c67732a..18c7f65faf2e4 100644
--- a/task-sdk/tests/task_sdk/definitions/test_taskgroup.py
+++ b/task-sdk/tests/task_sdk/definitions/test_taskgroup.py
@@ -774,8 +774,8 @@ def test_mapped_task_group_id_prefix_task_id():
     assert t1.task_id == "t1"
     assert t2.task_id == "g.t2"
 
-    dag.get_task("t1") == t1
-    dag.get_task("g.t2") == t2
+    assert dag.get_task("t1") == t1
+    assert dag.get_task("g.t2") == t2
 
 
 def test_pass_taskgroup_output_to_task():