style: fix error [SIM117] - Use a single with statement with multiple contexts instead of nested with statements.

Used Ruff (vscode and pre-commit) to:
- Black-compatible code formatting.
- fix all auto-fixable violations.
- isort-compatible import sorting.
- flake8-simplify manual fixes.

Signed-off-by: Paulo Vital <paulo.vital@ibm.com>

Author: pvital

tests/clients/boto3/test_boto3_s3.py

@@ -4,10 +4,10 @@
 
 import os
 from io import BytesIO
+from typing import Generator
 
-import pytest
 import boto3
-from typing import Generator
+import pytest
 from moto import mock_aws
 
 from instana.singletons import agent, get_tracer
@@ -157,7 +157,7 @@ def test_s3_upload_file(self) -> None:
     def test_s3_upload_file_obj(self) -> None:
         self.s3.create_bucket(Bucket=self.bucket_name)
 
-        with self.tracer.start_as_current_span("test"):
+        with self.tracer.start_as_current_span("test"):  # noqa: SIM117
             with open(upload_filename, "rb") as fd:
                 self.s3.upload_fileobj(fd, self.bucket_name, self.object_name)
 
@@ -214,7 +214,7 @@ def test_s3_download_file_obj(self) -> None:
         self.s3.create_bucket(Bucket=self.bucket_name)
         self.s3.upload_file(upload_filename, self.bucket_name, self.object_name)
 
-        with self.tracer.start_as_current_span("test"):
+        with self.tracer.start_as_current_span("test"):  # noqa: SIM117
             with open(download_target_filename, "wb") as fd:
                 self.s3.download_fileobj(self.bucket_name, self.object_name, fd)
 

tests/clients/test_google-cloud-storage.py

@@ -406,12 +406,10 @@ def test_objects_compose(self, mock_requests: Mock) -> None:
         )
 
         with self.tracer.start_as_current_span("test"):
-            client.bucket("test bucket").blob("dest object").compose(
-                [
-                    storage.blob.Blob("object 1", "test bucket"),
-                    storage.blob.Blob("object 2", "test bucket"),
-                ]
-            )
+            client.bucket("test bucket").blob("dest object").compose([
+                storage.blob.Blob("object 1", "test bucket"),
+                storage.blob.Blob("object 2", "test bucket"),
+            ])
 
         spans = self.recorder.queued_spans()
 
@@ -1046,10 +1044,9 @@ def test_batch_operation(self, mock_requests: Mock) -> None:
         )
         bucket = client.bucket("test-bucket")
 
-        with self.tracer.start_as_current_span("test"):
-            with client.batch():
-                for obj in ["obj1", "obj2"]:
-                    bucket.delete_blob(obj)
+        with self.tracer.start_as_current_span("test"), client.batch():
+            for obj in ["obj1", "obj2"]:
+                bucket.delete_blob(obj)
 
         spans = self.recorder.queued_spans()
 
@@ -1064,9 +1061,12 @@ def test_execute_with_instana_without_tags(self, mock_requests: Mock) -> None:
         client = self._client(
             credentials=AnonymousCredentials(), project="test-project"
         )
