mirror of
https://github.com/simonw/sqlite-utils.git
synced 2026-07-23 09:24:31 +02:00
cffe30fd2f`ValueTracker.evaluate()` used `if not value` to skip NULL values, but this also skipped empty strings (`""`), `0`, and other falsy values. A CSV column containing only empty strings was incorrectly inferred as INTEGER (the initial default) instead of TEXT. Replace the falsy check with an explicit `if value is None` so that only NULL is skipped and all other values — including `""` — are evaluated against the type tests.
120 lines
3.6 KiB
Python
120 lines
3.6 KiB
Python
from sqlite_utils import utils
|
|
from sqlite_utils.utils import TypeTracker
|
|
import csv
|
|
import io
|
|
import pytest
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"input,expected,should_be_is",
|
|
[
|
|
({}, None, True),
|
|
({"foo": "bar"}, None, True),
|
|
(
|
|
{"content": {"$base64": True, "encoded": "aGVsbG8="}},
|
|
{"content": b"hello"},
|
|
False,
|
|
),
|
|
],
|
|
)
|
|
def test_decode_base64_values(input, expected, should_be_is):
|
|
actual = utils.decode_base64_values(input)
|
|
if should_be_is:
|
|
assert actual is input
|
|
else:
|
|
assert actual == expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"size,expected",
|
|
(
|
|
(1, [["a"], ["b"], ["c"], ["d"]]),
|
|
(2, [["a", "b"], ["c", "d"]]),
|
|
(3, [["a", "b", "c"], ["d"]]),
|
|
(4, [["a", "b", "c", "d"]]),
|
|
),
|
|
)
|
|
def test_chunks(size, expected):
|
|
input = ["a", "b", "c", "d"]
|
|
chunks = list(map(list, utils.chunks(input, size)))
|
|
assert chunks == expected
|
|
|
|
|
|
def test_hash_record():
|
|
expected = "d383e7c0ba88f5ffcdd09be660de164b3847401a"
|
|
assert utils.hash_record({"name": "Cleo", "twitter": "CleoPaws"}) == expected
|
|
assert (
|
|
utils.hash_record(
|
|
{"name": "Cleo", "twitter": "CleoPaws", "age": 7}, keys=("name", "twitter")
|
|
)
|
|
== expected
|
|
)
|
|
assert (
|
|
utils.hash_record({"name": "Cleo", "twitter": "CleoPaws", "age": 7}) != expected
|
|
)
|
|
|
|
|
|
def test_maximize_csv_field_size_limit():
|
|
# Reset to default in case other tests have changed it
|
|
csv.field_size_limit(utils.ORIGINAL_CSV_FIELD_SIZE_LIMIT)
|
|
long_value = "a" * 131073
|
|
long_csv = "id,text\n1,{}".format(long_value)
|
|
fp = io.BytesIO(long_csv.encode("utf-8"))
|
|
# Using rows_from_file should error
|
|
with pytest.raises(csv.Error):
|
|
rows, _ = utils.rows_from_file(fp, utils.Format.CSV)
|
|
list(rows)
|
|
# But if we call maximize_csv_field_size_limit() first it should be OK:
|
|
utils.maximize_csv_field_size_limit()
|
|
fp2 = io.BytesIO(long_csv.encode("utf-8"))
|
|
rows2, _ = utils.rows_from_file(fp2, utils.Format.CSV)
|
|
rows_list2 = list(rows2)
|
|
assert len(rows_list2) == 1
|
|
assert rows_list2[0]["id"] == "1"
|
|
assert rows_list2[0]["text"] == long_value
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"rows,expected_types",
|
|
[
|
|
([{"a": ""}], {"a": "text"}),
|
|
([{"a": ""}, {"a": ""}], {"a": "text"}),
|
|
([{"a": "1"}, {"a": ""}], {"a": "text"}),
|
|
([{"a": "0"}], {"a": "integer"}),
|
|
([{"a": "1"}], {"a": "integer"}),
|
|
([{"a": None}], {"a": "integer"}),
|
|
],
|
|
)
|
|
def test_type_tracker_empty_strings(rows, expected_types):
|
|
tracker = TypeTracker()
|
|
list(tracker.wrap(iter(rows)))
|
|
assert tracker.types == expected_types
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"input,expected",
|
|
(
|
|
({"foo": {"bar": 1}}, {"foo_bar": 1}),
|
|
({"foo": {"bar": [1, 2, {"baz": 3}]}}, {"foo_bar": [1, 2, {"baz": 3}]}),
|
|
({"foo": {"bar": 1, "baz": {"three": 3}}}, {"foo_bar": 1, "foo_baz_three": 3}),
|
|
),
|
|
)
|
|
def test_flatten(input, expected):
|
|
assert utils.flatten(input) == expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"input,expected",
|
|
(
|
|
([], []),
|
|
(["id", "name"], ["id", "name"]),
|
|
(["id", "id"], ["id", "id_2"]),
|
|
(["id", "id", "id"], ["id", "id_2", "id_3"]),
|
|
# A renamed duplicate must not clobber a real column called id_2
|
|
(["id", "id", "id_2"], ["id", "id_3", "id_2"]),
|
|
(["id_2", "id", "id"], ["id_2", "id", "id_3"]),
|
|
(["id", "id", "id_2", "id_2"], ["id", "id_3", "id_2", "id_2_2"]),
|
|
),
|
|
)
|
|
def test_dedupe_keys(input, expected):
|
|
assert utils.dedupe_keys(input) == expected
|