|
1 | 1 | import json |
2 | 2 | import logging |
3 | 3 | import os |
4 | | -import unittest |
5 | 4 |
|
| 5 | +import pytest |
6 | 6 | from jsonasobj2 import as_json_obj |
7 | 7 |
|
8 | 8 | from linkml_runtime.dumpers import csv_dumper, json_dumper, tsv_dumper, yaml_dumper |
@@ -31,90 +31,132 @@ def _json(obj) -> str: |
31 | 31 | return json.dumps(obj, indent=" ", sort_keys=True) |
32 | 32 |
|
33 | 33 |
|
34 | | -class CsvAndTsvGenTestCase(unittest.TestCase): |
35 | | - def test_object_model(self): |
36 | | - book = Book(id="B1", genres=["fantasy"], creator={}) |
37 | | - logger.debug(as_json_obj(book.genres[0])) |
38 | | - assert str(book.genres[0]) == "fantasy" |
39 | | - assert book.genres[0].code.text == "fantasy" |
40 | | - processed = remove_empty_items(book.genres) |
41 | | - assert processed[0] == "fantasy" |
42 | | - series = BookSeries(id="S1", creator=Author(name="Q. Writer"), reviews=[Review(rating=5)]) |
43 | | - series.books.append(book) |
44 | | - schemaview = SchemaView(SCHEMA) |
45 | | - shop = Shop() |
46 | | - shop.all_book_series.append(series) |
47 | | - |
48 | | - csvstr = csv_dumper.dumps(shop, index_slot="all_book_series", schemaview=schemaview) |
49 | | - assert "," in csvstr |
50 | | - assert "\t" not in csvstr |
51 | | - |
52 | | - tsvstr = tsv_dumper.dumps(shop, index_slot="all_book_series", schemaview=schemaview) |
53 | | - assert "\t" in tsvstr |
54 | | - |
55 | | - def test_csvgen_roundtrip(self): |
56 | | - schemaview = SchemaView(SCHEMA) |
57 | | - data = yaml_loader.load(DATA, target_class=Shop) |
58 | | - csv_dumper.dump(data, to_file=OUTPUT, index_slot="all_book_series", schemaview=schemaview) |
59 | | - roundtrip = csv_loader.load(OUTPUT, target_class=Shop, index_slot="all_book_series", schemaview=schemaview) |
60 | | - logger.debug(json_dumper.dumps(roundtrip)) |
61 | | - logger.debug(f"COMPARE 1: {roundtrip}") |
62 | | - logger.debug(f"COMPARE 2: {data}") |
63 | | - assert roundtrip == data |
64 | | - |
65 | | - def test_csvgen_roundtrip_to_dict(self): |
66 | | - schemaview = SchemaView(SCHEMA) |
67 | | - data = yaml_loader.load(DATA, target_class=Shop) |
68 | | - csv_dumper.dump(data, to_file=OUTPUT, index_slot="all_book_series", schemaview=schemaview) |
69 | | - roundtrip = csv_loader.load_as_dict(OUTPUT, index_slot="all_book_series", schemaview=schemaview) |
70 | | - assert roundtrip == json_dumper.to_dict(data) |
71 | | - |
72 | | - def test_tsvgen_roundtrip(self): |
73 | | - schemaview = SchemaView(SCHEMA) |
74 | | - data = yaml_loader.load(DATA, target_class=Shop) |
75 | | - tsv_dumper.dump(data, to_file=OUTPUT, index_slot="all_book_series", schemaview=schemaview) |
76 | | - roundtrip = tsv_loader.load(OUTPUT, target_class=Shop, index_slot="all_book_series", schemaview=schemaview) |
77 | | - assert roundtrip == data |
78 | | - |
79 | | - def test_tsvgen_roundtrip_to_dict(self): |
80 | | - schemaview = SchemaView(SCHEMA) |
81 | | - data = yaml_loader.load(DATA, target_class=Shop) |
82 | | - tsv_dumper.dump(data, to_file=OUTPUT, index_slot="all_book_series", schemaview=schemaview) |
83 | | - roundtrip = tsv_loader.load_as_dict(OUTPUT, index_slot="all_book_series", schemaview=schemaview) |
84 | | - assert roundtrip == json_dumper.to_dict(data) |
85 | | - |
86 | | - def test_csvgen_unroundtrippable(self): |
87 | | - schemaview = SchemaView(SCHEMA) |
88 | | - # schema = YAMLGenerator(SCHEMA).schema |
89 | | - data = yaml_loader.load(DATA2, target_class=Shop) |
90 | | - logger.debug(data.all_book_series[0]) |
91 | | - logger.debug(data.all_book_series[0].genres[0]) |
92 | | - assert str(data.all_book_series[0].genres[0]) == "fantasy" |
93 | | - logger.debug(yaml_dumper.dumps(data)) |
94 | | - logger.debug(json_dumper.dumps(data)) |
95 | | - processed = remove_empty_items(data) |
96 | | - logger.debug(f"PROC {processed['all_book_series']}") |
97 | | - asj = as_json_object(processed, None) |
98 | | - logger.debug(f"ASJ {asj['all_book_series']}") |
99 | | - reconstituted_json = json.loads(json_dumper.dumps(data)) |
100 | | - s0 = reconstituted_json["all_book_series"][0] |
101 | | - logger.debug(s0) |
102 | | - logger.debug(json_dumper.dumps(data)) |
103 | | - # logger.debug(csv_dumper.dumps(data, index_slot='all_book_series', schema=schema)) |
104 | | - csv_dumper.dump(data, to_file=OUTPUT2, index_slot="all_book_series", schemaview=schemaview) |
105 | | - # assert False |
106 | | - roundtrip = csv_loader.load(OUTPUT2, target_class=Shop, index_slot="all_book_series", schemaview=schemaview) |
107 | | - logger.debug(json_dumper.dumps(roundtrip)) |
108 | | - assert roundtrip == data |
109 | | - |
110 | | - def test_tsvgen_unroundtrippable(self): |
111 | | - schemaview = SchemaView(SCHEMA) |
112 | | - data = yaml_loader.load(DATA2, target_class=Shop) |
113 | | - assert str(data.all_book_series[0].genres[0]) == "fantasy" |
114 | | - tsv_dumper.dump(data, to_file=OUTPUT2, index_slot="all_book_series", schemaview=schemaview) |
115 | | - roundtrip = tsv_loader.load(OUTPUT2, target_class=Shop, index_slot="all_book_series", schemaview=schemaview) |
116 | | - assert roundtrip == data |
117 | | - |
118 | | - |
119 | | -if __name__ == "__main__": |
120 | | - unittest.main() |
| 34 | +@pytest.fixture |
| 35 | +def schema_view(): |
| 36 | + """SchemaView instance for books normalized schema.""" |
| 37 | + return SchemaView(SCHEMA) |
| 38 | + |
| 39 | + |
| 40 | +@pytest.fixture |
| 41 | +def test_data(schema_view): |
| 42 | + """Load test data from books_normalized_01.yaml.""" |
| 43 | + return yaml_loader.load(DATA, target_class=Shop) |
| 44 | + |
| 45 | + |
| 46 | +@pytest.fixture |
| 47 | +def test_data2(schema_view): |
| 48 | + """Load test data from books_normalized_02.yaml.""" |
| 49 | + return yaml_loader.load(DATA2, target_class=Shop) |
| 50 | + |
| 51 | + |
| 52 | +def test_object_model(schema_view): |
| 53 | + """Test basic object model and enum handling.""" |
| 54 | + book = Book(id="B1", genres=["fantasy"], creator={}) |
| 55 | + logger.debug(as_json_obj(book.genres[0])) |
| 56 | + assert str(book.genres[0]) == "fantasy" |
| 57 | + assert book.genres[0].code.text == "fantasy" |
| 58 | + processed = remove_empty_items(book.genres) |
| 59 | + assert processed[0] == "fantasy" |
| 60 | + |
| 61 | + # Create series and shop |
| 62 | + series = BookSeries(id="S1", creator=Author(name="Q. Writer"), reviews=[Review(rating=5)]) |
| 63 | + series.books.append(book) |
| 64 | + shop = Shop() |
| 65 | + shop.all_book_series.append(series) |
| 66 | + |
| 67 | + # Test CSV dumping |
| 68 | + csvstr = csv_dumper.dumps(shop, index_slot="all_book_series", schemaview=schema_view) |
| 69 | + assert "," in csvstr |
| 70 | + assert "\t" not in csvstr |
| 71 | + |
| 72 | + # Test TSV dumping |
| 73 | + tsvstr = tsv_dumper.dumps(shop, index_slot="all_book_series", schemaview=schema_view) |
| 74 | + assert "\t" in tsvstr |
| 75 | + |
| 76 | + |
| 77 | +def test_csvgen_roundtrip(schema_view, test_data, tmp_path): |
| 78 | + """Test CSV round-trip conversion (dump and load).""" |
| 79 | + output_file = tmp_path / "books_flattened.tsv" |
| 80 | + |
| 81 | + csv_dumper.dump(test_data, to_file=str(output_file), index_slot="all_book_series", schemaview=schema_view) |
| 82 | + roundtrip = csv_loader.load( |
| 83 | + str(output_file), target_class=Shop, index_slot="all_book_series", schemaview=schema_view |
| 84 | + ) |
| 85 | + |
| 86 | + logger.debug(json_dumper.dumps(roundtrip)) |
| 87 | + logger.debug(f"COMPARE 1: {roundtrip}") |
| 88 | + logger.debug(f"COMPARE 2: {test_data}") |
| 89 | + assert roundtrip == test_data |
| 90 | + |
| 91 | + |
| 92 | +def test_csvgen_roundtrip_to_dict(schema_view, test_data, tmp_path): |
| 93 | + """Test CSV round-trip conversion to dictionary.""" |
| 94 | + output_file = tmp_path / "books_flattened.tsv" |
| 95 | + |
| 96 | + csv_dumper.dump(test_data, to_file=str(output_file), index_slot="all_book_series", schemaview=schema_view) |
| 97 | + roundtrip = csv_loader.load_as_dict(str(output_file), index_slot="all_book_series", schemaview=schema_view) |
| 98 | + assert roundtrip == json_dumper.to_dict(test_data) |
| 99 | + |
| 100 | + |
| 101 | +def test_tsvgen_roundtrip(schema_view, test_data, tmp_path): |
| 102 | + """Test TSV round-trip conversion (dump and load).""" |
| 103 | + output_file = tmp_path / "books_flattened.tsv" |
| 104 | + |
| 105 | + tsv_dumper.dump(test_data, to_file=str(output_file), index_slot="all_book_series", schemaview=schema_view) |
| 106 | + roundtrip = tsv_loader.load( |
| 107 | + str(output_file), target_class=Shop, index_slot="all_book_series", schemaview=schema_view |
| 108 | + ) |
| 109 | + assert roundtrip == test_data |
| 110 | + |
| 111 | + |
| 112 | +def test_tsvgen_roundtrip_to_dict(schema_view, test_data, tmp_path): |
| 113 | + """Test TSV round-trip conversion to dictionary.""" |
| 114 | + output_file = tmp_path / "books_flattened.tsv" |
| 115 | + |
| 116 | + tsv_dumper.dump(test_data, to_file=str(output_file), index_slot="all_book_series", schemaview=schema_view) |
| 117 | + roundtrip = tsv_loader.load_as_dict(str(output_file), index_slot="all_book_series", schemaview=schema_view) |
| 118 | + assert roundtrip == json_dumper.to_dict(test_data) |
| 119 | + |
| 120 | + |
| 121 | +def test_csvgen_unroundtrippable(schema_view, test_data2, tmp_path): |
| 122 | + """Test CSV handling of complex/unroundtrippable data.""" |
| 123 | + output_file = tmp_path / "books_flattened_02.tsv" |
| 124 | + |
| 125 | + # Test initial data structure |
| 126 | + logger.debug(test_data2.all_book_series[0]) |
| 127 | + logger.debug(test_data2.all_book_series[0].genres[0]) |
| 128 | + assert str(test_data2.all_book_series[0].genres[0]) == "fantasy" |
| 129 | + logger.debug(yaml_dumper.dumps(test_data2)) |
| 130 | + logger.debug(json_dumper.dumps(test_data2)) |
| 131 | + |
| 132 | + # Test data processing |
| 133 | + processed = remove_empty_items(test_data2) |
| 134 | + logger.debug(f"PROC {processed['all_book_series']}") |
| 135 | + asj = as_json_object(processed, None) |
| 136 | + logger.debug(f"ASJ {asj['all_book_series']}") |
| 137 | + |
| 138 | + reconstituted_json = json.loads(json_dumper.dumps(test_data2)) |
| 139 | + s0 = reconstituted_json["all_book_series"][0] |
| 140 | + logger.debug(s0) |
| 141 | + logger.debug(json_dumper.dumps(test_data2)) |
| 142 | + |
| 143 | + # Test CSV dump and load |
| 144 | + csv_dumper.dump(test_data2, to_file=str(output_file), index_slot="all_book_series", schemaview=schema_view) |
| 145 | + roundtrip = csv_loader.load( |
| 146 | + str(output_file), target_class=Shop, index_slot="all_book_series", schemaview=schema_view |
| 147 | + ) |
| 148 | + logger.debug(json_dumper.dumps(roundtrip)) |
| 149 | + assert roundtrip == test_data2 |
| 150 | + |
| 151 | + |
| 152 | +def test_tsvgen_unroundtrippable(schema_view, test_data2, tmp_path): |
| 153 | + """Test TSV handling of complex/unroundtrippable data.""" |
| 154 | + output_file = tmp_path / "books_flattened_02.tsv" |
| 155 | + |
| 156 | + assert str(test_data2.all_book_series[0].genres[0]) == "fantasy" |
| 157 | + |
| 158 | + tsv_dumper.dump(test_data2, to_file=str(output_file), index_slot="all_book_series", schemaview=schema_view) |
| 159 | + roundtrip = tsv_loader.load( |
| 160 | + str(output_file), target_class=Shop, index_slot="all_book_series", schemaview=schema_view |
| 161 | + ) |
| 162 | + assert roundtrip == test_data2 |
0 commit comments