test: revert assertStartsWith and clean up pre-existing flake8 lints

- Revert assertStartsWith (Python 3.13+ only) back to
  self.assertTrue(msg.startswith(...), msg) at all 5 call sites so the
  shutdown checks work on the current test environment.
- Drop the file-level "# flake8: noqa" and fix the pre-existing lints
  in lifecycle_test.py that surfaced after #8897 corrected pre-commit
  flake8 arg parsing:
    * E402: add per-import "# noqa: E402" to imports that must live
      after sys.path.append("../common")
    * F841: drop unused "md" and "tensor_shape" locals
    * E266: normalize "##" block comments to "#"
    * E721: replace "type(x) == T" with isinstance(x, T)
    * E712: replace "== False" with "is False"
    * E711: replace "!= None" with "is not None"
    * F401: drop unused "HTTPConnectionClosed" local import

Author: Vinya567

qa/L0_lifecycle/lifecycle_test.py

@@ -26,32 +26,30 @@
 # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-# flake8: noqa
-
 import sys
 
 sys.path.append("../common")
 
-import base64
-import concurrent.futures
-import json
-import multiprocessing
-import os
-import shutil
-import signal
-import threading
-import time
-import unittest
-from builtins import range
-from functools import partial
-from pathlib import Path
-
-import infer_util as iu
-import numpy as np
-import test_util as tu
-import tritonclient.grpc as grpcclient
-import tritonclient.http as httpclient
-from tritonclient.utils import InferenceServerException
+import base64  # noqa: E402
+import concurrent.futures  # noqa: E402
+import json  # noqa: E402
+import multiprocessing  # noqa: E402
+import os  # noqa: E402
+import shutil  # noqa: E402
+import signal  # noqa: E402
+import threading  # noqa: E402
+import time  # noqa: E402
+import unittest  # noqa: E402
+from builtins import range  # noqa: E402
+from functools import partial  # noqa: E402
+from pathlib import Path  # noqa: E402
+
+import infer_util as iu  # noqa: E402
+import numpy as np  # noqa: E402
+import test_util as tu  # noqa: E402
+import tritonclient.grpc as grpcclient  # noqa: E402
+import tritonclient.http as httpclient  # noqa: E402
+from tritonclient.utils import InferenceServerException  # noqa: E402
 
 
 class LifeCycleTest(tu.TestResultCollector):
@@ -310,7 +308,7 @@ def test_parse_error_no_model_config(self):
                 self.assertTrue(triton_client.is_server_live())
                 self.assertTrue(triton_client.is_server_ready())
 
