@ -54,6 +54,14 @@ def main():
except ImportError :
from io import StringIO
try :
from importlib . util import spec_from_loader , module_from_spec
from importlib . machinery import SourceFileLoader , ModuleSpec # pylint: disable=unused-import
except ImportError :
has_py3_loader = False
else :
has_py3_loader = True
if collection_full_name :
# allow importing code from collections when testing a collection
from ansible . module_utils . common . text . converters import to_bytes , to_text , to_native , text_type
@ -149,12 +157,25 @@ def main():
self . loaded_modules = set ( )
self . restrict_to_module_paths = restrict_to_module_paths
def find_spec ( self , fullname , path = None , target = None ) : # pylint: disable=unused-argument
# type: (RestrictedModuleLoader, str, list[str], types.ModuleType | None ) -> ModuleSpec | None | ImportError
""" Return the spec from the loader or None """
loader = self . _get_loader ( fullname , path = path )
if loader is not None :
if has_py3_loader :
# loader is expected to be Optional[importlib.abc.Loader], but RestrictedModuleLoader does not inherit from importlib.abc.Loder
return spec_from_loader ( fullname , loader ) # type: ignore[arg-type]
raise ImportError ( " Failed to import ' %s ' due to a bug in ansible-test. Check importlib imports for typos. " % fullname )
return None
def find_module ( self , fullname , path = None ) :
""" Return self if the given fullname is restricted, otherwise return None.
: param fullname : str
: param path : str
: return : RestrictedModuleLoader | None
"""
# type: (RestrictedModuleLoader, str, list[str]) -> RestrictedModuleLoader | None
""" Return self if the given fullname is restricted, otherwise return None. """
return self . _get_loader ( fullname , path = path )
def _get_loader ( self , fullname , path = None ) :
# type: (RestrictedModuleLoader, str, list[str]) -> RestrictedModuleLoader | None
""" Return self if the given fullname is restricted, otherwise return None. """
if fullname in self . loaded_modules :
return None # ignore modules that are already being loaded
@ -191,27 +212,49 @@ def main():
# not a namespace we care about
return None
def create_module ( self , spec ) : # pylint: disable=unused-argument
# type: (RestrictedModuleLoader, ModuleSpec) -> None
""" Return None to use default module creation. """
return None
def exec_module ( self , module ) :
# type: (RestrictedModuleLoader, types.ModuleType) -> None | ImportError
""" Execute the module if the name is ansible.module_utils.basic and otherwise raise an ImportError """
fullname = module . __spec__ . name
if fullname == ' ansible.module_utils.basic ' :
self . loaded_modules . add ( fullname )
for path in convert_ansible_name_to_absolute_paths ( fullname ) :
if not os . path . exists ( path ) :
continue
loader = SourceFileLoader ( fullname , path )
spec = spec_from_loader ( fullname , loader )
real_module = module_from_spec ( spec )
loader . exec_module ( real_module )
real_module . AnsibleModule = ImporterAnsibleModule # type: ignore[attr-defined]
real_module . _load_params = lambda * args , * * kwargs : { } # type: ignore[attr-defined] # pylint: disable=protected-access
sys . modules [ fullname ] = real_module
return None
raise ImportError ( ' could not find " %s " ' % fullname )
raise ImportError ( ' import of " %s " is not allowed in this context ' % fullname )
def load_module ( self , fullname ) :
""" Raise an ImportError.
: type fullname : str
"""
# type: (RestrictedModuleLoader, str) -> types.ModuleType | ImportError
""" Return the module if the name is ansible.module_utils.basic and otherwise raise an ImportError. """
if fullname == ' ansible.module_utils.basic ' :
module = self . __load_module ( fullname )
# stop Ansible module execution during AnsibleModule instantiation
module . AnsibleModule = ImporterAnsibleModule
module . AnsibleModule = ImporterAnsibleModule # type: ignore[attr-defined]
# no-op for _load_params since it may be called before instantiating AnsibleModule
module . _load_params = lambda * args , * * kwargs : { } # pylint: disable=protected-access
module . _load_params = lambda * args , * * kwargs : { } # type: ignore[attr-defined] # pylint: disable=protected-access
return module
raise ImportError ( ' import of " %s " is not allowed in this context ' % fullname )
def __load_module ( self , fullname ) :
""" Load the requested module while avoiding infinite recursion.
: type fullname : str
: rtype : module
"""
# type: (RestrictedModuleLoader, str) -> types.ModuleType
""" Load the requested module while avoiding infinite recursion. """
self . loaded_modules . add ( fullname )
return import_module ( fullname )
@ -514,33 +557,6 @@ def main():
" ignore " ,
" Python 3.5 support will be dropped in the next release of cryptography. Please upgrade your Python. " )
if sys . version_info > = ( 3 , 10 ) :
# Temporary solution for Python 3.10 until find_spec is implemented in RestrictedModuleLoader.
# That implementation is dependent on find_spec being added to the controller's collection loader first.
# The warning text is: main.<locals>.RestrictedModuleLoader.find_spec() not found; falling back to find_module()
warnings . filterwarnings (
" ignore " ,
r " main \ .<locals> \ .RestrictedModuleLoader \ .find_spec \ ( \ ) not found; falling back to find_module \ ( \ ) " ,
)
# Temporary solution for Python 3.10 until exec_module is implemented in RestrictedModuleLoader.
# That implementation is dependent on exec_module being added to the controller's collection loader first.
# The warning text is: main.<locals>.RestrictedModuleLoader.exec_module() not found; falling back to load_module()
warnings . filterwarnings (
" ignore " ,
r " main \ .<locals> \ .RestrictedModuleLoader \ .exec_module \ ( \ ) not found; falling back to load_module \ ( \ ) " ,
)
# Temporary solution for Python 3.10 until find_spec is implemented in the controller's collection loader.
warnings . filterwarnings (
" ignore " ,
r " _Ansible.*Finder \ .find_spec \ ( \ ) not found; falling back to find_module \ ( \ ) " ,
)
# Temporary solution for Python 3.10 until exec_module is implemented in the controller's collection loader.
warnings . filterwarnings (
" ignore " ,
r " _Ansible.*Loader \ .exec_module \ ( \ ) not found; falling back to load_module \ ( \ ) " ,
)
try :
yield
finally :
Sloane Hertel
2 years ago
committed by