-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Use pytest-datadir to pass data dir to tests Remove use of unittest
- Loading branch information
Showing
13 changed files
with
423 additions
and
474 deletions.
There are no files selected for viewing
Large diffs are not rendered by default.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,39 +1,26 @@ | ||
import os | ||
import unittest | ||
from tsdf import read_tsdf | ||
import tsdf | ||
from tsdf import legacy_tsdf_utils | ||
from tsdf.constants import TestConstants as CONST | ||
|
||
|
||
class TestConversion(unittest.TestCase): | ||
def test_conversion(shared_datadir): | ||
"""Test whether the conversion from TSDB (legacy metadata format) to TSDF works. """ | ||
|
||
def test_conversion(self): | ||
path_to_file = os.path.join(CONST.TEST_DATA_DIR, "ppp_format_meta_legacy.json") | ||
path_to_new_file = os.path.join( | ||
CONST.TEST_OUTPUT_DATA_DIR, "tmp_test_ppp_format_meta.json" | ||
) | ||
path_to_file = shared_datadir / "ppp_format_meta_legacy.json" | ||
path_to_new_file = shared_datadir / "tmp_test_ppp_format_meta.json" | ||
path_to_existing_tsdf_file = shared_datadir / "ppp_format_meta.json" | ||
|
||
path_to_existing_tsdf_file = os.path.join( | ||
CONST.TEST_DATA_DIR, "ppp_format_meta.json" | ||
) | ||
# Generate a TSDF metadata file from TSDB | ||
legacy_tsdf_utils.generate_tsdf_metadata_from_tsdb(path_to_file, path_to_new_file) | ||
|
||
# Generate a TSDF metadata file from TSDB | ||
legacy_tsdf_utils.generate_tsdf_metadata_from_tsdb(path_to_file, path_to_new_file) | ||
# Load the generated metadata file | ||
new_meta = tsdf.load_metadata_from_path(path_to_new_file) | ||
|
||
# Load the generated metadata file | ||
new_meta = read_tsdf.load_metadata_from_path(path_to_new_file) | ||
# Load the existing metadata file | ||
existing_meta = tsdf.load_metadata_from_path(path_to_existing_tsdf_file) | ||
|
||
# Load the existing metadata file | ||
existing_meta = read_tsdf.load_metadata_from_path(path_to_existing_tsdf_file) | ||
# Compare the two metadata files (whether the mapped TSDFs fields are the same) | ||
assert(new_meta["ppp_format_time.bin"].get_plain_tsdf_dict_copy() == | ||
existing_meta["ppp_format_time.bin"].get_plain_tsdf_dict_copy()) | ||
|
||
# Compare the two metadata files (whether the mapped TSDFs fields are the same) | ||
self.assertEqual( | ||
new_meta["ppp_format_time.bin"].get_plain_tsdf_dict_copy(), | ||
existing_meta["ppp_format_time.bin"].get_plain_tsdf_dict_copy(), | ||
) | ||
|
||
self.assertEqual( | ||
new_meta["ppp_format_samples.bin"].get_plain_tsdf_dict_copy(), | ||
existing_meta["ppp_format_samples.bin"].get_plain_tsdf_dict_copy(), | ||
) | ||
assert(new_meta["ppp_format_samples.bin"].get_plain_tsdf_dict_copy() == | ||
existing_meta["ppp_format_samples.bin"].get_plain_tsdf_dict_copy()) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,41 +1,37 @@ | ||
import json | ||
import unittest | ||
import pytest | ||
from tsdf import parse_metadata | ||
from tsdf.tsdfmetadata import TSDFMetadataFieldError, TSDFMetadataFieldValueError | ||
from tsdf.constants import TestConstants as CONST | ||
|
||
|
||
class TestWrongFormatting(unittest.TestCase): | ||
"""Test whether the exceptions are thrown in case the metadata file is not well annotated.""" | ||
def test_load_wrong_version(shared_datadir): | ||
"""Test that a file with a wrong version raises an exception.""" | ||
|
||
path = shared_datadir / "wrongversion_meta_fail.json" | ||
with open(path, "r") as file: | ||
data = json.load(file) | ||
with pytest.raises(TSDFMetadataFieldValueError): | ||
parse_metadata.read_data(data, path) # This should trigger an exception | ||
|
||
def test_load_wrong_version(self): | ||
"""Test that a file with a wrong version raises an exception.""" | ||
path = CONST.TEST_DATA_FILES["wrongversion"] | ||
with open(path, "r") as file: | ||
with self.assertRaises(TSDFMetadataFieldValueError) as context: | ||
data = json.load(file) | ||
parse_metadata.read_data(data, path) # This should trigger an exception | ||
def test_load_missing_key(shared_datadir): | ||
"""Test that a file with a missing mandatory key raises an exception.""" | ||
|
||
def test_load_missing_key(self): | ||
"""Test that a file with a missing mandatory key raises an exception.""" | ||
path = CONST.TEST_DATA_FILES["missingkey"] | ||
with open(path, "r") as file: | ||
with self.assertRaises(TSDFMetadataFieldError) as context: | ||
data = json.load(file) | ||
parse_metadata.read_data(data, path) # This should trigger an exception | ||
path = shared_datadir / "missingkey_meta_fail.json" | ||
with open(path, "r") as file: | ||
data = json.load(file) | ||
with pytest.raises(TSDFMetadataFieldError): | ||
parse_metadata.read_data(data, path) # This should trigger an exception | ||
|
||
|
||
class TestTSDFMetadataParsing(unittest.TestCase): | ||
"""Test whether the TSDF objects are well specified are well defined.""" | ||
def test_load_flat_structure(shared_datadir): | ||
"""Test parsing of a flat TSDF metadata file.""" | ||
|
||
def test_load_flat_structure(self): | ||
"""Test parsing of a flat TSDF metadata file.""" | ||
path = CONST.TEST_DATA_FILES["flat"] | ||
with open(path, "r") as file: | ||
data = json.load(file) | ||
streams = parse_metadata.read_data(data, path) | ||
first_stream = parse_metadata.get_file_metadata_at_index(streams, 0) | ||
version: str = first_stream.metadata_version | ||
for key, value in data.items(): | ||
if parse_metadata.is_mandatory_type(key, version): | ||
self.assertTrue(value == first_stream.__getattribute__(key)) | ||
path = shared_datadir / "flat_meta.json" | ||
with open(path, "r") as file: | ||
data = json.load(file) | ||
streams = parse_metadata.read_data(data, path) | ||
first_stream = parse_metadata.get_file_metadata_at_index(streams, 0) | ||
version: str = first_stream.metadata_version | ||
for key, value in data.items(): | ||
if parse_metadata.is_mandatory_type(key, version): | ||
assert(value == first_stream.__getattribute__(key)) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,85 +1,51 @@ | ||
import os | ||
import unittest | ||
import numpy as np | ||
from tsdf.constants import TestConstants as CONST | ||
from tsdf import read_tsdf | ||
from tsdf import read_binary | ||
from pathlib import Path | ||
import tsdf | ||
from tsdf import parse_metadata | ||
|
||
|
||
def load_single_bin_file(dir_path: str, file_name: str) -> np.ndarray: | ||
""" | ||
Load a single binary file from the given directory path and file name. | ||
:param dir_path: The directory path where the binary file is located. | ||
:param file_name: The name of the binary file without the extension. | ||
:returns: The binary data as a numpy array. | ||
""" | ||
path = os.path.join(dir_path, file_name + CONST.METADATA_EXTENSION) | ||
metadata = read_tsdf.load_metadata_from_path(path) | ||
data = read_binary.load_binary_from_metadata(metadata[file_name + CONST.BINARY_EXTENSION] | ||
) | ||
return data | ||
|
||
|
||
class TestBinaryFileReading(unittest.TestCase): | ||
"""Test reading of binary files based on the TSDF metadata.""" | ||
|
||
def test_load_binary_float32(self): | ||
data = load_single_bin_file(CONST.TEST_DATA_DIR, "example_10_3_float32") | ||
|
||
self.assertEqual(data.shape, (10, 3)) | ||
self.assertEqual(data.dtype, "float32") | ||
|
||
def test_load_binary_float64(self): | ||
data = load_single_bin_file(CONST.TEST_DATA_DIR, "example_10_3_float64") | ||
|
||
self.assertEqual(data.shape, (10, 3)) | ||
self.assertEqual(data.dtype, "float64") | ||
|
||
# def test_load_binary_float64_fail(self): | ||
# """Should raise an exception on reading binary data""" | ||
# path = os.path.join(CONST.TEST_DATA_DIR, "example_10_3_float64_meta_fail.json") | ||
# metadata = load_tsdf.load_metadata_from_path(path) | ||
# with self.assertRaises(Exception) as exc_context: | ||
# io_binary.load_binary_from_metadata( | ||
# CONST.TEST_DATA_DIR, io_metadata.get_file_metadata_at_index(metadata, 0) | ||
# ) | ||
# self.assertEqual( | ||
# exc_context.exception.args[0], "Number of rows doesn't match file length." | ||
# ) | ||
|
||
def test_load_binary_int16(self): | ||
data = load_single_bin_file(CONST.TEST_DATA_DIR, "example_10_3_int16") | ||
|
||
self.assertEqual(data.shape, (10, 3)) | ||
self.assertEqual(data.dtype, "int16") | ||
|
||
def test_load_like_ppp(self): | ||
path = CONST.TEST_DATA_FILES["ppp"] | ||
metadata = read_tsdf.load_metadata_from_path(path) | ||
time_data = read_binary.load_binary_from_metadata(parse_metadata.get_file_metadata_at_index(metadata, 0) | ||
) | ||
self.assertEqual(time_data.shape, (17,)) | ||
# time data should be loaded as float64 | ||
self.assertEqual(time_data.dtype, "float32") | ||
|
||
sample_data = read_binary.load_binary_from_metadata(parse_metadata.get_file_metadata_at_index(metadata, 1) | ||
) | ||
self.assertEqual(sample_data.shape, (17, 6)) | ||
# sample data should be loaded as int16 | ||
self.assertEqual(sample_data.dtype, "int16") | ||
|
||
def test_random_access(self): | ||
# TODO: test the new random access functionality | ||
file_name = "example_10_3_int16" | ||
path = os.path.join(CONST.TEST_DATA_DIR, file_name + CONST.METADATA_EXTENSION) | ||
metadata = read_tsdf.load_metadata_from_path(path) | ||
data = read_binary.load_binary_from_metadata( | ||
metadata[file_name + CONST.BINARY_EXTENSION], | ||
2, | ||
6, | ||
) | ||
self.assertEqual(data.shape, (4, 3)) | ||
self.assertEqual(data.dtype, "int16") | ||
from utils import load_single_bin_file | ||
|
||
|
||
def test_load_binary_float32(shared_datadir): | ||
data = load_single_bin_file(shared_datadir, "example_10_3_float32") | ||
assert(data.shape == (10, 3)) | ||
assert(data.dtype == "float32") | ||
|
||
def test_load_binary_float64(shared_datadir): | ||
data = load_single_bin_file(shared_datadir, "example_10_3_float64") | ||
assert(data.shape == (10, 3)) | ||
assert(data.dtype == "float64") | ||
|
||
# def test_load_binary_float64_fail(shared_datadir): | ||
# """Should raise an exception on reading binary data""" | ||
# path = shared_datadir / "example_10_3_float64_meta_fail.json" | ||
# metadata = load_tsdf.load_metadata_from_path(path) | ||
# with self.assertRaises(Exception) as exc_context: | ||
# io_binary.load_binary_from_metadata( | ||
# shared_datadir, io_metadata.get_file_metadata_at_index(metadata, 0) | ||
# ) | ||
# assert(exc_context.exception.args[0] == "Number of rows doesn't match file length.") | ||
|
||
def test_load_binary_int16(shared_datadir): | ||
data = load_single_bin_file(shared_datadir, "example_10_3_int16") | ||
assert(data.shape == (10, 3)) | ||
assert(data.dtype == "int16") | ||
|
||
def test_load_like_ppp(shared_datadir): | ||
metadata = tsdf.load_metadata_from_path(shared_datadir / "ppp_format_meta.json") | ||
time_data = tsdf.load_binary_from_metadata(parse_metadata.get_file_metadata_at_index(metadata, 0)) | ||
assert(time_data.shape == (17,)) | ||
# time data should be loaded as float64 | ||
assert(time_data.dtype == "float32") | ||
|
||
sample_data = tsdf.load_binary_from_metadata(parse_metadata.get_file_metadata_at_index(metadata, 1)) | ||
assert(sample_data.shape == (17, 6)) | ||
# sample data should be loaded as int16 | ||
assert(sample_data.dtype == "int16") | ||
|
||
def test_random_access(shared_datadir): | ||
# TODO: test the new random access functionality | ||
name = "example_10_3_int16" | ||
metadata = tsdf.load_metadata_from_path(shared_datadir / (name + "_meta.json")) | ||
data = tsdf.load_binary_from_metadata(metadata[name + ".bin"], 2, 6) | ||
assert(data.shape == (4, 3)) | ||
assert(data.dtype == "int16") |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,34 +1,31 @@ | ||
import unittest | ||
from tsdf.constants import TestConstants as CONST | ||
from tsdf import read_tsdf | ||
class TestMetadataFileReading(unittest.TestCase): | ||
"""Test loading of the metadata file.""" | ||
import tsdf | ||
|
||
def test_load_metadata_file(self): | ||
"""Test that a json file gets loaded correctly.""" | ||
with open(CONST.TEST_DATA_FILES["hierarchical"], "r") as file: | ||
data = read_tsdf.load_metadata_file(file) | ||
self.assertEqual(len(data), 4) | ||
|
||
def test_load_metadata_legacy_file(self): | ||
"""Test that a json file gets loaded correctly.""" | ||
with open(CONST.TEST_DATA_FILES["legacy"], "r") as file: | ||
data = read_tsdf.load_metadata_legacy_file(file) | ||
self.assertEqual(len(data), 2) | ||
def test_load_metadata_file(shared_datadir): | ||
"""Test that a json file gets loaded correctly.""" | ||
with open(shared_datadir / "hierarchical_meta.json", "r") as file: | ||
data = tsdf.load_metadata_file(file) | ||
assert(len(data) == 4) | ||
|
||
def test_load_metadata_from_path(self): | ||
"""Test that a json file from a path gets loaded correctly.""" | ||
data = read_tsdf.load_metadata_from_path(CONST.TEST_DATA_FILES["hierarchical"]) | ||
self.assertEqual(len(data), 4) | ||
def test_load_metadata_legacy_file(shared_datadir): | ||
"""Test that a json file gets loaded correctly.""" | ||
with open(shared_datadir / "ppp_format_meta_legacy.json", "r") as file: | ||
data = tsdf.load_metadata_legacy_file(file) | ||
assert(len(data) == 2) | ||
|
||
def test_load_metadata_string(self): | ||
"""Test that a json object gets loaded from a string correctly.""" | ||
with open(CONST.TEST_DATA_FILES["hierarchical"], "r") as file: | ||
json_string = file.read() | ||
data = read_tsdf.load_metadata_string(json_string) | ||
self.assertEqual(len(data), 4) | ||
def test_load_metadata_from_path(shared_datadir): | ||
"""Test that a json file from a path gets loaded correctly.""" | ||
data = tsdf.load_metadata_from_path(shared_datadir / "hierarchical_meta.json") | ||
assert(len(data) == 4) | ||
|
||
def test_load_metadatas_from_dir(self): | ||
"""Test that all metadata files gets loaded from a directory correctly.""" | ||
data = read_tsdf.load_metadatas_from_dir(CONST.TEST_DATA_DIR) | ||
self.assertEqual(len(data), 6) | ||
def test_load_metadata_string(shared_datadir): | ||
"""Test that a json object gets loaded from a string correctly.""" | ||
with open(shared_datadir / "hierarchical_meta.json", "r") as file: | ||
json_string = file.read() | ||
data = tsdf.load_metadata_string(json_string) | ||
assert(len(data) == 4) | ||
|
||
def test_load_metadatas_from_dir(shared_datadir): | ||
"""Test that all metadata files gets loaded from a directory correctly.""" | ||
data = tsdf.load_metadatas_from_dir(shared_datadir) | ||
assert(len(data) == 6) |
Oops, something went wrong.