mirror of
https://github.com/OpenKMIP/PyKMIP.git
synced 2025-04-08 19:25:06 +02:00
Update payloads for the Activate operation
This change updates the request and response payloads for the Activate operation, bringing them up-to-date with current coding style. The unit test suites for the payloads have been overhauled to also match current standards. Payload usage in the client and server also now comply with the updated implementations.
This commit is contained in:
Peter Hamilton
parent
278a54320c
commit
09473e6c1c
@ -13,4 +13,12 @@
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
__all__ = ['create', 'destroy', 'get', 'locate', 'register']
|
||||
from kmip.core.messages.payloads.activate import (
|
||||
ActivateRequestPayload, ActivateResponsePayload
|
||||
)
|
||||
|
||||
|
||||
__all__ = [
|
||||
'ActivateRequestPayload',
|
||||
'ActivateResponsePayload'
|
||||
]
|
||||
|
||||
@ -13,151 +13,226 @@
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
from kmip.core import attributes
|
||||
import six
|
||||
|
||||
from kmip.core import enums
|
||||
|
||||
from kmip.core.primitives import Struct
|
||||
|
||||
from kmip.core.utils import BytearrayStream
|
||||
from kmip.core import primitives
|
||||
from kmip.core import utils
|
||||
|
||||
|
||||
class ActivateRequestPayload(Struct):
|
||||
class ActivateRequestPayload(primitives.Struct):
|
||||
"""
|
||||
A request payload for the Activate operation.
|
||||
|
||||
The payload contains a UUID of a cryptographic object that that server
|
||||
should activate. See Section 4.19 of the KMIP 1.1 specification for more
|
||||
information.
|
||||
|
||||
Attributes:
|
||||
unique_identifier: The UUID of a managed cryptographic object
|
||||
unique_identifier: The unique ID of the managed object to activate
|
||||
on the server.
|
||||
"""
|
||||
def __init__(self,
|
||||
unique_identifier=None):
|
||||
"""
|
||||
Construct a ActivateRequestPayload object.
|
||||
Construct an Activate request payload struct.
|
||||
|
||||
Args:
|
||||
unique_identifier (UniqueIdentifier): The UUID of a managed
|
||||
cryptographic object.
|
||||
unique_identifier (string): The ID of the managed object (e.g., a
|
||||
symmetric key) to activate. Optional, defaults to None.
|
||||
"""
|
||||
super(ActivateRequestPayload, self).__init__(
|
||||
tag=enums.Tags.REQUEST_PAYLOAD)
|
||||
self.unique_identifier = unique_identifier
|
||||
self.validate()
|
||||
|
||||
def read(self, istream):
|
||||
self._unique_identifier = None
|
||||
|
||||
self.unique_identifier = unique_identifier
|
||||
|
||||
@property
|
||||
def unique_identifier(self):
|
||||
if self._unique_identifier:
|
||||
return self._unique_identifier.value
|
||||
else:
|
||||
return None
|
||||
|
||||
@unique_identifier.setter
|
||||
def unique_identifier(self, value):
|
||||
if value is None:
|
||||
self._unique_identifier = None
|
||||
elif isinstance(value, six.string_types):
|
||||
self._unique_identifier = primitives.TextString(
|
||||
value=value,
|
||||
tag=enums.Tags.UNIQUE_IDENTIFIER
|
||||
)
|
||||
else:
|
||||
raise TypeError("Unique identifier must be a string.")
|
||||
|
||||
def read(self, input_stream):
|
||||
"""
|
||||
Read the data encoding the ActivateRequestPayload object and decode it
|
||||
Read the data encoding the Activate request payload and decode it
|
||||
into its constituent parts.
|
||||
Args:
|
||||
istream (Stream): A data stream containing encoded object data,
|
||||
supporting a read method; usually a BytearrayStream object.
|
||||
input_stream (stream): A data stream containing encoded object
|
||||
data, supporting a read method; usually a BytearrayStream
|
||||
object.
|
||||
"""
|
||||
super(ActivateRequestPayload, self).read(istream)
|
||||
tstream = BytearrayStream(istream.read(self.length))
|
||||
super(ActivateRequestPayload, self).read(input_stream)
|
||||
local_stream = utils.BytearrayStream(input_stream.read(self.length))
|
||||
|
||||
self.unique_identifier = attributes.UniqueIdentifier()
|
||||
self.unique_identifier.read(tstream)
|
||||
if self.is_tag_next(enums.Tags.UNIQUE_IDENTIFIER, local_stream):
|
||||
self._unique_identifier = primitives.TextString(
|
||||
tag=enums.Tags.UNIQUE_IDENTIFIER
|
||||
)
|
||||
self._unique_identifier.read(local_stream)
|
||||
|
||||
self.is_oversized(tstream)
|
||||
self.validate()
|
||||
self.is_oversized(local_stream)
|
||||
|
||||
def write(self, ostream):
|
||||
def write(self, output_stream):
|
||||
"""
|
||||
Write the data encoding the ActivateRequestPayload object to a stream.
|
||||
Write the data encoding the Activate request payload to a stream.
|
||||
|
||||
Args:
|
||||
ostream (Stream): A data stream in which to encode object data,
|
||||
supporting a write method; usually a BytearrayStream object.
|
||||
output_stream (stream): A data stream in which to encode object
|
||||
data, supporting a write method; usually a BytearrayStream
|
||||
object.
|
||||
"""
|
||||
tstream = BytearrayStream()
|
||||
local_stream = utils.BytearrayStream()
|
||||
|
||||
# Write the contents of the request payload
|
||||
if self.unique_identifier is not None:
|
||||
self.unique_identifier.write(tstream)
|
||||
if self._unique_identifier is not None:
|
||||
self._unique_identifier.write(local_stream)
|
||||
|
||||
# Write the length and value of the request payload
|
||||
self.length = tstream.length()
|
||||
super(ActivateRequestPayload, self).write(ostream)
|
||||
ostream.write(tstream.buffer)
|
||||
self.length = local_stream.length()
|
||||
super(ActivateRequestPayload, self).write(output_stream)
|
||||
output_stream.write(local_stream.buffer)
|
||||
|
||||
def validate(self):
|
||||
"""
|
||||
Error check the attributes of the ActivateRequestPayload object.
|
||||
"""
|
||||
if self.unique_identifier is not None:
|
||||
if not isinstance(self.unique_identifier,
|
||||
attributes.UniqueIdentifier):
|
||||
msg = "invalid unique identifier"
|
||||
raise TypeError(msg)
|
||||
def __eq__(self, other):
|
||||
if isinstance(other, ActivateRequestPayload):
|
||||
if self.unique_identifier != other.unique_identifier:
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
else:
|
||||
return NotImplemented
|
||||
|
||||
def __ne__(self, other):
|
||||
if isinstance(other, ActivateRequestPayload):
|
||||
return not (self == other)
|
||||
else:
|
||||
return NotImplemented
|
||||
|
||||
def __repr__(self):
|
||||
arg = "unique_identifier='{0}'".format(self.unique_identifier)
|
||||
return "ActivateRequestPayload({0})".format(arg)
|
||||
|
||||
def __str__(self):
|
||||
return str({'unique_identifier': self.unique_identifier})
|
||||
|
||||
|
||||
class ActivateResponsePayload(Struct):
|
||||
class ActivateResponsePayload(primitives.Struct):
|
||||
"""
|
||||
A response payload for the Activate operation.
|
||||
|
||||
The payload contains the server response to the initial Activate request.
|
||||
See Section 4.19 of the KMIP 1.1 specification for more information.
|
||||
|
||||
Attributes:
|
||||
unique_identifier: The UUID of a managed cryptographic object.
|
||||
unique_identifier: The unique ID of the managed object that was
|
||||
activated on the server.
|
||||
"""
|
||||
def __init__(self,
|
||||
unique_identifier=None):
|
||||
"""
|
||||
Construct a ActivateResponsePayload object.
|
||||
Construct an Activate response payload struct.
|
||||
|
||||
Args:
|
||||
unique_identifier (UniqueIdentifier): The UUID of a managed
|
||||
cryptographic object.
|
||||
unique_identifier (string): The ID of the managed object (e.g., a
|
||||
symmetric key) that was activated. Optional, defaults to None.
|
||||
Required for read/write.
|
||||
"""
|
||||
super(ActivateResponsePayload, self).__init__(
|
||||
tag=enums.Tags.RESPONSE_PAYLOAD)
|
||||
if unique_identifier is None:
|
||||
self.unique_identifier = attributes.UniqueIdentifier()
|
||||
else:
|
||||
self.unique_identifier = unique_identifier
|
||||
self.validate()
|
||||
|
||||
def read(self, istream):
|
||||
self._unique_identifier = None
|
||||
|
||||
self.unique_identifier = unique_identifier
|
||||
|
||||
@property
|
||||
def unique_identifier(self):
|
||||
if self._unique_identifier:
|
||||
return self._unique_identifier.value
|
||||
else:
|
||||
return None
|
||||
|
||||
@unique_identifier.setter
|
||||
def unique_identifier(self, value):
|
||||
if value is None:
|
||||
self._unique_identifier = None
|
||||
elif isinstance(value, six.string_types):
|
||||
self._unique_identifier = primitives.TextString(
|
||||
value=value,
|
||||
tag=enums.Tags.UNIQUE_IDENTIFIER
|
||||
)
|
||||
else:
|
||||
raise TypeError("Unique identifier must be a string.")
|
||||
|
||||
def read(self, input_stream):
|
||||
"""
|
||||
Read the data encoding the ActivateResponsePayload object and decode it
|
||||
Read the data encoding the Activate response payload and decode it
|
||||
into its constituent parts.
|
||||
|
||||
Args:
|
||||
istream (Stream): A data stream containing encoded object data,
|
||||
supporting a read method; usually a BytearrayStream object.
|
||||
input_stream (stream): A data stream containing encoded object
|
||||
data, supporting a read method; usually a BytearrayStream
|
||||
object.
|
||||
"""
|
||||
super(ActivateResponsePayload, self).read(istream)
|
||||
tstream = BytearrayStream(istream.read(self.length))
|
||||
super(ActivateResponsePayload, self).read(input_stream)
|
||||
local_stream = utils.BytearrayStream(input_stream.read(self.length))
|
||||
|
||||
self.unique_identifier = attributes.UniqueIdentifier()
|
||||
self.unique_identifier.read(tstream)
|
||||
if self.is_tag_next(enums.Tags.UNIQUE_IDENTIFIER, local_stream):
|
||||
self._unique_identifier = primitives.TextString(
|
||||
tag=enums.Tags.UNIQUE_IDENTIFIER
|
||||
)
|
||||
self._unique_identifier.read(local_stream)
|
||||
else:
|
||||
raise ValueError(
|
||||
"Parsed payload encoding is missing the unique identifier "
|
||||
"field."
|
||||
)
|
||||
|
||||
self.is_oversized(tstream)
|
||||
self.validate()
|
||||
self.is_oversized(local_stream)
|
||||
|
||||
def write(self, ostream):
|
||||
def write(self, output_stream):
|
||||
"""
|
||||
Write the data encoding the ActivateResponsePayload object to a stream.
|
||||
Write the data encoding the Activate response payload to a stream.
|
||||
|
||||
Args:
|
||||
ostream (Stream): A data stream in which to encode object data,
|
||||
supporting a write method; usually a BytearrayStream object.
|
||||
output_stream (stream): A data stream in which to encode object
|
||||
data, supporting a write method; usually a BytearrayStream
|
||||
object.
|
||||
"""
|
||||
tstream = BytearrayStream()
|
||||
local_stream = utils.BytearrayStream()
|
||||
|
||||
# Write the contents of the response payload
|
||||
self.unique_identifier.write(tstream)
|
||||
if self.unique_identifier:
|
||||
self._unique_identifier.write(local_stream)
|
||||
else:
|
||||
raise ValueError(
|
||||
"Payload is missing the unique identifier field."
|
||||
)
|
||||
self.length = local_stream.length()
|
||||
super(ActivateResponsePayload, self).write(output_stream)
|
||||
output_stream.write(local_stream.buffer)
|
||||
|
||||
# Write the length and value of the request payload
|
||||
self.length = tstream.length()
|
||||
super(ActivateResponsePayload, self).write(ostream)
|
||||
ostream.write(tstream.buffer)
|
||||
def __eq__(self, other):
|
||||
if isinstance(other, ActivateResponsePayload):
|
||||
if self.unique_identifier != other.unique_identifier:
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
else:
|
||||
return NotImplemented
|
||||
|
||||
def validate(self):
|
||||
"""
|
||||
Error check the attributes of the ActivateRequestPayload object.
|
||||
"""
|
||||
if not isinstance(self.unique_identifier, attributes.UniqueIdentifier):
|
||||
msg = "invalid unique identifier"
|
||||
raise TypeError(msg)
|
||||
def __ne__(self, other):
|
||||
if isinstance(other, ActivateResponsePayload):
|
||||
return not (self == other)
|
||||
else:
|
||||
return NotImplemented
|
||||
|
||||
def __repr__(self):
|
||||
arg = "unique_identifier='{0}'".format(self.unique_identifier)
|
||||
return "ActivateResponsePayload({0})".format(arg)
|
||||
|
||||
def __str__(self):
|
||||
return str({'unique_identifier': self.unique_identifier})
|
||||
|
||||
@ -884,11 +884,9 @@ class KMIPProxy(KMIP):
|
||||
def _activate(self, unique_identifier=None, credential=None):
|
||||
operation = Operation(OperationEnum.ACTIVATE)
|
||||
|
||||
uuid = None
|
||||
if unique_identifier is not None:
|
||||
uuid = attr.UniqueIdentifier(unique_identifier)
|
||||
|
||||
payload = activate.ActivateRequestPayload(unique_identifier=uuid)
|
||||
payload = activate.ActivateRequestPayload(
|
||||
unique_identifier=unique_identifier
|
||||
)
|
||||
|
||||
batch_item = messages.RequestBatchItem(operation=operation,
|
||||
request_payload=payload)
|
||||
|
||||
@ -1775,7 +1775,7 @@ class KmipEngine(object):
|
||||
self._logger.info("Processing operation: Activate")
|
||||
|
||||
if payload.unique_identifier:
|
||||
unique_identifier = payload.unique_identifier.value
|
||||
unique_identifier = payload.unique_identifier
|
||||
else:
|
||||
unique_identifier = self._id_placeholder
|
||||
|
||||
@ -1802,7 +1802,7 @@ class KmipEngine(object):
|
||||
self._data_session.commit()
|
||||
|
||||
response_payload = activate.ActivateResponsePayload(
|
||||
unique_identifier=attributes.UniqueIdentifier(unique_identifier)
|
||||
unique_identifier=unique_identifier
|
||||
)
|
||||
|
||||
return response_payload
|
||||
|
||||
@ -13,177 +13,522 @@
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
from testtools import TestCase
|
||||
import testtools
|
||||
|
||||
from kmip.core import utils
|
||||
from kmip.core import attributes
|
||||
|
||||
from kmip.core.messages.payloads import activate
|
||||
from kmip.core.messages import payloads
|
||||
|
||||
|
||||
class TestActivateRequestPayload(TestCase):
|
||||
class TestActivateRequestPayload(testtools.TestCase):
|
||||
"""
|
||||
Test suite for the ActivateRequestPayload class.
|
||||
|
||||
Test encodings obtained from Sections 4.2 of the KMIP 1.1 Test
|
||||
Cases documentation.
|
||||
Test suite for the Activate request payload class.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
super(TestActivateRequestPayload, self).setUp()
|
||||
|
||||
self.uuid = attributes.UniqueIdentifier(
|
||||
'668eff89-3010-4258-bc0e-8c402309c746')
|
||||
# Encoding obtained from the KMIP 1.1 testing document, Section 4.1.
|
||||
#
|
||||
# This encoding matches the following set of values:
|
||||
# Request Payload
|
||||
# Unique Identifier - 668eff89-3010-4258-bc0e-8c402309c746
|
||||
|
||||
self.encoding_a = utils.BytearrayStream((
|
||||
b'\x42\x00\x79\x01\x00\x00\x00\x30\x42\x00\x94\x07\x00\x00\x00\x24'
|
||||
self.full_encoding = utils.BytearrayStream(
|
||||
b'\x42\x00\x79\x01\x00\x00\x00\x30'
|
||||
b'\x42\x00\x94\x07\x00\x00\x00\x24'
|
||||
b'\x36\x36\x38\x65\x66\x66\x38\x39\x2D\x33\x30\x31\x30\x2D\x34\x32'
|
||||
b'\x35\x38\x2D\x62\x63\x30\x65\x2D\x38\x63\x34\x30\x32\x33\x30\x39'
|
||||
b'\x63\x37\x34\x36\x00\x00\x00\x00'))
|
||||
b'\x63\x37\x34\x36\x00\x00\x00\x00'
|
||||
)
|
||||
|
||||
self.empty_encoding = utils.BytearrayStream(
|
||||
b'\x42\x00\x79\x01\x00\x00\x00\x00'
|
||||
)
|
||||
|
||||
def tearDown(self):
|
||||
super(TestActivateRequestPayload, self).tearDown()
|
||||
|
||||
def test_init_with_none(self):
|
||||
def test_init(self):
|
||||
"""
|
||||
Test that a ActivateRequestPayload object can be constructed with no
|
||||
specified value.
|
||||
Test that an Activate request payload can be constructed with no
|
||||
arguments.
|
||||
"""
|
||||
activate.ActivateRequestPayload()
|
||||
payload = payloads.ActivateRequestPayload()
|
||||
|
||||
self.assertEqual(None, payload.unique_identifier)
|
||||
|
||||
def test_init_with_args(self):
|
||||
"""
|
||||
Test that a ActivateRequestPayload object can be constructed with valid
|
||||
Test that an Activate request payload can be constructed with valid
|
||||
values.
|
||||
"""
|
||||
activate.ActivateRequestPayload(unique_identifier=self.uuid)
|
||||
payload = payloads.ActivateRequestPayload(
|
||||
unique_identifier='00000000-2222-4444-6666-888888888888'
|
||||
)
|
||||
|
||||
def test_validate_with_bad_uuid_type(self):
|
||||
self.assertEqual(
|
||||
'00000000-2222-4444-6666-888888888888',
|
||||
payload.unique_identifier
|
||||
)
|
||||
|
||||
def test_invalid_unique_identifier(self):
|
||||
"""
|
||||
Test that a TypeError exception is raised when an invalid UUID type
|
||||
is used to construct a ActivateRequestPayload object.
|
||||
Test that a TypeError is raised when an invalid value is used to set
|
||||
the unique identifier of an Activate request payload.
|
||||
"""
|
||||
kwargs = {'unique_identifier': 0}
|
||||
self.assertRaisesRegexp(
|
||||
TypeError, "invalid unique identifier",
|
||||
activate.ActivateRequestPayload, "not-a-uuid")
|
||||
TypeError,
|
||||
"Unique identifier must be a string.",
|
||||
payloads.ActivateRequestPayload,
|
||||
**kwargs
|
||||
)
|
||||
|
||||
def test_read_with_known_uuid(self):
|
||||
"""
|
||||
Test that a ActivateRequestPayload object with known UUID can be read
|
||||
from a data stream.
|
||||
"""
|
||||
payload = activate.ActivateRequestPayload()
|
||||
payload.read(self.encoding_a)
|
||||
expected = '668eff89-3010-4258-bc0e-8c402309c746'
|
||||
observed = payload.unique_identifier.value
|
||||
args = (payloads.ActivateRequestPayload(), 'unique_identifier', 0)
|
||||
self.assertRaisesRegexp(
|
||||
TypeError,
|
||||
"Unique identifier must be a string.",
|
||||
setattr,
|
||||
*args
|
||||
)
|
||||
|
||||
msg = "activate UUID value mismatch"
|
||||
msg += "; expected {0}, received {1}".format(
|
||||
expected, observed)
|
||||
self.assertEqual(expected, observed, msg)
|
||||
def test_read(self):
|
||||
"""
|
||||
Test that an Activate request payload struct can be read from a data
|
||||
stream.
|
||||
"""
|
||||
payload = payloads.ActivateRequestPayload()
|
||||
|
||||
def test_write_with_known_uuid(self):
|
||||
self.assertEqual(None, payload.unique_identifier)
|
||||
|
||||
payload.read(self.full_encoding)
|
||||
|
||||
self.assertEqual(
|
||||
'668eff89-3010-4258-bc0e-8c402309c746',
|
||||
payload.unique_identifier
|
||||
)
|
||||
|
||||
def test_read_empty(self):
|
||||
"""
|
||||
Test that a ActivateRequestPayload object with a known UUID can be
|
||||
written to a data stream.
|
||||
Test that an Activate request payload struct can be read from an empty
|
||||
data stream.
|
||||
"""
|
||||
payload = payloads.ActivateRequestPayload()
|
||||
|
||||
self.assertEqual(None, payload.unique_identifier)
|
||||
|
||||
payload.read(self.empty_encoding)
|
||||
|
||||
self.assertEqual(None, payload.unique_identifier)
|
||||
|
||||
def test_write(self):
|
||||
"""
|
||||
Test that an Activate request payload struct can be written to a data
|
||||
stream.
|
||||
"""
|
||||
payload = payloads.ActivateRequestPayload(
|
||||
unique_identifier='668eff89-3010-4258-bc0e-8c402309c746'
|
||||
)
|
||||
stream = utils.BytearrayStream()
|
||||
payload = activate.ActivateRequestPayload(self.uuid)
|
||||
|
||||
payload.write(stream)
|
||||
|
||||
length_expected = len(self.encoding_a)
|
||||
length_received = len(stream)
|
||||
self.assertEqual(len(self.full_encoding), len(stream))
|
||||
self.assertEqual(str(self.full_encoding), str(stream))
|
||||
|
||||
msg = "encoding lengths not equal"
|
||||
msg += "; expected {0}, received {1}".format(
|
||||
length_expected, length_received)
|
||||
self.assertEqual(length_expected, length_received, msg)
|
||||
def test_write_empty(self):
|
||||
"""
|
||||
Test that an empty Activate request payload struct can be written to a
|
||||
data stream.
|
||||
"""
|
||||
payload = payloads.ActivateRequestPayload()
|
||||
stream = utils.BytearrayStream()
|
||||
|
||||
msg = "encoding mismatch"
|
||||
msg += ";\nexpected:\n{0}\nreceived:\n{1}".format(self.encoding_a,
|
||||
stream)
|
||||
payload.write(stream)
|
||||
|
||||
self.assertEqual(self.encoding_a, stream, msg)
|
||||
self.assertEqual(len(self.empty_encoding), len(stream))
|
||||
self.assertEqual(str(self.empty_encoding), str(stream))
|
||||
|
||||
def test_equal_on_equal(self):
|
||||
"""
|
||||
Test that the equality operator returns True when comparing two
|
||||
Activate request payload structs with the same data.
|
||||
"""
|
||||
a = payloads.ActivateRequestPayload()
|
||||
b = payloads.ActivateRequestPayload()
|
||||
|
||||
self.assertTrue(a == b)
|
||||
self.assertTrue(b == a)
|
||||
|
||||
a = payloads.ActivateRequestPayload(
|
||||
unique_identifier='668eff89-3010-4258-bc0e-8c402309c746'
|
||||
)
|
||||
b = payloads.ActivateRequestPayload(
|
||||
unique_identifier='668eff89-3010-4258-bc0e-8c402309c746'
|
||||
)
|
||||
|
||||
self.assertTrue(a == b)
|
||||
self.assertTrue(b == a)
|
||||
|
||||
def test_equal_on_not_equal_unique_identifier(self):
|
||||
"""
|
||||
Test that the equality operator returns False when comparing two
|
||||
Activate request payload structs with different unique identifiers.
|
||||
"""
|
||||
a = payloads.ActivateRequestPayload(
|
||||
unique_identifier='668eff89-3010-4258-bc0e-8c402309c746'
|
||||
)
|
||||
b = payloads.ActivateRequestPayload(
|
||||
unique_identifier='49a1ca88-6bea-4fb2-b450-7e58802c303f'
|
||||
)
|
||||
|
||||
self.assertFalse(a == b)
|
||||
self.assertFalse(b == a)
|
||||
|
||||
def test_equal_on_type_mismatch(self):
|
||||
"""
|
||||
Test that the equality operator returns False when comparing two
|
||||
Activate request payload structs with different types.
|
||||
"""
|
||||
a = payloads.ActivateRequestPayload()
|
||||
b = 'invalid'
|
||||
|
||||
self.assertFalse(a == b)
|
||||
self.assertFalse(b == a)
|
||||
|
||||
def test_not_equal_on_equal(self):
|
||||
"""
|
||||
Test that the inequality operator returns False when comparing two
|
||||
Activate request payload structs with the same data.
|
||||
"""
|
||||
a = payloads.ActivateRequestPayload()
|
||||
b = payloads.ActivateRequestPayload()
|
||||
|
||||
self.assertFalse(a != b)
|
||||
self.assertFalse(b != a)
|
||||
|
||||
a = payloads.ActivateRequestPayload(
|
||||
unique_identifier='668eff89-3010-4258-bc0e-8c402309c746'
|
||||
)
|
||||
b = payloads.ActivateRequestPayload(
|
||||
unique_identifier='668eff89-3010-4258-bc0e-8c402309c746'
|
||||
)
|
||||
|
||||
self.assertFalse(a != b)
|
||||
self.assertFalse(b != a)
|
||||
|
||||
def test_not_equal_on_not_equal_unique_identifier(self):
|
||||
"""
|
||||
Test that the inequality operator returns True when comparing two
|
||||
Activate request payload structs with different unique identifiers.
|
||||
"""
|
||||
a = payloads.ActivateRequestPayload(
|
||||
unique_identifier='668eff89-3010-4258-bc0e-8c402309c746'
|
||||
)
|
||||
b = payloads.ActivateRequestPayload(
|
||||
unique_identifier='49a1ca88-6bea-4fb2-b450-7e58802c303f'
|
||||
)
|
||||
|
||||
self.assertTrue(a != b)
|
||||
self.assertTrue(b != a)
|
||||
|
||||
def test_not_equal_on_type_mismatch(self):
|
||||
"""
|
||||
Test that the inequality operator returns True when comparing two
|
||||
Activate request payload structs with different types.
|
||||
"""
|
||||
a = payloads.ActivateRequestPayload()
|
||||
b = 'invalid'
|
||||
|
||||
self.assertTrue(a != b)
|
||||
self.assertTrue(b != a)
|
||||
|
||||
def test_repr(self):
|
||||
"""
|
||||
Test that repr can be applied to an Activate request payload struct.
|
||||
"""
|
||||
payload = payloads.ActivateRequestPayload(
|
||||
unique_identifier='49a1ca88-6bea-4fb2-b450-7e58802c3038'
|
||||
)
|
||||
|
||||
expected = (
|
||||
"ActivateRequestPayload("
|
||||
"unique_identifier='49a1ca88-6bea-4fb2-b450-7e58802c3038')"
|
||||
)
|
||||
observed = repr(payload)
|
||||
|
||||
self.assertEqual(expected, observed)
|
||||
|
||||
def test_str(self):
|
||||
"""
|
||||
Test that str can be applied to an Activate request payload struct.
|
||||
"""
|
||||
payload = payloads.ActivateRequestPayload(
|
||||
unique_identifier='49a1ca88-6bea-4fb2-b450-7e58802c3038'
|
||||
)
|
||||
|
||||
expected = str({
|
||||
'unique_identifier': '49a1ca88-6bea-4fb2-b450-7e58802c3038'
|
||||
})
|
||||
observed = str(payload)
|
||||
|
||||
self.assertEqual(expected, observed)
|
||||
|
||||
|
||||
class TestActivateResponsePayload(TestCase):
|
||||
class TestActivateResponsePayload(testtools.TestCase):
|
||||
"""
|
||||
Test encodings obtained from Sections 4.2 of the KMIP 1.1 Test
|
||||
Cases documentation.
|
||||
Test suite for the Activate request payload class.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
super(TestActivateResponsePayload, self).setUp()
|
||||
|
||||
self.uuid = attributes.UniqueIdentifier(
|
||||
'668eff89-3010-4258-bc0e-8c402309c746')
|
||||
# Encoding obtained from the KMIP 1.1 testing document, Section 4.1.
|
||||
#
|
||||
# This encoding matches the following set of values:
|
||||
# Request Payload
|
||||
# Unique Identifier - 668eff89-3010-4258-bc0e-8c402309c746
|
||||
|
||||
self.encoding_a = utils.BytearrayStream((
|
||||
b'\x42\x00\x7C\x01\x00\x00\x00\x30\x42\x00\x94\x07\x00\x00\x00\x24'
|
||||
self.full_encoding = utils.BytearrayStream(
|
||||
b'\x42\x00\x7C\x01\x00\x00\x00\x30'
|
||||
b'\x42\x00\x94\x07\x00\x00\x00\x24'
|
||||
b'\x36\x36\x38\x65\x66\x66\x38\x39\x2D\x33\x30\x31\x30\x2D\x34\x32'
|
||||
b'\x35\x38\x2D\x62\x63\x30\x65\x2D\x38\x63\x34\x30\x32\x33\x30\x39'
|
||||
b'\x63\x37\x34\x36\x00\x00\x00\x00'))
|
||||
b'\x63\x37\x34\x36\x00\x00\x00\x00'
|
||||
)
|
||||
|
||||
self.empty_encoding = utils.BytearrayStream(
|
||||
b'\x42\x00\x7C\x01\x00\x00\x00\x00'
|
||||
)
|
||||
|
||||
def tearDown(self):
|
||||
super(TestActivateResponsePayload, self).tearDown()
|
||||
|
||||
def test_init_with_none(self):
|
||||
def test_init(self):
|
||||
"""
|
||||
Test that a ActivateResponsePayload object can be constructed with no
|
||||
specified value.
|
||||
Test that an Activate response payload can be constructed with no
|
||||
arguments.
|
||||
"""
|
||||
activate.ActivateResponsePayload()
|
||||
payload = payloads.ActivateResponsePayload()
|
||||
|
||||
self.assertEqual(None, payload.unique_identifier)
|
||||
|
||||
def test_init_with_args(self):
|
||||
"""
|
||||
Test that a ActivateResponsePayload object can be constructed with
|
||||
valid values.
|
||||
Test that an Activate response payload can be constructed with valid
|
||||
values.
|
||||
"""
|
||||
activate.ActivateResponsePayload(unique_identifier=self.uuid)
|
||||
payload = payloads.ActivateResponsePayload(
|
||||
unique_identifier='00000000-2222-4444-6666-888888888888'
|
||||
)
|
||||
|
||||
def test_validate_with_invalid_uuid(self):
|
||||
self.assertEqual(
|
||||
'00000000-2222-4444-6666-888888888888',
|
||||
payload.unique_identifier
|
||||
)
|
||||
|
||||
def test_invalid_unique_identifier(self):
|
||||
"""
|
||||
Test that a TypeError exception is raised when an invalid Operations
|
||||
list is used to construct a ActivateResponsePayload object.
|
||||
Test that a TypeError is raised when an invalid value is used to set
|
||||
the unique identifier of an Activate response payload.
|
||||
"""
|
||||
kwargs = {'unique_identifier': 0}
|
||||
self.assertRaisesRegexp(
|
||||
TypeError, "invalid unique identifier",
|
||||
activate.ActivateResponsePayload, "not-a-uuid")
|
||||
TypeError,
|
||||
"Unique identifier must be a string.",
|
||||
payloads.ActivateResponsePayload,
|
||||
**kwargs
|
||||
)
|
||||
|
||||
def test_read_with_known_uuid(self):
|
||||
"""
|
||||
Test that a ActivateResponsePayload object with known UUID can be read
|
||||
from a data stream.
|
||||
"""
|
||||
payload = activate.ActivateResponsePayload()
|
||||
payload.read(self.encoding_a)
|
||||
expected = '668eff89-3010-4258-bc0e-8c402309c746'
|
||||
observed = payload.unique_identifier.value
|
||||
args = (payloads.ActivateResponsePayload(), 'unique_identifier', 0)
|
||||
self.assertRaisesRegexp(
|
||||
TypeError,
|
||||
"Unique identifier must be a string.",
|
||||
setattr,
|
||||
*args
|
||||
)
|
||||
|
||||
msg = "activate UUID value mismatch"
|
||||
msg += "; expected {0}, received {1}".format(
|
||||
expected, observed)
|
||||
self.assertEqual(expected, observed, msg)
|
||||
def test_read(self):
|
||||
"""
|
||||
Test that an Activate response payload struct can be read from a data
|
||||
stream.
|
||||
"""
|
||||
payload = payloads.ActivateResponsePayload()
|
||||
|
||||
def test_write_with_known_uuid(self):
|
||||
self.assertEqual(None, payload.unique_identifier)
|
||||
|
||||
payload.read(self.full_encoding)
|
||||
|
||||
self.assertEqual(
|
||||
'668eff89-3010-4258-bc0e-8c402309c746',
|
||||
payload.unique_identifier
|
||||
)
|
||||
|
||||
def test_read_missing_unique_identifier(self):
|
||||
"""
|
||||
Test that a ActivateResponsePayload object with a known UUID can be
|
||||
written to a data stream.
|
||||
Test that a ValueError gets raised when a required Activate request
|
||||
payload field is missing when decoding the struct.
|
||||
"""
|
||||
payload = payloads.ActivateResponsePayload()
|
||||
args = (self.empty_encoding, )
|
||||
self.assertRaisesRegexp(
|
||||
ValueError,
|
||||
"Parsed payload encoding is missing the unique identifier field.",
|
||||
payload.read,
|
||||
*args
|
||||
)
|
||||
|
||||
def test_write(self):
|
||||
"""
|
||||
Test that an Activate response payload struct can be written to a data
|
||||
stream.
|
||||
"""
|
||||
payload = payloads.ActivateResponsePayload(
|
||||
unique_identifier='668eff89-3010-4258-bc0e-8c402309c746'
|
||||
)
|
||||
stream = utils.BytearrayStream()
|
||||
payload = activate.ActivateResponsePayload(self.uuid)
|
||||
|
||||
payload.write(stream)
|
||||
|
||||
length_expected = len(self.encoding_a)
|
||||
length_received = len(stream)
|
||||
self.assertEqual(len(self.full_encoding), len(stream))
|
||||
self.assertEqual(str(self.full_encoding), str(stream))
|
||||
|
||||
msg = "encoding lengths not equal"
|
||||
msg += "; expected {0}, received {1}".format(
|
||||
length_expected, length_received)
|
||||
self.assertEqual(length_expected, length_received, msg)
|
||||
def test_write_missing_unique_identifier(self):
|
||||
"""
|
||||
Test that a ValueError gets raised when a required Activate response
|
||||
payload field is missing when encoding the struct.
|
||||
"""
|
||||
payload = payloads.ActivateResponsePayload()
|
||||
stream = utils.BytearrayStream()
|
||||
args = (stream, )
|
||||
self.assertRaisesRegexp(
|
||||
ValueError,
|
||||
"Payload is missing the unique identifier field.",
|
||||
payload.write,
|
||||
*args
|
||||
)
|
||||
|
||||
msg = "encoding mismatch"
|
||||
msg += ";\nexpected:\n{0}\nreceived:\n{1}".format(self.encoding_a,
|
||||
stream)
|
||||
def test_equal_on_equal(self):
|
||||
"""
|
||||
Test that the equality operator returns True when comparing two
|
||||
Activate response payload structs with the same data.
|
||||
"""
|
||||
a = payloads.ActivateResponsePayload()
|
||||
b = payloads.ActivateResponsePayload()
|
||||
|
||||
self.assertEqual(self.encoding_a, stream, msg)
|
||||
self.assertTrue(a == b)
|
||||
self.assertTrue(b == a)
|
||||
|
||||
a = payloads.ActivateResponsePayload(
|
||||
unique_identifier='668eff89-3010-4258-bc0e-8c402309c746'
|
||||
)
|
||||
b = payloads.ActivateResponsePayload(
|
||||
unique_identifier='668eff89-3010-4258-bc0e-8c402309c746'
|
||||
)
|
||||
|
||||
self.assertTrue(a == b)
|
||||
self.assertTrue(b == a)
|
||||
|
||||
def test_equal_on_not_equal_unique_identifier(self):
|
||||
"""
|
||||
Test that the equality operator returns False when comparing two
|
||||
Activate response payload structs with different unique identifiers.
|
||||
"""
|
||||
a = payloads.ActivateResponsePayload(
|
||||
unique_identifier='668eff89-3010-4258-bc0e-8c402309c746'
|
||||
)
|
||||
b = payloads.ActivateResponsePayload(
|
||||
unique_identifier='49a1ca88-6bea-4fb2-b450-7e58802c303f'
|
||||
)
|
||||
|
||||
self.assertFalse(a == b)
|
||||
self.assertFalse(b == a)
|
||||
|
||||
def test_equal_on_type_mismatch(self):
|
||||
"""
|
||||
Test that the equality operator returns False when comparing two
|
||||
Activate response payload structs with different types.
|
||||
"""
|
||||
a = payloads.ActivateResponsePayload()
|
||||
b = 'invalid'
|
||||
|
||||
self.assertFalse(a == b)
|
||||
self.assertFalse(b == a)
|
||||
|
||||
def test_not_equal_on_equal(self):
|
||||
"""
|
||||
Test that the inequality operator returns False when comparing two
|
||||
Activate response payload structs with the same data.
|
||||
"""
|
||||
a = payloads.ActivateResponsePayload()
|
||||
b = payloads.ActivateResponsePayload()
|
||||
|
||||
self.assertFalse(a != b)
|
||||
self.assertFalse(b != a)
|
||||
|
||||
a = payloads.ActivateResponsePayload(
|
||||
unique_identifier='668eff89-3010-4258-bc0e-8c402309c746'
|
||||
)
|
||||
b = payloads.ActivateResponsePayload(
|
||||
unique_identifier='668eff89-3010-4258-bc0e-8c402309c746'
|
||||
)
|
||||
|
||||
self.assertFalse(a != b)
|
||||
self.assertFalse(b != a)
|
||||
|
||||
def test_not_equal_on_not_equal_unique_identifier(self):
|
||||
"""
|
||||
Test that the inequality operator returns True when comparing two
|
||||
Activate response payload structs with different unique identifiers.
|
||||
"""
|
||||
a = payloads.ActivateResponsePayload(
|
||||
unique_identifier='668eff89-3010-4258-bc0e-8c402309c746'
|
||||
)
|
||||
b = payloads.ActivateResponsePayload(
|
||||
unique_identifier='49a1ca88-6bea-4fb2-b450-7e58802c303f'
|
||||
)
|
||||
|
||||
self.assertTrue(a != b)
|
||||
self.assertTrue(b != a)
|
||||
|
||||
def test_not_equal_on_type_mismatch(self):
|
||||
"""
|
||||
Test that the inequality operator returns True when comparing two
|
||||
Activate response payload structs with different types.
|
||||
"""
|
||||
a = payloads.ActivateResponsePayload()
|
||||
b = 'invalid'
|
||||
|
||||
self.assertTrue(a != b)
|
||||
self.assertTrue(b != a)
|
||||
|
||||
def test_repr(self):
|
||||
"""
|
||||
Test that repr can be applied to an Activate response payload struct.
|
||||
"""
|
||||
payload = payloads.ActivateResponsePayload(
|
||||
unique_identifier='49a1ca88-6bea-4fb2-b450-7e58802c3038'
|
||||
)
|
||||
|
||||
expected = (
|
||||
"ActivateResponsePayload("
|
||||
"unique_identifier='49a1ca88-6bea-4fb2-b450-7e58802c3038')"
|
||||
)
|
||||
observed = repr(payload)
|
||||
|
||||
self.assertEqual(expected, observed)
|
||||
|
||||
def test_str(self):
|
||||
"""
|
||||
Test that str can be applied to an Activate response payload struct.
|
||||
"""
|
||||
payload = payloads.ActivateResponsePayload(
|
||||
unique_identifier='49a1ca88-6bea-4fb2-b450-7e58802c3038'
|
||||
)
|
||||
|
||||
expected = str({
|
||||
'unique_identifier': '49a1ca88-6bea-4fb2-b450-7e58802c3038'
|
||||
})
|
||||
observed = str(payload)
|
||||
|
||||
self.assertEqual(expected, observed)
|
||||
|
||||
@ -39,7 +39,7 @@ from kmip.core.factories import attributes as factory
|
||||
from kmip.core.messages import contents
|
||||
from kmip.core.messages import messages
|
||||
|
||||
from kmip.core.messages.payloads import activate
|
||||
from kmip.core.messages import payloads
|
||||
from kmip.core.messages.payloads import revoke
|
||||
from kmip.core.messages.payloads import create
|
||||
from kmip.core.messages.payloads import create_key_pair
|
||||
@ -5580,10 +5580,7 @@ class TestKmipEngine(testtools.TestCase):
|
||||
|
||||
object_id = str(managed_object.unique_identifier)
|
||||
|
||||
# Test by specifying the ID of the object to activate.
|
||||
payload = activate.ActivateRequestPayload(
|
||||
unique_identifier=attributes.UniqueIdentifier(object_id)
|
||||
)
|
||||
payload = payloads.ActivateRequestPayload(unique_identifier=object_id)
|
||||
|
||||
response_payload = e._process_activate(payload)
|
||||
e._data_session.commit()
|
||||
@ -5592,10 +5589,7 @@ class TestKmipEngine(testtools.TestCase):
|
||||
e._logger.info.assert_any_call(
|
||||
"Processing operation: Activate"
|
||||
)
|
||||
self.assertEqual(
|
||||
str(object_id),
|
||||
response_payload.unique_identifier.value
|
||||
)
|
||||
self.assertEqual(str(object_id), response_payload.unique_identifier)
|
||||
|
||||
symmetric_key = e._data_session.query(
|
||||
pie_objects.SymmetricKey
|
||||
@ -5616,7 +5610,7 @@ class TestKmipEngine(testtools.TestCase):
|
||||
|
||||
# Test that the ID placeholder can also be used to specify activation.
|
||||
e._id_placeholder = str(object_id)
|
||||
payload = activate.ActivateRequestPayload()
|
||||
payload = payloads.ActivateRequestPayload()
|
||||
args = (payload,)
|
||||
regex = "The object state is not pre-active and cannot be activated."
|
||||
self.assertRaisesRegexp(
|
||||
@ -5647,10 +5641,7 @@ class TestKmipEngine(testtools.TestCase):
|
||||
|
||||
object_id = str(managed_object.unique_identifier)
|
||||
|
||||
# Test by specifying the ID of the object to activate.
|
||||
payload = activate.ActivateRequestPayload(
|
||||
unique_identifier=attributes.UniqueIdentifier(object_id)
|
||||
)
|
||||
payload = payloads.ActivateRequestPayload(unique_identifier=object_id)
|
||||
|
||||
args = (payload,)
|
||||
name = enums.ObjectType.OPAQUE_DATA.name
|
||||
@ -5689,10 +5680,7 @@ class TestKmipEngine(testtools.TestCase):
|
||||
|
||||
object_id = str(managed_object.unique_identifier)
|
||||
|
||||
# Test by specifying the ID of the object to activate.
|
||||
payload = activate.ActivateRequestPayload(
|
||||
unique_identifier=attributes.UniqueIdentifier(object_id)
|
||||
)
|
||||
payload = payloads.ActivateRequestPayload(unique_identifier=object_id)
|
||||
|
||||
args = (payload,)
|
||||
regex = "The object state is not pre-active and cannot be activated."
|
||||
@ -5722,11 +5710,8 @@ class TestKmipEngine(testtools.TestCase):
|
||||
e._data_session = e._data_store_session_factory()
|
||||
|
||||
id_a = str(obj_a.unique_identifier)
|
||||
payload = activate.ActivateRequestPayload(
|
||||
unique_identifier=attributes.UniqueIdentifier(id_a)
|
||||
)
|
||||
payload = payloads.ActivateRequestPayload(unique_identifier=id_a)
|
||||
|
||||
# Test by specifying the ID of the object to activate.
|
||||
args = [payload]
|
||||
self.assertRaisesRegex(
|
||||
exceptions.ItemNotFound,
|
||||
@ -7991,9 +7976,7 @@ class TestKmipEngine(testtools.TestCase):
|
||||
e._logger.reset_mock()
|
||||
|
||||
# Activate the symmetric key
|
||||
payload = activate.ActivateRequestPayload(
|
||||
attributes.UniqueIdentifier(uuid)
|
||||
)
|
||||
payload = payloads.ActivateRequestPayload(uuid)
|
||||
|
||||
response_payload = e._process_activate(payload)
|
||||
e._data_session.commit()
|
||||
@ -8003,7 +7986,7 @@ class TestKmipEngine(testtools.TestCase):
|
||||
"Processing operation: Activate"
|
||||
)
|
||||
|
||||
activated_uuid = response_payload.unique_identifier.value
|
||||
activated_uuid = response_payload.unique_identifier
|
||||
self.assertEqual(uuid, activated_uuid)
|
||||
|
||||
# Encrypt some data using the symmetric key
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user