-
Notifications
You must be signed in to change notification settings - Fork 7
/
test_dataframe_base.py
276 lines (217 loc) · 7.8 KB
/
test_dataframe_base.py
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
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
# pylint: disable=redefined-outer-name,protected-access
# pylint: disable=missing-function-docstring,missing-module-docstring,missing-class-docstring
# pylint: disable=global-statement
import pandas as pd
import param
import pytest
from bokeh.models.sources import ColumnDataSource
from awesome_panel_extensions.widgets.dataframe_base import (
DataFrameWithStreamAndPatchBaseWidget as DFWidget,
)
VALUE_CHANGED_COUNT = 0
# region value
@pytest.fixture
def data():
return {"x": [1, 2, 3, 4], "y": ["a", "b", "c", "d"], "z": [True, False, True, False]}
@pytest.fixture
def dataframe(data):
return pd.DataFrame(data)
def test_constructor(dataframe):
# When
component = DFWidget(value=dataframe)
# Then
assert component.value is dataframe
assert isinstance(component._source, ColumnDataSource)
pd.testing.assert_frame_equal(component._source.to_df(), dataframe.reset_index())
def test_constructor_no_value():
# When
component = DFWidget()
# Then
assert isinstance(component._source, ColumnDataSource)
def test_change_value(dataframe):
# Given
component = DFWidget()
# When
component.value = dataframe
# Then
pd.testing.assert_frame_equal(component._source.to_df(), dataframe.reset_index())
# endregion value
# region stream
def test_stream_dataframe_dataframe_value():
# Given
value = pd.DataFrame({"x": [1, 2], "y": ["a", "b"]})
tabulator = DFWidget(value=value)
stream_value = pd.DataFrame({"x": [3, 4], "y": ["c", "d"]})
# Used to test that value event is triggered
global VALUE_CHANGED_COUNT
VALUE_CHANGED_COUNT = 0
@param.depends(tabulator.param.value, watch=True)
def _inc(*_):
global VALUE_CHANGED_COUNT
VALUE_CHANGED_COUNT += 1
# When
tabulator.stream(stream_value)
# Then
tabulator_source_df = tabulator._source.to_df().drop(columns=["index"])
expected = pd.DataFrame({"x": [1, 2, 3, 4], "y": ["a", "b", "c", "d"]})
pd.testing.assert_frame_equal(tabulator.value, expected)
pd.testing.assert_frame_equal(tabulator_source_df, expected)
assert VALUE_CHANGED_COUNT == 1
def test_stream_dataframe_series_value():
# Given
value = pd.DataFrame({"x": [1, 2], "y": ["a", "b"]})
tabulator = DFWidget(value=value)
stream_value = pd.DataFrame({"x": [3, 4], "y": ["c", "d"]}).loc[1]
# Used to test that value event is triggered
global VALUE_CHANGED_COUNT
VALUE_CHANGED_COUNT = 0
@param.depends(tabulator.param.value, watch=True)
def _inc(*_):
global VALUE_CHANGED_COUNT
VALUE_CHANGED_COUNT += 1
# When
tabulator.stream(stream_value)
# Then
tabulator_source_df = tabulator._source.to_df().drop(columns=["index"])
expected = pd.DataFrame({"x": [1, 2, 4], "y": ["a", "b", "d"]})
pd.testing.assert_frame_equal(tabulator.value, expected)
pd.testing.assert_frame_equal(
tabulator_source_df, expected, check_column_type=False, check_dtype=False
)
assert VALUE_CHANGED_COUNT == 1
def test_stream_dataframe_dictionary_value_multi():
# Given
value = pd.DataFrame({"x": [1, 2], "y": ["a", "b"]})
tabulator = DFWidget(value=value)
stream_value = {"x": [3, 4], "y": ["c", "d"]}
# Used to test that value event is triggered
global VALUE_CHANGED_COUNT
VALUE_CHANGED_COUNT = 0
@param.depends(tabulator.param.value, watch=True)
def _inc(*_):
global VALUE_CHANGED_COUNT
VALUE_CHANGED_COUNT += 1
# When PROVIDING A DICTIONARY OF COLUMNS
tabulator.stream(stream_value)
# Then
tabulator_source_df = tabulator._source.to_df().drop(columns=["index"])
expected = pd.DataFrame({"x": [1, 2, 3, 4], "y": ["a", "b", "c", "d"]})
pd.testing.assert_frame_equal(tabulator.value, expected)
pd.testing.assert_frame_equal(
tabulator_source_df, expected, check_column_type=False, check_dtype=False
)
assert VALUE_CHANGED_COUNT == 1
def test_stream_dataframe_dictionary_value_single():
# Given
value = pd.DataFrame({"x": [1, 2], "y": ["a", "b"]})
tabulator = DFWidget(value=value)
stream_value = {"x": 4, "y": "d"}
# Used to test that value event is triggered
global VALUE_CHANGED_COUNT
VALUE_CHANGED_COUNT = 0
@param.depends(tabulator.param.value, watch=True)
def _inc(*_):
global VALUE_CHANGED_COUNT
VALUE_CHANGED_COUNT += 1
# When PROVIDING A DICTIONARY ROW
tabulator.stream(stream_value)
# Then
tabulator_source_df = tabulator._source.to_df().drop(columns=["index"])
expected = pd.DataFrame({"x": [1, 2, 4], "y": ["a", "b", "d"]})
pd.testing.assert_frame_equal(tabulator.value, expected)
pd.testing.assert_frame_equal(
tabulator_source_df, expected, check_column_type=False, check_dtype=False
)
assert VALUE_CHANGED_COUNT == 1
# endregion Stream
# region Patch
def test_patch_dataframe_dataframe_value():
# Given
value = pd.DataFrame({"x": [1, 2], "y": ["a", "b"]})
tabulator = DFWidget(value=value)
patch_value = pd.DataFrame({"x": [3, 4], "y": ["c", "d"]})
# Used to test that value event is triggered
global VALUE_CHANGED_COUNT
VALUE_CHANGED_COUNT = 0
@param.depends(tabulator.param.value, watch=True)
def _inc(*_):
global VALUE_CHANGED_COUNT
VALUE_CHANGED_COUNT += 1
# When
tabulator.patch(patch_value)
# Then
tabulator_source_df = tabulator._source.to_df().drop(columns=["index"])
expected = pd.DataFrame({"x": [3, 4], "y": ["c", "d"]})
pd.testing.assert_frame_equal(tabulator.value, expected)
pd.testing.assert_frame_equal(tabulator_source_df, expected)
assert VALUE_CHANGED_COUNT == 1
# endregion Patch
def test_patch_from_partial_dataframe():
data = pd.DataFrame({"x": [1, 2, 3, 4], "y": ["a", "b", "c", "d"]})
data1 = data.loc[
0:1,
]
data2 = data.loc[2:4]
# When
tabulator = DFWidget(value=data1)
tabulator.value = data2.reset_index(drop=True)
patch_value = tabulator.value["x"] + 2
tabulator.patch(patch_value)
# Then
expected = pd.DataFrame({"x": [5, 6], "y": ["c", "d"]})
pd.testing.assert_frame_equal(tabulator.value, expected)
def test_range_index_of_dataframe_value():
# Given
data = pd.DataFrame({"x": [1, 2, 3, 4], "y": ["a", "b", "c", "d"]})
data2 = data.loc[2:4]
# When
with pytest.raises(ValueError) as error:
DFWidget(value=data2)
assert str(error.value) == (
"Please provide a DataFrame with RangeIndex starting at 0 and with step 1"
)
def test_patch_and_reset():
"""I experienced some strange behaviour which I test below.
The code actually worked as it should. The problem was that I patched the original
data so I could never "reset" back to the original data
"""
# Given
data = pd.DataFrame({"x": [1, 2, 3, 4], "y": ["a", "b", "c", "d"]})
data_copy = data.copy(deep=True)
tabulator = DFWidget(value=data_copy)
patch = tabulator.value["x"] + 2
# When patch Then
tabulator.patch(patch_value=patch)
assert set(tabulator._source.data["x"]) == {3, 4, 5, 6}
# When reset Then
tabulator.value = data
assert set(tabulator._source.data["x"]) == {1, 2, 3, 4}
def test_replace_stream_and_reset():
# Given
data = pd.DataFrame({"x": [1, 2, 3, 4, 5], "y": ["a", "b", "c", "d", "e"]})
data1 = (
data.copy(deep=True)
.loc[
0:1,
]
.reset_index(drop=True)
)
data2 = (
data.copy(deep=True)
.loc[
2:3,
]
.reset_index(drop=True)
)
data3 = data.copy(deep=True).loc[
4:4,
]
tabulator = DFWidget(value=data1)
# When replace, stream and reset
tabulator.value = data2
tabulator.stream(stream_value=data3)
tabulator.value = data.copy(deep=True).loc[
0:1,
]
# Then
assert set(tabulator._source.data["x"]) == {1, 2}