Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_memory_management(self):
"""Messages in BufrFile can be opened and closed properly."""
with BufrFile(TESTBUFR) as bufr_file:
self.assertEqual(len(bufr_file), 3)
for i in range(len(bufr_file)):
msg = BufrMessage(bufr_file)
self.assertEqual(msg["bufrHeaderCentre"], 98)
self.assertEqual(msg['count'], i + 1)
self.assertEqual(len(bufr_file.open_messages), 3)
self.assertEqual(len(bufr_file.open_messages), 0)
def test_metadata(self):
"""Metadata is read correctly from BufrMessage."""
with BufrFile(TESTBUFR) as bufr_file:
msg = BufrMessage(bufr_file)
msg.unpack()
msg_keys = msg.keys()
self.assertEqual(len(msg_keys), 140)
for key in KNOWN_BUFR_KEYS:
assert key in msg_keys
# Size of message in bytes
self.assertEqual(msg.size(), 220)
self.assertEqual(len(msg.keys()), len(msg))
def test_clone(self):
"""Messages can be used to produce clone Messages."""
with BufrFile(TESTBUFR) as bufr_file:
msg = BufrMessage(bufr_file)
msg2 = BufrMessage(clone=msg)
self.assertSequenceEqual(msg.keys(), msg2.keys())
def test_read_invalid_file(self):
"""Trying to open message on nonexistent file raises IOError."""
with NamedTemporaryFile(mode='r') as f:
with BufrFile(f.name) as bufr_file:
self.assertRaises(IOError, lambda: BufrMessage(bufr_file))
def test_value_setting(self):
"""Keys can be set properly."""
with BufrFile(TESTBUFR) as bufr_file:
msg = BufrMessage(bufr_file)
key, val = "localLongitude", 5
msg[key] = val
self.assertEqual(msg[key], val)
def test_iterator_protocol(self):
"""The BufrFile allows pythonic iteration over all messages."""
latitudes = []
with BufrFile(TESTBUFR) as bufr_file:
for msg in bufr_file:
latitudes.append(msg["localLatitude"])
self.assertSequenceEqual(latitudes, [70.93, 77, 78.92])
def test_read_past_last_message(self):
"""Trying to open message on exhausted BUFR file raises IOError."""
with BufrFile(TESTBUFR) as bufr_file:
for _ in range(len(bufr_file)):
BufrMessage(bufr_file)
self.assertRaises(IOError, lambda: BufrMessage(bufr_file))
def test_read_past_last_message(self):
"""Trying to open message on exhausted BUFR file raises IOError."""
with BufrFile(TESTBUFR) as bufr_file:
for _ in range(len(bufr_file)):
BufrMessage(bufr_file)
self.assertRaises(IOError, lambda: BufrMessage(bufr_file))
def test_value_setting(self):
"""Keys can be set properly."""
with BufrFile(TESTBUFR) as bufr_file:
msg = BufrMessage(bufr_file)
key, val = "localLongitude", 5
msg[key] = val
self.assertEqual(msg[key], val)
def test_serialize(self):
"""Message can be serialized to file."""
with BufrFile(TESTBUFR) as bufr_file:
msg = BufrMessage(bufr_file)
with open(TEST_OUTPUT, "w") as test:
msg.write(test)
os.unlink(TEST_OUTPUT)