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_metadata(self):
"""Metadata is read correctly from GribMessage."""
with GribFile(TESTGRIB) as grib_file:
msg = GribMessage(grib_file)
msg_keys = list(msg.keys())
for key in KNOWN_GRIB_KEYS:
assert key in msg_keys, "key '%s' not found" % key
# Size of message in bytes
self.assertEqual(msg.size(), 160219)
self.assertEqual(len(list(msg.keys())), len(msg))
def test_clone(self):
"""Messages can be used to produce clone Messages."""
with GribFile(TESTGRIB) as grib_file:
msg = GribMessage(grib_file)
msg2 = GribMessage(clone=msg)
self.assertSequenceEqual(list(msg.keys()), list(msg2.keys()))
def test_multi_value_setting(self):
"""Multiple keys/values can be set properly."""
msg = GribMessage(sample='GRIB1')
msg['paramId', 'stepType', 'edition'] = 49, 'avg', 2
self.assertEqual(msg['shortName'], '10fg')
# Another test
with GribFile(TESTGRIB) as grib_file:
msg = GribMessage(grib_file)
msg['setLocalDefinition', 'localDefinitionNumber'] = 1, 25
msg['typeOfFirstFixedSurface', 'typeOfSecondFixedSurface'] = 1, 8
msg[('typeOfFirstFixedSurface', 'typeOfSecondFixedSurface')] = (1, 8) # Also works
self.assertEqual(msg['localDefinitionNumber'], 25)
self.assertEqual(msg['typeOfLevel'], 'entireAtmosphere')
def test_message_counting_works(self):
"""The GribFile is aware of its messages."""
with GribFile(TESTGRIB) as grib_file:
msg_count = len(grib_file)
self.assertEqual(msg_count, 5)
def test_clone(self):
"""Messages can be used to produce clone Messages."""
with GribFile(TESTGRIB) as grib_file:
msg = GribMessage(grib_file)
msg2 = GribMessage(clone=msg)
self.assertSequenceEqual(msg.keys(), msg2.keys())
def test_memory_management(self):
"""Messages in GribFile can be opened and closed properly."""
with GribFile(TESTGRIB) as grib_file:
self.assertEqual(len(grib_file), 5)
for i in range(len(grib_file)):
msg = GribMessage(grib_file)
self.assertEqual(msg["shortName"], "msl")
self.assertEqual(msg['count'], i + 1)
self.assertEqual(len(grib_file.open_messages), 5)
self.assertEqual(len(grib_file.open_messages), 0)
def test_memory_management(self):
"""Messages in GribFile can be opened and closed properly."""
with GribFile(TESTGRIB) as grib_file:
self.assertEqual(len(grib_file), 5)
for i in range(len(grib_file)):
msg = GribMessage(grib_file)
self.assertEqual(msg["shortName"], "msl")
self.assertEqual(msg['count'], i+1)
self.assertEqual(len(grib_file.open_messages), 5)
self.assertEqual(len(grib_file.open_messages), 0)
def test_iterator_protocol(self):
"""The GribFile allows pythonic iteration over all messages."""
step_ranges = []
with GribFile(TESTGRIB) as grib_file:
for msg in grib_file:
step_ranges.append(msg["stepRange"])
self.assertSequenceEqual(step_ranges, ["0", "6", "12", "18", "24"])
def test_read_past_last_message(self):
"""Trying to open message on exhausted GRIB file raises IOError."""
with GribFile(TESTGRIB) as grib_file:
for _ in range(len(grib_file)):
GribMessage(grib_file)
self.assertRaises(IOError, lambda: GribMessage(grib_file))
def test_iterator_protocol(self):
"""The GribFile allows pythonic iteration over all messages."""
step_ranges = []
with GribFile(TESTGRIB) as grib_file:
for msg in grib_file:
step_ranges.append(msg["stepRange"])
self.assertSequenceEqual(step_ranges, ["0", "6", "12", "18", "24"])