Skip to content

Commit

Permalink
Replaced use of 2.7-and-later 'total_seconds' from timedelta.
Browse files Browse the repository at this point in the history
  • Loading branch information
@da4089
da4089 committed Dec 15, 2017
1 parent 2f7a126 commit fa1b6d1
Show file tree
Hide file tree
Showing 2 changed files with 78 additions and 50 deletions.
6 changes: 4 additions & 2 deletions simplefix/message.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -315,7 +315,8 @@ def append_tz_timestamp(self, tag, timestamp=None, precision=3,
# Get offset of local timezone east of UTC.
utc = datetime.datetime.utcfromtimestamp(now)
local = datetime.datetime.fromtimestamp(now)
offset = int((local - utc).total_seconds() / 60)
td = local - utc
offset = int(((td.days * 86400) + td.seconds) / 60)

s = local.strftime("%Y%m%d-%H:%M:%S")
if precision == 3:
Expand Down Expand Up @@ -358,7 +359,8 @@ def append_tz_time_only(self, tag, timestamp=None, precision=3,

now = time.mktime(t.timetuple()) + (t.microsecond * 1e-6)
utc = datetime.datetime.utcfromtimestamp(now)
offset = int((t - utc).total_seconds() / 60)
td = t - utc
offset = int(((td.days * 86400) + td.seconds) / 60)

s = t.strftime("%H:%M")
if precision == 0:
Expand Down
122 changes: 74 additions & 48 deletions test/test_message.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,16 @@
from simplefix import FixMessage


# NOTE: RHEL6 ships with Python 2.6, which is increasingly difficult to
# support as both 2.7 and 3.x have added numerous features that seem like
# they've been around forever, but actually aren't in 2.6.
#
# While in a few cases, I try to monkey-patch stuff up to 2.7-equivalent,
# in a lot of the test code, I just skip the tests if they're run on
# a 2.6 interpreter -- or I would but unittest in 2.6 doesn't understand
# skipping, so they just pass without testing anything.
#
# This constant is used to check the version and alter behaviour to suit.
VERSION = (sys.version_info[0] * 10) + sys.version_info[1]


Expand Down Expand Up @@ -58,7 +68,8 @@ def test_string_with_equals(self):

def test_string_without_equals(self):
"""Test field set with string not containing equals sign"""
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
with self.assertRaises(ValueError):
Expand All @@ -67,7 +78,8 @@ def test_string_without_equals(self):

def test_string_with_bad_tag(self):
"""Test field set with bad tag in tag=value string"""
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
with self.assertRaises(ValueError):
Expand Down Expand Up @@ -101,7 +113,8 @@ def test_set_session_version(self):

def test_get_repeating(self):
"""Test retrieval of repeating field's value"""
if VERSION == 26: return
if VERSION == 26:
return

pkt = FixMessage()
pkt.append_pair(42, "a")
Expand Down Expand Up @@ -138,7 +151,8 @@ def test_raw_msg_type(self):

def test_empty_message(self):
"""Test encoding of empty message"""
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
with self.assertRaises(ValueError):
Expand All @@ -147,7 +161,8 @@ def test_empty_message(self):

def test_encode_no_35(self):
"""Test encoding without MessageType(35) field"""
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
msg.append_pair(8, "FIX.4.2")
Expand All @@ -157,7 +172,8 @@ def test_encode_no_35(self):

def test_encode_no_8(self):
"""Test encoding without BeginString(8) field"""
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
msg.append_pair(35, "D")
Expand Down Expand Up @@ -318,7 +334,8 @@ def test_time_seconds_only(self):

def test_time_bad_precision(self):
"""Test bad time precision values"""
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
t = 1484581872.933458
Expand Down Expand Up @@ -392,7 +409,8 @@ def test_utcts_seconds_only(self):

def test_utcts_bad_precision(self):
"""Test UTCTimestamp bad time precision values"""
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
t = 1484581872.933458
Expand Down Expand Up @@ -448,7 +466,8 @@ def test_utcto_seconds_only(self):

def test_utcto_bad_precision(self):
"""Test UTCTimeOnly bad time precision values"""
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
t = 1484581872.933458
Expand All @@ -475,7 +494,8 @@ def test_utcto_parts_15_51_12_933_458(self):
return

def test_utcto_parts_bad_hour(self):
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
with self.assertRaises(ValueError):
Expand All @@ -487,7 +507,8 @@ def test_utcto_parts_bad_hour(self):
return

def test_utcto_parts_bad_minute(self):
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
with self.assertRaises(ValueError):
Expand All @@ -499,7 +520,8 @@ def test_utcto_parts_bad_minute(self):
return

def test_utcto_parts_bad_seconds(self):
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
with self.assertRaises(ValueError):
Expand All @@ -511,7 +533,8 @@ def test_utcto_parts_bad_seconds(self):
return

def test_utcto_parts_bad_ms(self):
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
with self.assertRaises(ValueError):
Expand All @@ -523,7 +546,8 @@ def test_utcto_parts_bad_ms(self):
return

def test_utcto_parts_bad_us(self):
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
with self.assertRaises(ValueError):
Expand All @@ -535,7 +559,8 @@ def test_utcto_parts_bad_us(self):
return

def test_offset_range(self):
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
with self.assertRaises(ValueError):
Expand Down Expand Up @@ -582,6 +607,14 @@ def test_append_tzts_none(self):
msg.append_tz_timestamp(1253, None)
return

@staticmethod
def calculate_tz_offset(t):
local = datetime.datetime.fromtimestamp(t)
utc = datetime.datetime.utcfromtimestamp(t)
td = local - utc
offset = int(((td.days * 86400) + td.seconds) / 60)
return offset

def test_append_tzts_float(self):
msg = FixMessage()
t = 1484581872.933458
Expand All @@ -592,9 +625,7 @@ def test_append_tzts_float(self):
(test.tm_year, test.tm_mon, test.tm_mday,
test.tm_hour, test.tm_min, test.tm_sec,
int((t - int(t)) * 1000))
offset = int((datetime.datetime.fromtimestamp(t) -
datetime.datetime.utcfromtimestamp(t)).total_seconds()
/ 60)
offset = self.calculate_tz_offset(t)
if offset == 0:
s += "Z"
else:
Expand All @@ -619,9 +650,7 @@ def test_append_tzts_datetime(self):
(test.tm_year, test.tm_mon, test.tm_mday,
test.tm_hour, test.tm_min, test.tm_sec,
int((t - int(t)) * 1000))
offset = int((datetime.datetime.fromtimestamp(t) -
datetime.datetime.utcfromtimestamp(t)).total_seconds()
/ 60)
offset = self.calculate_tz_offset(t)
if offset == 0:
s += "Z"
else:
Expand All @@ -646,9 +675,7 @@ def test_tzts_microseconds(self):
(test.tm_year, test.tm_mon, test.tm_mday,
test.tm_hour, test.tm_min, test.tm_sec,
int((t % 1) * 1e6))
offset = int((datetime.datetime.fromtimestamp(t) -
datetime.datetime.utcfromtimestamp(t)).total_seconds()
/ 60)
offset = self.calculate_tz_offset(t)
if offset == 0:
s += "Z"
else:
Expand All @@ -672,9 +699,7 @@ def test_tzts_seconds_only(self):
s = "%04u%02u%02u-%02u:%02u:%02u" % \
(test.tm_year, test.tm_mon, test.tm_mday,
test.tm_hour, test.tm_min, test.tm_sec)
offset = int((datetime.datetime.fromtimestamp(t) -
datetime.datetime.utcfromtimestamp(t)).total_seconds()
/ 60)
offset = self.calculate_tz_offset(t)
if offset == 0:
s += "Z"
else:
Expand All @@ -690,7 +715,8 @@ def test_tzts_seconds_only(self):

def test_tzts_bad_precision(self):
"""Test bad TZTimestamp precision value"""
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
t = 1484581872.933458
Expand All @@ -706,9 +732,7 @@ def test_tzto_datetime(self):
test = time.localtime(t)
s = "%02u:%02u:%02u.%03u" % \
(test.tm_hour, test.tm_min, test.tm_sec, int((t % 1) * 1e3))
offset = int((datetime.datetime.fromtimestamp(t) -
datetime.datetime.utcfromtimestamp(t)).total_seconds()
/ 60)
offset = self.calculate_tz_offset(t)
if offset == 0:
s += "Z"
else:
Expand All @@ -730,9 +754,7 @@ def test_tzto_minutes(self):

test = time.localtime(t)
s = "%02u:%02u" % (test.tm_hour, test.tm_min)
offset = int((datetime.datetime.fromtimestamp(t) -
datetime.datetime.utcfromtimestamp(t)).total_seconds()
/ 60)
offset = self.calculate_tz_offset(t)
if offset == 0:
s += "Z"
else:
Expand All @@ -755,9 +777,7 @@ def test_tzto_microseconds(self):
test = time.localtime(t)
s = "%02u:%02u:%02u.%06u" % \
(test.tm_hour, test.tm_min, test.tm_sec, int((t % 1) * 1e6))
offset = int((datetime.datetime.fromtimestamp(t) -
datetime.datetime.utcfromtimestamp(t)).total_seconds()
/ 60)
offset = self.calculate_tz_offset(t)
if offset == 0:
s += "Z"
else:
Expand All @@ -780,9 +800,7 @@ def test_tzto_seconds_only(self):
test = time.localtime(t)
s = "%02u:%02u:%02u" % \
(test.tm_hour, test.tm_min, test.tm_sec)
offset = int((datetime.datetime.fromtimestamp(t) -
datetime.datetime.utcfromtimestamp(t)).total_seconds()
/ 60)
offset = self.calculate_tz_offset(t)
if offset == 0:
s += "Z"
else:
Expand All @@ -798,7 +816,8 @@ def test_tzto_seconds_only(self):

def test_tzto_bad_precision(self):
"""Test bad TZTimeOnly precision value"""
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
t = 1484581872.933458
Expand Down Expand Up @@ -855,7 +874,8 @@ def test_tzto_parts_15_51_12_933_458_150(self):

def test_tzto_parts_bad_hour(self):
"""Test TZTimeOnly with out-of-range hour components"""
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
with self.assertRaises(ValueError):
Expand All @@ -868,7 +888,8 @@ def test_tzto_parts_bad_hour(self):

def test_tzto_parts_bad_minute(self):
"""Test TZTimeOnly with out-of-range minute components"""
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
with self.assertRaises(ValueError):
Expand All @@ -881,7 +902,8 @@ def test_tzto_parts_bad_minute(self):

def test_tzto_parts_bad_seconds(self):
"""Test TZTimeOnly with out-of-range seconds components"""
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
with self.assertRaises(ValueError):
Expand All @@ -894,7 +916,8 @@ def test_tzto_parts_bad_seconds(self):

def test_tzto_parts_bad_ms(self):
"""Test TZTimeOnly with out-of-range milliseconds components"""
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
with self.assertRaises(ValueError):
Expand All @@ -907,7 +930,8 @@ def test_tzto_parts_bad_ms(self):

def test_tzto_parts_bad_us(self):
"""Test TZTimeOnly with out-of-range microseconds components"""
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
with self.assertRaises(ValueError):
Expand Down Expand Up @@ -943,7 +967,8 @@ def test_strings(self):

def test_contains(self):
"""Test use of 'in' and 'not in' operators"""
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
msg.append_strings(["8=FIX.4.4", "35=0"])
Expand All @@ -955,7 +980,8 @@ def test_contains(self):

def test_none_value(self):
"""Test encoding of None value"""
if VERSION == 26: return
if VERSION == 26:
return

msg = FixMessage()
msg.append_pair(99999, None)
Expand Down

0 comments on commit fa1b6d1

Please sign in to comment.