Remove globals in policy module

The policy module creates several globals on the module level, which
makes it very difficult to test code which interacts with the policies.

This patch creates a `WheelPolicies` class which encapsulates the
policy processing code and exposes functions to query the parsed
policies. Unfortunately, a lot of functions directly operate on
policies and these all need a new argument to take the new policy
object. This change should make testing substantially easier as policies
can be easily created on the fly and passed down to functions which
require these.

src/auditwheel/main_repair.py

@@ -6,20 +6,15 @@
 
 from auditwheel.patcher import Patchelf
 
-from .policy import (
-    POLICY_PRIORITY_HIGHEST,
-    get_policy_by_name,
-    get_policy_name,
-    get_priority_by_name,
-    load_policies,
-)
+from .policy import WheelPolicies
 from .tools import EnvironmentDefault
 
 logger = logging.getLogger(__name__)
 
 
 def configure_parser(sub_parsers):
-    policies = load_policies()
+    wheel_policy = WheelPolicies.load()
+    policies = wheel_policy.policies
     policy_names = [p["name"] for p in policies]
     policy_names += [alias for p in policies for alias in p["aliases"]]
     epilog = """PLATFORMS:
@@ -32,7 +27,7 @@ def configure_parser(sub_parsers):
         if len(p["aliases"]) > 0:
             epilog += f" (aliased by {', '.join(p['aliases'])})"
         epilog += "\n"
-    highest_policy = get_policy_name(POLICY_PRIORITY_HIGHEST)
+    highest_policy = wheel_policy.get_policy_name(wheel_policy.priority_highest)
     help = """Vendor in external shared library dependencies of a wheel.
 If multiple wheels are specified, an error processing one
 wheel will abort processing of subsequent wheels.
@@ -114,6 +109,8 @@ def execute(args, p):
     from .repair import repair_wheel
     from .wheel_abi import NonPlatformWheel, analyze_wheel_abi
 
+    wheel_policy = WheelPolicies.load()
+
     for wheel_file in args.WHEEL_FILE:
         if not isfile(wheel_file):
             p.error("cannot access %s. No such file" % wheel_file)
@@ -124,23 +121,23 @@ def execute(args, p):
             os.makedirs(args.WHEEL_DIR)
 
         try:
-            wheel_abi = analyze_wheel_abi(wheel_file)
+            wheel_abi = analyze_wheel_abi(wheel_policy, wheel_file)
         except NonPlatformWheel:
             logger.info(NonPlatformWheel.LOG_MESSAGE)
             return 1
 
-        policy = get_policy_by_name(args.PLAT)
+        policy = wheel_policy.get_policy_by_name(args.PLAT)
         reqd_tag = policy["priority"]
 
-        if reqd_tag > get_priority_by_name(wheel_abi.sym_tag):
+        if reqd_tag > wheel_policy.get_priority_by_name(wheel_abi.sym_tag):
             msg = (
                 'cannot repair "%s" to "%s" ABI because of the presence '
                 "of too-recent versioned symbols. You'll need to compile "
                 "the wheel on an older toolchain." % (wheel_file, args.PLAT)
             )
             p.error(msg)
 
-        if reqd_tag > get_priority_by_name(wheel_abi.ucs_tag):
+        if reqd_tag > wheel_policy.get_priority_by_name(wheel_abi.ucs_tag):
             msg = (
                 'cannot repair "%s" to "%s" ABI because it was compiled '
                 "against a UCS2 build of Python. You'll need to compile "
@@ -149,7 +146,7 @@ def execute(args, p):
             )
             p.error(msg)
 
