Block non-abstract transport imports

tools/pylint-extensions/pylint-guidelines-checker/README.md

@@ -74,4 +74,6 @@ In the case of a false positive, use the disable command to remove the pylint er
 | client-accepts-api-version-keyword                | Ensure that the client constructor accepts a keyword-only api_version argument.                                                                       | # pylint:disable=client-accepts-api-version-keyword                                  | [link](https://azure.github.io/azure-sdk/python_design.html#specifying-the-service-version)                     |
 | enum-must-be-uppercase                 |  The enum name must be all uppercase.                                                                                                           | # pylint:disable=enum-must-be-uppercase                                      | [link](https://azure.github.io/azure-sdk/python_design.html#enumerations)                     |
 | enum-must-inherit-case-insensitive-enum-meta                 |  The enum should inherit from CaseInsensitiveEnumMeta.                                                                                                           | # pylint:disable=enum-must-inherit-case-insensitive-enum-meta                                      | [link](https://azure.github.io/azure-sdk/python_implementation.html#extensible-enumerations)                     |
-| networking-import-outside-azure-core-transport                 | This import is only allowed in azure.core.pipeline.transport.                                                                                                          | # pylint:disable=networking-import-outside-azure-core-transport                                     | [link](https://github.com/Azure/azure-sdk-for-python/issues/24989)                     |
+| networking-import-outside-azure-core-transport                 | This import is only allowed in azure.core.pipeline.transport.                                                                                                          | # pylint:disable=networking-import-outside-azure-core-transport                                     | [link](https://github.com/Azure/azure-sdk-for-python/issues/24989)                     |
+| non-abstract-transport-import                      | Only import abstract transports. Let core or end-user decide which transport to use.                                                                                                            | # pylint:disable=non-abstract-transport-import                                     | [link](https://github.com/Azure/azure-sdk-for-python/issues/25533)                     |
+

tools/pylint-extensions/pylint-guidelines-checker/pylint_guidelines_checker.py

@@ -1956,6 +1956,34 @@ def _check_import(self, name, node):
                     msgid=f"networking-import-outside-azure-core-transport", node=node, confidence=None
                 )
 
+class NonAbstractTransportImport(BaseChecker):
+    """Rule to check that we aren't import transports outside of `azure.core.pipeline.transport`.
+    Transport selection should be up to `azure.core` or the end-user, not individual SDKs.
+    """
+    name = "non-abstract-transport-import"
+    priority = -1
+    msgs = {
+        "C4750": (
+            "Only import abstract transports.",
+            "non-abstract-transport-import",
+            "Only import abstract transports. Let core or end-user decide which transport to use.",
+        ),
+    }
+    AZURE_CORE_TRANSPORT_NAME = "azure.core.pipeline.transport"
+    ABSTRACT_CLASSES = {"HttpTransport", "HttpRequest", "HttpResponse", "AsyncHttpTransport", "AsyncHttpResponse"}
+
+    def visit_importfrom(self, node):
+        """Check that we aren't import from a blocked package."""
+        if node.root().name.startswith(self.AZURE_CORE_TRANSPORT_NAME): 
+            return
+        if node.modname == self.AZURE_CORE_TRANSPORT_NAME: 
+            for name, _ in node.names:
+                if name not in self.ABSTRACT_CLASSES:
+                    self.add_message(
+                        msgid=f"non-abstract-transport-import",
+                        node=node,
+                        confidence=None,
+                    )
 
 # if a linter is registered in this function then it will be checked with pylint
 def register(linter):
@@ -1979,6 +2007,7 @@ def register(linter):
     linter.register_checker(CheckEnum(linter))
     linter.register_checker(NonCoreNetworkImport(linter))
     linter.register_checker(ClientListMethodsUseCorePaging(linter))
+    linter.register_checker(NonAbstractTransportImport(linter))
 
 
 

tools/pylint-extensions/pylint-guidelines-checker/tests/test_pylint_custom_plugins.py

@@ -3073,3 +3073,45 @@ def test_allowed_imports(self):
         importfrom_node.root().name = "azure.core.pipeline.transport._private_module"
         with self.assertNoMessages():
             self.checker.visit_importfrom(importfrom_node)
+
+
+class TestCheckNonAbstractTransportImport(pylint.testutils.CheckerTestCase):
+    """Test that we are blocking disallowed imports and allowing allowed imports."""
+    CHECKER_CLASS = checker.NonAbstractTransportImport
+
+    def test_disallowed_imports(self):
+        """Check that illegal imports raise warnings"""
+        importfrom_node = astroid.extract_node("from azure.core.pipeline.transport import RequestsTransport")
+        with self.assertAddsMessages(
+                pylint.testutils.MessageTest(
+                    msg_id="non-abstract-transport-import",
+                    line=1,
+                    node=importfrom_node,
+                    col_offset=0,
+                )
+        ):
+            self.checker.visit_importfrom(importfrom_node)
+
+    def test_allowed_imports(self):
+        """Check that allowed imports don't raise warnings."""
+        # import not in the blocked list.
+        importfrom_node = astroid.extract_node("from math import PI")
+        with self.assertNoMessages():
+            self.checker.visit_importfrom(importfrom_node)
+
+        # from import not in the blocked list.
+        importfrom_node = astroid.extract_node("from azure.core.pipeline import Pipeline")
+        with self.assertNoMessages():
+            self.checker.visit_importfrom(importfrom_node)
+
+        # Import abstract classes
+        importfrom_node = astroid.extract_node("from azure.core.pipeline.transport import HttpTransport, HttpRequest, HttpResponse, AsyncHttpTransport, AsyncHttpResponse")
+        with self.assertNoMessages():
+            self.checker.visit_importfrom(importfrom_node)
+
+        # Import non-abstract classes, but from in `azure.core.pipeline.transport`.
+        importfrom_node = astroid.extract_node("from azure.core.pipeline.transport import RequestsTransport, AioHttpTransport, AioHttpTransportResponse")
+        importfrom_node.root().name = "azure.core.pipeline.transport._private_module"
+        with self.assertNoMessages():
+            self.checker.visit_importfrom(importfrom_node)
+