Make tests runnable on Windows and add appveyor.yaml

appveyor.yml

@@ -0,0 +1,19 @@
+environment:
+  matrix:
+  - PYTHON: C:\\Python33
+  - PYTHON: C:\\Python34
+  - PYTHON: C:\\Python35
+  - PYTHON: C:\\Python33-x64
+  - PYTHON: C:\\Python34-x64
+  - PYTHON: C:\\Python35-x64
+install:
+  - "git submodule update --init typeshed"
+  - "SET PATH=%PYTHON%;%PYTHON%\\Scripts;C:\\Python27;%PATH%"
+  - "REN C:\\Python27\\python.exe python2.exe"
+  - "python --version"
+  - "python2 --version"
+build_script:
+  - "pip install -r test-requirements.txt"
+  - "python setup.py install"
+test_script:
+- cmd: python runtests.py -v

mypy/build.py

@@ -832,10 +832,14 @@ def write_cache(id: str, path: str, tree: MypyFile,
     with open(meta_json_tmp, 'w') as f:
         json.dump(meta, f, sort_keys=True)
         f.write('\n')
-    # TODO: On Windows, os.rename() may not be atomic, and we could
-    # use os.replace().  However that's new in Python 3.3.
-    os.rename(data_json_tmp, data_json)
-    os.rename(meta_json_tmp, meta_json)
+    # TODO: This is a temporary change until Python 3.2 support is dropped, see #1504
+    # os.rename will raise an exception rather than replace files on Windows
+    try:
+        replace = os.replace
+    except AttributeError:
+        replace = os.rename
+    replace(data_json_tmp, data_json)
+    replace(meta_json_tmp, meta_json)
 
 
 """Dependency manager.

mypy/test/data.py

@@ -16,7 +16,8 @@ def parse_test_cases(
         perform: Callable[['DataDrivenTestCase'], None],
         base_path: str = '.',
         optional_out: bool = False,
-        include_path: str = None) -> List['DataDrivenTestCase']:
+        include_path: str = None,
+        native_sep: bool = False) -> List['DataDrivenTestCase']:
     """Parse a file with test case descriptions.
 
     Return an array of test cases.
@@ -65,6 +66,8 @@ def parse_test_cases(
             tcout = []  # type: List[str]
             if i < len(p) and p[i].id == 'out':
                 tcout = p[i].data
+                if native_sep and os.path.sep == '\\':
+                    tcout = [fix_win_path(line) for line in tcout]
                 ok = True
                 i += 1
             elif optional_out:
@@ -291,7 +294,7 @@ def expand_includes(a: List[str], base_path: str) -> List[str]:
     return res
 
 
-def expand_errors(input, output, fnam):
+def expand_errors(input: List[str], output: List[str], fnam: str) -> None:
     """Transform comments such as '# E: message' in input.
 
     The result is lines like 'fnam:line: error: message'.
@@ -302,3 +305,17 @@ def expand_errors(input, output, fnam):
         if m:
             severity = 'error' if m.group(1) == 'E' else 'note'
             output.append('{}:{}: {}: {}'.format(fnam, i + 1, severity, m.group(2)))
+
+
+def fix_win_path(line: str) -> str:
+    r"""Changes paths to Windows paths in error messages.
+
+    E.g. foo/bar.py -> foo\bar.py.
+    """
+    m = re.match(r'^([\S/]+):(\d+:)?(\s+.*)', line)
+    if not m:
+        return line
+    else:
+        filename, lineno, message = m.groups()
+        return '{}:{}{}'.format(filename.replace('/', '\\'),
+                                lineno or '', message)

mypy/test/data/pythoneval.test

@@ -16,7 +16,7 @@ import re
 from typing import Sized, Sequence, Iterator, Iterable, Mapping, AbstractSet
 
 def check(o, t):
-    rep = re.sub('0x[0-9a-f]+', '0x...', repr(o))
+    rep = re.sub('0x[0-9a-fA-F]+', '0x...', repr(o))
     rep = rep.replace('sequenceiterator', 'str_iterator')
     trep = str(t).replace('_abcoll.Sized', 'collections.abc.Sized')
     print(rep, trep, isinstance(o, t))

mypy/test/testcheck.py

@@ -153,7 +153,8 @@ def find_error_paths(self, a: List[str]) -> Set[str]:
         for line in a:
             m = re.match(r'([^\s:]+):\d+: error:', line)
             if m:
-                hits.add(m.group(1))
+                p = m.group(1).replace('/', os.path.sep)
+                hits.add(p)
         return hits
 
     def find_module_files(self) -> Dict[str, str]:

mypy/test/testcmdline.py

@@ -29,7 +29,10 @@ def cases(self) -> List[DataDrivenTestCase]:
         c = []  # type: List[DataDrivenTestCase]
         for f in cmdline_files:
             c += parse_test_cases(os.path.join(test_data_prefix, f),
-                                  test_python_evaluation, test_temp_dir, True)
+                                  test_python_evaluation,
+                                  base_path=test_temp_dir,
+                                  optional_out=True,
+                                  native_sep=True)
         return c
 
 

mypy/test/testsemanal.py

@@ -36,7 +36,10 @@ def cases(self):
         c = []
         for f in semanal_files:
             c += parse_test_cases(os.path.join(test_data_prefix, f),
-                                  test_semanal, test_temp_dir, optional_out=True)
+                                  test_semanal,
+                                  base_path=test_temp_dir,
+                                  optional_out=True,
+                                  native_sep=True)
         return c
 
 

mypy/test/teststubgen.py

@@ -108,34 +108,35 @@ def test_stubgen(testcase):
         sys.path.insert(0, 'stubgen-test-path')
     os.mkdir('stubgen-test-path')
     source = '\n'.join(testcase.input)
-    handle = tempfile.NamedTemporaryFile(prefix='prog_', suffix='.py', dir='stubgen-test-path')
+    handle = tempfile.NamedTemporaryFile(prefix='prog_', suffix='.py', dir='stubgen-test-path',
+                                         delete=False)
     assert os.path.isabs(handle.name)
     path = os.path.basename(handle.name)
     name = path[:-3]
     path = os.path.join('stubgen-test-path', path)
     out_dir = '_out'
     os.mkdir(out_dir)
     try:
-        with open(path, 'w') as file:
-            file.write(source)
-            file.close()
-            # Without this we may sometimes be unable to import the module below, as importlib
-            # caches os.listdir() results in Python 3.3+ (Guido explained this to me).
-            reset_importlib_caches()
-            try:
-                if testcase.name.endswith('_import'):
-                    generate_stub_for_module(name, out_dir, quiet=True)
-                else:
-                    generate_stub(path, out_dir)
-                a = load_output(out_dir)
-            except CompileError as e:
-                a = e.messages
-            assert_string_arrays_equal(testcase.output, a,
-                                       'Invalid output ({}, line {})'.format(
-                                           testcase.file, testcase.line))
+        handle.write(bytes(source, 'ascii'))
+        handle.close()
+        # Without this we may sometimes be unable to import the module below, as importlib
+        # caches os.listdir() results in Python 3.3+ (Guido explained this to me).
+        reset_importlib_caches()
+        try:
+            if testcase.name.endswith('_import'):
+                generate_stub_for_module(name, out_dir, quiet=True)
+            else:
+                generate_stub(path, out_dir)
+            a = load_output(out_dir)
+        except CompileError as e:
+            a = e.messages
+        assert_string_arrays_equal(testcase.output, a,
+                                   'Invalid output ({}, line {})'.format(
+                                       testcase.file, testcase.line))
     finally:
-        shutil.rmtree(out_dir)
         handle.close()
+        os.unlink(handle.name)
+        shutil.rmtree(out_dir)
 
 
 def reset_importlib_caches():

mypy/test/testtransform.py

@@ -30,7 +30,9 @@ def cases(self):
         c = []
         for f in self.transform_files:
             c += parse_test_cases(os.path.join(test_data_prefix, f),
-                                  test_transform, test_temp_dir)
+                                  test_transform,
+                                  base_path=test_temp_dir,
+                                  native_sep=True)
         return c
 
 

mypy/waiter.py

@@ -31,42 +31,23 @@ def __init__(self, name: str, args: List[str], *, cwd: str = None,
         self.end_time = None  # type: float
 
     def start(self) -> None:
-        self.outfile = tempfile.NamedTemporaryFile()
+        self.outfile = tempfile.TemporaryFile()
         self.start_time = time.time()
         self.process = Popen(self.args, cwd=self.cwd, env=self.env,
                              stdout=self.outfile, stderr=STDOUT)
         self.pid = self.process.pid
 
-    def handle_exit_status(self, status: int) -> None:
-        """Update process exit status received via an external os.waitpid() call."""
-        # Inlined subprocess._handle_exitstatus, it's not a public API.
-        # TODO(jukka): I'm not quite sure why this is implemented like this.
-        self.end_time = time.time()
-        process = self.process
-        assert process.returncode is None
-        if os.WIFSIGNALED(status):
-            process.returncode = -os.WTERMSIG(status)
-        elif os.WIFEXITED(status):
-            process.returncode = os.WEXITSTATUS(status)
-        else:
-            # Should never happen
-            raise RuntimeError("Unknown child exit status!")
-        assert process.returncode is not None
-
     def wait(self) -> int:
         return self.process.wait()
 
     def status(self) -> Optional[int]:
         return self.process.returncode
 
     def read_output(self) -> str:
-        with open(self.outfile.name, 'rb') as file:
-            # Assume it's ascii to avoid unicode headaches (and portability issues).
-            return file.read().decode('ascii')
-
-    def cleanup(self) -> None:
-        self.outfile.close()
-        assert not os.path.exists(self.outfile.name)
+        file = self.outfile
+        file.seek(0)
+        # Assume it's ascii to avoid unicode headaches (and portability issues).
+        return file.read().decode('ascii')
 
     @property
     def elapsed_time(self) -> float:
@@ -178,17 +159,25 @@ def _record_time(self, name: str, elapsed_time: float) -> None:
         name2 = re.sub('( .*?) .*', r'\1', name)  # First two words.
         self.times2[name2] = elapsed_time + self.times2.get(name2, 0)
 
+    def _poll_current(self) -> Tuple[int, int]:
+        while True:
+            time.sleep(.25)
+            for pid in self.current:
+                cmd = self.current[pid][1]
+                code = cmd.process.poll()
+                if code is not None:
+                    cmd.end_time = time.time()
+                    return pid, code
+
     def _wait_next(self) -> Tuple[List[str], int, int]:
         """Wait for a single task to finish.
 
         Return tuple (list of failed tasks, number test cases, number of failed tests).
         """
-        pid, status = os.waitpid(-1, 0)
+        pid, status = self._poll_current()
         num, cmd = self.current.pop(pid)
         name = cmd.name
 
-        cmd.handle_exit_status(status)
-
         self._record_time(cmd.name, cmd.elapsed_time)
 
         rc = cmd.wait()
@@ -223,7 +212,6 @@ def _wait_next(self) -> Tuple[List[str], int, int]:
 
         # Get task output.
         output = cmd.read_output()
-        cmd.cleanup()
         num_tests, num_tests_failed = parse_test_stats_from_output(output, fail_type)
 
         if fail_type is not None or self.verbosity >= 1: