Merge branch 'feature/6' into dev

ID: 6
Type: feature
Title: implement importlib.resources handler for urllib
Status: done
Priority: high
Created: 2025-06-20
Description: A handler that can be registered with an urllib.request
             OpenerDirector to open importlib.resources package files.

Makefile

@@ -16,7 +16,7 @@ configure: configure.ac
 	.venv/bin/python3 -m pip install --upgrade pip
 	.venv/bin/pip install -r requirements-dev.txt
 
-test-reports: test-reports/unit test-reports/static
+test-reports: test-reports/unit test-reports/static test-reports/integration
 
 test-reports/unit:
 	python3 -m pipenv run -v test-unit

TODO

@@ -113,7 +113,7 @@ Description: Implement my custom algorithm for doing rolling hash string search
 ID: 6
 Type: feature
 Title: implement importlib.resources handler for urllib
-Status: in-progress
+Status: done
 Priority: high
 Created: 2025-06-20
 Description: A handler that can be registered with an urllib.request

src/byteb4rb1e/utils/testing/pytest/__init__.py

@@ -0,0 +1,14 @@
+import os
+from pathlib import Path
+from typing import Tuple
+
+
+def get_current_test() -> Tuple[Path, str]:
+    current_test_env = os.getenv("PYTEST_CURRENT_TEST")
+    if current_test_env is None:
+        raise RuntimeError("PYTEST_CURRENT_TEST not set. Must be run under pytest.")
+
+    suite_path, case_name = current_test_env.split('::', 1)
+    case_name = case_name.split(' ', 1)[0]
+    return Path(suite_path).resolve(), case_name
+

src/byteb4rb1e/utils/testing/pytest/decorators.py

@@ -0,0 +1,47 @@
+from functools import wraps
+from pathlib import Path
+import os
+import subprocess
+import sys
+
+from byteb4rb1e.utils.testing.pytest import get_current_test
+
+
+def run_in_subprocess_once():
+    """
+    A decorator that reruns th test in a subprocess if not already inside one.
+    Requires pytest to be installed and test to be run by pytest.
+
+    For what? Anything that can't be done in a thread-safe manner, e.g. modifying PYTHON_PATH
+    """
+    def decorator(test_func):
+        @wraps(test_func)
+        def wrapper(*args, **kwargs):
+            if os.environ.get("XPYTEST_INSIDE_SUBPROCESS") == "1":
+                return test_func(*args, **kwargs)
+
+            suite_path, case_name = get_current_test()
+
+            cmd = [
+                sys.executable,
+                "-m", "pytest",
+                f"{suite_path}::{case_name}",
+            ] 
+
+            result = subprocess.run(
+                cmd,
+                env={**os.environ, "XPYTEST_INSIDE_SUBPROCESS": "1"},
+                capture_output=True,
+                text=True,
+            )
+
+            if result.returncode != 0:
+                print(' '.join(cmd))
+                print("==== Subprocess stdout ====")
+                print(result.stdout)
+                print("==== Subprocess stderr ====")
+                print(result.stderr)
+                raise AssertionError(f"Subprocess test failed with exit code {result.returncode}")
+        return wrapper
+    return decorator
+

src/byteb4rb1e/utils/testing/pytest/fixtures.py

@@ -1,14 +1,40 @@
 import os
 from pathlib import Path
-from typing import Tuple
+import sys
+from typing import Dict, Tuple, Union
 
 import pytest
 
+from byteb4rb1e.utils.testing.pytest import get_current_test
+
 
 @pytest.fixture
 def current_test() -> Tuple[Path, str]:
     """
     """
-    suite_path, case_name = os.getenv('PYTEST_CURRENT_TEST').split('::', 1)
+    return get_current_test()
-    case_name = case_name.split(' ', 1)[0]
+
-    return Path(suite_path).resolve(), case_name
+
+@pytest.fixture
+def mock_pkg(tmp_path):
+    def _create(name: str, files: Dict[str, Union[str, bytes]]):
+        pkg_path = tmp_path / name.replace('.', os.path.sep)
+        pkg_path.mkdir(parents=True)
+        (pkg_path / "__init__.py").touch()
+
+        for fname, content in files.items():
+            fpath = (pkg_path / fname)
+            fpath.parent.mkdir(parents=True, exist_ok=True)
+            if isinstance(content, str):
+                fpath.write_text(content)
+            else:
+                fpath.write_bytes(content)
+
+        sys.path.insert(0, str(tmp_path))
+        return name, pkg_path
+
+    yield _create
+
+    # cleanup sys.path after test
+    if str(tmp_path) in sys.path:
+        sys.path.remove(str(tmp_path))

src/byteb4rb1e/utils/urllib/request.py

@@ -0,0 +1,38 @@
+import email
+import importlib.resources
+import mimetypes
+from urllib.request import URLError
+import urllib.request
+
+
+class PkgHandler(urllib.request.BaseHandler):
+    """
+    """
+    def pkg_open(self, req) -> urllib.request.addinfourl:
+        pkg_files = importlib.resources.files(req.host)
+
+        try:
+            fh = next(
+                pkg_files.glob(req.selector.lstrip('//'))
+            ).open('rb')
+        except Exception as e:
+            raise URLError(f'{e.__class__.__name__}: {e}') from e
+
+        fh.seek(0, 2);
+        size = fh.tell();
+        fh.seek(0);
+
+        mtype, _ = mimetypes.guess_type(req.selector)
+
+        headers = email.message_from_string(
+            'Content-Type: %s\nContent-Length: %d\n' %
+            (mtype or 'text/plain', size)
+        )
+
+        if not mtype or mtype.startswith('text/'):
+            fh.close()
+            fh = next(
+                pkg_files.glob(req.selector.lstrip('//'))
+            ).open('r')
+
+        return urllib.request.addinfourl(fh, headers, None)

