forked from pandas-dev/pandas
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest_functions.py
83 lines (67 loc) · 2.92 KB
/
test_functions.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
import numpy as np
from pandas import (
DataFrame,
Series,
concat,
)
import pandas._testing as tm
from pandas.tests.copy_view.util import get_array
def test_concat_frames(using_copy_on_write):
df = DataFrame({"b": ["a"] * 3})
df2 = DataFrame({"a": ["a"] * 3})
df_orig = df.copy()
result = concat([df, df2], axis=1)
if using_copy_on_write:
assert np.shares_memory(get_array(result, "b"), get_array(df, "b"))
assert np.shares_memory(get_array(result, "a"), get_array(df2, "a"))
else:
assert not np.shares_memory(get_array(result, "b"), get_array(df, "b"))
assert not np.shares_memory(get_array(result, "a"), get_array(df2, "a"))
result.iloc[0, 0] = "d"
if using_copy_on_write:
assert not np.shares_memory(get_array(result, "b"), get_array(df, "b"))
assert np.shares_memory(get_array(result, "a"), get_array(df2, "a"))
result.iloc[0, 1] = "d"
if using_copy_on_write:
assert not np.shares_memory(get_array(result, "a"), get_array(df2, "a"))
tm.assert_frame_equal(df, df_orig)
def test_concat_frames_updating_input(using_copy_on_write):
df = DataFrame({"b": ["a"] * 3})
df2 = DataFrame({"a": ["a"] * 3})
result = concat([df, df2], axis=1)
if using_copy_on_write:
assert np.shares_memory(get_array(result, "b"), get_array(df, "b"))
assert np.shares_memory(get_array(result, "a"), get_array(df2, "a"))
else:
assert not np.shares_memory(get_array(result, "b"), get_array(df, "b"))
assert not np.shares_memory(get_array(result, "a"), get_array(df2, "a"))
expected = result.copy()
df.iloc[0, 0] = "d"
if using_copy_on_write:
assert not np.shares_memory(get_array(result, "b"), get_array(df, "b"))
assert np.shares_memory(get_array(result, "a"), get_array(df2, "a"))
df2.iloc[0, 0] = "d"
if using_copy_on_write:
assert not np.shares_memory(get_array(result, "a"), get_array(df2, "a"))
tm.assert_frame_equal(result, expected)
def test_concat_series(using_copy_on_write):
ser = Series([1, 2], name="a")
ser2 = Series([3, 4], name="b")
ser_orig = ser.copy()
ser2_orig = ser2.copy()
result = concat([ser, ser2], axis=1)
if using_copy_on_write:
assert np.shares_memory(get_array(result, "a"), ser.values)
assert np.shares_memory(get_array(result, "b"), ser2.values)
else:
assert not np.shares_memory(get_array(result, "a"), ser.values)
assert not np.shares_memory(get_array(result, "b"), ser2.values)
result.iloc[0, 0] = 100
if using_copy_on_write:
assert not np.shares_memory(get_array(result, "a"), ser.values)
assert np.shares_memory(get_array(result, "b"), ser2.values)
result.iloc[0, 1] = 1000
if using_copy_on_write:
assert not np.shares_memory(get_array(result, "b"), ser2.values)
tm.assert_series_equal(ser, ser_orig)
tm.assert_series_equal(ser2, ser2_orig)