tyler.durden/PyKMIP
1
0
Fork 0
mirror of https://github.com/OpenKMIP/PyKMIP.git synced 2025-04-08 19:25:06 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #372 from OpenKMIP/feat/add-obtain-lease-payloads

Browse Source 
Add payloads for the ObtainLease operation
This commit is contained in:
Peter Hamilton 2017-12-12 14:44:22 -05:00 committed by GitHub
commit 21fbb6ab3f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
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)