Source code for pywbem_mock._wbemconnection_mock

#
# (C) Copyright 2018 InovaDevelopment.com
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this program; if not, write to the Free Software
# Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
#
# Author: Karl  Schopmeyer <inovadevelopment.com>
#

"""
Mock support for the WBEMConnection class to allow pywbem users to test the
pywbem client without requiring a running WBEM server.

For documentation, see mocksupport.rst.
"""

from __future__ import absolute_import, print_function

import os
import time
import re
from xml.dom import minidom
# try:
#    from unittest.mock import Mock
# except ImportError:
#    from mock import Mock
import six

from pywbem import WBEMConnection, CIMClass, CIMClassName, \
    CIMInstance, CIMInstanceName, CIMParameter, CIMQualifierDeclaration, \
    cimtype, CIMError, CIM_ERR_FAILED, DEFAULT_NAMESPACE, MOFCompiler, \
    DEFAULT_TIMEOUT
from pywbem._nocasedict import NocaseDict
from pywbem._utils import _format
from ._mainprovider import MainProvider

from ._inmemoryrepository import InMemoryRepository

from ._mockmofwbemconnection import _MockMOFWBEMConnection

from ._providerregistry import ProviderRegistry
from ._providerdependentregistry import ProviderDependentRegistry

from ._providerdispatcher import ProviderDispatcher

from ._dmtf_cim_schema import build_schema_mof
from ._utils import _uprint

from ._namespaceprovider import CIMNamespaceProvider

from ._subscriptionproviders import CIMIndicationSubscriptionProvider, \
    CIMListenerDestinationProvider, CIMIndicationFilterProvider

__all__ = ['FakedWBEMConnection']

# Fake Server default values for parameters that apply to repo and operations

# allowed output formats for the repository display
OUTPUT_FORMATS = ['mof', 'xml', 'repr']

# Issue #2065. We have not considered that iq and ico are deprecated in
# on DSP0200 for get_instance, etc. We could  set up a default to ignore these
# parameters for the operations in which they are deprecated and we
# should/could ignore them. We need to document our behavior in relation to the
# spec.


def _pretty_xml(xml_string):
    """
    Common function to produce pretty xml string from an input xml_string.

    This function is NOT intended to be used in major code paths since it
    uses the minidom to produce the prettified xml and that uses a lot
    of memory
    """

    result_dom = minidom.parseString(xml_string)
    pretty_result = result_dom.toprettyxml(indent='  ')

    # remove extra empty lines
    return re.sub(r'>( *[\r\n]+)+( *)<', r'>\n\2<', pretty_result)


def _cvt_rqd_classname(classname):
    """Convert required classname to string"""
    if isinstance(classname, CIMClassName):
        classname = classname.classname
    return classname


def _cvt_opt_classname(classname):
    """Convert optional classname to string if it exists"""
    if classname is None:
        return classname
    if isinstance(classname, CIMClassName):
        return classname.classname
    return classname


def _cvt_obj_name(objname):
    """Convert objectname to string if classname or return if inst name"""
    if isinstance(objname, CIMInstanceName):
        return objname
    if isinstance(objname, CIMClassName):
        return objname.classname
    return objname