-        if reqd_tag > get_priority_by_name(wheel_abi.blacklist_tag):
+        if reqd_tag > wheel_policy.get_priority_by_name(wheel_abi.blacklist_tag):
             msg = (
                 'cannot repair "%s" to "%s" ABI because it depends on '
                 "black-listed symbols." % (wheel_file, args.PLAT)
@@ -158,7 +155,7 @@ def execute(args, p):
 
         abis = [policy["name"]] + policy["aliases"]
         if not args.ONLY_PLAT:
-            if reqd_tag < get_priority_by_name(wheel_abi.overall_tag):
+            if reqd_tag < wheel_policy.get_priority_by_name(wheel_abi.overall_tag):
                 logger.info(
                     (
                         "Wheel is eligible for a higher priority tag. "
@@ -168,11 +165,12 @@ def execute(args, p):
                     args.PLAT,
                     wheel_abi.overall_tag,
                 )
-                higher_policy = get_policy_by_name(wheel_abi.overall_tag)
+                higher_policy = wheel_policy.get_policy_by_name(wheel_abi.overall_tag)
                 abis = [higher_policy["name"]] + higher_policy["aliases"] + abis
 
         patcher = Patchelf()
         out_wheel = repair_wheel(
+            wheel_policy,
             wheel_file,
             abis=abis,
             lib_sdir=args.LIB_SDIR,

src/auditwheel/main_show.py

@@ -2,6 +2,8 @@
 
 import logging
 
+from auditwheel.policy import WheelPolicies
+
 logger = logging.getLogger(__name__)
 
 
@@ -23,22 +25,17 @@ def execute(args, p):
     import json
     from os.path import basename, isfile
 
-    from .policy import (
-        POLICY_PRIORITY_HIGHEST,
-        POLICY_PRIORITY_LOWEST,
-        get_policy_name,
-        get_priority_by_name,
-        load_policies,
-    )
     from .wheel_abi import NonPlatformWheel, analyze_wheel_abi
 
+    wheel_policy = WheelPolicies.load()
+
     fn = basename(args.WHEEL_FILE)
 
     if not isfile(args.WHEEL_FILE):
         p.error("cannot access %s. No such file" % args.WHEEL_FILE)
 
     try:
-        winfo = analyze_wheel_abi(args.WHEEL_FILE)
+        winfo = analyze_wheel_abi(wheel_policy, args.WHEEL_FILE)
     except NonPlatformWheel:
         logger.info(NonPlatformWheel.LOG_MESSAGE)
         return 1
@@ -52,7 +49,10 @@ def execute(args, p):
         % (fn, winfo.overall_tag)
     )
 
-    if get_priority_by_name(winfo.pyfpe_tag) < POLICY_PRIORITY_HIGHEST:
+    if (
+        wheel_policy.get_priority_by_name(winfo.pyfpe_tag)
+        < wheel_policy.priority_highest
+    ):
         printp(
             "This wheel uses the PyFPE_jbuf function, which is not compatible with the"
             " manylinux1 tag. (see https://www.python.org/dev/peps/pep-0513/"
@@ -61,7 +61,7 @@ def execute(args, p):
         if args.verbose < 1:
             return
 
-    if get_priority_by_name(winfo.ucs_tag) < POLICY_PRIORITY_HIGHEST:
+    if wheel_policy.get_priority_by_name(winfo.ucs_tag) < wheel_policy.priority_highest:
         printp(
             "This wheel is compiled against a narrow unicode (UCS2) "
             "version of Python, which is not compatible with the "
@@ -81,7 +81,7 @@ def execute(args, p):
             "system-provided shared libraries: %s" % ", ".join(libs_with_versions)
         )
 
-    if get_priority_by_name(winfo.sym_tag) < POLICY_PRIORITY_HIGHEST:
+    if wheel_policy.get_priority_by_name(winfo.sym_tag) < wheel_policy.priority_highest:
         printp(
             (
                 'This constrains the platform tag to "%s". '
@@ -95,15 +95,17 @@ def execute(args, p):
         if args.verbose < 1:
             return
 
-    libs = winfo.external_refs[get_policy_name(POLICY_PRIORITY_LOWEST)]["libs"]
+    libs = winfo.external_refs[
+        wheel_policy.get_policy_name(wheel_policy.priority_lowest)
+    ]["libs"]
     if len(libs) == 0:
         printp("The wheel requires no external shared libraries! :)")
     else:
         printp("The following external shared libraries are required " "by the wheel:")
         print(json.dumps(dict(sorted(libs.items())), indent=4))
 
-    for p in sorted(load_policies(), key=lambda p: p["priority"]):
-        if p["priority"] > get_priority_by_name(winfo.overall_tag):
+    for p in sorted(wheel_policy.policies, key=lambda p: p["priority"]):
+        if p["priority"] > wheel_policy.get_priority_by_name(winfo.overall_tag):
             libs = winfo.external_refs[p["name"]]["libs"]
             if len(libs):
                 printp(

src/auditwheel/policy/__init__.py

@@ -18,6 +18,87 @@
 # https://docs.python.org/3/library/platform.html#platform.architecture
 bits = 8 * (8 if sys.maxsize > 2**32 else 4)
 
+_POLICY_JSON_MAP = {
+    Libc.GLIBC: _HERE / "manylinux-policy.json",
+    Libc.MUSL: _HERE / "musllinux-policy.json",
+}
+
+
+class WheelPolicies:
+    @staticmethod
+    def load():
+        libc_variant = get_libc()
+        policies_path = _POLICY_JSON_MAP[libc_variant]
+        policy_dict = json.loads(policies_path.read_text())
+        return WheelPolicies(policy_dict)
+
+    def __init__(self, policies: dict) -> None:
+        self._policies = []
+        self._musl_policy = _get_musl_policy()
+        self._arch_name = get_arch_name()
+        self._libc_variant = get_libc()
+
+        _validate_pep600_compliance(policies)
+        for policy in policies:
+            if self._musl_policy is not None and policy["name"] not in {
+                "linux",
+                self._musl_policy,
+            }:
+                continue
+            if (
+                self._arch_name in policy["symbol_versions"].keys()
+                or policy["name"] == "linux"
+            ):
+                if policy["name"] != "linux":
+                    policy["symbol_versions"] = policy["symbol_versions"][
+                        self._arch_name
+                    ]
+                policy["name"] = policy["name"] + "_" + self._arch_name
+                policy["aliases"] = [
+                    alias + "_" + self._arch_name for alias in policy["aliases"]
+                ]
+                policy["lib_whitelist"] = _fixup_musl_libc_soname(
+                    policy["lib_whitelist"]
+                )
+                self._policies.append(policy)
+
+        if self._libc_variant == Libc.MUSL:
+            assert len(self._policies) == 2, self._policies
+
+    @property
+    def policies(self):
+        return self._policies
+
+    @property
+    def priority_highest(self):
+        return max(p["priority"] for p in self._policies)
+
+    @property
+    def priority_lowest(self):
+        return min(p["priority"] for p in self._policies)
+
+    def get_policy_by_name(self, name: str) -> dict | None:
+        matches = [
+            p for p in self._policies if p["name"] == name or name in p["aliases"]
+        ]
+        if len(matches) == 0:
+            return None
+        if len(matches) > 1:
+            raise RuntimeError("Internal error. Policies should be unique")
+        return matches[0]
+
+    def get_policy_name(self, priority: int) -> str | None:
+        matches = [p["name"] for p in self._policies if p["priority"] == priority]
+        if len(matches) == 0:
+            return None
+        if len(matches) > 1:
+            raise RuntimeError("Internal error. priorities should be unique")
+        return matches[0]
+
+    def get_priority_by_name(self, name: str) -> int | None:
+        policy = self.get_policy_by_name(name)
+        return None if policy is None else policy["priority"]
+
 
 def get_arch_name() -> str:
     machine = _platform_module.machine()
@@ -65,22 +146,13 @@ def _validate_pep600_compliance(policies) -> None:
             symbol_versions[arch] = symbol_versions_arch
 
 
-_POLICY_JSON_MAP = {
-    Libc.GLIBC: _HERE / "manylinux-policy.json",
-    Libc.MUSL: _HERE / "musllinux-policy.json",
-}
-
-
 def _get_musl_policy():
     if _LIBC != Libc.MUSL:
         return None
     musl_version = get_musl_version(find_musl_libc())
     return f"musllinux_{musl_version.major}_{musl_version.minor}"
 
 
-_MUSL_POLICY = _get_musl_policy()
-
-
 def _fixup_musl_libc_soname(whitelist):
     if _LIBC != Libc.MUSL:
         return whitelist
@@ -105,60 +177,6 @@ def _fixup_musl_libc_soname(whitelist):
     return new_whitelist
 
 
-with _POLICY_JSON_MAP[_LIBC].open() as f:
-    _POLICIES = []
-    _policies_temp = json.load(f)
-    _validate_pep600_compliance(_policies_temp)
-    for _p in _policies_temp:
-        if _MUSL_POLICY is not None and _p["name"] not in {"linux", _MUSL_POLICY}:
-            continue
-        if _ARCH_NAME in _p["symbol_versions"].keys() or _p["name"] == "linux":
-            if _p["name"] != "linux":
-                _p["symbol_versions"] = _p["symbol_versions"][_ARCH_NAME]
-            _p["name"] = _p["name"] + "_" + _ARCH_NAME
-            _p["aliases"] = [alias + "_" + _ARCH_NAME for alias in _p["aliases"]]
-            _p["lib_whitelist"] = _fixup_musl_libc_soname(_p["lib_whitelist"])
-            _POLICIES.append(_p)
-    if _LIBC == Libc.MUSL:
-        assert len(_POLICIES) == 2, _POLICIES
-
-POLICY_PRIORITY_HIGHEST = max(p["priority"] for p in _POLICIES)
-POLICY_PRIORITY_LOWEST = min(p["priority"] for p in _POLICIES)
-
-
-def load_policies():
-    return _POLICIES
-
-
-def _load_policy_schema():
-    with open(join(dirname(abspath(__file__)), "policy-schema.json")) as f_:
-        schema = json.load(f_)
-    return schema
-
-
-def get_policy_by_name(name: str) -> dict | None:
-    matches = [p for p in _POLICIES if p["name"] == name or name in p["aliases"]]
-    if len(matches) == 0:
-        return None
-    if len(matches) > 1:
-        raise RuntimeError("Internal error. Policies should be unique")
-    return matches[0]
-
-
-def get_policy_name(priority: int) -> str | None:
-    matches = [p["name"] for p in _POLICIES if p["priority"] == priority]
-    if len(matches) == 0:
-        return None
-    if len(matches) > 1:
-        raise RuntimeError("Internal error. priorities should be unique")
-    return matches[0]
-
-
-def get_priority_by_name(name: str) -> int | None:
-    policy = get_policy_by_name(name)
-    return None if policy is None else policy["priority"]
-
-
 def get_replace_platforms(name: str) -> list[str]:
     """Extract platform tag replacement rules from policy
 
@@ -185,10 +203,14 @@ def get_replace_platforms(name: str) -> list[str]:
 from .external_references import lddtree_external_references  # noqa
 from .versioned_symbols import versioned_symbols_policy  # noqa
 
+def _load_policy_schema():
+    with open(join(dirname(abspath(__file__)), "policy-schema.json")) as f_:
+        schema = json.load(f_)
+    return schema
+
+
 __all__ = [
     "lddtree_external_references",
     "versioned_symbols_policy",
-    "load_policies",
-    "POLICY_PRIORITY_HIGHEST",
-    "POLICY_PRIORITY_LOWEST",
+    "WheelPolicies",
 ]