1
1
"""Test module for classes in pandas.api.typing."""
2
2
3
+ from pathlib import Path
4
+
3
5
import numpy as np
4
6
import pandas as pd
7
+ from pandas ._testing import ensure_clean
5
8
from pandas .api .typing import (
6
9
DataFrameGroupBy ,
7
10
DatetimeIndexResamplerGroupby ,
11
+ Expanding ,
12
+ ExpandingGroupby ,
13
+ ExponentialMovingWindow ,
14
+ ExponentialMovingWindowGroupby ,
15
+ JsonReader ,
16
+ NaTType ,
8
17
NAType ,
18
+ PeriodIndexResamplerGroupby ,
19
+ Resampler ,
20
+ Rolling ,
21
+ RollingGroupby ,
9
22
SeriesGroupBy ,
23
+ TimedeltaIndexResamplerGroupby ,
24
+ TimeGrouper ,
25
+ Window ,
10
26
)
27
+ import pytest
11
28
from typing_extensions import assert_type
12
29
13
30
from tests import check
14
31
32
+ from pandas .io .json ._json import read_json
33
+ from pandas .io .stata import StataReader
34
+
15
35
16
36
def test_dataframegroupby ():
17
37
df = pd .DataFrame ({"a" : [1 , 2 , 3 ]})
18
- check (
19
- assert_type (df .groupby ("a" ), DataFrameGroupBy ),
20
- DataFrameGroupBy ,
21
- )
38
+ group = df .groupby ("a" )
39
+
40
+ def f1 (gb : DataFrameGroupBy ):
41
+ check (gb , DataFrameGroupBy )
42
+
43
+ f1 (group )
22
44
23
45
24
46
def test_seriesgroupby ():
25
- sr : pd .Series [int ] = pd .Series ([1 , 2 , 3 ], index = pd .Index (["a" , "b" , "a" ]))
26
- check (
27
- assert_type (sr .groupby (level = 0 ), SeriesGroupBy ),
28
- SeriesGroupBy ,
29
- )
47
+ sr = pd .Series ([1 , 2 , 3 ], index = pd .Index (["a" , "b" , "a" ]))
48
+
49
+ def f1 (gb : SeriesGroupBy ):
50
+ check (gb , SeriesGroupBy )
51
+
52
+ f1 (sr .groupby (level = 0 ))
30
53
31
54
32
55
def tests_datetimeindexersamplergroupby () -> None :
@@ -35,13 +58,100 @@ def tests_datetimeindexersamplergroupby() -> None:
35
58
np .random .standard_normal ((365 , 2 )), index = idx , columns = ["col1" , "col2" ]
36
59
)
37
60
gb_df = df .groupby ("col2" )
38
- check (
39
- assert_type (gb_df .resample ("ME" ), DatetimeIndexResamplerGroupby ),
40
- DatetimeIndexResamplerGroupby ,
41
- pd .DataFrame ,
61
+ check (gb_df .resample ("ME" ), DatetimeIndexResamplerGroupby )
62
+
63
+
64
+ def test_timedeltaindexresamplergroupby () -> None :
65
+ idx = pd .TimedeltaIndex (["0 days" , "1 days" , "2 days" , "3 days" , "4 days" ])
66
+ df = pd .DataFrame (
67
+ np .random .standard_normal ((5 , 2 )), index = idx , columns = ["col1" , "col2" ]
42
68
)
69
+ gb_df = df .groupby ("col2" )
70
+ check (gb_df .resample ("1D" ), TimedeltaIndexResamplerGroupby )
43
71
44
72
45
73
def test_natype () -> None :
46
74
i64dt = pd .Int64Dtype ()
47
75
check (assert_type (i64dt .na_value , NAType ), NAType )
76
+
77
+
78
+ def test_nattype () -> None :
79
+ td = pd .Timedelta ("1 day" )
80
+ as_nat = pd .NaT
81
+ check (assert_type (td + as_nat , NaTType ), NaTType )
82
+
83
+
84
+ def test_expanding () -> None :
85
+ df = pd .DataFrame ({"B" : [0 , 1 , 2 , np .nan , 4 ]})
86
+ check (df .expanding (), Expanding )
87
+
88
+
89
+ def test_expanding_groubpy () -> None :
90
+ df = pd .DataFrame ({"B" : [0 , 1 , 2 , np .nan , 4 ]})
91
+ check (df .groupby ("B" ).expanding (), ExpandingGroupby )
92
+
93
+
94
+ def test_ewm () -> None :
95
+ df = pd .DataFrame ({"B" : [0 , 1 , 2 , np .nan , 4 ]})
96
+ check (df .ewm (2 ), ExponentialMovingWindow )
97
+
98
+
99
+ def test_ewm_groubpy () -> None :
100
+ df = pd .DataFrame ({"B" : [0 , 1 , 2 , np .nan , 4 ]})
101
+ check (df .groupby ("B" ).ewm (2 ), ExponentialMovingWindowGroupby )
102
+
103
+
104
+ def test_json_reader () -> None :
105
+ df = pd .DataFrame ({"B" : [0 , 1 , 2 , np .nan , 4 ]})
106
+
107
+ with ensure_clean () as path :
108
+ check (assert_type (df .to_json (path ), None ), type (None ))
109
+ json_reader = read_json (path , chunksize = 1 , lines = True )
110
+ check (json_reader , JsonReader )
111
+ json_reader .close ()
112
+
113
+
114
+ @pytest .mark .skip ("Resampling with a PeriodIndex is deprecated." )
115
+ def test_periodindexresamplergroupby () -> None :
116
+ idx = pd .period_range ("2020-01-28 09:00" , periods = 4 , freq = "D" )
117
+ df = pd .DataFrame (data = 4 * [range (2 )], index = idx , columns = ["a" , "b" ])
118
+ check (
119
+ df .groupby ("a" ).resample ("3min" ),
120
+ PeriodIndexResamplerGroupby ,
121
+ )
122
+
123
+
124
+ def test_resampler () -> None :
125
+ s = pd .Series ([1 , 2 , 3 , 4 , 5 ], index = pd .date_range ("20130101" , periods = 5 , freq = "s" ))
126
+ check (s .resample ("3min" ), Resampler )
127
+
128
+
129
+ def test_rolling () -> None :
130
+ df = pd .DataFrame ({"B" : [0 , 1 , 2 , np .nan , 4 ]})
131
+ check (df .rolling (2 ), Rolling )
132
+
133
+
134
+ def test_rolling_groupby () -> None :
135
+ df = pd .DataFrame ({"B" : [0 , 1 , 2 , np .nan , 4 ]})
136
+ check (df .groupby ("B" ).rolling (2 ), RollingGroupby )
137
+
138
+
139
+ def test_timegrouper () -> None :
140
+ check (pd .Grouper (key = "Publish date" , freq = "1W" ), TimeGrouper )
141
+
142
+
143
+ def test_window () -> None :
144
+ ser = pd .Series ([0 , 1 , 5 , 2 , 8 ])
145
+ check (ser .rolling (2 , win_type = "gaussian" ), Window )
146
+
147
+
148
+ def test_statereader (tmp_path : Path ) -> None :
149
+ df = pd .DataFrame ([[1 , 2 ], [3 , 4 ]], columns = ["col_1" , "col_2" ])
150
+ time_stamp = pd .Timestamp (2000 , 2 , 29 , 14 , 21 )
151
+ variable_labels = {"col_1" : "This is an example" }
152
+ path = tmp_path / "file"
153
+ df .to_stata (
154
+ path , time_stamp = time_stamp , variable_labels = variable_labels , version = None
155
+ )
156
+ with StataReader (path ) as reader :
157
+ check (reader , StataReader )
0 commit comments