#
# 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.
#
"""
Operation recorder interface and implementations.
"""
from __future__ import print_function, absolute_import
from collections import namedtuple
try:
from collections import OrderedDict
except ImportError:
from ordereddict import OrderedDict
from datetime import datetime, timedelta
import yaml
import six
from .cim_obj import CIMInstance, CIMInstanceName, CIMClass, CIMClassName, \
CIMProperty, CIMMethod, CIMParameter, CIMQualifier, \
CIMQualifierDeclaration, NocaseDict
from .cim_types import CIMInt, CIMFloat, CIMDateTime
from .exceptions import CIMError
__all__ = ['BaseOperationRecorder', 'TestClientRecorder',
'OpArgs', 'OpResult', 'HttpRequest', 'HttpResponse']
if six.PY2:
_Longint = long
else:
_Longint = int
OpArgs_tuple = namedtuple("OpArgs_tuple", ["method", "args"])
def _represent_ordereddict(dump, tag, mapping, flow_style=None):
"""PyYAML representer function for OrderedDict.
This is needed for yaml.safe_dump() to support OrderedDict.
Courtesy:
http://blog.elsdoerfer.name/2012/07/26/make-pyyaml-output-an-ordereddict/
"""
value = []
node = yaml.MappingNode(tag, value, flow_style=flow_style)
if dump.alias_key is not None:
dump.represented_objects[dump.alias_key] = node
best_style = True
if hasattr(mapping, 'items'):
mapping = mapping.items()
for item_key, item_value in mapping:
node_key = dump.represent_data(item_key)
node_value = dump.represent_data(item_value)
if not (isinstance(node_key, yaml.ScalarNode) \
and not node_key.style):
best_style = False
if not (isinstance(node_value, yaml.ScalarNode) \
and not node_value.style):
best_style = False
value.append((node_key, node_value))
if flow_style is None:
if dump.default_flow_style is not None:
node.flow_style = dump.default_flow_style
else:
node.flow_style = best_style
return node
yaml.SafeDumper.add_representer(
OrderedDict,
lambda dumper, value:
_represent_ordereddict(dumper, u'tag:yaml.org,2002:map', value))
[docs]class OpArgs(OpArgs_tuple):
"""
A named tuple representing the name and input arguments of the invocation
of a :class:`~pywbem.WBEMConnection` method, with the following named fields
and attributes:
Attributes:
method (:term:`unicode string`):
Name of the :class:`~pywbem.WBEMConnection` method.
args (:class:`py:dict`):
Dictionary of input arguments (both positional and keyword-based).
"""
__slots__ = ()
def __repr__(self):
return "OpArgs(method={s.method!r}, args={s.args!r})".format(s=self)
OpResult_tuple = namedtuple("OpResult_tuple", ["ret", "exc"])
[docs]class OpResult(OpResult_tuple):
"""
A named tuple representing the result of the invocation of a
:class:`~pywbem.WBEMConnection` method, with the following named fields
and attributes:
Attributes:
ret (:class:`py:object`):
Return value, if the method returned.
`None`, if the method raised an exception.
Note that `None` may be a legitimate return value, so the test for
exceptions should be done based upon the :attr:`exc` variable.
exc (:exc:`~py:exceptions.Exception`):
Exception object, if the method raised an exception.
`None`, if the method returned.
"""
__slots__ = ()
def __repr__(self):
return "OpResult(ret={s.ret!r}, exc={s.exc!r})".format(s=self)
HttpRequest_tuple = namedtuple("HttpRequest_tuple",
["version", "url", "target", "method", "headers",
"payload"])
[docs]class HttpRequest(HttpRequest_tuple):
"""
A named tuple representing the HTTP request sent by the WBEM client, with
the following named fields and attributes:
Attributes:
version (:term:`number`):
HTTP version from the request line (10 for HTTP/1.0, 11 for HTTP/1.1).
url (:term:`unicode string`):
URL of the WBEM server (e.g. 'https://myserver.acme.com:15989').
target (:term:`unicode string`):
Target URL segment as stated in request line (e.g. '/cimom').
method (:term:`unicode string`):
HTTP method as stated in the request line (e.g. "POST").
headers (:class:`py:dict`):
A dictionary of all HTTP header fields:
* key (:term:`unicode string`): Name of the header field
* value (:term:`unicode string`): Value of the header field
payload (:term:`unicode string`):
HTTP payload, i.e. the CIM-XML string.
"""
__slots__ = ()
def __repr__(self):
return "HttpRequest(version={s.version!r}, url={s.url!r}, " \
"target={s.target!r}, method={s.method!r}, " \
"headers={s.headers!r}, payload={s.payload!r})" \
.format(s=self)
HttpResponse_tuple = namedtuple("HttpResponse_tuple",
["version", "status", "reason", "headers",
"payload"])
[docs]class HttpResponse(HttpResponse_tuple):
"""
A named tuple representing the HTTP response received by the WBEM client,
with the following named fields and attributes:
Attributes:
version (:term:`number`):
HTTP version from the response line (10 for HTTP/1.0, 11 for HTTP/1.1).
status (:term:`number`):
HTTP status code from the response line (e.g. 200).
reason (:term:`unicode string`):
HTTP reason phrase from the response line (e.g. "OK").
headers (:class:`py:dict`):
A dictionary of all HTTP header fields:
* key (:term:`unicode string`): Name of the header field
* value (:term:`unicode string`): Value of the header field
payload (:term:`unicode string`):
HTTP payload, i.e. the CIM-XML string.
"""
__slots__ = ()
def __repr__(self):
return "HttpResponse(version={s.version!r}, status={s.status!r}, " \
"reason={s.reason!r}, headers={s.headers!r}, " \
"payload={s.payload!r})".format(s=self)
[docs]class BaseOperationRecorder(object):
"""
Abstract base class defining the interface to an operation recorder,
that records the WBEM operations executed in a connection to a WBEM
server.
An operation recorder can be registered by setting the
:attr:`~pywbem.WBEMConnection.operation_recorder` instance
attribute of the :class:`~pywbem.WBEMConnection` object to an
object of a subclass of this base class.
When an operation recorder is registered on a connection, each operation
that is executed on the connection will cause the :meth:`record`
method of the operation recorder object to be called.
"""
def __init__(self):
self.reset()
def reset(self):
self._pywbem_method = None
self._pywbem_args = None
self._pywbem_result_ret = None
self._pywbem_result_exc = None
self._http_request_version = None
self._http_request_url = None
self._http_request_target = None
self._http_request_method = None
self._http_request_headers = None
self._http_request_payload = None
self._http_response_version = None
self._http_response_status = None
self._http_response_reason = None
self._http_response_headers = None
self._http_response_payload = None
def stage_pywbem_args(self, method, **kwargs):
self._pywbem_method = method
self._pywbem_args = kwargs
def stage_pywbem_result(self, ret, exc):
self._pywbem_result_ret = ret
self._pywbem_result_exc = exc
def stage_http_request(self, version, url, target, method, headers,
payload):
self._http_request_version = version
self._http_request_url = url
self._http_request_target = target
self._http_request_method = method
self._http_request_headers = headers
self._http_request_payload = payload
def stage_http_response1(self, version, status, reason, headers):
self._http_response_version = version
self._http_response_status = status
self._http_response_reason = reason
self._http_response_headers = headers
def stage_http_response2(self, payload):
self._http_response_payload = payload
def record_staged(self):
pwargs = OpArgs(
self._pywbem_method,
self._pywbem_args)
pwresult = OpResult(
self._pywbem_result_ret,
self._pywbem_result_exc)
httpreq = HttpRequest(
self._http_request_version,
self._http_request_url,
self._http_request_target,
self._http_request_method,
self._http_request_headers,
self._http_request_payload)
httpresp = HttpResponse(
self._http_response_version,
self._http_response_status,
self._http_response_reason,
self._http_response_headers,
self._http_response_payload)
self.record(pwargs, pwresult, httpreq, httpresp)
[docs] def record(self, pywbem_args, pywbem_result, http_request, http_response):
"""
Function that is called to record a single WBEM operation, i.e. the
invocation of a single :class:`~pywbem.WBEMConnection` method.
Parameters:
pywbem_args (:class:`~pywbem.OpArgs`):
The name and input arguments of the :class:`~pywbem.WBEMConnection`
method that is recorded.
pywbem_result (:class:`~pywbem.OpResult`):
The result (return value or exception) of the
:class:`~pywbem.WBEMConnection` method that is recorded.
http_request (:class:`~pywbem.HttpRequest`):
The HTTP request sent by the :class:`~pywbem.WBEMConnection` method
that is recorded.
`None`, if no HTTP request had been sent (e.g. because an exception
was raised before getting there).
http_response (:class:`~pywbem.HttpResponse`):
The HTTP response received by the :class:`~pywbem.WBEMConnection`
method that is recorded.
`None`, if no HTTP response had been received (e.g. because an
exception was raised before getting there).
"""
raise NotImplementedError
[docs]class TestClientRecorder(BaseOperationRecorder):
"""
An operation recorder that generates test cases for each recorded
operation. The test cases are in the YAML format suitable for the
`test_client` unit test module of the pywbem project.
"""
# HTTP header fields to exclude when creating the testcase
# (in lower case)
EXCLUDE_REQUEST_HEADERS = [
'authorization',
'content-length',
'content-type',
]
EXCLUDE_RESPONSE_HEADERS = [
'content-length',
'content-type',
]
# Dummy server URL and credentials for use in generated test case
TESTCASE_URL = 'http://acme.com:80'
TESTCASE_USER = 'username'
TESTCASE_PASSWORD = 'password'
def __init__(self, fp):
"""
Parameters:
fp (file):
An open file that each test case will be written to.
"""
super(TestClientRecorder, self).__init__()
self._fp = fp
[docs] def record(self, pywbem_args, pywbem_result, http_request, http_response):
"""
Function that records the invocation of a single
:class:`~pywbem.WBEMConnection` method, by appending a corresponding
test case to the file.
Parameters: See :meth:`pywbem.BaseOperationRecorder.record`.
"""
testcase = OrderedDict()
testcase['name'] = pywbem_args.method
testcase['description'] = 'Generated by TestClientRecorder'
tc_pywbem_request = OrderedDict()
tc_pywbem_request['url'] = TestClientRecorder.TESTCASE_URL
tc_pywbem_request['creds'] = [TestClientRecorder.TESTCASE_USER,
TestClientRecorder.TESTCASE_PASSWORD]
tc_pywbem_request['namespace'] = 'root/cimv2'
tc_pywbem_request['timeout'] = 10
tc_pywbem_request['debug'] = False
tc_operation = OrderedDict()
tc_operation['pywbem_method'] = pywbem_args.method
for arg_name in pywbem_args.args:
tc_operation[arg_name] = self.toyaml(pywbem_args.args[arg_name])
tc_pywbem_request['operation'] = tc_operation
testcase['pywbem_request'] = tc_pywbem_request
tc_pywbem_response = OrderedDict()
if pywbem_result.ret is not None:
tc_pywbem_response['result'] = self.toyaml(pywbem_result.ret)
if pywbem_result.exc is not None:
exc = pywbem_result.exc
if isinstance(exc, CIMError):
tc_pywbem_response['cim_status'] = self.toyaml(exc.status_code)
else:
tc_pywbem_response['exception'] = self.toyaml(
exc.__class__.__name__)
testcase['pywbem_response'] = tc_pywbem_response
tc_http_request = OrderedDict()
if http_request is not None:
tc_http_request['verb'] = http_request.method
tc_http_request['url'] = TestClientRecorder.TESTCASE_URL + \
http_request.target
tc_request_headers = OrderedDict()
if http_request.headers is not None:
for hdr_name in http_request.headers:
if hdr_name.lower() not in \
TestClientRecorder.EXCLUDE_REQUEST_HEADERS:
tc_request_headers[hdr_name] = \
http_request.headers[hdr_name]
tc_http_request['headers'] = tc_request_headers
tc_http_request['data'] = http_request.payload
testcase['http_request'] = tc_http_request
tc_http_response = OrderedDict()
if http_response is not None:
tc_http_response['status'] = http_response.status
tc_response_headers = OrderedDict()
if http_response.headers is not None:
for hdr_name in http_response.headers:
if hdr_name.lower() not in \
TestClientRecorder.EXCLUDE_RESPONSE_HEADERS:
tc_response_headers[hdr_name] = \
http_response.headers[hdr_name]
tc_http_response['headers'] = tc_response_headers
if http_response.payload is not None:
data = http_response.payload.replace("\n\n", "\n")
else:
data = None
tc_http_response['data'] = data
else:
tc_http_response['exception'] = "# Change this to a callback " \
"function that causes this " \
"condition."
testcase['http_response'] = tc_http_response
testcases = []
testcases.append(testcase)
self._fp.write(yaml.safe_dump(testcases,
encoding='utf-8',
allow_unicode=True,
default_flow_style=False))
self._fp.flush()
[docs] def toyaml(self, obj):
"""
Convert any allowable input argument to or return value from an
operation method to an object that is ready for serialization into
test_client yaml format.
"""
if isinstance(obj, (list, tuple)):
ret = []
for item in obj:
ret.append(self.toyaml(item))
return ret
elif isinstance(obj, (dict, NocaseDict)):
ret = OrderedDict()
for key in obj:
ret[key] = self.toyaml(obj[key])
return ret
elif obj is None:
return obj
elif isinstance(obj, six.binary_type):
return obj.decode("utf-8")
elif isinstance(obj, six.text_type):
return obj
elif isinstance(obj, (bool, int)):
return obj
elif isinstance(obj, CIMInt):
return _Longint(obj)
elif isinstance(obj, CIMFloat):
return float(obj)
elif isinstance(obj, CIMDateTime):
return str(obj)
elif isinstance(obj, (datetime, timedelta)):
return CIMDateTime(obj)
elif isinstance(obj, CIMInstance):
ret = OrderedDict()
ret['pywbem_object'] = 'CIMInstance'
ret['classname'] = self.toyaml(obj.classname)
ret['properties'] = self.toyaml(obj.properties)
ret['path'] = self.toyaml(obj.path)
return ret
elif isinstance(obj, CIMInstanceName):
ret = OrderedDict()
ret['pywbem_object'] = 'CIMInstanceName'
ret['classname'] = self.toyaml(obj.classname)
ret['namespace'] = self.toyaml(obj.namespace)
ret['keybindings'] = self.toyaml(obj.keybindings)
return ret
elif isinstance(obj, CIMClass):
ret = OrderedDict()
ret['pywbem_object'] = 'CIMClass'
ret['classname'] = self.toyaml(obj.classname)
ret['superclass'] = self.toyaml(obj.superclass)
ret['properties'] = self.toyaml(obj.properties)
ret['methods'] = self.toyaml(obj.methods)
ret['qualifiers'] = self.toyaml(obj.qualifiers)
return ret
elif isinstance(obj, CIMClassName):
ret = OrderedDict()
ret['pywbem_object'] = 'CIMClassName'
ret['classname'] = self.toyaml(obj.classname)
ret['host'] = self.toyaml(obj.host)
ret['namespace'] = self.toyaml(obj.namespace)
return ret
elif isinstance(obj, CIMProperty):
ret = OrderedDict()
ret['pywbem_object'] = 'CIMProperty'
ret['name'] = self.toyaml(obj.name)
ret['value'] = self.toyaml(obj.value)
ret['type'] = self.toyaml(obj.type)
ret['reference_class'] = self.toyaml(obj.reference_class)
ret['embedded_object'] = self.toyaml(obj.embedded_object)
ret['is_array'] = self.toyaml(obj.is_array)
ret['array_size'] = self.toyaml(obj.array_size)
ret['class_origin'] = self.toyaml(obj.class_origin)
ret['propagated'] = self.toyaml(obj.propagated)
ret['qualifiers'] = self.toyaml(obj.qualifiers)
return ret
elif isinstance(obj, CIMMethod):
ret = OrderedDict()
ret['pywbem_object'] = 'CIMMethod'
ret['name'] = self.toyaml(obj.name)
ret['return_type'] = self.toyaml(obj.return_type)
ret['class_origin'] = self.toyaml(obj.class_origin)
ret['propagated'] = self.toyaml(obj.propagated)
ret['parameters'] = self.toyaml(obj.parameters)
ret['qualifiers'] = self.toyaml(obj.qualifiers)
return ret
elif isinstance(obj, CIMParameter):
ret = OrderedDict()
ret['pywbem_object'] = 'CIMParameter'
ret['name'] = self.toyaml(obj.name)
ret['type'] = self.toyaml(obj.type)
ret['reference_class'] = self.toyaml(obj.reference_class)
ret['is_array'] = self.toyaml(obj.is_array)
ret['array_size'] = self.toyaml(obj.array_size)
ret['qualifiers'] = self.toyaml(obj.qualifiers)
return ret
elif isinstance(obj, CIMQualifier):
ret = OrderedDict()
ret['pywbem_object'] = 'CIMQualifier'
ret['name'] = self.toyaml(obj.name)
ret['value'] = self.toyaml(obj.value)
ret['type'] = self.toyaml(obj.type)
ret['propagated'] = self.toyaml(obj.propagated)
ret['tosubclass'] = self.toyaml(obj.tosubclass)
ret['toinstance'] = self.toyaml(obj.toinstance)
ret['overridable'] = self.toyaml(obj.overridable)
ret['translatable'] = self.toyaml(obj.translatable)
return ret
elif isinstance(obj, CIMQualifierDeclaration):
ret = OrderedDict()
ret['pywbem_object'] = 'CIMQualifierDeclaration'
ret['name'] = self.toyaml(obj.name)
ret['type'] = self.toyaml(obj.type)
ret['value'] = self.toyaml(obj.value)
ret['is_array'] = self.toyaml(obj.is_array)
ret['array_size'] = self.toyaml(obj.array_size)
ret['scopes'] = self.toyaml(obj.scopes)
ret['tosubclass'] = self.toyaml(obj.tosubclass)
ret['toinstance'] = self.toyaml(obj.toinstance)
ret['overridable'] = self.toyaml(obj.overridable)
ret['translatable'] = self.toyaml(obj.translatable)
return ret
else:
raise TypeError("Invalid type in TestClientRecorder.toyaml(): " \
"%s %s" % (obj.__class__.__name__, type(obj)))