Spaces:
Runtime error
Runtime error
File size: 977 Bytes
e4f9cbe |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 |
"""Tests for the CSV source."""
import csv
import os
import pathlib
from ...schema import schema
from .csv_source import LINE_NUMBER_COLUMN, CSVDataset
from .source import SourceSchema
def test_csv(tmp_path: pathlib.Path) -> None:
csv_rows = [{'x': 1, 'y': 'ten'}, {'x': 2, 'y': 'twenty'}]
filename = 'test-dataset.csv'
filepath = os.path.join(tmp_path, filename)
with open(filepath, 'w') as f:
writer = csv.DictWriter(f, fieldnames=list(csv_rows[0].keys()))
writer.writeheader()
writer.writerows(csv_rows)
source = CSVDataset(filepaths=[filepath])
source.setup()
source_schema = source.source_schema()
assert source_schema == SourceSchema(
fields=schema({
LINE_NUMBER_COLUMN: 'int64',
'x': 'int64',
'y': 'string'
}).fields, num_items=2)
items = list(source.process())
assert items == [{
LINE_NUMBER_COLUMN: 0,
'x': 1,
'y': 'ten'
}, {
LINE_NUMBER_COLUMN: 1,
'x': 2,
'y': 'twenty'
}]
|