Source code for pyicat_plus.tests.servers.coilmq.pkg_resources

from importlib.resources import as_file
from importlib.resources import files
from typing import BinaryIO

_ENTERED_CONTEXT_MANAGERS = []




def resource_filename(package: str, resource: str) -> str:
    """
    Replacement for pkg_resources.resource_filename
    Returns a real filesystem path (even if package is zipped)
    """
    ref = files(package) / resource
    cm = as_file(ref)
    path = cm.__enter__()
    _ENTERED_CONTEXT_MANAGERS.append(cm)
    return str(path)





def resource_stream(package: str, resource: str) -> BinaryIO:
    """
    Replacement for pkg_resources.resource_stream
    Returns a file-like object (rb)
    """
    return (files(package) / resource).open("rb")





def cleanup() -> None:
    """
    Close any open context managers
    """
    global _ENTERED_CONTEXT_MANAGERS
    for cm in _ENTERED_CONTEXT_MANAGERS:
        try:
            cm.__exit__(None, None, None)
        except Exception:
            pass
    _ENTERED_CONTEXT_MANAGERS = []