-        with self.tracer.start_as_current_span("test"), patch(
-            "instana.instrumentation.google.cloud.storage._collect_attributes",
-            return_value=None,
+        with (
+            self.tracer.start_as_current_span("test"),
+            patch(
+                "instana.instrumentation.google.cloud.storage._collect_attributes",
+                return_value=None,
+            ),
         ):
             buckets = client.list_buckets()
             for b in buckets:
@@ -1077,9 +1077,12 @@ def test_execute_with_instana_is_tracing_off(self) -> None:
         client = self._client(
             credentials=AnonymousCredentials(), project="test-project"
         )
-        with self.tracer.start_as_current_span("test"), patch(
-            "instana.instrumentation.google.cloud.storage.get_tracer_tuple",
-            return_value=(None, None, None),
+        with (
+            self.tracer.start_as_current_span("test"),
+            patch(
+                "instana.instrumentation.google.cloud.storage.get_tracer_tuple",
+                return_value=(None, None, None),
+            ),
         ):
             response = client.list_buckets()
             assert isinstance(response.client, storage.Client)
@@ -1096,12 +1099,16 @@ def test_download_with_instana_is_tracing_off(self, mock_requests: Mock) -> None
         client = self._client(
             credentials=AnonymousCredentials(), project="test-project"
         )
-        with self.tracer.start_as_current_span("test"), patch(
-            "instana.instrumentation.google.cloud.storage.get_tracer_tuple",
-            return_value=(None, None, None),
+        with (
+            self.tracer.start_as_current_span("test"),
+            patch(
+                "instana.instrumentation.google.cloud.storage.get_tracer_tuple",
+                return_value=(None, None, None),
+            ),
         ):
             response = (
-                client.bucket("test bucket")
+                client
+                .bucket("test bucket")
                 .blob("test object")
                 .download_to_file(
                     io.BytesIO(),
@@ -1120,12 +1127,16 @@ def test_upload_with_instana_is_tracing_off(self, mock_requests: Mock) -> None:
             credentials=AnonymousCredentials(), project="test-project"
         )
 
-        with self.tracer.start_as_current_span("test"), patch(
-            "instana.instrumentation.google.cloud.storage.get_tracer_tuple",
-            return_value=(None, None, None),
+        with (
+            self.tracer.start_as_current_span("test"),
+            patch(
+                "instana.instrumentation.google.cloud.storage.get_tracer_tuple",
+                return_value=(None, None, None),
+            ),
         ):
             response = (
-                client.bucket("test bucket")
+                client
+                .bucket("test bucket")
                 .blob("test object")
                 .upload_from_string("CONTENT")
             )
@@ -1144,14 +1155,17 @@ def test_finish_batch_operation_is_tracing_off(self, mock_requests: Mock) -> Non
         )
         bucket = client.bucket("test-bucket")
 
-        with self.tracer.start_as_current_span("test"), patch(
-            "instana.instrumentation.google.cloud.storage.get_tracer_tuple",
-            return_value=(None, None, None),
+        with (
+            self.tracer.start_as_current_span("test"),
+            patch(
+                "instana.instrumentation.google.cloud.storage.get_tracer_tuple",
+                return_value=(None, None, None),
+            ),
+            client.batch() as batch_response,
         ):
-            with client.batch() as batch_response:
-                for obj in ["obj1", "obj2"]:
-                    bucket.delete_blob(obj)
-                assert batch_response
+            for obj in ["obj1", "obj2"]:
+                bucket.delete_blob(obj)
+            assert batch_response
 
     def _client(self, *args, **kwargs) -> storage.Client:
         # override the HTTP client to bypass the authorization

tests/clients/test_logging.py

@@ -100,12 +100,14 @@ def test_root_exit_span(self) -> None:
         assert spans[0].data["log"].get("message") == "foo bar"
 
     def test_exception(self) -> None:
-        with self.tracer.start_as_current_span("test"):
-            with patch(
+        with (
+            self.tracer.start_as_current_span("test"),
+            patch(
                 "instana.span.span.InstanaSpan.add_event",
                 side_effect=Exception("mocked error"),
-            ):
-                self.logger.warning("foo %s", "bar")
+            ),
+        ):
+            self.logger.warning("foo %s", "bar")
 
         spans = self.recorder.queued_spans()
 

tests/clients/test_mysqlclient.py

@@ -3,6 +3,7 @@
 
 
 import sys
+
 import MySQLdb
 import pytest
 
@@ -229,10 +230,9 @@ def test_error_capture(self):
         assert db_span.data["mysql"]["port"] == testenv["mysql_port"]
 
     def test_connect_cursor_ctx_mgr(self):
-        with self.tracer.start_as_current_span("test"):
-            with self.db as connection:
-                with connection.cursor() as cursor:
-                    affected_rows = cursor.execute("""SELECT * from users""")
+        with self.tracer.start_as_current_span("test"), self.db as connection:  # noqa: SIM117
+            with connection.cursor() as cursor:
+                affected_rows = cursor.execute("""SELECT * from users""")
 
         assert affected_rows == 1
         spans = self.recorder.queued_spans()
@@ -254,10 +254,9 @@ def test_connect_cursor_ctx_mgr(self):
         assert db_span.data["mysql"]["port"] == testenv["mysql_port"]
 
     def test_connect_ctx_mgr(self):
-        with self.tracer.start_as_current_span("test"):
-            with self.db as connection:
-                cursor = connection.cursor()
-                cursor.execute("""SELECT * from users""")
+        with self.tracer.start_as_current_span("test"), self.db as connection:
+            cursor = connection.cursor()
+            cursor.execute("""SELECT * from users""")
 
         spans = self.recorder.queued_spans()
         assert len(spans) == 2

tests/clients/test_pymysql.py

@@ -2,13 +2,13 @@
 # (c) Copyright Instana Inc. 2020
 
 
-import pytest
+from typing import Generator
 
 import pymysql
+import pytest
 
-from typing import Generator
-from tests.helpers import testenv
 from instana.singletons import agent, get_tracer
+from tests.helpers import testenv
 
 
 class TestPyMySQL:
@@ -255,10 +255,9 @@ def test_error_capture(self) -> None:
         assert db_span.data["mysql"]["port"] == testenv["mysql_port"]
 
     def test_connect_cursor_ctx_mgr(self) -> None:
-        with self.tracer.start_as_current_span("test"):
-            with self.db as connection:
-                with connection.cursor() as cursor:
-                    affected_rows = cursor.execute("""SELECT * from users""")
+        with self.tracer.start_as_current_span("test"), self.db as connection:  # noqa: SIM117
+            with connection.cursor() as cursor:
+                affected_rows = cursor.execute("""SELECT * from users""")
 
         assert affected_rows == 1
         spans = self.recorder.queued_spans()
@@ -280,10 +279,9 @@ def test_connect_cursor_ctx_mgr(self) -> None:
         assert db_span.data["mysql"]["port"] == testenv["mysql_port"]
 
     def test_connect_ctx_mgr(self) -> None:
-        with self.tracer.start_as_current_span("test"):
-            with self.db as connection:
-                cursor = connection.cursor()
-                cursor.execute("""SELECT * from users""")
+        with self.tracer.start_as_current_span("test"), self.db as connection:
+            cursor = connection.cursor()
+            cursor.execute("""SELECT * from users""")
 
         spans = self.recorder.queued_spans()
         assert len(spans) == 2

tests/test_fsm_cmdline.py

@@ -121,13 +121,13 @@ def test_get_cmdline_linux_proc(
 
     def test_get_cmdline_linux_proc_file_not_found(self) -> None:
         """Test _get_cmdline_linux_proc when file doesn't exist."""
-        with patch("builtins.open", side_effect=FileNotFoundError()):
+        with patch("builtins.open", side_effect=FileNotFoundError()):  # noqa: SIM117
             with pytest.raises(FileNotFoundError):
                 self.machine._get_cmdline_linux_proc()
 
     def test_get_cmdline_linux_proc_permission_error(self) -> None:
         """Test _get_cmdline_linux_proc with permission error."""
-        with patch("builtins.open", side_effect=PermissionError()):
+        with patch("builtins.open", side_effect=PermissionError()):  # noqa: SIM117
             with pytest.raises(PermissionError):
                 self.machine._get_cmdline_linux_proc()
 
@@ -187,11 +187,13 @@ def test_get_cmdline_unix_ps_empty_output(self) -> None:
 
     def test_get_cmdline_unix_ps_subprocess_error(self) -> None:
         """Test _get_cmdline_unix_ps when subprocess fails."""
-        with patch(
-            "subprocess.Popen", side_effect=subprocess.SubprocessError("Test error")
+        with (
+            patch(
+                "subprocess.Popen", side_effect=subprocess.SubprocessError("Test error")
+            ),
+            pytest.raises(subprocess.SubprocessError),
         ):
-            with pytest.raises(subprocess.SubprocessError):
-                self.machine._get_cmdline_unix_ps(1234)
+            self.machine._get_cmdline_unix_ps(1234)
 
     @pytest.mark.parametrize(
         "proc_exists,proc_content,expected_output",
@@ -263,18 +265,28 @@ def test_get_cmdline_platform_detection(
 
     def test_get_cmdline_windows_exception_fallback(self) -> None:
         """Test _get_cmdline falls back to sys.argv on Windows exception."""
-        with patch("instana.fsm.is_windows", return_value=True), patch.object(
-            self.machine, "_get_cmdline_windows", side_effect=Exception("Test error")
-        ), patch("instana.fsm.logger.debug") as mock_logger:
+        with (
+            patch("instana.fsm.is_windows", return_value=True),
+            patch.object(
+                self.machine,
+                "_get_cmdline_windows",
+                side_effect=Exception("Test error"),
+            ),
+            patch("instana.fsm.logger.debug") as mock_logger,
+        ):
             result = self.machine._get_cmdline(1234)
             assert result == sys.argv
             mock_logger.assert_called_once()
 
     def test_get_cmdline_unix_exception_fallback(self) -> None:
         """Test _get_cmdline falls back to sys.argv on Unix exception."""
-        with patch("instana.fsm.is_windows", return_value=False), patch.object(
-            self.machine, "_get_cmdline_unix", side_effect=Exception("Test error")
-        ), patch("instana.fsm.logger.debug") as mock_logger:
+        with (
+            patch("instana.fsm.is_windows", return_value=False),
+            patch.object(
+                self.machine, "_get_cmdline_unix", side_effect=Exception("Test error")
+            ),
+            patch("instana.fsm.logger.debug") as mock_logger,
+        ):
             result = self.machine._get_cmdline(1234)
             assert result == sys.argv
             mock_logger.assert_called_once()
@@ -298,18 +310,26 @@ def test_get_cmdline_unix_exception_fallback(self) -> None:
     )
     def test_get_cmdline_various_exceptions(self, exception_type: type) -> None:
         """Test _get_cmdline handles various exception types gracefully."""
-        with patch("instana.fsm.is_windows", return_value=False), patch.object(
-            self.machine, "_get_cmdline_unix", side_effect=exception_type("Test error")
+        with (
+            patch("instana.fsm.is_windows", return_value=False),
+            patch.object(
+                self.machine,
+                "_get_cmdline_unix",
+                side_effect=exception_type("Test error"),
+            ),
         ):
             result = self.machine._get_cmdline(1234)
             assert result == sys.argv
 
     def test_get_cmdline_with_actual_pid(self) -> None:
         """Test _get_cmdline with actual process ID."""
         current_pid = os.getpid()
-        with patch("instana.fsm.is_windows", return_value=False), patch.object(
-            self.machine, "_get_cmdline_unix", return_value=["test_cmd"]
-        ) as mock_method:
+        with (
+            patch("instana.fsm.is_windows", return_value=False),
+            patch.object(
+                self.machine, "_get_cmdline_unix", return_value=["test_cmd"]
+            ) as mock_method,
+        ):
             result = self.machine._get_cmdline(current_pid)
             assert result == ["test_cmd"]
             mock_method.assert_called_once_with(current_pid)