Coverage for src/bluetooth_sig/utils/prewarm_catalog.py: 98%

1"""Authoritative registry prewarm catalogue. 

2 

3The catalogue derives prewarm targets from registry base-class descendants, 

4then appends explicit outliers that do not follow the shared registry base 

5hierarchy. 

6""" 

7 

8from __future__ import annotations 

9 

10import functools 

11import importlib 

12import pkgutil 

13import re 

14from collections.abc import Callable 

15from typing import Any 

16 

17from ..gatt.uuid_registry import get_uuid_registry 

18from ..registry.base import BaseGenericRegistry, BaseUUIDClassRegistry, BaseUUIDRegistry 

19 

20_REGISTRY_PACKAGE_ROOT = "bluetooth_sig.registry" 

21# These modules host BaseUUIDClassRegistry descendants used by public APIs. 

22_EXTRA_DISCOVERY_MODULES: tuple[str, ...] = ( 

23 "bluetooth_sig.gatt.characteristics.registry", 

24 "bluetooth_sig.gatt.services.registry", 

25) 

26 

27RegistryLoader = tuple[str, Callable[[], Any]] 

28 

29 

30def _to_snake_case(name: str) -> str: 

31 """Convert CamelCase/PascalCase names to snake_case.""" 

32 first_pass = re.sub(r"(.)([A-Z][a-z]+)", r"\1_\2", name) 

33 return re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", first_pass).lower() 

34 

35 

36def _iter_subclasses(root: type[Any]) -> list[type[Any]]: 

37 """Return all recursive subclasses of *root* (excluding the root itself).""" 

38 discovered: list[type[Any]] = [] 

39 stack = list(root.__subclasses__()) 

40 seen: set[type[Any]] = set() 

41 

42 while stack: 

43 cls = stack.pop() 

44 if cls in seen: 

45 continue 

46 seen.add(cls) 

47 discovered.append(cls) 

48 stack.extend(cls.__subclasses__()) 

49 

50 return discovered 

51 

52 

53def _import_registry_modules() -> None: 

54 """Import registry modules so subclass discovery has complete coverage.""" 

55 registry_pkg = importlib.import_module(_REGISTRY_PACKAGE_ROOT) 

56 if hasattr(registry_pkg, "__path__"): 

57 for module_info in pkgutil.walk_packages(registry_pkg.__path__, f"{_REGISTRY_PACKAGE_ROOT}."): 

58 importlib.import_module(module_info.name) 

59 

60 for module_name in _EXTRA_DISCOVERY_MODULES: 

61 importlib.import_module(module_name) 

62 

63 

64def _build_discovered_registry_loaders() -> tuple[RegistryLoader, ...]: 

65 """Build ensure-loaded loaders from registry base-class descendants.""" 

66 _import_registry_modules() 

67 

68 registry_classes: set[type[Any]] = set() 

69 registry_classes.update(_iter_subclasses(BaseGenericRegistry)) 

70 registry_classes.update(_iter_subclasses(BaseUUIDRegistry)) 

71 registry_classes.update(_iter_subclasses(BaseUUIDClassRegistry)) 

72 

73 loaders: list[RegistryLoader] = [] 

74 for registry_cls in sorted(registry_classes, key=lambda cls: f"{cls.__module__}.{cls.__qualname__}"): 

75 loader_name = _to_snake_case(registry_cls.__name__.replace("Registry", "")) + "_registry" 

76 loader = registry_cls.get_instance().ensure_loaded 

77 loaders.append((loader_name, loader)) 

78 

79 return tuple(loaders) 

80 

81 

82def _build_outlier_loaders() -> tuple[RegistryLoader, ...]: 

83 """Build loaders for registries that do not use shared base registry classes.""" 

84 return (("uuid_registry", lambda: get_uuid_registry().ensure_loaded()),) 

85 

86 

87@functools.lru_cache(maxsize=1) 

88def _cached_prewarm_loaders() -> tuple[RegistryLoader, ...]: 

89 """Build and cache the full prewarm loader list.""" 

90 return _build_discovered_registry_loaders() + _build_outlier_loaders() 

91 

92 

93def get_prewarm_loaders() -> tuple[RegistryLoader, ...]: 

94 """Return cached prewarm loaders, building discovery state on first call.""" 

95 return _cached_prewarm_loaders()