diff --git a/kmip/core/messages/payloads/__init__.py b/kmip/core/messages/payloads/__init__.py index 4f89fd3..075e4ed 100644 --- a/kmip/core/messages/payloads/__init__.py +++ b/kmip/core/messages/payloads/__init__.py @@ -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' +] diff --git a/kmip/core/messages/payloads/activate.py b/kmip/core/messages/payloads/activate.py index d33228d..919a4e4 100644 --- a/kmip/core/messages/payloads/activate.py +++ b/kmip/core/messages/payloads/activate.py @@ -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}) diff --git a/kmip/services/kmip_client.py b/kmip/services/kmip_client.py index bc22d23..d3454a6 100644 --- a/kmip/services/kmip_client.py +++ b/kmip/services/kmip_client.py @@ -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) diff --git a/kmip/services/server/engine.py b/kmip/services/server/engine.py index 3f72182..5bc609e 100644 --- a/kmip/services/server/engine.py +++ b/kmip/services/server/engine.py @@ -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 diff --git a/kmip/tests/unit/core/messages/payloads/test_activate.py b/kmip/tests/unit/core/messages/payloads/test_activate.py index 9c7719a..89205f7 100644 --- a/kmip/tests/unit/core/messages/payloads/test_activate.py +++ b/kmip/tests/unit/core/messages/payloads/test_activate.py @@ -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) diff --git a/kmip/tests/unit/services/server/test_engine.py b/kmip/tests/unit/services/server/test_engine.py index 0c40711..2955773 100644 --- a/kmip/tests/unit/services/server/test_engine.py +++ b/kmip/tests/unit/services/server/test_engine.py @@ -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