1
# Copyright (C) 2015 Yahoo! Inc. All Rights Reserved.
3
# Licensed under the Apache License, Version 2.0 (the "License"); you may
4
# not use this file except in compliance with the License. You may obtain
5
# a copy of the License at
7
# http://www.apache.org/licenses/LICENSE-2.0
9
# Unless required by applicable law or agreed to in writing, software
10
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12
# License for the specific language governing permissions and limitations
21
from oslotest import base as test_base
22
from pytz import timezone
24
import six.moves.xmlrpc_client as xmlrpclib
27
from oslo_serialization import msgpackutils
29
# NOTE(harlowja): itertools.count only started to take a step value
30
# in python 2.7+ so we can't use it in 2.6...
31
if sys.version_info[0:2] == (2, 6):
37
_TZ_FMT = '%Y-%m-%d %H:%M:%S %Z%z'
40
def _dumps_loads(obj):
41
obj = msgpackutils.dumps(obj)
42
return msgpackutils.loads(obj)
45
class MsgPackUtilsTestMixin(test_base.BaseTestCase):
47
self.assertEqual(_dumps_loads([1, 2, 3]), [1, 2, 3])
49
def test_empty_list(self):
50
self.assertEqual(_dumps_loads([]), [])
53
# Seems like we do lose whether it was a tuple or not...
55
# Maybe fixed someday:
57
# https://github.com/msgpack/msgpack-python/issues/98
58
self.assertEqual(_dumps_loads((1, 2, 3)), [1, 2, 3])
61
self.assertEqual(_dumps_loads(dict(a=1, b=2, c=3)),
64
def test_empty_dict(self):
65
self.assertEqual(_dumps_loads({}), {})
67
def test_complex_dict(self):
69
'now': datetime.datetime(1920, 2, 3, 4, 5, 6, 7),
70
'later': datetime.datetime(1921, 2, 3, 4, 5, 6, 9),
78
'today': datetime.date.today(),
80
self.assertEqual(_dumps_loads(src), src)
82
def test_itercount(self):
83
it = itertools.count(1)
86
it2 = _dumps_loads(it)
87
self.assertEqual(six.next(it), six.next(it2))
89
it = itertools.count(0)
90
it2 = _dumps_loads(it)
91
self.assertEqual(six.next(it), six.next(it2))
93
@testtools.skipIf(_PY26, 'itertools.count step not supported')
94
def test_itercount_step(self):
95
it = itertools.count(1, 3)
96
it2 = _dumps_loads(it)
97
self.assertEqual(six.next(it), six.next(it2))
100
self.assertEqual(_dumps_loads(set([1, 2])), set([1, 2]))
102
def test_empty_set(self):
103
self.assertEqual(_dumps_loads(set([])), set([]))
105
def test_frozenset(self):
106
self.assertEqual(_dumps_loads(frozenset([1, 2])), frozenset([1, 2]))
108
def test_empty_frozenset(self):
109
self.assertEqual(_dumps_loads(frozenset([])), frozenset([]))
111
def test_datetime_preserve(self):
112
x = datetime.datetime(1920, 2, 3, 4, 5, 6, 7)
113
self.assertEqual(_dumps_loads(x), x)
115
def test_datetime(self):
116
x = xmlrpclib.DateTime()
117
x.decode("19710203T04:05:06")
118
self.assertEqual(_dumps_loads(x), x)
120
def test_ipaddr(self):
121
thing = {'ip_addr': netaddr.IPAddress('1.2.3.4')}
122
self.assertEqual(_dumps_loads(thing), thing)
124
def test_today(self):
125
today = datetime.date.today()
126
self.assertEqual(today, _dumps_loads(today))
128
def test_datetime_tz_clone(self):
129
eastern = timezone('US/Eastern')
130
now = datetime.datetime.now()
131
e_dt = eastern.localize(now)
132
e_dt2 = _dumps_loads(e_dt)
133
self.assertEqual(e_dt, e_dt2)
134
self.assertEqual(e_dt.strftime(_TZ_FMT), e_dt2.strftime(_TZ_FMT))
136
def test_datetime_tz_different(self):
137
eastern = timezone('US/Eastern')
138
pacific = timezone('US/Pacific')
139
now = datetime.datetime.now()
141
e_dt = eastern.localize(now)
142
p_dt = pacific.localize(now)
144
self.assertNotEqual(e_dt, p_dt)
145
self.assertNotEqual(e_dt.strftime(_TZ_FMT), p_dt.strftime(_TZ_FMT))
147
e_dt2 = _dumps_loads(e_dt)
148
p_dt2 = _dumps_loads(p_dt)
150
self.assertNotEqual(e_dt2, p_dt2)
151
self.assertNotEqual(e_dt2.strftime(_TZ_FMT), p_dt2.strftime(_TZ_FMT))
153
self.assertEqual(e_dt, e_dt2)
154
self.assertEqual(p_dt, p_dt2)