tyler.durden
/
PyKMIP
mirror of https://github.com/OpenKMIP/PyKMIP.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add payloads for the ObtainLease operation 

This change adds request and response payloads for the ObtainLease
operation. Unit test suites are included for both payloads and
both payloads can be imported directly from the payloads package.
This commit is contained in:
Peter Hamilton 2017-12-04 13:53:48 -05:00
parent fe52d7a03d
commit 42da7f1d15
3 changed files with 1046 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
kmip/core/messages/payloads/__init__.py
View File

@ -81,6 +81,10 @@ from kmip.core.messages.payloads.mac import (
MACRequestPayload,
MACResponsePayload
)
from kmip.core.messages.payloads.obtain_lease import (
ObtainLeaseRequestPayload,
ObtainLeaseResponsePayload
)
from kmip.core.messages.payloads.query import (
QueryRequestPayload,
QueryResponsePayload
@ -146,6 +150,8 @@ __all__ = [
"LocateResponsePayload",
"MACRequestPayload",
"MACResponsePayload",
"ObtainLeaseRequestPayload",
"ObtainLeaseResponsePayload",
"QueryRequestPayload",
"QueryResponsePayload",
"RecoverRequestPayload",

328
kmip/core/messages/payloads/obtain_lease.py Normal file
View File

@ -0,0 +1,328 @@
# Copyright (c) 2017 The Johns Hopkins University/Applied Physics Laboratory
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import six
from kmip import enums
from kmip.core import primitives
from kmip.core import utils
class ObtainLeaseRequestPayload(primitives.Struct):
"""
A request payload for the ObtainLease operation.
Attributes:
unique_identifier: The unique ID of the object to be leased.
"""
def __init__(self, unique_identifier=None):
"""
Construct an ObtainLease request payload struct.
Args:
unique_identifier (string): The ID of the managed object (e.g.,
a public key) to obtain a lease for. Optional, defaults to
None.
"""
super(ObtainLeaseRequestPayload, self).__init__(
enums.Tags.REQUEST_PAYLOAD
)
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 ObtainLease request payload and decode it
into its constituent parts.
Args:
input_stream (stream): A data stream containing encoded object
data, supporting a read method; usually a BytearrayStream
object.
Raises:
ValueError: Raised if the data attribute is missing from the
encoded payload.
"""
super(ObtainLeaseRequestPayload, self).read(input_stream)
local_stream = utils.BytearrayStream(input_stream.read(self.length))
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(local_stream)
def write(self, output_stream):
"""
Write the data encoding the ObtainLease request payload to a stream.
Args:
output_stream (stream): A data stream in which to encode object
data, supporting a write method; usually a BytearrayStream
object.
Raises:
ValueError: Raised if the data attribute is not defined.
"""
local_stream = utils.BytearrayStream()
if self._unique_identifier:
self._unique_identifier.write(local_stream)
self.length = local_stream.length()
super(ObtainLeaseRequestPayload, self).write(output_stream)
output_stream.write(local_stream.buffer)
def __eq__(self, other):
if isinstance(other, ObtainLeaseRequestPayload):
if self.unique_identifier != other.unique_identifier:
return False
else:
return True
else:
return NotImplemented
def __ne__(self, other):
if isinstance(other, ObtainLeaseRequestPayload):
return not (self == other)
else:
return NotImplemented
def __repr__(self):
args = "unique_identifier='{0}'".format(self.unique_identifier)
return "ObtainLeaseRequestPayload({0})".format(args)
def __str__(self):
return str({
'unique_identifier': self.unique_identifier
})
class ObtainLeaseResponsePayload(primitives.Struct):
"""
A response payload for the ObtainLease operation.
Attributes:
unique_identifier: The unique ID of the object that was leased.
lease_time: The amount of time, in seconds, that the object lease is
in effect.
last_change_date: The date, in seconds since the epoch, representing
the last time a change was made to the object or one of its
attributes.
"""
def __init__(self,
unique_identifier=None,
lease_time=None,
last_change_date=None):
"""
Construct an ObtainLease response payload struct.
Args:
unique_identifier (string): The ID of the managed object (e.g.,
a public key) a lease was obtained for. Optional, defaults to
None.
lease_time (int): The amount of time, in seconds, that the object
lease is in effect for. Optional, defaults to None.
last_change_date (int): The date, in seconds since the epoch,
when the last change was made to the object or one of its
attributes. Optional, defaults to None.
"""
super(ObtainLeaseResponsePayload, self).__init__(
enums.Tags.RESPONSE_PAYLOAD
)
self._unique_identifier = None
self._lease_time = None
self._last_change_date = None
self.unique_identifier = unique_identifier
self.lease_time = lease_time
self.last_change_date = last_change_date
@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.")
@property
def lease_time(self):
if self._lease_time:
return self._lease_time.value
else:
return None
@lease_time.setter
def lease_time(self, value):
if value is None:
self._lease_time = None
elif isinstance(value, six.integer_types):
self._lease_time = primitives.Interval(
value=value,
tag=enums.Tags.LEASE_TIME
)
else:
raise TypeError("Lease time must be an integer.")
@property
def last_change_date(self):
if self._last_change_date:
return self._last_change_date.value
else:
return None
@last_change_date.setter
def last_change_date(self, value):
if value is None:
self._last_change_date = None
elif isinstance(value, six.integer_types):
self._last_change_date = primitives.DateTime(
value=value,
tag=enums.Tags.LAST_CHANGE_DATE
)
else:
raise TypeError("Last change date must be an integer.")
def read(self, input_stream):
"""
Read the data encoding the ObtainLease response payload and decode it
into its constituent parts.
Args:
input_stream (stream): A data stream containing encoded object
data, supporting a read method; usually a BytearrayStream
object.
Raises:
ValueError: Raised if the data attribute is missing from the
encoded payload.
"""
super(ObtainLeaseResponsePayload, self).read(input_stream)
local_stream = utils.BytearrayStream(input_stream.read(self.length))
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)
if self.is_tag_next(enums.Tags.LEASE_TIME, local_stream):
self._lease_time = primitives.Interval(
tag=enums.Tags.LEASE_TIME
)
self._lease_time.read(local_stream)
if self.is_tag_next(enums.Tags.LAST_CHANGE_DATE, local_stream):
self._last_change_date = primitives.DateTime(
tag=enums.Tags.LAST_CHANGE_DATE
)
self._last_change_date.read(local_stream)
self.is_oversized(local_stream)
def write(self, output_stream):
"""
Write the data encoding the ObtainLease response payload to a stream.
Args:
output_stream (stream): A data stream in which to encode object
data, supporting a write method; usually a BytearrayStream
object.
Raises:
ValueError: Raised if the data attribute is not defined.
"""
local_stream = utils.BytearrayStream()
if self._unique_identifier:
self._unique_identifier.write(local_stream)
if self._lease_time:
self._lease_time.write(local_stream)
if self._last_change_date:
self._last_change_date.write(local_stream)
self.length = local_stream.length()
super(ObtainLeaseResponsePayload, self).write(output_stream)
output_stream.write(local_stream.buffer)
def __eq__(self, other):
if isinstance(other, ObtainLeaseResponsePayload):
if self.unique_identifier != other.unique_identifier:
return False
elif self.lease_time != other.lease_time:
return False
elif self.last_change_date != other.last_change_date:
return False
else:
return True
else:
return NotImplemented
def __ne__(self, other):
if isinstance(other, ObtainLeaseResponsePayload):
return not (self == other)
else:
return NotImplemented
def __repr__(self):
args = ", ".join([
"unique_identifier='{0}'".format(self.unique_identifier),
"lease_time={0}".format(self.lease_time),
"last_change_date={0}".format(self.last_change_date)
])
return "ObtainLeaseResponsePayload({0})".format(args)
def __str__(self):
return str({
'unique_identifier': self.unique_identifier,
'lease_time': self.lease_time,
'last_change_date': self.last_change_date
})

712
kmip/tests/unit/core/messages/payloads/test_obtain_lease.py Normal file
View File

@ -0,0 +1,712 @@
# Copyright (c) 2017 The Johns Hopkins University/Applied Physics Laboratory
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import testtools
from kmip.core import utils
from kmip.core.messages import payloads
class TestObtainLeaseRequestPayload(testtools.TestCase):
"""
Test suite for the ObtainLease request payload.
"""
def setUp(self):
super(TestObtainLeaseRequestPayload, self).setUp()
# Encoding obtained from the KMIP 1.1 testing document, Section 9.5.
#
# This encoding matches the following set of values:
# Request Payload
# Unique Identifier - f4152f17-9312-431a-b3fb-4fe86a86a7a1
self.full_encoding = utils.BytearrayStream(
b'\x42\x00\x79\x01\x00\x00\x00\x30'
b'\x42\x00\x94\x07\x00\x00\x00\x24'
b'\x66\x34\x31\x35\x32\x66\x31\x37\x2D\x39\x33\x31\x32\x2D\x34\x33'
b'\x31\x61\x2D\x62\x33\x66\x62\x2D\x34\x66\x65\x38\x36\x61\x38\x36'
b'\x61\x37\x61\x31\x00\x00\x00\x00'
)
self.empty_encoding = utils.BytearrayStream(
b'\x42\x00\x79\x01\x00\x00\x00\x00'
)
def tearDown(self):
super(TestObtainLeaseRequestPayload, self).tearDown()
def test_init(self):
"""
Test that an ObtainLease request payload can be constructed with no
arguments.
"""
payload = payloads.ObtainLeaseRequestPayload()
self.assertEqual(None, payload.unique_identifier)
def test_init_with_args(self):
"""
Test that an ObtainLease request payload can be constructed with valid
values.
"""
payload = payloads.ObtainLeaseRequestPayload(
unique_identifier='00000000-1111-2222-3333-444444444444'
)
self.assertEqual(
'00000000-1111-2222-3333-444444444444',
payload.unique_identifier
)
def test_invalid_unique_identifier(self):
"""
Test that a TypeError is raised when an invalid value is used to set
the unique identifier of an ObtainLease request payload.
"""
kwargs = {'unique_identifier': 0}
self.assertRaisesRegexp(
TypeError,
"Unique identifier must be a string.",
payloads.ObtainLeaseRequestPayload,
**kwargs
)
payload = payloads.ObtainLeaseRequestPayload()
args = (payload, 'unique_identifier', 0)
self.assertRaisesRegexp(
TypeError,
"Unique identifier must be a string.",
setattr,
*args
)
def test_read(self):
"""
Test that an ObtainLease request payload can be read from a data
stream.
"""
payload = payloads.ObtainLeaseRequestPayload()
self.assertEqual(None, payload.unique_identifier)
payload.read(self.full_encoding)
self.assertEqual(
'f4152f17-9312-431a-b3fb-4fe86a86a7a1',
payload.unique_identifier
)
def test_read_empty(self):
"""
Test that an ObtainLease request payload can be read from an empty
data stream.
"""
payload = payloads.ObtainLeaseRequestPayload()
self.assertEqual(None, payload.unique_identifier)
payload.read(self.empty_encoding)
self.assertEqual(None, payload.unique_identifier)
def test_write(self):
"""
Test that an ObtainLease request payload can be written to a data
stream.
"""
payload = payloads.ObtainLeaseRequestPayload(
unique_identifier='f4152f17-9312-431a-b3fb-4fe86a86a7a1'
)
stream = utils.BytearrayStream()
payload.write(stream)
self.assertEqual(len(self.full_encoding), len(stream))
self.assertEqual(str(self.full_encoding), str(stream))
def test_write_empty(self):
"""
Test that an empty ObtainLease request payload can be written to a
data stream.
"""
payload = payloads.ObtainLeaseRequestPayload()
stream = utils.BytearrayStream()
payload.write(stream)
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
ObtainLease request payloads with the same data.
"""
a = payloads.ObtainLeaseRequestPayload()
b = payloads.ObtainLeaseRequestPayload()
self.assertTrue(a == b)
self.assertTrue(b == a)
a = payloads.ObtainLeaseRequestPayload(
unique_identifier='f4152f17-9312-431a-b3fb-4fe86a86a7a1'
)
b = payloads.ObtainLeaseRequestPayload(
unique_identifier='f4152f17-9312-431a-b3fb-4fe86a86a7a1'
)
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
ObtainLease request payloads with different unique identifiers.
"""
a = payloads.ObtainLeaseRequestPayload(
unique_identifier='a'
)
b = payloads.ObtainLeaseRequestPayload(
unique_identifier='b'
)
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
ObtainLease request payloads with different types.
"""
a = payloads.ObtainLeaseRequestPayload()
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
ObtainLease request payloads with the same data.
"""
a = payloads.ObtainLeaseRequestPayload()
b = payloads.ObtainLeaseRequestPayload()
self.assertFalse(a != b)
self.assertFalse(b != a)
a = payloads.ObtainLeaseRequestPayload(
unique_identifier='f4152f17-9312-431a-b3fb-4fe86a86a7a1'
)
b = payloads.ObtainLeaseRequestPayload(
unique_identifier='f4152f17-9312-431a-b3fb-4fe86a86a7a1'
)
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
ObtainLease request payloads with different unique identifiers.
"""
a = payloads.ObtainLeaseRequestPayload(
unique_identifier='a'
)
b = payloads.ObtainLeaseRequestPayload(
unique_identifier='b'
)
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
ObtainLease request payloads with different types.
"""
a = payloads.ObtainLeaseRequestPayload()
b = 'invalid'
self.assertTrue(a != b)
self.assertTrue(b != a)
def test_repr(self):
"""
Test that repr can be applied to an ObtainLease request payload.
"""
payload = payloads.ObtainLeaseRequestPayload(
unique_identifier='49a1ca88-6bea-4fb2-b450-7e58802c3038'
)
expected = (
"ObtainLeaseRequestPayload("
"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 ObtainLease request payload
"""
payload = payloads.ObtainLeaseRequestPayload(
unique_identifier='49a1ca88-6bea-4fb2-b450-7e58802c3038'
)
expected = str({
'unique_identifier': '49a1ca88-6bea-4fb2-b450-7e58802c3038'
})
observed = str(payload)
self.assertEqual(expected, observed)
class TestObtainLeaseResponsePayload(testtools.TestCase):
"""
Test suite for the ObtainLease response payload.
"""
def setUp(self):
super(TestObtainLeaseResponsePayload, self).setUp()
# Encoding obtained from the KMIP 1.1 testing document, Section 9.5.
#
# This encoding matches the following set of values:
# Response Payload
# Unique Identifier - f4152f17-9312-431a-b3fb-4fe86a86a7a1
# Lease Time - 0
# Last Change Date - 0x4F9A5564 (Fri Apr 27 10:14:28 CEST 2012)
self.full_encoding = utils.BytearrayStream(
b'\x42\x00\x7C\x01\x00\x00\x00\x50'
b'\x42\x00\x94\x07\x00\x00\x00\x24'
b'\x66\x34\x31\x35\x32\x66\x31\x37\x2D\x39\x33\x31\x32\x2D\x34\x33'
b'\x31\x61\x2D\x62\x33\x66\x62\x2D\x34\x66\x65\x38\x36\x61\x38\x36'
b'\x61\x37\x61\x31\x00\x00\x00\x00'
b'\x42\x00\x49\x0A\x00\x00\x00\x04\x00\x00\x00\x00\x00\x00\x00\x00'
b'\x42\x00\x48\x09\x00\x00\x00\x08\x00\x00\x00\x00\x4F\x9A\x55\x64'
)
# This encoding matches the following set of values:
# Response Payload
# Unique Identifier - f4152f17-9312-431a-b3fb-4fe86a86a7a1
self.partial_encoding = utils.BytearrayStream(
b'\x42\x00\x7C\x01\x00\x00\x00\x30'
b'\x42\x00\x94\x07\x00\x00\x00\x24'
b'\x66\x34\x31\x35\x32\x66\x31\x37\x2D\x39\x33\x31\x32\x2D\x34\x33'
b'\x31\x61\x2D\x62\x33\x66\x62\x2D\x34\x66\x65\x38\x36\x61\x38\x36'
b'\x61\x37\x61\x31\x00\x00\x00\x00'
)
self.empty_encoding = utils.BytearrayStream(
b'\x42\x00\x7C\x01\x00\x00\x00\x00'
)
def tearDown(self):
super(TestObtainLeaseResponsePayload, self).tearDown()
def test_init(self):
"""
Test that an ObtainLease response payload can be constructed with no
arguments.
"""
payload = payloads.ObtainLeaseResponsePayload()
self.assertEqual(None, payload.unique_identifier)
self.assertEqual(None, payload.lease_time)
self.assertEqual(None, payload.last_change_date)
def test_init_with_args(self):
"""
Test that an ObtainLease response payload can be constructed with valid
values.
"""
payload = payloads.ObtainLeaseResponsePayload(
unique_identifier='00000000-1111-2222-3333-444444444444',
lease_time=1000000000,
last_change_date=1512400153
)
self.assertEqual(
'00000000-1111-2222-3333-444444444444',
payload.unique_identifier
)
self.assertEqual(1000000000, payload.lease_time)
self.assertEqual(1512400153, payload.last_change_date)
def test_invalid_unique_identifier(self):
"""
Test that a TypeError is raised when an invalid value is used to set
the unique identifier of an ObtainLease response payload.
"""
kwargs = {'unique_identifier': 0}
self.assertRaisesRegexp(
TypeError,
"Unique identifier must be a string.",
payloads.ObtainLeaseResponsePayload,
**kwargs
)
payload = payloads.ObtainLeaseResponsePayload()
args = (payload, 'unique_identifier', 0)
self.assertRaisesRegexp(
TypeError,
"Unique identifier must be a string.",
setattr,
*args
)
def test_invalid_lease_time(self):
"""
Test that a TypeError is raised when an invalid value is used to set
the lease time of an ObtainLease response payload.
"""
kwargs = {'lease_time': 'invalid'}
self.assertRaisesRegexp(
TypeError,
"Lease time must be an integer.",
payloads.ObtainLeaseResponsePayload,
**kwargs
)
payload = payloads.ObtainLeaseResponsePayload()
args = (payload, 'lease_time', 'invalid')
self.assertRaisesRegexp(
TypeError,
"Lease time must be an integer.",
setattr,
*args
)
def test_invalid_last_change_date(self):
"""
Test that a TypeError is raised when an invalid value is used to set
the last change date of an ObtainLease response payload.
"""
kwargs = {'last_change_date': 'invalid'}
self.assertRaisesRegexp(
TypeError,
"Last change date must be an integer.",
payloads.ObtainLeaseResponsePayload,
**kwargs
)
payload = payloads.ObtainLeaseResponsePayload()
args = (payload, 'last_change_date', 'invalid')
self.assertRaisesRegexp(
TypeError,
"Last change date must be an integer.",
setattr,
*args
)
def test_read(self):
"""
Test that an ObtainLease response payload can be read from a data
stream.
"""
payload = payloads.ObtainLeaseResponsePayload()
self.assertEqual(None, payload.unique_identifier)
self.assertEqual(None, payload.lease_time)
self.assertEqual(None, payload.last_change_date)
payload.read(self.full_encoding)
self.assertEqual(
'f4152f17-9312-431a-b3fb-4fe86a86a7a1',
payload.unique_identifier
)
self.assertEqual(0, payload.lease_time)
self.assertEqual(1335514468, payload.last_change_date)
def test_read_partial(self):
"""
Test that an ObtainLease response payload can be read from a partial
data stream.
"""
payload = payloads.ObtainLeaseResponsePayload()
self.assertEqual(None, payload.unique_identifier)
self.assertEqual(None, payload.lease_time)
self.assertEqual(None, payload.last_change_date)
payload.read(self.partial_encoding)
self.assertEqual(
'f4152f17-9312-431a-b3fb-4fe86a86a7a1',
payload.unique_identifier
)
self.assertEqual(None, payload.lease_time)
self.assertEqual(None, payload.last_change_date)
def test_read_empty(self):
"""
Test that an ObtainLease response payload can be read from an empty
data stream.
"""
payload = payloads.ObtainLeaseResponsePayload()
self.assertEqual(None, payload.unique_identifier)
self.assertEqual(None, payload.lease_time)
self.assertEqual(None, payload.last_change_date)
payload.read(self.empty_encoding)
self.assertEqual(None, payload.unique_identifier)
self.assertEqual(None, payload.lease_time)
self.assertEqual(None, payload.last_change_date)
def test_write(self):
"""
Test that an ObtainLease response payload can be written to a data
stream.
"""
payload = payloads.ObtainLeaseResponsePayload(
unique_identifier='f4152f17-9312-431a-b3fb-4fe86a86a7a1',
lease_time=0,
last_change_date=1335514468
)
stream = utils.BytearrayStream()
payload.write(stream)
self.assertEqual(len(self.full_encoding), len(stream))
self.assertEqual(str(self.full_encoding), str(stream))
def test_write_partial(self):
"""
Test that a partial ObtainLease response payload can be written to a
data stream.
"""
payload = payloads.ObtainLeaseResponsePayload(
unique_identifier='f4152f17-9312-431a-b3fb-4fe86a86a7a1'
)
stream = utils.BytearrayStream()
payload.write(stream)
self.assertEqual(len(self.partial_encoding), len(stream))
self.assertEqual(str(self.partial_encoding), str(stream))
def test_write_empty(self):
"""
Test that an empty ObtainLease response payload can be written to a
data stream.
"""
payload = payloads.ObtainLeaseResponsePayload()
stream = utils.BytearrayStream()
payload.write(stream)
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
ObtainLease response payloads with the same data.
"""
a = payloads.ObtainLeaseResponsePayload()
b = payloads.ObtainLeaseResponsePayload()
self.assertTrue(a == b)
self.assertTrue(b == a)
a = payloads.ObtainLeaseResponsePayload(
unique_identifier='49a1ca88-6bea-4fb2-b450-7e58802c3038',
lease_time=1511882848,
last_change_date=1512410153
)
b = payloads.ObtainLeaseResponsePayload(
unique_identifier='49a1ca88-6bea-4fb2-b450-7e58802c3038',
lease_time=1511882848,
last_change_date=1512410153
)
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
ObtainLease response payloads with different unique identifiers.
"""
a = payloads.ObtainLeaseResponsePayload(
unique_identifier='a'
)
b = payloads.ObtainLeaseResponsePayload(
unique_identifier='b'
)
self.assertFalse(a == b)
self.assertFalse(b == a)
def test_equal_on_not_equal_lease_time(self):
"""
Test that the equality operator returns False when comparing two
ObtainLease response payloads with different lease times.
"""
a = payloads.ObtainLeaseResponsePayload(
lease_time=0
)
b = payloads.ObtainLeaseResponsePayload(
lease_time=1511882848
)
self.assertFalse(a == b)
self.assertFalse(b == a)
def test_equal_on_not_equal_last_change_date(self):
"""
Test that the equality operator returns False when comparing two
ObtainLease response payloads with different last change dates.
"""
a = payloads.ObtainLeaseResponsePayload(
last_change_date=0
)
b = payloads.ObtainLeaseResponsePayload(
last_change_date=1511882848
)
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
ObtainLease response payloads with different types.
"""
a = payloads.ObtainLeaseResponsePayload()
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
ObtainLease response payloads with the same data.
"""
a = payloads.ObtainLeaseResponsePayload()
b = payloads.ObtainLeaseResponsePayload()
self.assertFalse(a != b)
self.assertFalse(b != a)
a = payloads.ObtainLeaseResponsePayload(
unique_identifier='49a1ca88-6bea-4fb2-b450-7e58802c3038',
lease_time=1511882848,
last_change_date=0
)
b = payloads.ObtainLeaseResponsePayload(
unique_identifier='49a1ca88-6bea-4fb2-b450-7e58802c3038',
lease_time=1511882848,
last_change_date=0
)
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
ObtainLease response payloads with different unique identifiers.
"""
a = payloads.ObtainLeaseResponsePayload(
unique_identifier='a'
)
b = payloads.ObtainLeaseResponsePayload(
unique_identifier='b'
)
self.assertTrue(a != b)
self.assertTrue(b != a)
def test_not_equal_on_not_equal_lease_time(self):
"""
Test that the inequality operator returns True when comparing two
ObtainLease response payloads with different lease times.
"""
a = payloads.ObtainLeaseResponsePayload(
lease_time=0
)
b = payloads.ObtainLeaseResponsePayload(
lease_time=1511882848
)
self.assertTrue(a != b)
self.assertTrue(b != a)
def test_not_equal_on_not_equal_last_change_time(self):
"""
Test that the inequality operator returns True when comparing two
ObtainLease response payloads with different last change time.
"""
a = payloads.ObtainLeaseResponsePayload(
lease_time=0
)
b = payloads.ObtainLeaseResponsePayload(
lease_time=1511882848
)
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
ObtainLease response payloads with different types.
"""
a = payloads.ObtainLeaseResponsePayload()
b = 'invalid'
self.assertTrue(a != b)
self.assertTrue(b != a)
def test_repr(self):
"""
Test that repr can be applied to an ObtainLease response payload.
"""
payload = payloads.ObtainLeaseResponsePayload(
unique_identifier='49a1ca88-6bea-4fb2-b450-7e58802c3038',
lease_time=1511882898,
last_change_date=1512410153
)
expected = (
"ObtainLeaseResponsePayload("
"unique_identifier='49a1ca88-6bea-4fb2-b450-7e58802c3038', "
"lease_time=1511882898, "
"last_change_date=1512410153)"
)
observed = repr(payload)
self.assertEqual(expected, observed)
def test_str(self):
"""
Test that str can be applied to an ObtainLease response payload.
"""
payload = payloads.ObtainLeaseResponsePayload(
unique_identifier='49a1ca88-6bea-4fb2-b450-7e58802c3038',
lease_time=1511882898,
last_change_date=1512410153
)
expected = str({
'unique_identifier': '49a1ca88-6bea-4fb2-b450-7e58802c3038',
'lease_time': 1511882898,
'last_change_date': 1512410153
})
observed = str(payload)
self.assertEqual(expected, observed)