Skip to content

Commit

Permalink
black format tests/
Browse files Browse the repository at this point in the history
  • Loading branch information
Teque5 committed Apr 12, 2024
1 parent 446a67b commit bdbfa07
Show file tree
Hide file tree
Showing 4 changed files with 114 additions and 101 deletions.
2 changes: 1 addition & 1 deletion tests/test_archive.py
Expand Up @@ -83,7 +83,7 @@ def test_unwritable_fileobj_throws_fileerror(test_sigmffile):
def test_unwritable_name_throws_fileerror(test_sigmffile):
# Cannot assume /root/ is unwritable (e.g. Docker environment)
# so use invalid filename
unwritable_file = '/bad_name/'
unwritable_file = "/bad_name/"
with pytest.raises(error.SigMFFileError):
test_sigmffile.archive(name=unwritable_file)

Expand Down
113 changes: 59 additions & 54 deletions tests/test_sigmffile.py
Expand Up @@ -51,8 +51,8 @@ def test_iterator_basic(self):
count += 1
self.assertEqual(count, len(self.sigmf_object))

class TestAnnotationHandling(unittest.TestCase):

class TestAnnotationHandling(unittest.TestCase):
def test_get_annotations_with_index(self):
"""Test that only annotations containing index are returned from get_annotations()"""
smf = SigMFFile(copy.deepcopy(TEST_METADATA))
Expand Down Expand Up @@ -87,7 +87,7 @@ def test_set_data_file_without_annotations(self):
TEST_FLOAT32_DATA.tofile(temp_path_data)
smf.set_data_file(temp_path_data)
samples = smf.read_samples()
self.assertTrue(len(samples)==16)
self.assertTrue(len(samples) == 16)

