Fix smoke tests on Windows.

tests/fake_repos.py

 #!/usr/bin/python
 # Copyright (c) 2010 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 """Generate fake repositories for testing."""
 
 import atexit
+import errno
 import logging
 import os
 import pprint
 import re
-import shutil
+import stat
 import subprocess
 import sys
+import time
 import unittest
 
 
 ## Utility functions
 
 
 def addKill():
   """Add kill() method to subprocess.Popen for python <2.6"""
   if getattr(subprocess.Popen, 'kill', None):
     return
   if sys.platform.startswith('win'):
     def kill_win(process):
       import win32process
       return win32process.TerminateProcess(process._handle, -1)
     subprocess.kill = kill_win
   else:
     def kill_nix(process):
       import signal
       return os.kill(process.pid, signal.SIGKILL)
     subprocess.kill = kill_nix
 
 
-def rmtree(path):
-  """Delete a directory."""
-  if os.path.exists(path):
-    shutil.rmtree(path)
+def rmtree(*path):
+  """Recursively removes a directory, even if it's marked read-only.
+
+  Remove the directory located at *path, if it exists.
+
+  shutil.rmtree() doesn't work on Windows if any of the files or directories
+  are read-only, which svn repositories and some .svn files are.  We need to
+  be able to force the files to be writable (i.e., deletable) as we traverse
+  the tree.
+
+  Even with all this, Windows still sometimes fails to delete a file, citing
+  a permission error (maybe something to do with antivirus scans or disk
+  indexing).  The best suggestion any of the user forums had was to wait a
+  bit and try again, so we do that too.  It's hand-waving, but sometimes it
+  works. :/
+  """
+  file_path = os.path.join(*path)
+  if not os.path.exists(file_path):
+    return
+
+  def RemoveWithRetry_win(rmfunc, path):
+    os.chmod(path, stat.S_IWRITE)
+    if win32_api_avail:
+      win32api.SetFileAttributes(path, win32con.FILE_ATTRIBUTE_NORMAL)
+    try:
+      return rmfunc(path)
+    except EnvironmentError, e:
+      if e.errno != errno.EACCES:
+        raise
+      print 'Failed to delete %s: trying again' % repr(path)
+      time.sleep(0.1)
+      return rmfunc(path)
+
+  def RemoveWithRetry_non_win(rmfunc, path):
+    if os.path.islink(path):
+      return os.remove(path)
+    else:
+      return rmfunc(path)
+
+  win32_api_avail = False
+  remove_with_retry = None
+  if sys.platform.startswith('win'):

[bradn, 2010/06/01 01:12:04]

Pretty sure this is always win32, but anyways. Did you want to include cygwin
too?

[M-A Ruel, 2010/06/01 01:26:27]

You are right, it's just a copy-paste from scripts/common/chromium_utils.py

+    # Some people don't have the APIs installed. In that case we'll do without.
+    try:
+      win32api = __import__('win32api')
+      win32con = __import__('win32con')
+      win32_api_avail = True
+    except ImportError:
+      pass
+    remove_with_retry = RemoveWithRetry_win
+  else:
+    remove_with_retry = RemoveWithRetry_non_win
+
+  for root, dirs, files in os.walk(file_path, topdown=False):
+    # For POSIX:  making the directory writable guarantees removability.
+    # Windows will ignore the non-read-only bits in the chmod value.
+    os.chmod(root, 0770)
+    for name in files:
+      remove_with_retry(os.remove, os.path.join(root, name))
+    for name in dirs:
+      remove_with_retry(os.rmdir, os.path.join(root, name))
+
+  remove_with_retry(os.rmdir, file_path)
 
 
 def write(path, content):
   f = open(path, 'wb')
   f.write(content)
   f.close()
 
 
 join = os.path.join
 
@@ skipping 10 matching lines @@
   return subprocess.Popen(*args, **kwargs)
 
 
 def read_tree(tree_root):
   """Returns a dict of all the files in a tree. Defaults to self.root_dir."""
   tree = {}
   for root, dirs, files in os.walk(tree_root):
     for d in filter(lambda x: x.startswith('.'), dirs):
       dirs.remove(d)
     for f in [join(root, f) for f in files if not f.startswith('.')]:
-      tree[f[len(tree_root) + 1:]] = open(join(root, f), 'rb').read()
+      filepath = f[len(tree_root) + 1:].replace(os.sep, '/')
+      assert len(filepath), f
+      tree[filepath] = open(join(root, f), 'rU').read()
   return tree
 
 
 def dict_diff(dict1, dict2):
   diff = {}
   for k, v in dict1.iteritems():
     if k not in dict2:
       diff[k] = v
     elif v != dict2[k]:
       diff[k] = (v, dict2[k])
   for k, v in dict2.iteritems():
     if k not in dict1:
       diff[k] = v
   return diff
 
 
 def mangle_svn_tree(*args):
   result = {}
   for old_root, new_root, tree in args:
     for k, v in tree.iteritems():
       if not k.startswith(old_root):
         continue
-      result[join(new_root, k[len(old_root) + 1:])] = v
+      result[join(new_root, k[len(old_root) + 1:]).replace(os.sep, '/')] = v
   return result
 
 
 def mangle_git_tree(*args):
   result = {}
   for new_root, tree in args:
     for k, v in tree.iteritems():
       result[join(new_root, k)] = v
   return result
 
@@ skipping 217 matching lines @@
     'pattern': 'nonexistent',
     'action': ['python', '-c',
                'open(\\'src/svn_hooked2\\', \\'w\\').write(\\'svn_hooked2\\')'],
   },
 ]
 """ % { 'host': self.HOST }))
 
   def setUpGIT(self):
     """Creates git repositories and start the servers."""
     if self.gitdaemon:
-      return
+      return True
     self.setUp()
+    if sys.platform == 'win32':
+      return False
     for repo in ['repo_%d' % r for r in range(1, 5)]:
       check_call(['git', 'init', '-q', join(self.git_root, repo)])
       self.git_hashes[repo] = []
 
     # Testing:
     # - dependency disapear
     # - dependency renamed
     # - versioned and unversioned reference
     # - relative and full reference
     # - deps_os
@@ skipping 72 matching lines @@
       },
       'origin': "git/repo_1@2\n"
     })
 
     # Start the daemon.
     cmd = ['git', 'daemon', '--export-all', '--base-path=' + self.repos_dir]
     if self.HOST == '127.0.0.1':
       cmd.append('--listen=127.0.0.1')
     logging.debug(cmd)
     self.gitdaemon = Popen(cmd, cwd=self.repos_dir)
+    return True
 
   def _commit_svn(self, tree):
     self._genTree(self.svn_root, tree)
     commit_svn(self.svn_root)
     if self.svn_revs and self.svn_revs[-1]:
       new_tree = self.svn_revs[-1].copy()
       new_tree.update(tree)
     else:
       new_tree = tree.copy()
     self.svn_revs.append(new_tree)
@@ skipping 74 matching lines @@
     fake.setUp()
     print('Fake setup, press enter to quit or Ctrl-C to keep the checkouts.')
     sys.stdin.readline()
   except KeyboardInterrupt:
     fake.SHOULD_LEAK = True
   return 0
 
 
 if __name__ == '__main__':
   sys.exit(main(sys.argv))