1+ import argparse
2+ import os
3+ import sys
4+ import logging
5+ import tempfile
6+
7+ from typing import Optional , List
8+ from subprocess import CalledProcessError , check_call
9+
10+ from .Check import Check
11+ from ci_tools .parsing import ParsedSetup
12+ from ci_tools .functions import discover_targeted_packages
13+ from ci_tools .scenario .generation import create_package_and_install
14+ from ci_tools .variables import in_ci , set_envvar_defaults
15+ from ci_tools .environment_exclusions import (
16+ is_check_enabled , is_typing_ignored
17+ )
18+
19+ logging .getLogger ().setLevel (logging .INFO )
20+
21+ PYTHON_VERSION = "3.9"
22+ MYPY_VERSION = "1.14.1"
23+
24+ class mypy (Check ):
25+ def __init__ (self ) -> None :
26+ super ().__init__ ()
27+
28+ def register (self , subparsers : "argparse._SubParsersAction" , parent_parsers : Optional [List [argparse .ArgumentParser ]] = None ) -> None :
29+ """Register the `mypy` check. The mypy check installs mypy and runs mypy against the target package.
30+ """
31+ parents = parent_parsers or []
32+ p = subparsers .add_parser ("mypy" , parents = parents , help = "Run the mypy check" )
33+ p .set_defaults (func = self .run )
34+
35+ p .add_argument (
36+ "--next" ,
37+ default = False ,
38+ help = "Next version of mypy is being tested" ,
39+ required = False
40+ )
41+
42+ def run (self , args : argparse .Namespace ) -> int :
43+ """Run the mypy check command."""
44+ print ("Running mypy check in isolated venv..." )
45+
46+ set_envvar_defaults ()
47+
48+ targeted = self .get_targeted_directories (args )
49+
50+ results : List [int ] = []
51+
52+ for parsed in targeted :
53+ package_dir = parsed .folder
54+ package_name = parsed .name
55+ print (f"Processing { package_name } )
56+
57+ staging_area = tempfile .mkdtemp ()
58+ create_package_and_install (
59+ distribution_directory = staging_area ,
60+ target_setup = package_dir ,
61+ skip_install = False ,
62+ cache_dir = None ,
63+ work_dir = staging_area ,
64+ force_create = False ,
65+ package_type = "wheel" ,
66+ pre_download_disabled = False ,
67+ )
68+
69+ # install mypy
70+ try :
71+ if (args .next ):
72+ # use latest version of mypy
73+ check_call ([sys .executable , "-m" , "pip" , "install" , "mypy" ])
74+ else :
75+ check_call ([sys .executable , "-m" , "pip" , "install" , f"mypy=={ MYPY_VERSION } ])
76+ except CalledProcessError as e :
77+ print ("Failed to install mypy:" , e )
78+ return e .returncode
79+
80+ logging .info (f"Running mypy against { package_name } )
81+
82+ if not args .next and in_ci ():
83+ if not is_check_enabled (package_dir , "mypy" , True ) or is_typing_ignored (package_name ):
84+ logging .info (
85+ f"Package { package_name }
86+ )
87+ continue
88+
89+ top_level_module = parsed .namespace .split ("." )[0 ]
90+
91+ commands = [
92+ sys .executable ,
93+ "-m" ,
94+ "mypy" ,
95+ "--python-version" ,
96+ PYTHON_VERSION ,
97+ "--show-error-codes" ,
98+ "--ignore-missing-imports" ,
99+ ]
100+ src_code = [* commands , os .path .join (package_dir , top_level_module )]
101+ src_code_error = None
102+ sample_code_error = None
103+ try :
104+ logging .info (
105+ f"Running mypy commands on src code: { src_code }
106+ )
107+ results .append (check_call (src_code ))
108+ logging .info ("Verified mypy, no issues found" )
109+ except CalledProcessError as src_error :
110+ src_code_error = src_error
111+ results .append (src_error .returncode )
112+
113+ if not args .next and in_ci () and not is_check_enabled (package_dir , "type_check_samples" , True ):
114+ logging .info (
115+ f"Package { package_name }
116+ )
117+ continue
118+ else :
119+ # check if sample dirs exists, if not, skip sample code check
120+ samples = os .path .exists (os .path .join (package_dir , "samples" ))
121+ generated_samples = os .path .exists (os .path .join (package_dir , "generated_samples" ))
122+ if not samples and not generated_samples :
123+ logging .info (
124+ f"Package { package_name }
125+ )
126+ else :
127+ sample_code = [
128+ * commands ,
129+ "--check-untyped-defs" ,
130+ "--follow-imports=silent" ,
131+ os .path .join (package_dir , "samples" if samples else "generated_samples" ),
132+ ]
133+ try :
134+ logging .info (
135+ f"Running mypy commands on sample code: { sample_code }
136+ )
137+ results .append (check_call (sample_code ))
138+ except CalledProcessError as sample_error :
139+ sample_code_error = sample_error
140+ results .append (sample_error .returncode )
141+
142+ if args .next and in_ci () and not is_typing_ignored (package_name ):
143+ from gh_tools .vnext_issue_creator import create_vnext_issue , close_vnext_issue
144+ if src_code_error or sample_code_error :
145+ create_vnext_issue (package_dir , "mypy" )
146+ else :
147+ close_vnext_issue (package_name , "mypy" )
148+
149+ return max (results ) if results else 0
150+
0 commit comments