fix: replace pkg_resources with importlib.resources/metadata for setuptools 82 compat

Author: tagaws

setup.py

@@ -57,6 +57,7 @@ def find_version(*file_paths):
         "cfn_flip>=1.2.3",
         "nested-lookup",
         "botocore>=1.31.17",
+        "importlib_resources>=5.0;python_version<'3.9'",
     ],
     entry_points={
         "console_scripts": ["cfn-cli = rpdk.core.cli:main", "cfn = rpdk.core.cli:main"]

src/rpdk/core/data_loaders.py

@@ -6,7 +6,10 @@
 from io import TextIOWrapper
 from pathlib import Path
 
-import pkg_resources
+try:
+    from importlib.resources import files as importlib_resources_files
+except ImportError:  # Python < 3.9: importlib.resources.files() added in 3.9
+    from importlib_resources import files as importlib_resources_files
 import referencing
 import referencing.exceptions
 import yaml
@@ -38,7 +41,9 @@ def resource_stream(package_name, resource_name, encoding="utf-8"):
     Decoding errors raise :exc:`ValueError`. :term:`universal newlines`
     are enabled. Can be used in a ``with`` statement.
     """
-    f = pkg_resources.resource_stream(package_name, resource_name)
+    import sys
+    pkg = sys.modules[package_name].__spec__.parent or package_name
+    f = importlib_resources_files(pkg).joinpath(resource_name).open("rb")
     return TextIOWrapper(f, encoding=encoding)
 
 
@@ -55,8 +60,10 @@ def resource_yaml(package_name, resource_name):
 
 
 def copy_resource(package_name, resource_name, out_path):
-    with pkg_resources.resource_stream(
-        package_name, resource_name
+    import sys
+    pkg = sys.modules[package_name].__spec__.parent or package_name
+    with importlib_resources_files(pkg).joinpath(resource_name).open(
+        "rb"
     ) as fsrc, out_path.open("wb") as fdst:
         shutil.copyfileobj(fsrc, fdst)
 

src/rpdk/core/plugin_registry.py

@@ -1,35 +1,32 @@
-import pkg_resources
+try:
+    from importlib.metadata import entry_points as importlib_entry_points
+except ImportError:  # Python < 3.8
+    from importlib_metadata import entry_points as importlib_entry_points
+
+
+def _iter_entry_points(group):
+    eps = importlib_entry_points()
+    if hasattr(eps, "select"):  # Python 3.12+
+        return eps.select(group=group)
+    return eps.get(group, [])
+
 
 PLUGIN_REGISTRY = {
     entry_point.name: entry_point.load
-    for entry_point in pkg_resources.iter_entry_points("rpdk.v1.languages")
+    for entry_point in _iter_entry_points("rpdk.v1.languages")
 }
 
 
 def get_plugin_choices():
-    plugin_choices = [
-        entry_point.name
-        for entry_point in pkg_resources.iter_entry_points("rpdk.v1.languages")
-    ]
-    return sorted(set(plugin_choices))
+    return sorted({ep.name for ep in _iter_entry_points("rpdk.v1.languages")})
 
 
 def get_parsers():
-    parsers = {
-        entry_point.name: entry_point.load
-        for entry_point in pkg_resources.iter_entry_points("rpdk.v1.parsers")
-    }
-
-    return parsers
+    return {ep.name: ep.load for ep in _iter_entry_points("rpdk.v1.parsers")}
 
 
 def get_extensions():
-    extensions = {
-        entry_point.name: entry_point.load
-        for entry_point in pkg_resources.iter_entry_points("rpdk.v1.extensions")
-    }
-
-    return extensions
+    return {ep.name: ep.load for ep in _iter_entry_points("rpdk.v1.extensions")}
 
 
 def load_plugin(language):

tests/functional_importlib_compat.py

@@ -0,0 +1,68 @@
+"""
+Integration test: validates that data_loaders and plugin_registry work correctly
+without pkg_resources, using only importlib.resources / importlib.metadata.
+
+Run with:
+    pytest tests/functional_importlib_compat.py -v
+"""
+import sys
+import importlib
+import pytest
+
+
+def test_no_pkg_resources_imported_by_data_loaders():
+    """data_loaders must not import pkg_resources at all."""
+    # Force reimport to catch top-level imports
+    if "rpdk.core.data_loaders" in sys.modules:
+        del sys.modules["rpdk.core.data_loaders"]
+
+    import unittest.mock as mock
+    with mock.patch.dict("sys.modules", {"pkg_resources": None}):
+        # Should not raise ModuleNotFoundError
+        import rpdk.core.data_loaders  # noqa: F401
+
+
+def test_no_pkg_resources_imported_by_plugin_registry():
+    """plugin_registry must not import pkg_resources at all."""
+    if "rpdk.core.plugin_registry" in sys.modules:
+        del sys.modules["rpdk.core.plugin_registry"]
+
+    import unittest.mock as mock
+    with mock.patch.dict("sys.modules", {"pkg_resources": None}):
+        import rpdk.core.plugin_registry  # noqa: F401
+
+
+def test_resource_json_loads_real_schema():
+    """resource_json must load an actual bundled schema file end-to-end."""
+    from rpdk.core.data_loaders import resource_json
+    schema = resource_json(
+        "rpdk.core", "data/schema/provider.definition.schema.v1.json"
+    )
+    assert "$schema" in schema or "properties" in schema
+
+
+def test_resource_stream_returns_readable_content():
+    """resource_stream must return a readable text stream for a bundled file."""
+    from rpdk.core.data_loaders import resource_stream
+    with resource_stream(
+        "rpdk.core", "data/schema/provider.definition.schema.v1.json"
+    ) as f:
+        content = f.read()
+    assert len(content) > 0
+    assert "$schema" in content or "properties" in content
+
+
+def test_plugin_registry_get_plugin_choices_does_not_raise():
+    """get_plugin_choices must not raise even with no plugins installed."""
+    from rpdk.core.plugin_registry import get_plugin_choices
+    choices = get_plugin_choices()
+    assert isinstance(choices, list)
+
+
+def test_importlib_resources_files_available():
+    """Verify the compat shim resolves correctly on this Python version."""
+    if sys.version_info >= (3, 9):
+        from importlib.resources import files
+    else:
+        from importlib_resources import files
+    assert callable(files)

tests/test_data_loaders.py

@@ -6,7 +6,7 @@
 from io import BytesIO, StringIO
 from pathlib import Path
 from subprocess import check_output
-from unittest.mock import ANY, create_autospec, patch
+from unittest.mock import ANY, Mock, create_autospec, patch
 
 import pytest
 import yaml
@@ -620,10 +620,18 @@ def plugin():
 
 def mock_pkg_resource_stream(bytes_in, func=resource_stream):
     resource_name = "data/test.utf-8"
-    target = "rpdk.core.data_loaders.pkg_resources.resource_stream"
-    with patch(target, autospec=True, return_value=BytesIO(bytes_in)) as mock_stream:
+    target = "rpdk.core.data_loaders.importlib_resources_files"
+    mock_open = Mock(return_value=BytesIO(bytes_in))
+    mock_path = Mock()
+    mock_path.open = mock_open
+    mock_joinpath = Mock(return_value=mock_path)
+    mock_files = Mock()
+    mock_files.joinpath = mock_joinpath
+    with patch(target, return_value=mock_files) as mock_stream:
         f = func(__name__, resource_name)
-    mock_stream.assert_called_once_with(__name__, resource_name)
+    mock_stream.assert_called_once_with("tests")
+    mock_joinpath.assert_called_once_with(resource_name)
+    mock_open.assert_called_once_with("rb")
     return f
 
 

tests/test_plugin_registry.py

@@ -18,7 +18,7 @@ def test_get_extensions():
     mock_entrypoint_2 = Mock()
 
     patch_iter_entry_points = patch(
-        "rpdk.core.plugin_registry.pkg_resources.iter_entry_points"
+        "rpdk.core.plugin_registry._iter_entry_points"
     )
     with patch_iter_entry_points as mock_iter_entry_points:
         mock_iter_entry_points.return_value = [mock_entrypoint_1, mock_entrypoint_2]

tests/utils.py

@@ -5,7 +5,8 @@
 from random import sample
 from unittest.mock import Mock, patch
 
-import pkg_resources
+import importlib.metadata
+from unittest.mock import patch as _patch
 
 from rpdk.core.project import Project
 
@@ -74,14 +75,13 @@ def chdir(path):
 
 
 def add_dummy_language_plugin():
-    distribution = pkg_resources.Distribution(__file__)
-    entry_point = pkg_resources.EntryPoint.parse(
-        "dummy = rpdk.dummy:DummyLanguagePlugin", dist=distribution
+    ep = importlib.metadata.EntryPoint(
+        name="dummy", value="rpdk.dummy:DummyLanguagePlugin", group="rpdk.v1.languages"
     )
-    distribution._ep_map = {  # pylint: disable=protected-access
-        "rpdk.v1.languages": {"dummy": entry_point}
-    }
-    pkg_resources.working_set.add(distribution)
+    _patch(
+        "rpdk.core.plugin_registry._iter_entry_points",
+        side_effect=lambda group: [ep] if group == "rpdk.v1.languages" else [],
+    ).start()
 
 
 def get_mock_project():