def test_set_data_file_with_annotations(self):
"""
Expand All @@ -104,7 +104,8 @@ def test_set_data_file_with_annotations(self):
# Issues warning since file ends before the final annotatio
smf.set_data_file(temp_path_data)
samples = smf.read_samples()
self.assertTrue(len(samples)==16)
self.assertTrue(len(samples) == 16)


def simulate_capture(sigmf_md, n, capture_len):
start_index = capture_len * n
Expand All @@ -118,9 +119,7 @@ def simulate_capture(sigmf_md, n, capture_len):
"core:longitude": -105.0 + 0.0001 * n,
}

sigmf_md.add_annotation(start_index=start_index,
length=capture_len,
metadata=annotation_md)
sigmf_md.add_annotation(start_index=start_index, length=capture_len, metadata=annotation_md)


def test_default_constructor():
Expand All @@ -129,7 +128,7 @@ def test_default_constructor():

def test_set_non_required_global_field():
sigf = SigMFFile()
sigf.set_global_field('this_is:not_in_the_schema', None)
sigf.set_global_field("this_is:not_in_the_schema", None)


def test_add_capture():
Expand Down Expand Up @@ -226,89 +225,95 @@ def test_multichannel_seek(self):


def test_key_validity():
'''assure the keys in test metadata are valid'''
"""assure the keys in test metadata are valid"""
for top_key, top_val in TEST_METADATA.items():
if type(top_val) is dict:
for core_key in top_val.keys():
assert core_key in vars(SigMFFile)[f'VALID_{top_key.upper()}_KEYS']
assert core_key in vars(SigMFFile)[f"VALID_{top_key.upper()}_KEYS"]
elif type(top_val) is list:
# annotations are in a list
for annot in top_val:
for core_key in annot.keys():
assert core_key in SigMFFile.VALID_ANNOTATION_KEYS
else:
raise ValueError('expected list or dict')
raise ValueError("expected list or dict")


def test_ordered_metadata():
'''check to make sure the metadata is sorted as expected'''
"""check to make sure the metadata is sorted as expected"""
sigf = SigMFFile()
top_sort_order = ['global', 'captures', 'annotations']
top_sort_order = ["global", "captures", "annotations"]
for kdx, key in enumerate(sigf.ordered_metadata()):
assert kdx == top_sort_order.index(key)


def test_captures_checking():
'''
"""
these tests make sure the various captures access tools work properly
'''
np.array(TEST_U8_DATA0, dtype=np.uint8).tofile('/tmp/d0.sigmf-data')
with open('/tmp/d0.sigmf-meta','w') as f0: json.dump(TEST_U8_META0, f0)
np.array(TEST_U8_DATA1, dtype=np.uint8).tofile('/tmp/d1.sigmf-data')
with open('/tmp/d1.sigmf-meta','w') as f1: json.dump(TEST_U8_META1, f1)
np.array(TEST_U8_DATA2, dtype=np.uint8).tofile('/tmp/d2.sigmf-data')
with open('/tmp/d2.sigmf-meta','w') as f2: json.dump(TEST_U8_META2, f2)
np.array(TEST_U8_DATA3, dtype=np.uint8).tofile('/tmp/d3.sigmf-data')
with open('/tmp/d3.sigmf-meta','w') as f3: json.dump(TEST_U8_META3, f3)
np.array(TEST_U8_DATA4, dtype=np.uint8).tofile('/tmp/d4.sigmf-data')
with open('/tmp/d4.sigmf-meta','w') as f4: json.dump(TEST_U8_META4, f4)

sigmf0 = sigmffile.fromfile('/tmp/d0.sigmf-meta', skip_checksum=True)
sigmf1 = sigmffile.fromfile('/tmp/d1.sigmf-meta', skip_checksum=True)
sigmf2 = sigmffile.fromfile('/tmp/d2.sigmf-meta', skip_checksum=True)
sigmf3 = sigmffile.fromfile('/tmp/d3.sigmf-meta', skip_checksum=True)
sigmf4 = sigmffile.fromfile('/tmp/d4.sigmf-meta', skip_checksum=True)
"""
np.array(TEST_U8_DATA0, dtype=np.uint8).tofile("/tmp/d0.sigmf-data")
with open("/tmp/d0.sigmf-meta", "w") as f0:
json.dump(TEST_U8_META0, f0)
np.array(TEST_U8_DATA1, dtype=np.uint8).tofile("/tmp/d1.sigmf-data")
with open("/tmp/d1.sigmf-meta", "w") as f1:
json.dump(TEST_U8_META1, f1)
np.array(TEST_U8_DATA2, dtype=np.uint8).tofile("/tmp/d2.sigmf-data")
with open("/tmp/d2.sigmf-meta", "w") as f2:
json.dump(TEST_U8_META2, f2)
np.array(TEST_U8_DATA3, dtype=np.uint8).tofile("/tmp/d3.sigmf-data")
with open("/tmp/d3.sigmf-meta", "w") as f3:
json.dump(TEST_U8_META3, f3)
np.array(TEST_U8_DATA4, dtype=np.uint8).tofile("/tmp/d4.sigmf-data")
with open("/tmp/d4.sigmf-meta", "w") as f4:
json.dump(TEST_U8_META4, f4)

sigmf0 = sigmffile.fromfile("/tmp/d0.sigmf-meta", skip_checksum=True)
sigmf1 = sigmffile.fromfile("/tmp/d1.sigmf-meta", skip_checksum=True)
sigmf2 = sigmffile.fromfile("/tmp/d2.sigmf-meta", skip_checksum=True)
sigmf3 = sigmffile.fromfile("/tmp/d3.sigmf-meta", skip_checksum=True)
sigmf4 = sigmffile.fromfile("/tmp/d4.sigmf-meta", skip_checksum=True)

assert sigmf0._count_samples() == 256
assert sigmf0._is_conforming_dataset()
assert (0,0) == sigmf0.get_capture_byte_boundarys(0)
assert (0,256) == sigmf0.get_capture_byte_boundarys(1)
assert (0, 0) == sigmf0.get_capture_byte_boundarys(0)
assert (0, 256) == sigmf0.get_capture_byte_boundarys(1)
assert np.array_equal(TEST_U8_DATA0, sigmf0.read_samples(autoscale=False))
assert np.array_equal(np.array([]), sigmf0.read_samples_in_capture(0))
assert np.array_equal(TEST_U8_DATA0, sigmf0.read_samples_in_capture(1,autoscale=False))
assert np.array_equal(TEST_U8_DATA0, sigmf0.read_samples_in_capture(1, autoscale=False))

assert sigmf1._count_samples() == 192
assert not sigmf1._is_conforming_dataset()
assert (32,160) == sigmf1.get_capture_byte_boundarys(0)
assert (160,224) == sigmf1.get_capture_byte_boundarys(1)
assert np.array_equal(np.array(range(128)), sigmf1.read_samples_in_capture(0,autoscale=False))
assert np.array_equal(np.array(range(128,192)), sigmf1.read_samples_in_capture(1,autoscale=False))
assert (32, 160) == sigmf1.get_capture_byte_boundarys(0)
assert (160, 224) == sigmf1.get_capture_byte_boundarys(1)
assert np.array_equal(np.array(range(128)), sigmf1.read_samples_in_capture(0, autoscale=False))
assert np.array_equal(np.array(range(128, 192)), sigmf1.read_samples_in_capture(1, autoscale=False))

assert sigmf2._count_samples() == 192
assert not sigmf2._is_conforming_dataset()
assert (32,160) == sigmf2.get_capture_byte_boundarys(0)
assert (176,240) == sigmf2.get_capture_byte_boundarys(1)
assert np.array_equal(np.array(range(128)), sigmf2.read_samples_in_capture(0,autoscale=False))
assert np.array_equal(np.array(range(128,192)), sigmf2.read_samples_in_capture(1,autoscale=False))
assert (32, 160) == sigmf2.get_capture_byte_boundarys(0)
assert (176, 240) == sigmf2.get_capture_byte_boundarys(1)
assert np.array_equal(np.array(range(128)), sigmf2.read_samples_in_capture(0, autoscale=False))
assert np.array_equal(np.array(range(128, 192)), sigmf2.read_samples_in_capture(1, autoscale=False))

assert sigmf3._count_samples() == 192
assert not sigmf3._is_conforming_dataset()
assert (32,64) == sigmf3.get_capture_byte_boundarys(0)
assert (64,160) == sigmf3.get_capture_byte_boundarys(1)
assert (192,256) == sigmf3.get_capture_byte_boundarys(2)
assert np.array_equal(np.array(range(32)), sigmf3.read_samples_in_capture(0,autoscale=False))
assert np.array_equal(np.array(range(32,128)), sigmf3.read_samples_in_capture(1,autoscale=False))
assert np.array_equal(np.array(range(128,192)), sigmf3.read_samples_in_capture(2,autoscale=False))
assert (32, 64) == sigmf3.get_capture_byte_boundarys(0)
assert (64, 160) == sigmf3.get_capture_byte_boundarys(1)
assert (192, 256) == sigmf3.get_capture_byte_boundarys(2)
assert np.array_equal(np.array(range(32)), sigmf3.read_samples_in_capture(0, autoscale=False))
assert np.array_equal(np.array(range(32, 128)), sigmf3.read_samples_in_capture(1, autoscale=False))
assert np.array_equal(np.array(range(128, 192)), sigmf3.read_samples_in_capture(2, autoscale=False))

assert sigmf4._count_samples() == 96
assert not sigmf4._is_conforming_dataset()
assert (32,160) == sigmf4.get_capture_byte_boundarys(0)
assert (160,224) == sigmf4.get_capture_byte_boundarys(1)
assert np.array_equal(np.array(range(64)), sigmf4.read_samples_in_capture(0,autoscale=False)[:,0])
assert np.array_equal(np.array(range(64,96)), sigmf4.read_samples_in_capture(1,autoscale=False)[:,1])
assert (32, 160) == sigmf4.get_capture_byte_boundarys(0)
assert (160, 224) == sigmf4.get_capture_byte_boundarys(1)
assert np.array_equal(np.array(range(64)), sigmf4.read_samples_in_capture(0, autoscale=False)[:, 0])
assert np.array_equal(np.array(range(64, 96)), sigmf4.read_samples_in_capture(1, autoscale=False)[:, 1])


def test_slicing():
'''Test __getitem___ builtin for sigmffile, make sure slicing and indexing works as expected.'''
"""Test __getitem___ builtin for sigmffile, make sure slicing and indexing works as expected."""
_, temp_data0 = tempfile.mkstemp()
np.array(TEST_U8_DATA0, dtype=np.uint8).tofile(temp_data0)
sigmf0 = SigMFFile(metadata=TEST_U8_META0, data_file=temp_data0)
Expand All @@ -326,8 +331,8 @@ def test_slicing():
_, temp_data2 = tempfile.mkstemp()
np.array(TEST_U8_DATA4, dtype=np.uint8).tofile(temp_data2)
sigmf2 = SigMFFile(TEST_U8_META4, data_file=temp_data2)
channelized = np.array(TEST_U8_DATA4).reshape((128,2))
channelized = np.array(TEST_U8_DATA4).reshape((128, 2))
assert np.array_equal(channelized, sigmf2[:][:])
assert np.array_equal(sigmf2[10:20, 91:112], sigmf2.read_samples(autoscale=False)[10:20, 91:112])
assert np.array_equal(sigmf2[0], channelized[0])
assert np.array_equal(sigmf2[1,:], channelized[1,:])
assert np.array_equal(sigmf2[1, :], channelized[1, :])
40 changes: 17 additions & 23 deletions tests/test_validation.py
Expand Up @@ -18,50 +18,49 @@


def test_valid_data():
'''assure the supplied metadata is OK'''
"""assure the supplied metadata is OK"""
invalid_metadata = dict(TEST_METADATA)
SigMFFile(TEST_METADATA).validate()


class FailingCases(unittest.TestCase):
'''Cases where the validator should throw an exception.'''
"""Cases where the validator should throw an exception."""

def setUp(self):
self.metadata = dict(TEST_METADATA)

def test_no_version(self):
'''core:version must be present'''
"""core:version must be present"""
del self.metadata[SigMFFile.GLOBAL_KEY][SigMFFile.VERSION_KEY]
with self.assertRaises(ValidationError):
SigMFFile(self.metadata).validate()

def test_extra_top_level_key(self):
'''no extra keys allowed on the top level'''
self.metadata['extra'] = 0
"""no extra keys allowed on the top level"""
self.metadata["extra"] = 0
with self.assertRaises(ValidationError):
SigMFFile(self.metadata).validate()

def test_extra_top_level_key(self):
'''label must be less than 20 chars'''
self.metadata[SigMFFile.ANNOTATION_KEY][0][SigMFFile.LABEL_KEY] = 'a' * 21
"""label must be less than 20 chars"""
self.metadata[SigMFFile.ANNOTATION_KEY][0][SigMFFile.LABEL_KEY] = "a" * 21
with self.assertRaises(ValidationError):
SigMFFile(self.metadata).validate()

def test_invalid_type(self):
'''license key must be string'''
"""license key must be string"""
self.metadata[SigMFFile.GLOBAL_KEY][SigMFFile.LICENSE_KEY] = 1
with self.assertRaises(ValidationError):
SigMFFile(self.metadata).validate()

def test_invalid_capture_order(self):
'''metadata must have captures in order'''
self.metadata[SigMFFile.CAPTURE_KEY] = [
{SigMFFile.START_INDEX_KEY: 10},
{SigMFFile.START_INDEX_KEY: 9}
]
"""metadata must have captures in order"""
self.metadata[SigMFFile.CAPTURE_KEY] = [{SigMFFile.START_INDEX_KEY: 10}, {SigMFFile.START_INDEX_KEY: 9}]
with self.assertRaises(ValidationError):
SigMFFile(self.metadata).validate()

def test_invalid_annotation_order(self):
'''metadata must have annotations in order'''
"""metadata must have annotations in order"""
self.metadata[SigMFFile.ANNOTATION_KEY] = [
{
SigMFFile.START_INDEX_KEY: 2,
Expand All @@ -70,24 +69,19 @@ def test_invalid_annotation_order(self):
{
SigMFFile.START_INDEX_KEY: 1,
SigMFFile.LENGTH_INDEX_KEY: 120000,
}
},
]
with self.assertRaises(ValidationError):
SigMFFile(self.metadata).validate()

def test_annotation_without_sample_count(self):
'''annotation without length should be accepted'''
self.metadata[SigMFFile.ANNOTATION_KEY] = [
{
SigMFFile.START_INDEX_KEY: 2
}
]
"""annotation without length should be accepted"""
self.metadata[SigMFFile.ANNOTATION_KEY] = [{SigMFFile.START_INDEX_KEY: 2}]
SigMFFile(self.metadata).validate()


def test_invalid_hash(self):
_, temp_path = tempfile.mkstemp()
TEST_FLOAT32_DATA.tofile(temp_path)
self.metadata[SigMFFile.GLOBAL_KEY][SigMFFile.HASH_KEY] = 'derp'
self.metadata[SigMFFile.GLOBAL_KEY][SigMFFile.HASH_KEY] = "derp"
with self.assertRaises(sigmf.error.SigMFFileError):
SigMFFile(metadata=self.metadata, data_file=temp_path)
60 changes: 37 additions & 23 deletions tests/testdata.py
Expand Up @@ -16,51 +16,65 @@
SigMFFile.ANNOTATION_KEY: [{SigMFFile.LENGTH_INDEX_KEY: 16, SigMFFile.START_INDEX_KEY: 0}],
SigMFFile.CAPTURE_KEY: [{SigMFFile.START_INDEX_KEY: 0}],
SigMFFile.GLOBAL_KEY: {
SigMFFile.DATATYPE_KEY: 'rf32_le',
SigMFFile.HASH_KEY: 'f4984219b318894fa7144519185d1ae81ea721c6113243a52b51e444512a39d74cf41a4cec3c5d000bd7277cc71232c04d7a946717497e18619bdbe94bfeadd6',
SigMFFile.DATATYPE_KEY: "rf32_le",
SigMFFile.HASH_KEY: "f4984219b318894fa7144519185d1ae81ea721c6113243a52b51e444512a39d74cf41a4cec3c5d000bd7277cc71232c04d7a946717497e18619bdbe94bfeadd6",
SigMFFile.NUM_CHANNELS_KEY: 1,
SigMFFile.VERSION_KEY: __specification__,
}
},
}

# Data0 is a test of a compliant two capture recording
TEST_U8_DATA0 = list(range(256))
TEST_U8_META0 = {
SigMFFile.ANNOTATION_KEY: [],
SigMFFile.CAPTURE_KEY: [ {SigMFFile.START_INDEX_KEY: 0, SigMFFile.HEADER_BYTES_KEY: 0},
{SigMFFile.START_INDEX_KEY: 0, SigMFFile.HEADER_BYTES_KEY: 0} ], # very strange..but technically legal?
SigMFFile.GLOBAL_KEY: {SigMFFile.DATATYPE_KEY: 'ru8', SigMFFile.TRAILING_BYTES_KEY: 0}
SigMFFile.CAPTURE_KEY: [
{SigMFFile.START_INDEX_KEY: 0, SigMFFile.HEADER_BYTES_KEY: 0},
{SigMFFile.START_INDEX_KEY: 0, SigMFFile.HEADER_BYTES_KEY: 0},
], # very strange..but technically legal?
SigMFFile.GLOBAL_KEY: {SigMFFile.DATATYPE_KEY: "ru8", SigMFFile.TRAILING_BYTES_KEY: 0},
}
# Data1 is a test of a two capture recording with header_bytes and trailing_bytes set
TEST_U8_DATA1 = [0xfe]*32 + list(range(192)) + [0xff]*32
TEST_U8_DATA1 = [0xFE] * 32 + list(range(192)) + [0xFF] * 32
TEST_U8_META1 = {
SigMFFile.ANNOTATION_KEY: [],
SigMFFile.CAPTURE_KEY: [ {SigMFFile.START_INDEX_KEY: 0, SigMFFile.HEADER_BYTES_KEY: 32},
{SigMFFile.START_INDEX_KEY: 128} ],
SigMFFile.GLOBAL_KEY: {SigMFFile.DATATYPE_KEY: 'ru8', SigMFFile.TRAILING_BYTES_KEY: 32}
SigMFFile.CAPTURE_KEY: [
{SigMFFile.START_INDEX_KEY: 0, SigMFFile.HEADER_BYTES_KEY: 32},
{SigMFFile.START_INDEX_KEY: 128},
],
SigMFFile.GLOBAL_KEY: {SigMFFile.DATATYPE_KEY: "ru8", SigMFFile.TRAILING_BYTES_KEY: 32},
}
# Data2 is a test of a two capture recording with multiple header_bytes set
TEST_U8_DATA2 = [0xfe]*32 + list(range(128)) + [0xfe]*16 + list(range(128,192)) + [0xff]*16
TEST_U8_DATA2 = [0xFE] * 32 + list(range(128)) + [0xFE] * 16 + list(range(128, 192)) + [0xFF] * 16
TEST_U8_META2 = {
SigMFFile.ANNOTATION_KEY: [],
SigMFFile.CAPTURE_KEY: [ {SigMFFile.START_INDEX_KEY: 0, SigMFFile.HEADER_BYTES_KEY: 32},
{SigMFFile.START_INDEX_KEY: 128, SigMFFile.HEADER_BYTES_KEY: 16} ],
SigMFFile.GLOBAL_KEY: {SigMFFile.DATATYPE_KEY: 'ru8', SigMFFile.TRAILING_BYTES_KEY: 16}
SigMFFile.CAPTURE_KEY: [
{SigMFFile.START_INDEX_KEY: 0, SigMFFile.HEADER_BYTES_KEY: 32},
{SigMFFile.START_INDEX_KEY: 128, SigMFFile.HEADER_BYTES_KEY: 16},
],
SigMFFile.GLOBAL_KEY: {SigMFFile.DATATYPE_KEY: "ru8", SigMFFile.TRAILING_BYTES_KEY: 16},
}
# Data3 is a test of a three capture recording with multiple header_bytes set
TEST_U8_DATA3 = [0xfe]*32 + list(range(128)) + [0xfe]*32 + list(range(128,192))
TEST_U8_DATA3 = [0xFE] * 32 + list(range(128)) + [0xFE] * 32 + list(range(128, 192))
TEST_U8_META3 = {
SigMFFile.ANNOTATION_KEY: [],
SigMFFile.CAPTURE_KEY: [ {SigMFFile.START_INDEX_KEY: 0, SigMFFile.HEADER_BYTES_KEY: 32},
{SigMFFile.START_INDEX_KEY: 32},
{SigMFFile.START_INDEX_KEY: 128, SigMFFile.HEADER_BYTES_KEY: 32} ],
SigMFFile.GLOBAL_KEY: {SigMFFile.DATATYPE_KEY: 'ru8'}
SigMFFile.CAPTURE_KEY: [
{SigMFFile.START_INDEX_KEY: 0, SigMFFile.HEADER_BYTES_KEY: 32},
{SigMFFile.START_INDEX_KEY: 32},
{SigMFFile.START_INDEX_KEY: 128, SigMFFile.HEADER_BYTES_KEY: 32},
],
SigMFFile.GLOBAL_KEY: {SigMFFile.DATATYPE_KEY: "ru8"},
}
# Data4 is a two channel version of Data0
TEST_U8_DATA4 = [0xfe]*32 + [y for y in list(range(96)) for i in [0,1]] + [0xff]*32
TEST_U8_DATA4 = [0xFE] * 32 + [y for y in list(range(96)) for i in [0, 1]] + [0xFF] * 32
TEST_U8_META4 = {
SigMFFile.ANNOTATION_KEY: [],
SigMFFile.CAPTURE_KEY: [ {SigMFFile.START_INDEX_KEY: 0, SigMFFile.HEADER_BYTES_KEY: 32},
{SigMFFile.START_INDEX_KEY: 64} ],
SigMFFile.GLOBAL_KEY: {SigMFFile.DATATYPE_KEY: 'ru8', SigMFFile.TRAILING_BYTES_KEY: 32, SigMFFile.NUM_CHANNELS_KEY: 2}
SigMFFile.CAPTURE_KEY: [
{SigMFFile.START_INDEX_KEY: 0, SigMFFile.HEADER_BYTES_KEY: 32},
{SigMFFile.START_INDEX_KEY: 64},
],
SigMFFile.GLOBAL_KEY: {
SigMFFile.DATATYPE_KEY: "ru8",
SigMFFile.TRAILING_BYTES_KEY: 32,
SigMFFile.NUM_CHANNELS_KEY: 2,
},
}

0 comments on commit bdbfa07

Please sign in to comment.