10
10
import tempfile
11
11
import venv
12
12
from pathlib import Path
13
- from typing import NoReturn
13
+ from typing import TYPE_CHECKING , NoReturn
14
14
15
15
import tomli
16
16
17
+ if TYPE_CHECKING :
18
+
19
+ def colored (__str : str , __style : str ) -> str :
20
+ ...
21
+
22
+ else :
23
+ try :
24
+ from termcolor import colored
25
+ except ImportError :
26
+
27
+ def colored (s : str , _ : str ) -> str :
28
+ return s
29
+
17
30
18
31
@functools .lru_cache ()
19
32
def get_mypy_req () -> str :
20
33
with open ("requirements-tests.txt" ) as f :
21
34
return next (line .strip () for line in f if "mypy" in line )
22
35
23
36
24
- def run_stubtest (dist : Path ) -> bool :
37
+ def run_stubtest (dist : Path , * , verbose : bool = False ) -> bool :
25
38
with open (dist / "METADATA.toml" ) as f :
26
39
metadata = dict (tomli .loads (f .read ()))
27
40
41
+ print (f"{ dist .name } ..." , end = "" )
42
+
28
43
if not metadata .get ("stubtest" , True ):
29
- print (f"Skipping stubtest for { dist .name } \n \n " )
44
+ print (colored (" skipping" , "yellow" ))
45
+ print ()
30
46
return True
31
47
32
48
with tempfile .TemporaryDirectory () as tmp :
@@ -47,9 +63,7 @@ def run_stubtest(dist: Path) -> bool:
47
63
pip_cmd = [pip_exe , "install" , "-r" , str (req_path )]
48
64
subprocess .run (pip_cmd , check = True , capture_output = True )
49
65
except subprocess .CalledProcessError as e :
50
- print (f"Failed to install requirements for { dist .name } " , file = sys .stderr )
51
- print (e .stdout .decode (), file = sys .stderr )
52
- print (e .stderr .decode (), file = sys .stderr )
66
+ print_command_failure ("Failed to install requirements" , e )
53
67
return False
54
68
55
69
# We need stubtest to be able to import the package, so install mypy into the venv
@@ -58,18 +72,15 @@ def run_stubtest(dist: Path) -> bool:
58
72
dists_to_install = [dist_req , get_mypy_req ()]
59
73
dists_to_install .extend (metadata .get ("requires" , []))
60
74
pip_cmd = [pip_exe , "install" ] + dists_to_install
61
- print (" " .join (pip_cmd ), file = sys .stderr )
62
75
try :
63
76
subprocess .run (pip_cmd , check = True , capture_output = True )
64
77
except subprocess .CalledProcessError as e :
65
- print (f"Failed to install { dist .name } " , file = sys .stderr )
66
- print (e .stdout .decode (), file = sys .stderr )
67
- print (e .stderr .decode (), file = sys .stderr )
78
+ print_command_failure ("Failed to install" , e )
68
79
return False
69
80
70
81
packages_to_check = [d .name for d in dist .iterdir () if d .is_dir () and d .name .isidentifier ()]
71
82
modules_to_check = [d .stem for d in dist .iterdir () if d .is_file () and d .suffix == ".pyi" ]
72
- cmd = [
83
+ stubtest_cmd = [
73
84
python_exe ,
74
85
"-m" ,
75
86
"mypy.stubtest" ,
@@ -85,35 +96,62 @@ def run_stubtest(dist: Path) -> bool:
85
96
]
86
97
allowlist_path = dist / "@tests/stubtest_allowlist.txt"
87
98
if allowlist_path .exists ():
88
- cmd .extend (["--allowlist" , str (allowlist_path )])
99
+ stubtest_cmd .extend (["--allowlist" , str (allowlist_path )])
89
100
90
101
try :
91
- print ( f"MYPYPATH= { dist } " , " " . join ( cmd ), file = sys . stderr )
92
- subprocess .run ( cmd , env = { "MYPYPATH" : str ( dist ), "MYPY_FORCE_COLOR" : "1" }, check = True )
93
- except subprocess . CalledProcessError :
94
- print ( f"stubtest failed for { dist . name } " , file = sys . stderr )
95
- print ( " \n \n " , file = sys . stderr )
102
+ subprocess . run ( stubtest_cmd , env = { "MYPYPATH" : str ( dist ) , "MYPY_FORCE_COLOR" : "1" }, check = True , capture_output = True )
103
+ except subprocess .CalledProcessError as e :
104
+ print ( colored ( " fail" , "red" ))
105
+ print_commands ( dist , pip_cmd , stubtest_cmd )
106
+ print_command_output ( e )
96
107
97
108
print ("Ran with the following environment:" , file = sys .stderr )
98
- subprocess .run ([pip_exe , "freeze" ])
109
+ ret = subprocess .run ([pip_exe , "freeze" ], capture_output = True )
110
+ print_command_output (ret )
99
111
100
112
if allowlist_path .exists ():
101
113
print (
102
114
f'To fix "unused allowlist" errors, remove the corresponding entries from { allowlist_path } ' , file = sys .stderr
103
115
)
116
+ print (file = sys .stderr )
104
117
else :
105
118
print (f"Re-running stubtest with --generate-allowlist.\n Add the following to { allowlist_path } :" , file = sys .stderr )
106
- subprocess .run (cmd + ["--generate-allowlist" ], env = {"MYPYPATH" : str (dist )})
107
- print ("\n \n " , file = sys .stderr )
119
+ ret = subprocess .run (stubtest_cmd + ["--generate-allowlist" ], env = {"MYPYPATH" : str (dist )}, capture_output = True )
120
+ print_command_output (ret )
121
+
108
122
return False
109
123
else :
110
- print (f"stubtest succeeded for { dist .name } " , file = sys .stderr )
111
- print ("\n \n " , file = sys .stderr )
124
+ print (colored (" success" , "green" ))
125
+
126
+ if verbose :
127
+ print_commands (dist , pip_cmd , stubtest_cmd )
128
+
112
129
return True
113
130
114
131
132
+ def print_commands (dist : Path , pip_cmd : list [str ], stubtest_cmd : list [str ]) -> None :
133
+ print (file = sys .stderr )
134
+ print (" " .join (pip_cmd ), file = sys .stderr )
135
+ print (f"MYPYPATH={ dist } " , " " .join (stubtest_cmd ), file = sys .stderr )
136
+ print (file = sys .stderr )
137
+
138
+
139
+ def print_command_failure (message : str , e : subprocess .CalledProcessError ) -> None :
140
+ print (colored (" fail" , "red" ))
141
+ print (file = sys .stderr )
142
+ print (message , file = sys .stderr )
143
+ print_command_output (e )
144
+
145
+
146
+ def print_command_output (e : subprocess .CalledProcessError | subprocess .CompletedProcess [bytes ]) -> None :
147
+ print (e .stdout .decode (), end = "" , file = sys .stderr )
148
+ print (e .stderr .decode (), end = "" , file = sys .stderr )
149
+ print (file = sys .stderr )
150
+
151
+
115
152
def main () -> NoReturn :
116
153
parser = argparse .ArgumentParser ()
154
+ parser .add_argument ("-v" , "--verbose" , action = "store_true" , help = "verbose output" )
117
155
parser .add_argument ("--num-shards" , type = int , default = 1 )
118
156
parser .add_argument ("--shard-index" , type = int , default = 0 )
119
157
parser .add_argument ("dists" , metavar = "DISTRIBUTION" , type = str , nargs = argparse .ZERO_OR_MORE )
@@ -129,10 +167,13 @@ def main() -> NoReturn:
129
167
for i , dist in enumerate (dists ):
130
168
if i % args .num_shards != args .shard_index :
131
169
continue
132
- if not run_stubtest (dist ):
170
+ if not run_stubtest (dist , verbose = args . verbose ):
133
171
result = 1
134
172
sys .exit (result )
135
173
136
174
137
175
if __name__ == "__main__" :
138
- main ()
176
+ try :
177
+ main ()
178
+ except KeyboardInterrupt :
179
+ pass
0 commit comments