tests/integration/byteb4rb1e/utils/testing/pytest/test_.py

@@ -0,0 +1,33 @@
+import os
+from pathlib import Path
+
+import pytest
+
+pytestmark = pytest.mark.pytest
+
+from byteb4rb1e.utils.testing.pytest import get_current_test
+from byteb4rb1e.utils.testing.pytest.decorators import run_in_subprocess_once
+
+
+class Test_get_current_test:
+    """
+    """
+
+    def test_default(self):
+        """
+        """
+        os.environ['PYTEST_CURRENT_TEST'] = 'foo::bar (something)'
+
+        result = get_current_test()
+
+        assert isinstance(result[0], Path)
+        assert str(result[0].name) == 'foo'
+
+        assert result[1] == 'bar'
+
+    def test_invalid(self):
+        """
+        """
+        del os.environ['PYTEST_CURRENT_TEST']
+        with pytest.raises(RuntimeError):
+            get_current_test()

tests/integration/byteb4rb1e/utils/testing/pytest/test_decorators.py

@@ -0,0 +1,21 @@
+from pathlib import Path
+
+import pytest
+
+pytestmark = pytest.mark.pytest
+
+from byteb4rb1e.utils.testing.pytest.decorators import run_in_subprocess_once
+
+
+@run_in_subprocess_once()
+def test_run_in_subprocess_once(tmp_path):
+    marker = tmp_path / "executed_in_subprocess.txt"
+
+    if marker.exists():
+        raise AssertionError("Marker file exists before test logic ran (shouldn't happen in parent process)")
+
+    # Create proof of execution
+    marker.write_text("Subprocess was here.")
+
+    # Now assert it
+    assert marker.exists()

tests/integration/byteb4rb1e/utils/testing/pytest/test_fixtures.py

@@ -1,10 +1,12 @@
 from pathlib import Path
+import importlib.resources
 
 import pytest
 
 pytestmark = pytest.mark.pytest
 
-from byteb4rb1e.utils.testing.pytest.fixtures import current_test
+from byteb4rb1e.utils.testing.pytest.decorators import run_in_subprocess_once
+from byteb4rb1e.utils.testing.pytest.fixtures import current_test, mock_pkg
 
 
 def test_current_test(current_test):
@@ -14,3 +16,18 @@ def test_current_test(current_test):
 
     assert str(Path(__file__)) == str(suite_path)
     assert case_name == "test_current_test"
+
+
+@run_in_subprocess_once()
+def test_mock_pkg(mock_pkg):
+    """
+    """
+    dummy_data = 'Hello'
+
+    mock_pkg('foobarpkg', {
+        'data.txt': dummy_data
+    })
+
+    result = next(importlib.resources.files('foobarpkg').glob('data.txt')).read_text()
+
+    assert result == dummy_data

tests/unit/byteb4rb1e/utils/urllib/test_request.py

@@ -0,0 +1,90 @@
+import os.path
+import sys
+import urllib.request
+
+import pytest
+
+from byteb4rb1e.utils.testing.pytest.decorators import run_in_subprocess_once
+from byteb4rb1e.utils.testing.pytest.fixtures import mock_pkg
+from byteb4rb1e.utils.urllib.request import PkgHandler
+
+
+class TestPkgHandler:
+    """
+    """
+    @run_in_subprocess_once()
+    def test_text(self, mock_pkg):
+        """
+        """
+        _opener: urllib.request.OpenerDirector =  urllib.request.build_opener(
+            PkgHandler()
+        )
+
+        dummy_data = 'Hello'
+
+        mock_pkg('foobarpkg', {
+            'data.txt': dummy_data
+        })
+
+        result = _opener.open('pkg://foobarpkg/data.txt').readline()
+
+        assert isinstance(result, str)
+        assert result == dummy_data
+
+
+    @run_in_subprocess_once()
+    def test_bytes(self, mock_pkg):
+        """
+        """
+        _opener: urllib.request.OpenerDirector =  urllib.request.build_opener(
+            PkgHandler()
+        )
+
+        dummy_data = b'foobar123'
+
+        mock_pkg('foobarpkg', {
+            'data.bin': dummy_data
+        })
+
+        result = _opener.open('pkg://foobarpkg/data.bin').readline()
+
+        assert isinstance(result, bytes)
+        assert result == dummy_data
+
+
+    @run_in_subprocess_once()
+    def test_subdir(self, mock_pkg):
+        """
+        """
+        _opener: urllib.request.OpenerDirector =  urllib.request.build_opener(
+            PkgHandler()
+        )
+
+        dummy_data = 'foobar123'
+
+        mock_pkg('foobarpkg', {
+            'foo/bar/data.txt': dummy_data
+        })
+
+        result = _opener.open('pkg://foobarpkg/foo/bar/data.txt').readline()
+
+        assert result == dummy_data
+
+
+    @run_in_subprocess_once()
+    def test_nested_module(self, mock_pkg):
+        """
+        """
+        _opener: urllib.request.OpenerDirector =  urllib.request.build_opener(
+            PkgHandler()
+        )
+
+        dummy_data = 'foobar123'
+
+        mock_pkg('foo.bar.pkg', {
+            'dummy/data.txt': dummy_data
+        })
+
+        result = _opener.open('pkg://foo.bar.pkg/dummy/data.txt').readline()
+
+        assert result == dummy_data