blob: 885c24938b2b17a2791789055deb783ab05f2c7e [file] [log] [blame]
# -*- coding: utf-8 -*-
# Protocol Buffers - Google's data interchange format
# Copyright 2008 Google Inc. All rights reserved.
#
# Use of this source code is governed by a BSD-style
# license that can be found in the LICENSE file or at
# https://developers.google.com/open-source/licenses/bsd
"""Tests proto Duration APIs."""
import datetime
import unittest
from google.protobuf import duration
from google.protobuf.internal import testing_refleaks
from google.protobuf.internal import well_known_types_test_pb2
from google.protobuf import duration_pb2
@testing_refleaks.TestCase
class DurationTest(unittest.TestCase):
def test_duration_integer_conversion(self):
self.assertEqual(1, duration.to_nanoseconds(duration.from_nanoseconds(1)))
self.assertEqual(-1, duration.to_seconds(duration.from_seconds(-1)))
self.assertEqual(
123, duration.to_milliseconds(duration.from_milliseconds(123))
)
self.assertEqual(
321, duration.to_microseconds(duration.from_microseconds(321))
)
def test_duration_json(self):
def check_duration(message, text):
self.assertEqual(text, duration.to_json_string(message))
parsed_duration = duration.from_json_string(text)
self.assertEqual(message, parsed_duration)
message = duration_pb2.Duration()
message.seconds = 0
message.nanos = 0
check_duration(message, '0s')
message.nanos = 10000000
check_duration(message, '0.010s')
message.nanos = 10000
check_duration(message, '0.000010s')
message.nanos = 10
check_duration(message, '0.000000010s')
def test_duration_timedelta(self):
message = duration.from_nanoseconds(1999999999)
td = duration.to_timedelta(message)
self.assertEqual(1, td.seconds)
self.assertEqual(999999, td.microseconds)
message = duration.from_microseconds(-1)
td = duration.to_timedelta(message)
converted_message = duration.from_timedelta(td)
self.assertEqual(message, converted_message)
def test_duration_construction(self):
expected_td = datetime.timedelta(microseconds=123)
message = well_known_types_test_pb2.WKTMessage(
optional_duration=expected_td
)
self.assertEqual(expected_td, message.optional_duration.ToTimedelta())
def test_repeated_duration_construction(self):
td0 = datetime.timedelta(microseconds=123)
td1 = datetime.timedelta(microseconds=456)
dr = duration_pb2.Duration()
message = well_known_types_test_pb2.WKTMessage(repeated_td=[td0, td1, dr])
self.assertEqual(td0, duration.to_timedelta(message.repeated_td[0]))
self.assertEqual(td1, duration.to_timedelta(message.repeated_td[1]))
self.assertEqual(dr, message.repeated_td[2])
def test_duration_sub_annotation(self):
dt = datetime.datetime.now()
dr = duration_pb2.Duration()
td = datetime.timedelta(microseconds=123)
# datetime - Duration
self.assertEqual(dt - dr, dt - duration.to_timedelta(dr))
# timedelta - Duration and Duration - Duration
self.assertEqual(td - dr, duration.from_timedelta(td) - dr)
# Duration - timedelta
self.assertEqual(dr - td, dr - duration.from_timedelta(td))
def test_duration_add_annotation(self):
dt = datetime.datetime.now()
dr = duration_pb2.Duration()
dr2 = duration_pb2.Duration(seconds=100)
# datetime + Duration and Duration + datetime
self.assertEqual(dt + dr, dr + dt)
message = well_known_types_test_pb2.WKTMessage(optional_timestamp=dt)
# Duration + Timestamp
self.assertEqual(dr + message.optional_timestamp, dr + dt)
td = datetime.timedelta(microseconds=123)
# Duration + timedelta and timedelta + Duration
self.assertEqual(dr + td, td + dr)
# Duration + Duration
self.assertEqual(dr + dr2, dr2 + dr)
def test_assign_datetime_to_duration(self):
message = well_known_types_test_pb2.WKTMessage()
with self.assertRaises((TypeError)):
message.optional_duration = datetime.datetime.now()
if __name__ == '__main__':
unittest.main()