wherobots-python-dbapi/tests/test_store_options.py at a7be0673c0c06a6b2b5132012c86a706bfb9c313 · james-willis/wherobots-python-dbapi · GitHub

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
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
"""Tests for Store options support."""

import json

from wherobots.db.models import Store
from wherobots.db.types import StorageFormat


class TestStoreOptions:
    """Tests for the options field on Store."""

    def test_default_options_is_none(self):
        store = Store(format=StorageFormat.PARQUET)
        assert store.options is None

    def test_options_set(self):
        store = Store(
            format=StorageFormat.CSV,
            options={"header": "false", "delimiter": "|"},
        )
        assert store.options == {"header": "false", "delimiter": "|"}

    def test_empty_options_normalized_to_none(self):
        store = Store(format=StorageFormat.PARQUET, options={})
        assert store.options is None

    def test_none_options_stays_none(self):
        store = Store(format=StorageFormat.PARQUET, options=None)
        assert store.options is None

    def test_options_defensively_copied(self):
        original = {"header": "false"}
        store = Store(format=StorageFormat.CSV, options=original)
        # Mutating the original should not affect the store
        original["delimiter"] = "|"
        assert "delimiter" not in store.options

    def test_options_dict_is_mutable(self):
        """Store is not frozen, so options dict can be mutated after construction."""
        store = Store(format=StorageFormat.CSV, options={"header": "false"})
        store.options["delimiter"] = "|"
        assert store.options == {"header": "false", "delimiter": "|"}


class TestStoreForDownloadWithOptions:
    """Tests for Store.for_download() with options parameter."""

    def test_for_download_default_no_options(self):
        store = Store.for_download()
        assert store.options is None

    def test_for_download_with_options(self):
        store = Store.for_download(options={"header": "false"})
        assert store.options == {"header": "false"}
        assert store.single is True
        assert store.generate_presigned_url is True

    def test_for_download_with_format_and_options(self):
        store = Store.for_download(
            format=StorageFormat.CSV,
            options={"header": "false", "delimiter": "|"},
        )
        assert store.format == StorageFormat.CSV
        assert store.options == {"header": "false", "delimiter": "|"}

    def test_for_download_empty_options_normalized(self):
        store = Store.for_download(options={})
        assert store.options is None


class TestStoreSerializationWithOptions:
    """Tests for store dict serialization matching connection.py's format."""

    def _serialize_store(self, store: Store) -> dict:
        """Replicate the serialization logic from Connection.__execute_sql."""
        store_dict = {
            "format": store.format.value,
            "single": str(store.single).lower(),
            "generate_presigned_url": str(store.generate_presigned_url).lower(),
        }
        if store.options:
            store_dict["options"] = store.options
        return store_dict

    def test_serialize_without_options(self):
        store = Store.for_download(format=StorageFormat.GEOJSON)
        d = self._serialize_store(store)
        assert d == {
            "format": "geojson",
            "single": "true",
            "generate_presigned_url": "true",
        }
        assert "options" not in d

    def test_serialize_with_options(self):
        store = Store.for_download(
            format=StorageFormat.CSV,
            options={"header": "false", "delimiter": "|"},
        )
        d = self._serialize_store(store)
        assert d == {
            "format": "csv",
            "single": "true",
            "generate_presigned_url": "true",
            "options": {"header": "false", "delimiter": "|"},
        }

    def test_serialize_empty_options_omitted(self):
        store = Store(format=StorageFormat.PARQUET, options={})
        d = self._serialize_store(store)
        assert "options" not in d

    def test_json_roundtrip_with_options(self):
        store = Store.for_download(
            format=StorageFormat.GEOJSON,
            options={"ignoreNullFields": "false"},
        )
        d = self._serialize_store(store)
        payload = json.dumps(d)
        parsed = json.loads(payload)
        assert parsed["options"] == {"ignoreNullFields": "false"}

    def test_full_request_shape(self):
        """Verify the full execute_sql request dict shape with store options."""
        store = Store.for_download(
            format=StorageFormat.CSV,
            options={"header": "false"},
        )
        request = {
            "kind": "execute_sql",
            "execution_id": "test-id",
            "statement": "SELECT 1",
        }
        store_dict = self._serialize_store(store)
        request["store"] = store_dict

        assert request == {
            "kind": "execute_sql",
            "execution_id": "test-id",
            "statement": "SELECT 1",
            "store": {
                "format": "csv",
                "single": "true",
                "generate_presigned_url": "true",
                "options": {"header": "false"},
            },
        }