[docs]class FakedWBEMConnection(WBEMConnection): """ A subclass of :class:`pywbem.WBEMConnection` that mocks the communication with a WBEM server by utilizing a local in-memory CIM repository to generate responses in the same way the WBEM server would. *New in pywbem 0.12 as experimental and finalized in 1.2.* Each :class:`~pywbem_mock.FakedWBEMConnection` object has its own CIM repository which contains multiple CIM namespaces, and each namespace may contain CIM qualifier types (declarations), CIM classes and CIM instances. This class provides only a subset of the init parameters of :class:`~pywbem.WBEMConnection` because it does not have a connection to a WBEM server. It uses a faked and fixed URL for the WBEM server (``http://FakedUrl``) as a means of identifying the connection by users. Logging of the faked operations is supported via the pywbem logging facility and can be controlled in the same way as for :class:`~pywbem.WBEMConnection`. For details, see :ref:`WBEM operation logging`. Some of the longer running methods of this class add time statistics to the :attr:`pywbem.WBEMConnection.statistics`. For details on how to get the statistics, see :ref:`WBEM operation statistics`. """ def __init__(self, default_namespace=DEFAULT_NAMESPACE, use_pull_operations=False, stats_enabled=False, timeout=DEFAULT_TIMEOUT, response_delay=None, disable_pull_operations=None, url=None): """ Parameters: default_namespace (:term:`string`): Default namespace. This parameter has the same characteristics as the same-named init parameter of :class:`~pywbem.WBEMConnection`. use_pull_operations (:class:`py:bool`): Flag to control whether pull or traditional operations are used in the iter... operations. This parameter has the same characteristics as the same-named init parameter of :class:`~pywbem.WBEMConnection`. timeout (:term:`number`): This parameter has the same characteristics as the same-named init parameter of :class:`~pywbem.WBEMConnection`. stats_enabled (:class:`py:bool`): Flag to enable operation statistics. This parameter has the same characteristics as the same-named init parameter of :class:`~pywbem.WBEMConnection`. response_delay (:term:`number`): Artifically created delay for each operation, in seconds. This must be a positive number. Delays less than a second or other fractional delays may be achieved with float numbers. `None` disables the delay. Note that the :attr:`~pywbem_mock.FakedWBEMConnection.response_delay` property can be used to set this delay subsequent to object creation. disable_pull_operations (:class:`py:bool`): Flag to allow user to disable the pull operations ( Open... and Pull.. requests). The default is None which enables pull operations to execute. Setting the flag to True causes pull operations to raise CIMError(CIM_ERR_NOT_SUPPORTED). The :attr:`~pywbem_mock.FakedWBEMConnection.disable_pull_operations` property can be used to set this variable. url (:term:`string`): Defines a url to replace the default http://FakedURL.5988 url which is passed to the superclass. The url must be an acceptable syntax for WBEMConnection initialization. This is useful when multiple mocks are required for testing. Raises: ValueError: Invalid arguments """ # Response delay in seconds. Any operation is delayed by this time. # Initialize before superclass init because otherwise logger may # fail with this attribute not found self._response_delay = response_delay # define attribute here to assure it is defined before CIM repository # created. Reset again after repository created. self._disable_pull_operations = disable_pull_operations super(FakedWBEMConnection, self).__init__( url or 'http://FakedUrl:5988', default_namespace=default_namespace, use_pull_operations=use_pull_operations, stats_enabled=stats_enabled, timeout=timeout) # See the cimrepository property for more information self._cimrepository = None # Provider registry defines user added providers. This is a dictionary # with key equal classname that contains entries for namespaces, # provider type, and provider for each class name defined self._provider_registry = ProviderRegistry() # Registry for provider dependent files. self._provider_dependent_registry = ProviderDependentRegistry() # Define the datastore to be used with an initial namespace, the client # connection default namespace. This is passed to the providerdispatcher # and mainprovider and not used further in this class. # Initiate instance of the ProviderDispatcher with required # parameters including the CIM repository. This call initializes # self.cimrepository. self._providerdispatcher = ProviderDispatcher( self.cimrepository, self._provider_registry) # Initiate the MainProvider with parameters required to execute self._mainprovider = MainProvider(self.host, self.disable_pull_operations, self.cimrepository, self._providerdispatcher) # Flag to allow or disallow the use of the Open... and Pull... # operations. Uses the setter method self.disable_pull_operations = disable_pull_operations # Defines the connection for the compiler. The compiler uses this # instance of this class as the client interface. self._mofwbemconnection = _MockMOFWBEMConnection(self) self._imethodcall = self._mock_imethodcall self._methodcall = self._mock_methodcall @property def namespaces(self): """ :term:`NocaseList` of :term:`string`: The names of the namespaces that exist in the CIM repository. """ return self._mainprovider.namespaces @property def interop_namespace_names(self): """ :term:`NocaseList` of :term:`string`: The valid Interop namespace names. Only these names may be the Interop namespace and only one Interop namespace may exist in a WBEM server environment. This list is defined in :attr:`pywbem.WBEMServer.INTEROP_NAMESPACES`. """ return self._mainprovider.interop_namespace_names @property def cimrepository(self): """ :class:`~pywbem_mock.InMemoryRepository`: The mocked in-memory CIM repository. The CIM repository is the data store for CIM classes, CIM instances, and CIM qualifier declarations, all partitioned by CIM namespaces. """ if self._cimrepository is None: self._cimrepository = InMemoryRepository(self.default_namespace) return self._cimrepository @property def provider_dependent_registry(self): """ :class:`~pywbem_mock.ProviderDependentRegistry`: The registry for provider dependent files, in context of a mock script. """ return self._provider_dependent_registry @property def response_delay(self): """ :term:`number`: Artifically created delay for each operation, in seconds. If `None`, there is no delay. This attribute is settable. For details, see the description of the same-named init parameter of :class:`this class <pywbem_mock.FakedWBEMConnection>`. """ return self._response_delay @response_delay.setter def response_delay(self, delay): """Setter method; for a description see the getter method.""" if isinstance(delay, (int, float)) and delay >= 0 or delay is None: self._response_delay = delay else: raise ValueError( _format("Invalid value for response_delay: {0!A}, must be a " "positive number", delay)) @property def disable_pull_operations(self): """ :class:`py:bool`: Boolean flag to set option to disable the execution of the open and pull operation request handlers in the CIM repository. This emulates the characteristic in some CIM servers that did not implement pull operations. The default is to allow pull operations. All pull operations requests may be forbidden from executing by setting disable_pull_operations to True. This attribute is settable. For details, see the description of the same-named init parameter of :class:`this class <pywbem_mock.FakedWBEMConnection>`. """ return self._disable_pull_operations @disable_pull_operations.setter def disable_pull_operations(self, disable): """Setter method; for a description see the getter method.""" # Attribute will always be boolean if disable is None: disable = False if isinstance(disable, bool): # pylint: disable=attribute-defined-outside-init self._disable_pull_operations = disable # modify the parameter in the mainprovider self._mainprovider.disable_pull_operations = disable else: raise ValueError( _format('Invalid type for disable_pull_operations: {0!A}, ' 'must be a boolean', disable))
[docs] def __str__(self): return _format( "FakedWBEMConnection(" "response_delay={s.response_delay}, " "super={super})", s=self, super=super(FakedWBEMConnection, self).__str__())
[docs] def __repr__(self): return _format( "FakedWBEMConnection(" "response_delay={s.response_delay}, " "disable_pull_operations={s.disable_pull_operations} " "super={super})", s=self, super=super(FakedWBEMConnection, self).__repr__())
[docs] def copy(self): """ *New in pywbem 1.3.* Return a deep copy of the object with internal state reset and reusing the same repository and registries. The returned object uses the same repository, provider registry and provider dependent registry objects as the original object. Besides that, all other user-specifiable attributes of the object are deep-copied, and all other internal state is reset. """ cpy = FakedWBEMConnection( default_namespace=self.default_namespace, use_pull_operations=self.use_pull_operations, stats_enabled=self.stats_enabled, timeout=self.timeout, response_delay=self._response_delay, disable_pull_operations=self._disable_pull_operations, url=self.url, ) # only immutable parameters # pylint: disable=protected-access # Reuse repository and registries of the original object cpy._cimrepository = self._cimrepository cpy._provider_registry = self._provider_registry cpy._provider_dependent_registry = self._provider_dependent_registry # These objects know about the repository and registries, but other # wise do not have internal state cpy._providerdispatcher = self._providerdispatcher cpy._mainprovider = self._mainprovider # pylint: enable=protected-access return cpy
# The namespace management methods must be in the this class directly # so they can be access with call to the methods from instance of # this class. they are considered part of the external API.
[docs] def add_namespace(self, namespace, verbose=False): """ Add a CIM namespace to the CIM repository of the faked connection. The namespace must not yet exist in the CIM repository. Parameters: namespace (:term:`string`): The name of the CIM namespace to be added to the CIM repository. Must not be `None`. Any leading or trailing slash characters are removed before the string is used to define the namespace name. verbose (:class:`py:bool`): Verbose mode: Print a message about the namespace creation. Raises: ValueError: Namespace argument must not be None. :exc:`~pywbem.CIMError`: CIM_ERR_ALREADY_EXISTS if the namespace already exists in the CIM repository. """ self._mainprovider.add_namespace(namespace, verbose=verbose)
[docs] def remove_namespace(self, namespace, verbose=False): """ Remove a CIM namespace from the CIM repository of the faked connection. The namespace must exist in the CIM repository and must be empty. Parameters: namespace (:term:`string`): The name of the CIM namespace in the CIM repository (case insensitive). Must not be `None`. Leading or trailing slash characters are ignored. verbose (:class:`py:bool`): Verbose mode: Print a message about the namespace deletion. Raises: ValueError: Namespace argument must not be None :exc:`~pywbem.CIMError`: (CIM_ERR_NOT_FOUND) if the namespace does not exist in the CIM repository. :exc:`~pywbem.CIMError`: (CIM_ERR_NAMESPACE_NOT_EMPTY) if the namespace is not empty. :exc:`~pywbem.CIMError`: (CIM_ERR_NAMESPACE_NOT_EMPTY) if attempting to delete the default connection namespace. This namespace cannot be deleted from the CIM repository """ self._mainprovider.remove_namespace(namespace, verbose=verbose)
[docs] def is_interop_namespace(self, namespace): """ Tests if a namespace name is a valid Interop namespace name. This method does not access the CIM repository for this test; it merely compares the specified namespace name against the list of valid Interop namespace names returned by :meth:`interop_namespace_names`. Parameters: namespace (:term:`string`): The namespace name that is to be tested. Returns: :class:`py:bool`: Indicates whether the namespace name is a valid Interop namespace name. """ return self._mainprovider.is_interop_namespace(namespace)
[docs] def find_interop_namespace(self): """ Find the Interop namespace in the CIM repository, or return `None`. The Interop namespace is identified by comparing all namespace names in the CIM repository against the list of valid Interop namespace names returned by :meth:`interop_namespace_names`. Returns: :term:`string`: The name of the Interop namespace if one exists in the CIM repository or otherwise `None`. """ return self._mainprovider.find_interop_namespace()
[docs] def install_namespace_provider(self, interop_namespace, schema_pragma_file=None, verbose=None): """ FakedWBEMConnection user method to install the namespace provider in the Interop namespace where the proposed interop_namespace is defined by the parameter interop_namespace Because this provider requires a set of classes from the DMTF schema, the schema_pragma_file install the schema is required. This method should only be called once at the creation of the mock environment. Parameters: interop_namespace (:term:`string`): The Interop namespace defined for this environment schema_pragma_file (:term:`string`): File path defining a CIM schema pragma file for the set of CIM classes that make up a schema such as the DMTF schema. This file must contain a pragma statement for each of the classes defined in the schema. If None, no attempt is made to any CIM classes required for the provider and it is assumed that the CIM classes are already installed verbose (:class:`py:bool`): If True, displays progress information as providers are installed. Raises: :exc:`~pywbem.CIMError`: with status code appropriate for any error encountered in the installation of the provider. """ # Determine if an interop namespace already exists and confirm that # we are using a valid interop namespace name to add the # new namespace. if not self.find_interop_namespace(): self.add_namespace(interop_namespace) provider = CIMNamespaceProvider(self.cimrepository) self.register_provider(provider, namespaces=interop_namespace, schema_pragma_files=schema_pragma_file, verbose=verbose)
[docs] def install_subscription_providers(self, interop_namespace, schema_pragma_file=None, verbose=None): """ FakedWBEMConnection user method to install the indication subscription providers in the Interop namespace where the proposed interop_namespace is defined by the parameter interop_namespace **Experimental:** *New in pywbem 1.3.0 as experimental.* Because these provider requires a set of classes from the DMTF schema, the schema_pragma_file install the schema is required. This method should only be called once at the creation of the mock environment. Parameters: interop_namespace (:term:`string`): The Interop namespace defined for this environment schema_pragma_file (:term:`string`): File path defining a CIM schema pragma file for the set of CIM classes that make up a schema such as the DMTF schema. This file must contain a pragma statement for each of the classes defined in the schema. If None, no attempt is made to any CIM classes required for the provider and it is assumed that the CIM classes are already installed verbose (:class:`py:bool`): If True, displays progress information as providers are installed. Raises: :exc:`~pywbem.CIMError`: with status code appropriate for any error encountered in the installation of the provider. """ # Determine if an interop namespace already exists and confirm that # we are using a valid interop namespace name to add the # new namespace. if not self.find_interop_namespace(): self.add_namespace(interop_namespace) provider = CIMListenerDestinationProvider(self.cimrepository) self.register_provider(provider, namespaces=interop_namespace, schema_pragma_files=schema_pragma_file, verbose=verbose) provider = CIMIndicationFilterProvider(self.cimrepository) self.register_provider(provider, namespaces=interop_namespace, schema_pragma_files=schema_pragma_file, verbose=verbose) provider = CIMIndicationSubscriptionProvider(self.cimrepository) self.register_provider(provider, namespaces=interop_namespace, schema_pragma_files=schema_pragma_file, verbose=verbose)
########################################################################### # # Methods to compile mof files into repository # ###########################################################################
[docs] def compile_mof_file(self, mof_file, namespace=None, search_paths=None, verbose=None): """ Compile the MOF definitions in the specified file (and its included files) and add the resulting CIM objects to the specified CIM namespace of the CIM repository. If the namespace does not exist, :exc:`~pywbem.CIMError` with status CIM_ERR_INVALID_NAMESPACE is raised. This method supports all MOF pragmas, and specifically the include pragma. If a CIM class or CIM qualifier type to be added already exists in the target namespace with the same name (comparing case insensitively), this method raises :exc:`~pywbem.CIMError`. If a CIM instance to be added already exists in the target namespace with the same keybinding values, this method raises :exc:`~pywbem.CIMError`. In all cases where this method raises an exception, the CIM repository remains unchanged. Parameters: mof_file (:term:`string`): Path name of the file containing the MOF definitions to be compiled. namespace (:term:`string`): The name of the CIM namespace in the associated CIM repository that is the target of the compilation, and is also used for lookup of any dependent CIM elements. If `None`, the default namespace of the connection is used. A namespace defined in a namespace pragma of the MOF superceeds this namespace from the point in the compilation unit(string/file) where it is declared. The namespace specified in this parameter or the MOF inamespace pragma must exist. search_paths (:term:`py:iterable` of :term:`string`): An iterable of directory path names where MOF dependent files will be looked up. See the description of the `search_path` init parameter of the :class:`~pywbem.MOFCompiler` class for more information on MOF dependent files. verbose (:class:`py:bool`): Controls whether to issue more detailed compiler messages. Raises: IOError: MOF file not found. :exc:`~pywbem.MOFCompileError`: Compile error in the MOF. """ # By default, error messages are always printed in addition to being # raised as MOFCompileError. We only want them to be printed in verbose # mode: log_func_kwargs = {} if not verbose: log_func_kwargs['log_func'] = None stats_name = "compile_mof_file(ns={!r})".format(namespace) with self.statistics(stats_name): namespace = namespace or self.default_namespace self._mainprovider.validate_namespace(namespace) # issue #2063 refactor this so there is cleaner interface to # WBEMConnection mofcomp = MOFCompiler(self._mofwbemconnection, search_paths=search_paths, verbose=verbose, **log_func_kwargs) mofcomp.compile_file(mof_file, namespace)
[docs] def compile_mof_string(self, mof_str, namespace=None, search_paths=None, verbose=None): """ Compile the MOF definitions in the specified string and add the resulting CIM objects to the specified CIM namespace of the CIM repository. If the namespace does not exist, :exc:`~pywbem.CIMError` with status CIM_ERR_INVALID_NAMESPACE is raised. This method supports all MOF pragmas, and specifically the include pragma. If a CIM class or CIM qualifier type to be added already exists in the target namespace with the same name (comparing case insensitively), this method raises :exc:`~pywbem.CIMError`. If a CIM instance to be added already exists in the target namespace with the same keybinding values, this method raises :exc:`~pywbem.CIMError`. In all cases where this method raises an exception, the CIM repository remains unchanged. Parameters: mof_str (:term:`string`): A string with the MOF definitions to be compiled. namespace (:term:`string`): The name of the CIM namespace in the associated CIM repository that is the target of the compilation, and is also used for lookup of any dependent CIM elements. If `None`, the default namespace of the connection is used. A namespace defined in a namespace pragma of the MOF superceeds this namespace from the point in the compilation unit(string/file) where it is declared. The namespace specified in this parameter or the MOF inamespace pragma must exist. search_paths (:term:`py:iterable` of :term:`string`): An iterable of directory path names where MOF dependent files will be looked up. See the description of the `search_path` init parameter of the :class:`~pywbem.MOFCompiler` class for more information on MOF dependent files. verbose (:class:`py:bool`): Controls whether to issue more detailed compiler messages. Raises: IOError: MOF file not found. :exc:`~pywbem.MOFCompileError`: Compile error in the MOF. """ # By default, error messages are always printed in addition to being # raised as MOFCompileError. We only want them to be printed in verbose # mode: log_func_kwargs = {} if not verbose: log_func_kwargs['log_func'] = None stats_name = "compile_mof_string(ns={!r})".format(namespace) with self.statistics(stats_name): namespace = namespace or self.default_namespace self._mainprovider.validate_namespace(namespace) mofcomp = MOFCompiler(self._mofwbemconnection, search_paths=search_paths, verbose=verbose, **log_func_kwargs) mofcomp.compile_string(mof_str, namespace)
[docs] def compile_schema_classes(self, class_names, schema_pragma_files, namespace=None, verbose=False): # pylint: disable=line-too-long """ Compile the classes defined by `class_names` and all of their dependences. The class names must be classes in the defined schema and with pragma statements in a schema pragma file. Each schema pragma file in the `schema_pragma_files` parameter must be in a directory that also encompasses the MOF files for all of the classes defined in the schema pragma file and the dependencies of those classes. While the relative paths of all of the CIM class files is defined in the `schema_pragma_file` the pywbem MOF compiler may also search for dependencies (ex. superclasses, references, etc.) that are not specifically listed in the `class_names` and the path of the `schema_pragma_file` is the top level directory for that search. The mof schema directory must include: 1. Qualifier declarations defined in a single file with the name `qualifiers.mof` defined within the directory defined by the `schema_mof_dir` parameter 2. The file `schema_pragma_file` that defines the location of all of the CIM class files within the a schema mof directory hierarchy. This is the `schema_pragma_file` attribute of the DMTFCIMSchema class. 3. The MOF files, one for each class, for the classes that could be compiled within the directory hierarchy defined by `schema_mof_dir`. Only the leaf class names need be included in the `class_names` list since the compiler will find all dependencies as part of the dependency resolution for the compile of each class in `class_names`. Parameters: class_names (:term:`string` or :class:`py:list` of :term:`string`): Class names of the classes to be compiled. These class names must be a subset of the classes defined in `schema_pragma_file`. schema_pragma_files (:term:`string` or :class:`py:list` of :term:`string`): Relative or absolute file path(s) of schema pragma files that include a MOF pragma include statement for each CIM class to be compiled. This file path is available from :attr:`pywbem_mock.DMTFCIMSchema.schema_pragma_file`. namespace (:term:`string`): The name of the CIM namespace in the associated CIM repository that is the target of the compilation, and is also used for lookup of any dependent CIM elements. If `None`, the default namespace of the connection is used. A namespace defined in a namespace pragma of the MOF superceeds this namespace from the point in the compilation unit(string/file) where it is declared. The namespace specified in this parameter or the MOF inamespace pragma must exist. verbose (:class:`py:bool`): If `True`, progress messages are output to stdout as the schema is downloaded and expanded. Default is `False`. Raises: :class:`~pywbem.MOFCompileError`: For errors in MOF parsing, finding MOF dependencies or issues with the CIM repository. :exc:`~pywbem.CIMError`: Other errors relating to the target server environment. """ # noqa: E501 # pylint: enable:line-too-long stats_name = "compile_schema_classes(ns={!r})".format(namespace) with self.statistics(stats_name): if isinstance(schema_pragma_files, six.string_types): schema_pragma_files = [schema_pragma_files] # Build the pragma file and compile for each pragma file in # schema_pragma_files. The search path for each compile is the # directory containing that schema_pragma_file for schema_pragma_file in schema_pragma_files: search_path = os.path.dirname(schema_pragma_file) compile_pragma = build_schema_mof( class_names, schema_pragma_file) self.compile_mof_string(compile_pragma, namespace=namespace, search_paths=search_path, verbose=verbose)
###################################################################### # # Add Pywbem CIM objects directly to the data store # ######################################################################
[docs] def add_cimobjects(self, objects, namespace=None): # pylint: disable=line-too-long """ Add CIM classes, instances and/or CIM qualifier types (declarations) to the specified CIM namespace of the CIM repository. This method adds a copy of the objects presented so that the user may modify the objects without impacting the repository. If the namespace does not exist, :exc:`~pywbem.CIMError` with status CIM_ERR_INVALID_NAMESPACE is raised. The method imposes very few limits on the objects added. It does require that the superclass exist for any class added and that instances added include a path component. If the qualifier flavor attributes are not set, it sets them to those defined in the Qualifier Declarations if those exist. If a CIM class or CIM qualifier type to be added already exists in the target namespace with the same name (comparing case insensitively), this method fails, and the CIM repository remains unchanged. If a CIM instance to be added already exists in the target namespace with the same keybinding values, this method fails, and the CIM repository remains unchanged. Parameters: objects (:class:`~pywbem.CIMClass` or :class:`~pywbem.CIMInstance` or :class:`~pywbem.CIMQualifierDeclaration`, or list of them): CIM object or objects to be added to the CIM repository. The list may contain different kinds of CIM objects. namespace (:term:`string`): The name of the target CIM namespace in the CIM repository. This namespace is also used for lookup of any existing or dependent CIM objects. If `None`, the default namespace of the connection is used. Raises: ValueError: Invalid input CIM object in `objects` parameter. TypeError: Invalid type in `objects` parameter. :exc:`~pywbem.CIMError`: CIM_ERR_INVALID_NAMESPACE: Namespace does not exist. :exc:`~pywbem.CIMError`: Failure related to the CIM objects in the CIM repository. """ # noqa: E501 # pylint: enable=line-too-long stats_name = "add_cimobjects(ns={!r})".format(namespace) with self.statistics(stats_name): namespace = namespace or self.default_namespace self._mainprovider.validate_namespace(namespace) if isinstance(objects, list): for obj in objects: self.add_cimobjects(obj, namespace=namespace) else: obj = objects if isinstance(obj, CIMClass): cc = obj.copy() if cc.superclass: if not self._mainprovider.class_exists( namespace, cc.superclass): raise ValueError( _format("Class {0!A} defines superclass {1!A} " "but the superclass does not exist in " "the repository.", cc.classname, cc.superclass)) # pylint: disable=protected-access cc1 = self._mainprovider._resolve_class( cc, namespace, self.cimrepository.get_qualifier_store(namespace), verbose=False) class_store = self.cimrepository.get_class_store(namespace) class_store.create(cc.classname, cc1.copy()) elif isinstance(obj, CIMInstance): inst = obj.copy() if inst.path is None: raise ValueError( _format("Instances added must include a path. " "Instance {0!A} does not include a path", inst)) if inst.path.namespace is None: inst.path.namespace = namespace if inst.path.host is not None: inst.path.host = None instance_store = \ self.cimrepository.get_instance_store(namespace) try: if instance_store.object_exists(inst.path): raise ValueError( _format("Instance {0!A} already exists in " "CIM repository", inst)) except CIMError as ce: raise CIMError( CIM_ERR_FAILED, _format("Internal failure of add_cimobject " "operation. Rcvd CIMError {0!A}", ce)) instance_store.create(inst.path, inst) elif isinstance(obj, CIMQualifierDeclaration): qual = obj.copy() qualifier_store = \ self.cimrepository.get_qualifier_store(namespace) qualifier_store.create(qual.name, qual) else: # Internal mocker error assert False, \ _format("Object to add_cimobjects. {0} invalid type", type(obj))
[docs] def display_repository(self, namespaces=None, dest=None, summary=False, output_format='mof'): """ Display the namespaces and objects in the CIM repository in one of multiple formats to a destination. Parameters: namespaces (:term:`string` or :class:`py:list` of :term:`string`): Limits display output to the specified CIM namespace or namespaces. If `None`, all namespaces of the CIM repository are displayed. dest (:term:`string`): File path of an output file. If `None`, the output is written to stdout. summary (:class:`py:bool`): Flag for summary mode. If `True`, only a summary count of CIM objects in the specified namespaces of the CIM repository is produced. If `False`, both the summary count and the details of the CIM objects are produced. output_format (:term:`string`): Output format, one of: 'mof', 'xml', or 'repr'. """ # The comments are line oriented. if output_format == 'mof': cmt_begin = '// ' cmt_end = '' elif output_format == 'xml': cmt_begin = '<!-- ' cmt_end = ' ->' else: cmt_begin = '' cmt_end = '' if output_format not in OUTPUT_FORMATS: raise ValueError( _format("Invalid output format definition {0!A}. " "{1!A} are valid.", output_format, OUTPUT_FORMATS)) _uprint(dest, _format(u"{0}========Mock Repo Display fmt={1} " u"namespaces={2} ========={3}\n", cmt_begin, output_format, ('all' if namespaces is None else _format("{0!A}", namespaces)), cmt_end)) # get all namespace names repo_ns = sorted(self.namespaces) for ns in repo_ns: _uprint(dest, _format(u"\n{0}NAMESPACE {1!A}{2}\n", cmt_begin, ns, cmt_end)) self._display_objects('Qualifier Declarations', self.cimrepository.get_qualifier_store(ns), ns, cmt_begin, cmt_end, dest=dest, summary=summary, output_format=output_format) self._display_objects('Classes', self.cimrepository.get_class_store(ns), ns, cmt_begin, cmt_end, dest=dest, summary=summary, output_format=output_format) self._display_objects('Instances', self.cimrepository.get_instance_store(ns), ns, cmt_begin, cmt_end, dest=dest, summary=summary, output_format=output_format) _uprint(dest, _format(u'{0}============End Repository================={1}', cmt_begin, cmt_end))
@staticmethod def _display_objects(obj_type, object_repo, namespace, cmt_begin, cmt_end, dest=None, summary=None, output_format=None): """ Display a set of objects of obj_type from the dictionary defined by the parameter object_repo. obj_type is a string that defines the type of object ('Classes', 'Instances', 'Qualifier Declarations', 'Methods'). """ # Issue #2062: Consider sorting to preserve order of # compile/add. Make this part of refactor to separate repository and # datastore because it may be data store dependent _uprint(dest, _format(u"{0}Namespace {1!A}: contains {2} {3} {4}\n", cmt_begin, namespace, object_repo.len(), obj_type, cmt_end)) if summary: return # instances are special because the inner struct is a list if obj_type == 'Instances': # Issue # 2062 - Consider sorting here in the future for inst in object_repo.iter_values(): if output_format == 'xml': _uprint(dest, _format(u"{0} Path={1} {2}\n{3}", cmt_begin, inst.path.to_wbem_uri(), cmt_end, _pretty_xml(inst.tocimxmlstr()))) elif output_format == 'repr': _uprint(dest, _format(u"Path:\n{0!A}\nInst:\n{1!A}\n", inst.path, inst)) else: _uprint(dest, _format(u"{0} Path={1} {2}\n{3}", cmt_begin, inst.path.to_wbem_uri(), cmt_end, inst.tomof())) else: # Display classes and qualifier declarations sorted assert obj_type in ['Classes', 'Qualifier Declarations'] names = list(object_repo.iter_names()) if names: for name in sorted(names): obj = object_repo.get(name) if output_format == 'xml': _uprint(dest, _pretty_xml(obj.tocimxmlstr())) elif output_format == 'repr': _uprint(dest, _format(u"{0!A}", obj)) else: _uprint(dest, obj.tomof())
[docs] def register_provider(self, provider, namespaces=None, schema_pragma_files=None, verbose=None): # pylint: disable=line-too-long """ Register the `provider` object for specific namespaces and CIM classes. Registering a provider tells the FakedWBEMConnection that the provider implementation provided with this method as the `provider` parameter is to be executed as the request response method for the namespaces defined in the `namespaces` parameter, the provider type defined in the 'provider_type` attribute of the `provider` and the class(es) defined in the provider `provider_classnames` attribute of the `provider`. The provider registration process includes: 1. Validation that the namespaces defined for the provider exist. 2. Validation that the superclass of the provider is consistent with the `provider_type` attribute defined in the provider. 3. Installation of any CIM classes defined by the provider `provider_classnames` attribute including dependencies for these classes using the `schema_pragma_files` parameter to define the MOF compiler search directories for dependencies. 4. Adding the provider to the registry of user_providers so that any of the request methods defined for the `provider_type` are passed to this provider in place of the default request processors. 5. Execute post_register_setup() call to the provider to allow the provider to perform any special setup functionality. Providers can only be registered for the following request response methods: 1. provider_type = 'instance': defines methods for CreateInstance, ModifyInstance, and DeleteInstance requests within a subclass of the `InstanceWriteProvider` class. 2. provider_type = 'method': defines a InvokeMethod method within a subclass of the `MethodProvider` class. Each classname in a particular namespace may have at most one provider registered Parameters: provider (instance of subclass of :class:`pywbem_mock.InstanceWriteProvider` or :class:`pywbem_mock.MethodProvider`): An instance of the user provider class which is a subclass of :class:`pywbem_mock.InstanceWriteProvider`. The methods in this subclass override the corresponding methods in InstanceWriteProvider. The method call parameters must be the same as the defult method in InstanceWriteProvider and it must return data in the same format if the default method returns data. namespaces (:term:`string` or :class:`py:list` of :term:`string`): Namespace or namespaces for which the provider is to be registered. If `None`, the default namespace of the connection will be used. schema_pragma_files (:term:`string` or :class:`py:list` of :term:`string`): File paths defining a schema pragma file MOF for the set of CIM classes that make up a schema such as the DMTF schema. These files must contain include pragma statements defining the file location of the classes to be compiled for the defined provider and for any dependencies required to compile those classes. The directory containing each schema pragma file is passed to the MOF compiler as the search path for compile dependencies. See :class:`pywbem.MOFCompiler` for more information on the `search_paths` parameter. verbose (:class:`py:bool`): Flag to enable detailed display of actions Raises: TypeError: Invalid provider_type retrieved from provider or provider_type does not match superlclass or the namespace parameter is invalid. ValueError: Provider_type retrieved from provider is not a valid string. ValueError: Classnames parameter retrieved from provider not a valid string or iterable or namespace does not exist in repository. """ # noqa: E501 # pylint: enable=line-too-long self._provider_registry.register_provider( self, provider, namespaces=namespaces, schema_pragma_files=schema_pragma_files, verbose=verbose)
[docs] def display_registered_providers(self): """ Display information on the currently registered providers. Parameters: dest (:term:`string`): File path of an output file. If `None`, the output is written to stdout. """ self._provider_registry.display_registered_providers()
######################################################################## # # Pywbem functions mocked. WBEMConnection only mocks the WBEMConnection # _imethodcall and _methodcall methods. This captures all calls # to the wbem server. # ######################################################################## def _mock_imethodcall(self, methodname, namespace, response_params_rqd=None, **params): # pylint: disable=unused-argument """ Mocks the WBEMConnection._imethodcall() method. This mock calls methods within this class that fake the processing in a WBEM server (at the CIM Object level) for the varisous CIM/XML methods and return. The methodname input parameters directly translate to the server request handler names (i.e. GetClass, ...) including capitalization """ # The statistics is already applied in WBEMConnection. # Create the local method name methodname = '_imeth_' + methodname methodnameattr = getattr(self, methodname) # Execute the named method result = methodnameattr(namespace, **params) # sleep for defined number of seconds if self._response_delay: time.sleep(self._response_delay) return result def _mock_methodcall(self, methodname, objectname, Params=None, **params): # pylint: disable=invalid-name """ Mocks the WBEMConnection._methodcall() method. This function performs the same checks and transformations as WBEMConnection._methodcall(). """ # The statistics is already applied in WBEMConnection. # Normalize the target object into either CIMInstanceName or # CIMClassName, both with namespace set. if isinstance(objectname, (CIMInstanceName, CIMClassName)): localobject = objectname.copy() if localobject.namespace is None: localobject.namespace = self.default_namespace localobject.host = None elif isinstance(objectname, six.string_types): # a string is always interpreted as a class name localobject = CIMClassName(objectname, namespace=self.default_namespace) else: raise TypeError( _format("The 'ObjectName' parameter of the WBEMConnection " "operation has invalid type {0} (must be a string, " "a CIMClassName, or a CIMInstanceName)", type(objectname))) # Merge the Params and **params into a single no-case dictionary # of name: CIMParameter params_dict = NocaseDict() if Params: for param in Params: if isinstance(param, CIMParameter): params_dict[param.name] = param elif isinstance(param, tuple): params_dict[param[0]] = CIMParameter(param[0], cimtype(param[1]), value=param[1]) else: raise TypeError( _format("InvokeMethod of method {0!A} on object {1!A}: " "A parameter in the Params argument has " "incorrect Python type {2} " "- expected one of tuple (name, value) or " "pywbem.CIMParameter", methodname, localobject, type(param))) if params: for pname, pvalue in params.items(): params_dict[pname] = CIMParameter( pname, cimtype(pvalue), value=pvalue) result = self._meth_InvokeMethod(methodname, localobject, params_dict) # Sleep for defined number of seconds if self._response_delay: time.sleep(self._response_delay) return result @staticmethod def _make_pull_imethod_resp(objs, eos, context_id): """ Create the correct imethod response for the open and pull methods """ eos_tuple = (u'EndOfSequence', None, eos) enum_ctxt_tuple = (u'EnumerationContext', None, context_id) return [("IRETURNVALUE", {}, objs), enum_ctxt_tuple, eos_tuple] @staticmethod def _make_openquery_imethod_resp(objs, eos, context_id, result_class): """ Create the correct imethod response for the OpenQueryInstances method """ eos_tuple = (u'EndOfSequence', None, eos) enum_ctxt_tuple = (u'EnumerationContext', None, context_id) result_class_tuple = (u'QueryResultClass', None, result_class) return [("IRETURNVALUE", {}, objs), enum_ctxt_tuple, eos_tuple, result_class_tuple] @staticmethod def _make_tuple(rtn_value): """ Change the return value from the value consistent with the definition in cim_operations.py into a tuple in accord with _imethodcall """ return [("IRETURNVALUE", {}, rtn_value)] def _return_assoc_tuple(self, objects): """ Create the property tuple for _imethod return of references, referencenames, associators, and associatornames methods. This is different than the get/enum imethod return tuples. It creates an OBJECTPATH for each object in the return list. _imethod call returns None when there are zero objects rather than a tuple with empty object path """ if objects: result = [(u'OBJECTPATH', {}, obj) for obj in objects] return self._make_tuple(result) return None ##################################################################### # # The following methods map the imethodcall interface (dictionary of # the method parameters to the arguments required by each corresponding # method in the MainProvider or InstanceWriteProvider and call that # method. the resultsare mapped back to be compatible with imethodcall # return tuple. # Methods that allow user providers are routed to the # InstanceWriteProvider. All other methods are rounted to the # MainProvider. # ##################################################################### # Instance Operations def _imeth_EnumerateInstanceNames(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.EnumerateInstanceNames` with parameters defined for that method and map response to tuple response for imethodcall. Parameters: params (class:`py:dict`): Dictionary of parameters for the method called. Returns: Tuple with instance paths comatible with imethodcall Raises: Error: Exceptions from the call """ instance_paths = self._mainprovider.EnumerateInstanceNames( ClassName=_cvt_rqd_classname(params['ClassName']), namespace=namespace) return self._make_tuple(instance_paths) def _imeth_EnumerateInstances(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.EnumerateInstances` with parameters defined for that method and map response to tuple response for imethodcall. Parameters: params (class:`py:dict`): Dictionary of parameters for the method called. Returns: Tuple with instances comatible with imethodcall Raises: Error: Exceptions from the call """ instances = self._mainprovider.EnumerateInstances( ClassName=_cvt_rqd_classname(params['ClassName']), namespace=namespace, LocalOnly=params.get('LocalOnly', None), DeepInheritance=params.get('DeepInheritance', None), IncludeQualifiers=params.get('IncludeQualifiers', None), IncludeClassOrigin=params.get('IncludeClassOrigin', None), PropertyList=params.get('PropertyList', None)) return self._make_tuple(instances) def _imeth_GetInstance(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.GetInstance` with parameters defined for that method and map response to tuple response for imethodcall. The method called includes the namespace within the InstanceName rather than as a separate element. Parameters: params (class:`py:dict`): Dictionary of parameters for the method called. Returns: Tuple with instance comatible with imethodcall Raises: Error: Exceptions from the call """ instance_name = params['InstanceName'] assert instance_name.namespace is None instance_name.namespace = namespace instance = self._mainprovider.GetInstance( InstanceName=instance_name, LocalOnly=params.get('LocalOnly', None), IncludeQualifiers=params.get('IncludeQualifiers', None), IncludeClassOrigin=params.get('IncludeClassOrigin', None), PropertyList=params.get('PropertyList', None)) return self._make_tuple([instance]) def _imeth_CreateInstance(self, namespace, **params): """ Call :meth:`~pywbem_mock.ProviderDispatcher.CreateInstance` with parameters defined for that method and map response to tuple response for imethodcall. This method allows user providers and therefore is passed to the :class:`ProviderDispatcher`. Parameters: params (class:`py:dict`): Dictionary of parameters for the method called. Raises: Error: Exceptions from the call """ new_instance_path = self._providerdispatcher.CreateInstance( namespace=namespace, NewInstance=params['NewInstance']) return self._make_tuple([new_instance_path]) def _imeth_ModifyInstance(self, namespace, **params): """ Call :meth:`~pywbem_mock.ProviderDispatcher.ModifyInstance` with parameters defined for that method and map response to tuple response for imethodcall. The method called includes the namespace within the ModifiedInstance rather than as a separate element. This method allows user providers and therefore is passed to the :class:`ProviderDispatcher`. Parameters: params (class:`py:dict`): Dictionary of parameters for the method called. Raises: Error: Exceptions from the call """ modified_instance = params['ModifiedInstance'] assert modified_instance.path.namespace is None modified_instance.path.namespace = namespace self._providerdispatcher.ModifyInstance( ModifiedInstance=modified_instance, IncludeQualifiers=params.get('IncludeQualifiers', None), PropertyList=params.get('PropertyList', None)) def _imeth_DeleteInstance(self, namespace, **params): """ Call :meth:`~pywbem_mock.ProviderDispatcher.DeleteInstance` with parameters defined for that method and map response to tuple response for imethodcall. The method called includes the namespace within the InstanceName rather than as a separate element. This method allows user providers and therefore is passed to the :class:`ProviderDispatcher`. Parameters: params (class:`py:dict`): Dictionary of parameters for the method called. Raises: Error: Exceptions from the call """ instance_name = params['InstanceName'] assert instance_name.namespace is None instance_name.namespace = namespace self._providerdispatcher.DeleteInstance( InstanceName=instance_name) def _imeth_ExecQuery(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.ExecQuery` with parameters defined for that method and map response to tuple response for imethodcall. Parameters: params (class:`py:dict`): Dictionary of parameters for the method called. Returns: Tuple with instances compatible with imethodcall Raises: Error: Exceptions from the call """ # pylint: disable=assignment-from-no-return instances = self._mainprovider.ExecQuery( namespace=namespace, QueryLanguage=params['QueryLanguage'], Query=params['Query']) # Issue 2064: The following is untested because the # mainprovider ExecQuery only generates exception. return self._make_tuple([instances]) # CIMClass operations def _imeth_EnumerateClasses(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.EnumerateClasses` with parameters defined for that method and map response to tuple response for imethodcall. """ classes = self._mainprovider.EnumerateClasses( params.get("namespace", namespace), ClassName=_cvt_opt_classname(params.get('ClassName', None)), DeepInheritance=params.get('DeepInheritance', None), LocalOnly=params.get('LocalOnly', None), IncludeQualifiers=params.get('IncludeQualifiers', None), IncludeClassOrigin=params.get('IncludeClassOrigin, None)', None)) return self._make_tuple(classes) def _imeth_EnumerateClassNames(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.EnumerateClasseNames` with parameters defined for that method and map response to tuple response for imethodcall. """ classnames = self._mainprovider.EnumerateClassNames( params.get("namespace", namespace), ClassName=_cvt_opt_classname(params.get('ClassName', None)), DeepInheritance=params.get('DeepInheritance', None)) # Map the class name strings to CIMClassName rtn_cim_classnames = [ CIMClassName(cn, namespace=namespace, host=self.host) for cn in classnames] return self._make_tuple(rtn_cim_classnames) def _imeth_GetClass(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.EGetClass` with parameters defined for that method and map response to tuple response for imethodcall. """ klass = self._mainprovider.GetClass( params.get("namespace", namespace), ClassName=_cvt_rqd_classname(params['ClassName']), LocalOnly=params.get('LocalOnly', None), IncludeQualifiers=params.get('IncludeQualifiers', None), IncludeClassOrigin=params.get('IncludeClassOrigin', None), PropertyList=params.get('PropertyList', None)) return self._make_tuple([klass]) def _imeth_CreateClass(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.CreateClass` with parameters defined for that method and map response to tuple response for imethodcall. """ self._mainprovider.CreateClass( namespace, NewClass=params['NewClass']) def _imeth_ModifyClass(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.ModifyClass` with parameters defined for that method and map response to tuple response for imethodcall. """ self._mainprovider.ModifyClass( namespace, ModifiedClass=params['ModifiedClass']) def _imeth_DeleteClass(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.DeleteClass` with parameters defined for that method and map response to tuple response for imethodcall. """ self._mainprovider.DeleteClass( namespace, ClassName=_cvt_rqd_classname(params['ClassName'])) # Qualifier declaration operations def _imeth_EnumerateQualifiers(self, namespace, **params): # pylint: disable=unused-argument """ Call :meth:`~pywbem_mock.MainProvider.EnumerateQualifiers` with parameters defined for that method and map response to tuple response for imethodcall. """ qualifiers = self._mainprovider.EnumerateQualifiers( namespace=namespace) return self._make_tuple(qualifiers) def _imeth_GetQualifier(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.GetQualifier` with parameters defined for that method and map response to tuple response for imethodcall. """ qualifier = self._mainprovider.GetQualifier( namespace=namespace, QualifierName=params['QualifierName']) return self._make_tuple([qualifier]) def _imeth_DeleteQualifier(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.DeleteQualifier` with parameters defined for that method and map response to tuple response for imethodcall. """ self._mainprovider.DeleteQualifier( namespace=namespace, QualifierName=params['QualifierName']) def _imeth_SetQualifier(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.SetQualifier` with parameters defined for that method and map response to tuple response for imethodcall. """ self._mainprovider.SetQualifier( namespace=namespace, QualifierDeclaration=params['QualifierDeclaration']) # Associator and Reference operations def _imeth_ReferenceNames(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.ReferenceNames` with parameters defined for that method and map response to tuple response for imethodcall., """ object_names = self._mainprovider.ReferenceNames( namespace=namespace, ObjectName=_cvt_obj_name(params['ObjectName']), ResultClass=_cvt_opt_classname(params.get('ResultClass', None)), Role=params.get('Role', None)) return self._return_assoc_tuple(object_names) def _imeth_References(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.References` with parameters defined for that method and map response to tuple response for imethodcall. """ objects = self._mainprovider.References( namespace=namespace, ObjectName=_cvt_obj_name(params['ObjectName']), ResultClass=_cvt_opt_classname(params.get('ResultClass', None)), Role=params.get('Role', None), IncludeQualifiers=params.get('IncludeQualifiers', None), IncludeClassOrigin=params.get('IncludeClassOrigin', None), PropertyList=params.get('PropertyList', None)) return self._return_assoc_tuple(objects) def _imeth_AssociatorNames(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.AssociatorNames` with parameters defined for that method and map response to tuple response for imethodcall. """ object_names = self._mainprovider.AssociatorNames( namespace=namespace, ObjectName=_cvt_obj_name(params['ObjectName']), AssocClass=_cvt_opt_classname(params.get('AssocClass', None)), ResultClass=_cvt_opt_classname(params.get('ResultClass', None)), Role=params.get('Role', None), ResultRole=params.get('ResultRole', None)) return self._return_assoc_tuple(object_names) def _imeth_Associators(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.Associators` with parameters defined for that method and map response to tuple response for imethodcall. """ objects = self._mainprovider.Associators( namespace=namespace, ObjectName=_cvt_obj_name(params['ObjectName']), AssocClass=_cvt_opt_classname(params.get('AssocClass', None)), ResultClass=_cvt_opt_classname(params.get('ResultClass', None)), Role=params.get('Role', None), ResultRole=params.get('ResultRole', None), IncludeQualifiers=params.get('IncludeQualifiers', None), IncludeClassOrigin=params.get('IncludeClassOrigin', None), PropertyList=params.get('PropertyList', None)) return self._return_assoc_tuple(objects) # The pull operations including Open..., Pull... and CloseEnumeration def _imeth_OpenEnumerateInstancePaths(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.OpenEnumerateInstancePaths` with parameters defined for that method and map response to tuple response for imethodcall. """ context_tuple = self._mainprovider.OpenEnumerateInstancePaths( namespace=namespace, ClassName=_cvt_rqd_classname(params['ClassName']), FilterQueryLanguage=params.get('FilterQueryLanguage', None), FilterQuery=params.get('FilterQuery', None), OperationTimeout=params.get('OperationTimeout', None), ContinueOnError=params.get('ContinueOnError', None), MaxObjectCount=params.get('MaxObjectCount', None)) return self._make_pull_imethod_resp(*context_tuple) def _imeth_OpenEnumerateInstances(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.OpenEnumerateInstances` with parameters defined for that method and map response to tuple response for imethodcall. """ context_tuple = self._mainprovider.OpenEnumerateInstances( namespace=namespace, ClassName=_cvt_rqd_classname(params['ClassName']), DeepInheritance=params.get('DeepInheritance', None), IncludeClassOrigin=params.get('IncludeClassOrigin', None), PropertyList=params.get('PropertyList', None), FilterQueryLanguage=params.get('FilterQueryLanguage', None), FilterQuery=params.get('FilterQuery', None), OperationTimeout=params.get('OperationTimeout', None), ContinueOnError=params.get('ContinueOnError', None), MaxObjectCount=params.get('MaxObjectCount', None)) return self._make_pull_imethod_resp(*context_tuple) def _imeth_OpenReferenceInstancePaths(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.OpenReferenceInstancePaths` with parameters defined for that method and map response to tuple response for imethodcall. """ context_tuple = self._mainprovider.OpenReferenceInstancePaths( namespace=namespace, InstanceName=params['InstanceName'], ResultClass=_cvt_opt_classname(params.get('ResultClass', None)), Role=params.get('Role', None), FilterQueryLanguage=params.get('FilterQueryLanguage', None), FilterQuery=params.get('FilterQuery', None), OperationTimeout=params.get('OperationTimeout', None), ContinueOnError=params.get('ContinueOnError', None), MaxObjectCount=params.get('MaxObjectCount', None)) return self._make_pull_imethod_resp(*context_tuple) def _imeth_OpenReferenceInstances(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.OpenReferenceInstances` with parameters defined for that method and map response to tuple response for imethodcall. """ context_tuple = self._mainprovider.OpenReferenceInstances( namespace=namespace, InstanceName=params['InstanceName'], ResultClass=_cvt_opt_classname(params.get('ResultClass', None)), Role=params.get('Role', None), IncludeClassOrigin=params.get('IncludeClassOrigin', None), PropertyList=params.get('PropertyList', None), FilterQueryLanguage=params.get('FilterQueryLanguage', None), FilterQuery=params.get('FilterQuery', None), OperationTimeout=params.get('OperationTimeout', None), ContinueOnError=params.get('ContinueOnError', None), MaxObjectCount=params.get('MaxObjectCount', None)) return self._make_pull_imethod_resp(*context_tuple) def _imeth_OpenAssociatorInstances(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.OpenAssociatorInstances` with parameters defined for that method and map response to tuple response for imethodcall. """ context_tuple = self._mainprovider.OpenAssociatorInstances( namespace=namespace, InstanceName=params['InstanceName'], AssocClass=_cvt_opt_classname(params.get('AssocClass', None)), ResultClass=_cvt_opt_classname(params.get('ResultClass', None)), Role=params.get('Role', None), ResultRole=params.get('ResultRole', None), IncludeClassOrigin=params.get('IncludeClassOrigin', None), PropertyList=params.get('PropertyList', None), FilterQueryLanguage=params.get('FilterQueryLanguage', None), FilterQuery=params.get('FilterQuery', None), OperationTimeout=params.get('OperationTimeout', None), ContinueOnError=params.get('ContinueOnError', None), MaxObjectCount=params.get('MaxObjectCount', None)) return self._make_pull_imethod_resp(*context_tuple) def _imeth_OpenAssociatorInstancePaths(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.OpenAssociatorInstancePaths` with parameters defined for that method and map response to tuple response for imethodcall. """ context_tuple = self._mainprovider.OpenAssociatorInstancePaths( namespace=namespace, InstanceName=params['InstanceName'], AssocClass=_cvt_opt_classname(params.get('AssocClass', None)), ResultClass=_cvt_opt_classname(params.get('ResultClass', None)), Role=params.get('Role', None), ResultRole=params.get('ResultRole', None), FilterQueryLanguage=params.get('FilterQueryLanguage', None), FilterQuery=params.get('FilterQuery', None), OperationTimeout=params.get('OperationTimeout', None), ContinueOnError=params.get('ContinueOnError', None), MaxObjectCount=params.get('MaxObjectCount', None)) return self._make_pull_imethod_resp(*context_tuple) def _imeth_OpenQueryInstances(self, namespace, **params): """ Call :meth:`~pywbem_mock.MainProvider.OpenQueryInstances` with parameters defined for that method and map response to tuple response for imethodcall. """ context_tuple = self._mainprovider.OpenQueryInstances( namespace=namespace, FilterQueryLanguage=params.get('FilterQueryLanguage', None), FilterQuery=params.get('FilterQuery', None), ReturnQueryResultClass=params.get('ReturnQueryResultClass', None), OperationTimeout=params.get('OperationTimeout', None), ContinueOnError=params.get('ContinueOnError', None), MaxObjectCount=params.get('MaxObjectCount', None)) return self._make_openquery_imethod_resp(*context_tuple) def _imeth_PullInstancePaths(self, namespace, **params): # pylint: disable=unused-argument """ Call :meth:`~pywbem_mock.MainProvider.PullInstancePaths` with parameters defined for that method and map response to tuple response for imethodcall. """ context_tuple = self._mainprovider.PullInstancePaths( EnumerationContext=params['EnumerationContext'], MaxObjectCount=params['MaxObjectCount']) return self._make_pull_imethod_resp(*context_tuple) def _imeth_PullInstancesWithPath(self, namespace, **params): # pylint: disable=unused-argument """ Call :meth:`~pywbem_mock.MainProvider.PullInstancesWithPath` with parameters defined for that method and map response to tuple response for imethodcall. """ context_tuple = self._mainprovider.PullInstancesWithPath( EnumerationContext=params['EnumerationContext'], MaxObjectCount=params['MaxObjectCount']) return self._make_pull_imethod_resp(*context_tuple) def _imeth_PullInstances(self, namespace, **params): # pylint: disable=unused-argument """ Call :meth:`~pywbem_mock.MainProvider.PullInstances` with parameters defined for that method and map response to tuple response for imethodcall. """ context_tuple = self._mainprovider.PullInstances( EnumerationContext=params['EnumerationContext'], MaxObjectCount=params['MaxObjectCount']) return self._make_pull_imethod_resp(*context_tuple) def _imeth_CloseEnumeration(self, namespace, **params): # pylint: disable=unused-argument """ Call :meth:`~pywbem_mock.MainProvider.CloseEnumeration` with parameters defined for that method. Returns nothing """ self._mainprovider.CloseEnumeration( EnumerationContext=params['EnumerationContext']) def _meth_InvokeMethod(self, methodname, localobject, params): # pylint: disable=invalid-name """ Call :meth:`~pywbem_mock.ProviderDispatcher.InvokeMethod`. """ result = self._providerdispatcher.InvokeMethod( methodname, localobject, params) return result