-                md = triton_client.get_model_metadata(model_name, "1")
+                triton_client.get_model_metadata(model_name, "1")
                 self.assertTrue(
                     False,
                     "expected model '"
@@ -2258,7 +2256,6 @@ def test_multiple_model_repository_control_startup_models(self):
     def test_model_repository_index(self):
         # use model control EXPLICIT and --load-model to load a subset of models
         # in model repository
-        tensor_shape = (1, 16)
         model_bases = ["plan", "libtorch", "simple_libtorch"]
 
         # Sanity check on loaded models
@@ -2580,8 +2577,8 @@ def test_file_override_security(self):
         self.assertTrue(os.path.exists(os.path.join(model_basepath, existing_file_rel)))
 
         # Symlinks
-        ## No easy way to inject symlink into generated temp model dir, so for
-        ## testing sake, make a fixed symlink path in /tmp.
+        # No easy way to inject symlink into generated temp model dir, so for
+        # testing sake, make a fixed symlink path in /tmp.
         escape_dir_symlink_rel = os.path.join("..", "escape_symlink")
         escape_dir_symlink_full = "/tmp/escape_symlink"
         self.assertEqual(
@@ -2676,7 +2673,7 @@ def callback(user_data, result, error):
             self.assertTrue(False, "expected error for new inference during shutdown")
         except InferenceServerException as ex:
             msg = ex.message()
-            self.assertStartsWith(self._GRPC_CONNECTION_REFUSED_PREFIX, msg)
+            self.assertTrue(msg.startswith(self._GRPC_CONNECTION_REFUSED_PREFIX), msg)
 
         # Wait until the results are available in user_data
         time_out = 30
@@ -2686,7 +2683,7 @@ def callback(user_data, result, error):
 
         # Previous requests should succeed
         for result in async_results:
-            if type(result) == InferenceServerException:
+            if isinstance(result, InferenceServerException):
                 raise result
             output_data = result.as_numpy("OUTPUT0")
             np.testing.assert_allclose(
@@ -2740,21 +2737,23 @@ def callback(user_data, result, error):
             # each subsequent request returns Connection refused
             msg = ex.message()
             if "CANCELLED" not in msg:
-                self.assertStartsWith(self._GRPC_CONNECTION_REFUSED_PREFIX, msg)
+                self.assertTrue(
+                    msg.startswith(self._GRPC_CONNECTION_REFUSED_PREFIX), msg
+                )
         # 2: New sequence with existing sequence ID
         try:
             triton_client.infer(model_name, inputs, sequence_id=1, sequence_start=True)
             self.assertTrue(False, "expected error for new inference during shutdown")
         except InferenceServerException as ex:
             msg = ex.message()
-            self.assertStartsWith(self._GRPC_CONNECTION_REFUSED_PREFIX, msg)
+            self.assertTrue(msg.startswith(self._GRPC_CONNECTION_REFUSED_PREFIX), msg)
         # 3: Continuing sequence after shutdown
         try:
             triton_client.infer(model_name, inputs, sequence_id=2, sequence_end=True)
             self.assertTrue(False, "expected error for new inference during shutdown")
         except InferenceServerException as ex:
             msg = ex.message()
-            self.assertStartsWith(self._GRPC_CONNECTION_REFUSED_PREFIX, msg)
+            self.assertTrue(msg.startswith(self._GRPC_CONNECTION_REFUSED_PREFIX), msg)
 
         # Wait until the results are available in user_data
         time_out = 30
@@ -2764,7 +2763,7 @@ def callback(user_data, result, error):
 
         # Previous requests should succeed
         for result in async_results:
-            if type(result) == InferenceServerException:
+            if isinstance(result, InferenceServerException):
                 raise result
             output_data = result.as_numpy("OUTPUT")
             np.testing.assert_allclose(
@@ -2814,7 +2813,7 @@ def callback(user_data, result, error):
             self.assertTrue(False, "expected error for new inference during shutdown")
         except InferenceServerException as ex:
             msg = ex.message()
-            self.assertStartsWith(self._GRPC_CONNECTION_REFUSED_PREFIX, msg)
+            self.assertTrue(msg.startswith(self._GRPC_CONNECTION_REFUSED_PREFIX), msg)
 
         # Wait until the results are available in user_data
         time_out = 10
@@ -2824,7 +2823,7 @@ def callback(user_data, result, error):
 
         # Previous requests should succeed
         for result in async_results:
-            if type(result) == InferenceServerException:
+            if isinstance(result, InferenceServerException):
                 raise result
             output_data = result.as_numpy("OUTPUT0")
             np.testing.assert_allclose(
@@ -3070,7 +3069,7 @@ def _load_unload():
         # This test can replicate a load while async unloading on machines with
         # sufficient concurrency. Regardless on whether it is replicated or not,
         # the server must not crash.
-        if load_before_unload_finish[0] == False:
+        if load_before_unload_finish[0] is False:
             # Track non-replication on test printout via statistics.
             warning_msg = "Cannot replicate a load while async unloading. CPU count: {}. num_threads: {}.".format(
                 multiprocessing.cpu_count(), num_threads
@@ -3296,7 +3295,9 @@ def test_model_config_overwite(self):
 """
 
         # Ensure the model has been loaded w/ the expected (different from override) config.
-        self.assertTrue(original_config != None and original_config != override_config)
+        self.assertTrue(
+            original_config is not None and original_config != override_config
+        )
 
         # Reload the model with the overriding configuration value.
         triton_client.load_model(model_name, config=override_config)
@@ -3359,7 +3360,6 @@ def test_shutdown_while_loading(self):
     def test_shutdown_with_live_connection(self):
         model_name = "add_sub"
         model_shape = (16,)
-        from geventhttpclient.response import HTTPConnectionClosed
 
         input_data = np.ones(shape=model_shape, dtype=np.float32)
         inputs = [