Revert "Split scm-specific functions out of gclient_scm.py to scm.py."

This reverts commit 7489404c56dcc725c7c13214f410b2d375a6630a.

git-svn-id: svn://svn.chromium.org/chrome/trunk/tools/depot_tools@31810 0039d316-1c4b-4281-b951-d872f2087c98
experimental/szager/collated-output
maruel@chromium.org 16 years ago
parent 8ce0a13a85
commit 0563d6c747

@ -13,10 +13,8 @@ UNIT_TESTS = [
'tests.gcl_unittest',
'tests.gclient_test',
'tests.gclient_scm_test',
'tests.gclient_utils_test',
'tests.presubmit_unittest',
'tests.revert_unittest',
'tests.scm_unittest',
'tests.trychange_unittest',
'tests.watchlists_unittest',
]

@ -12,7 +12,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
"""Gclient-specific SCM-specific operations."""
import logging
import os
@ -22,10 +21,9 @@ import sys
import xml.dom.minidom
import gclient_utils
# TODO(maruel): Temporary.
from scm import CaptureGit, CaptureGitStatus, CaptureSVN,
from scm import CaptureSVNHeadRevision, CaptureSVNInfo, CaptureSVNStatus,
from scm import RunSVN, RunSVNAndFilterOutput, RunSVNAndGetFileList
SVN_COMMAND = "svn"
GIT_COMMAND = "git"
### SCM abstraction layer
@ -488,3 +486,323 @@ class SVNWrapper(SCMWrapper):
filterer = DiffFilterer(self.relpath)
RunSVNAndFilterOutput(command, path, False, False, filterer.Filter)
# -----------------------------------------------------------------------------
# Git utils:
def CaptureGit(args, in_directory=None, print_error=True):
"""Runs git, capturing output sent to stdout as a string.
Args:
args: A sequence of command line parameters to be passed to git.
in_directory: The directory where git is to be run.
Returns:
The output sent to stdout as a string.
"""
c = [GIT_COMMAND]
c.extend(args)
# *Sigh*: Windows needs shell=True, or else it won't search %PATH% for
# the git.exe executable, but shell=True makes subprocess on Linux fail
# when it's called with a list because it only tries to execute the
# first string ("git").
stderr = None
if not print_error:
stderr = subprocess.PIPE
return subprocess.Popen(c,
cwd=in_directory,
shell=sys.platform.startswith('win'),
stdout=subprocess.PIPE,
stderr=stderr).communicate()[0]
def CaptureGitStatus(files, upstream_branch='origin'):
"""Returns git status.
@files can be a string (one file) or a list of files.
Returns an array of (status, file) tuples."""
command = ["diff", "--name-status", "-r", "%s.." % upstream_branch]
if not files:
pass
elif isinstance(files, basestring):
command.append(files)
else:
command.extend(files)
status = CaptureGit(command).rstrip()
results = []
if status:
for statusline in status.split('\n'):
m = re.match('^(\w)\t(.+)$', statusline)
if not m:
raise Exception("status currently unsupported: %s" % statusline)
results.append(('%s ' % m.group(1), m.group(2)))
return results
# -----------------------------------------------------------------------------
# SVN utils:
def RunSVN(args, in_directory):
"""Runs svn, sending output to stdout.
Args:
args: A sequence of command line parameters to be passed to svn.
in_directory: The directory where svn is to be run.
Raises:
Error: An error occurred while running the svn command.
"""
c = [SVN_COMMAND]
c.extend(args)
gclient_utils.SubprocessCall(c, in_directory)
def CaptureSVN(args, in_directory=None, print_error=True):
"""Runs svn, capturing output sent to stdout as a string.
Args:
args: A sequence of command line parameters to be passed to svn.
in_directory: The directory where svn is to be run.
Returns:
The output sent to stdout as a string.
"""
c = [SVN_COMMAND]
c.extend(args)
# *Sigh*: Windows needs shell=True, or else it won't search %PATH% for
# the svn.exe executable, but shell=True makes subprocess on Linux fail
# when it's called with a list because it only tries to execute the
# first string ("svn").
stderr = None
if not print_error:
stderr = subprocess.PIPE
return subprocess.Popen(c,
cwd=in_directory,
shell=(sys.platform == 'win32'),
stdout=subprocess.PIPE,
stderr=stderr).communicate()[0]
def RunSVNAndGetFileList(options, args, in_directory, file_list):
"""Runs svn checkout, update, or status, output to stdout.
The first item in args must be either "checkout", "update", or "status".
svn's stdout is parsed to collect a list of files checked out or updated.
These files are appended to file_list. svn's stdout is also printed to
sys.stdout as in RunSVN.
Args:
options: command line options to gclient
args: A sequence of command line parameters to be passed to svn.
in_directory: The directory where svn is to be run.
Raises:
Error: An error occurred while running the svn command.
"""
command = [SVN_COMMAND]
command.extend(args)
# svn update and svn checkout use the same pattern: the first three columns
# are for file status, property status, and lock status. This is followed
# by two spaces, and then the path to the file.
update_pattern = '^... (.*)$'
# The first three columns of svn status are the same as for svn update and
# svn checkout. The next three columns indicate addition-with-history,
# switch, and remote lock status. This is followed by one space, and then
# the path to the file.
status_pattern = '^...... (.*)$'
# args[0] must be a supported command. This will blow up if it's something
# else, which is good. Note that the patterns are only effective when
# these commands are used in their ordinary forms, the patterns are invalid
# for "svn status --show-updates", for example.
pattern = {
'checkout': update_pattern,
'status': status_pattern,
'update': update_pattern,
}[args[0]]
compiled_pattern = re.compile(pattern)
def CaptureMatchingLines(line):
match = compiled_pattern.search(line)
if match:
file_list.append(match.group(1))
RunSVNAndFilterOutput(args,
in_directory,
options.verbose,
True,
CaptureMatchingLines)
def RunSVNAndFilterOutput(args,
in_directory,
print_messages,
print_stdout,
filter):
"""Runs svn checkout, update, status, or diff, optionally outputting
to stdout.
The first item in args must be either "checkout", "update",
"status", or "diff".
svn's stdout is passed line-by-line to the given filter function. If
print_stdout is true, it is also printed to sys.stdout as in RunSVN.
Args:
args: A sequence of command line parameters to be passed to svn.
in_directory: The directory where svn is to be run.
print_messages: Whether to print status messages to stdout about
which Subversion commands are being run.
print_stdout: Whether to forward Subversion's output to stdout.
filter: A function taking one argument (a string) which will be
passed each line (with the ending newline character removed) of
Subversion's output for filtering.
Raises:
Error: An error occurred while running the svn command.
"""
command = [SVN_COMMAND]
command.extend(args)
gclient_utils.SubprocessCallAndFilter(command,
in_directory,
print_messages,
print_stdout,
filter=filter)
def CaptureSVNInfo(relpath, in_directory=None, print_error=True):
"""Returns a dictionary from the svn info output for the given file.
Args:
relpath: The directory where the working copy resides relative to
the directory given by in_directory.
in_directory: The directory where svn is to be run.
"""
output = CaptureSVN(["info", "--xml", relpath], in_directory, print_error)
dom = gclient_utils.ParseXML(output)
result = {}
if dom:
GetNamedNodeText = gclient_utils.GetNamedNodeText
GetNodeNamedAttributeText = gclient_utils.GetNodeNamedAttributeText
def C(item, f):
if item is not None: return f(item)
# /info/entry/
# url
# reposityory/(root|uuid)
# wc-info/(schedule|depth)
# commit/(author|date)
# str() the results because they may be returned as Unicode, which
# interferes with the higher layers matching up things in the deps
# dictionary.
# TODO(maruel): Fix at higher level instead (!)
result['Repository Root'] = C(GetNamedNodeText(dom, 'root'), str)
result['URL'] = C(GetNamedNodeText(dom, 'url'), str)
result['UUID'] = C(GetNamedNodeText(dom, 'uuid'), str)
result['Revision'] = C(GetNodeNamedAttributeText(dom, 'entry', 'revision'),
int)
result['Node Kind'] = C(GetNodeNamedAttributeText(dom, 'entry', 'kind'),
str)
result['Schedule'] = C(GetNamedNodeText(dom, 'schedule'), str)
result['Path'] = C(GetNodeNamedAttributeText(dom, 'entry', 'path'), str)
result['Copied From URL'] = C(GetNamedNodeText(dom, 'copy-from-url'), str)
result['Copied From Rev'] = C(GetNamedNodeText(dom, 'copy-from-rev'), str)
return result
def CaptureSVNHeadRevision(url):
"""Get the head revision of a SVN repository.
Returns:
Int head revision
"""
info = CaptureSVN(["info", "--xml", url], os.getcwd())
dom = xml.dom.minidom.parseString(info)
return dom.getElementsByTagName('entry')[0].getAttribute('revision')
def CaptureSVNStatus(files):
"""Returns the svn 1.5 svn status emulated output.
@files can be a string (one file) or a list of files.
Returns an array of (status, file) tuples."""
command = ["status", "--xml"]
if not files:
pass
elif isinstance(files, basestring):
command.append(files)
else:
command.extend(files)
status_letter = {
None: ' ',
'': ' ',
'added': 'A',
'conflicted': 'C',
'deleted': 'D',
'external': 'X',
'ignored': 'I',
'incomplete': '!',
'merged': 'G',
'missing': '!',
'modified': 'M',
'none': ' ',
'normal': ' ',
'obstructed': '~',
'replaced': 'R',
'unversioned': '?',
}
dom = gclient_utils.ParseXML(CaptureSVN(command))
results = []
if dom:
# /status/target/entry/(wc-status|commit|author|date)
for target in dom.getElementsByTagName('target'):
for entry in target.getElementsByTagName('entry'):
file_path = entry.getAttribute('path')
wc_status = entry.getElementsByTagName('wc-status')
assert len(wc_status) == 1
# Emulate svn 1.5 status ouput...
statuses = [' '] * 7
# Col 0
xml_item_status = wc_status[0].getAttribute('item')
if xml_item_status in status_letter:
statuses[0] = status_letter[xml_item_status]
else:
raise Exception('Unknown item status "%s"; please implement me!' %
xml_item_status)
# Col 1
xml_props_status = wc_status[0].getAttribute('props')
if xml_props_status == 'modified':
statuses[1] = 'M'
elif xml_props_status == 'conflicted':
statuses[1] = 'C'
elif (not xml_props_status or xml_props_status == 'none' or
xml_props_status == 'normal'):
pass
else:
raise Exception('Unknown props status "%s"; please implement me!' %
xml_props_status)
# Col 2
if wc_status[0].getAttribute('wc-locked') == 'true':
statuses[2] = 'L'
# Col 3
if wc_status[0].getAttribute('copied') == 'true':
statuses[3] = '+'
# Col 4
if wc_status[0].getAttribute('switched') == 'true':
statuses[4] = 'S'
# TODO(maruel): Col 5 and 6
item = (''.join(statuses), file_path)
results.append(item)
return results

336
scm.py

@ -1,336 +0,0 @@
# Copyright (c) 2006-2009 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.
"""SCM-specific functions."""
import os
import re
import subprocess
import sys
import xml.dom.minidom
import gclient_utils
SVN_COMMAND = "svn"
GIT_COMMAND = "git"
# -----------------------------------------------------------------------------
# Git utils:
def CaptureGit(args, in_directory=None, print_error=True):
"""Runs git, capturing output sent to stdout as a string.
Args:
args: A sequence of command line parameters to be passed to git.
in_directory: The directory where git is to be run.
Returns:
The output sent to stdout as a string.
"""
c = [GIT_COMMAND]
c.extend(args)
# *Sigh*: Windows needs shell=True, or else it won't search %PATH% for
# the git.exe executable, but shell=True makes subprocess on Linux fail
# when it's called with a list because it only tries to execute the
# first string ("git").
stderr = None
if not print_error:
stderr = subprocess.PIPE
return subprocess.Popen(c,
cwd=in_directory,
shell=sys.platform.startswith('win'),
stdout=subprocess.PIPE,
stderr=stderr).communicate()[0]
def CaptureGitStatus(files, upstream_branch='origin'):
"""Returns git status.
@files can be a string (one file) or a list of files.
Returns an array of (status, file) tuples."""
command = ["diff", "--name-status", "-r", "%s.." % upstream_branch]
if not files:
pass
elif isinstance(files, basestring):
command.append(files)
else:
command.extend(files)
status = CaptureGit(command).rstrip()
results = []
if status:
for statusline in status.split('\n'):
m = re.match('^(\w)\t(.+)$', statusline)
if not m:
raise Exception("status currently unsupported: %s" % statusline)
results.append(('%s ' % m.group(1), m.group(2)))
return results
# -----------------------------------------------------------------------------
# SVN utils:
def RunSVN(args, in_directory):
"""Runs svn, sending output to stdout.
Args:
args: A sequence of command line parameters to be passed to svn.
in_directory: The directory where svn is to be run.
Raises:
Error: An error occurred while running the svn command.
"""
c = [SVN_COMMAND]
c.extend(args)
gclient_utils.SubprocessCall(c, in_directory)
def CaptureSVN(args, in_directory=None, print_error=True):
"""Runs svn, capturing output sent to stdout as a string.
Args:
args: A sequence of command line parameters to be passed to svn.
in_directory: The directory where svn is to be run.
Returns:
The output sent to stdout as a string.
"""
c = [SVN_COMMAND]
c.extend(args)
# *Sigh*: Windows needs shell=True, or else it won't search %PATH% for
# the svn.exe executable, but shell=True makes subprocess on Linux fail
# when it's called with a list because it only tries to execute the
# first string ("svn").
stderr = None
if not print_error:
stderr = subprocess.PIPE
return subprocess.Popen(c,
cwd=in_directory,
shell=(sys.platform == 'win32'),
stdout=subprocess.PIPE,
stderr=stderr).communicate()[0]
def RunSVNAndGetFileList(options, args, in_directory, file_list):
"""Runs svn checkout, update, or status, output to stdout.
The first item in args must be either "checkout", "update", or "status".
svn's stdout is parsed to collect a list of files checked out or updated.
These files are appended to file_list. svn's stdout is also printed to
sys.stdout as in RunSVN.
Args:
options: command line options to gclient
args: A sequence of command line parameters to be passed to svn.
in_directory: The directory where svn is to be run.
Raises:
Error: An error occurred while running the svn command.
"""
command = [SVN_COMMAND]
command.extend(args)
# svn update and svn checkout use the same pattern: the first three columns
# are for file status, property status, and lock status. This is followed
# by two spaces, and then the path to the file.
update_pattern = '^... (.*)$'
# The first three columns of svn status are the same as for svn update and
# svn checkout. The next three columns indicate addition-with-history,
# switch, and remote lock status. This is followed by one space, and then
# the path to the file.
status_pattern = '^...... (.*)$'
# args[0] must be a supported command. This will blow up if it's something
# else, which is good. Note that the patterns are only effective when
# these commands are used in their ordinary forms, the patterns are invalid
# for "svn status --show-updates", for example.
pattern = {
'checkout': update_pattern,
'status': status_pattern,
'update': update_pattern,
}[args[0]]
compiled_pattern = re.compile(pattern)
def CaptureMatchingLines(line):
match = compiled_pattern.search(line)
if match:
file_list.append(match.group(1))
RunSVNAndFilterOutput(args,
in_directory,
options.verbose,
True,
CaptureMatchingLines)
def RunSVNAndFilterOutput(args,
in_directory,
print_messages,
print_stdout,
filter):
"""Runs svn checkout, update, status, or diff, optionally outputting
to stdout.
The first item in args must be either "checkout", "update",
"status", or "diff".
svn's stdout is passed line-by-line to the given filter function. If
print_stdout is true, it is also printed to sys.stdout as in RunSVN.
Args:
args: A sequence of command line parameters to be passed to svn.
in_directory: The directory where svn is to be run.
print_messages: Whether to print status messages to stdout about
which Subversion commands are being run.
print_stdout: Whether to forward Subversion's output to stdout.
filter: A function taking one argument (a string) which will be
passed each line (with the ending newline character removed) of
Subversion's output for filtering.
Raises:
Error: An error occurred while running the svn command.
"""
command = [SVN_COMMAND]
command.extend(args)
gclient_utils.SubprocessCallAndFilter(command,
in_directory,
print_messages,
print_stdout,
filter=filter)
def CaptureSVNInfo(relpath, in_directory=None, print_error=True):
"""Returns a dictionary from the svn info output for the given file.
Args:
relpath: The directory where the working copy resides relative to
the directory given by in_directory.
in_directory: The directory where svn is to be run.
"""
output = CaptureSVN(["info", "--xml", relpath], in_directory, print_error)
dom = gclient_utils.ParseXML(output)
result = {}
if dom:
GetNamedNodeText = gclient_utils.GetNamedNodeText
GetNodeNamedAttributeText = gclient_utils.GetNodeNamedAttributeText
def C(item, f):
if item is not None: return f(item)
# /info/entry/
# url
# reposityory/(root|uuid)
# wc-info/(schedule|depth)
# commit/(author|date)
# str() the results because they may be returned as Unicode, which
# interferes with the higher layers matching up things in the deps
# dictionary.
# TODO(maruel): Fix at higher level instead (!)
result['Repository Root'] = C(GetNamedNodeText(dom, 'root'), str)
result['URL'] = C(GetNamedNodeText(dom, 'url'), str)
result['UUID'] = C(GetNamedNodeText(dom, 'uuid'), str)
result['Revision'] = C(GetNodeNamedAttributeText(dom, 'entry', 'revision'),
int)
result['Node Kind'] = C(GetNodeNamedAttributeText(dom, 'entry', 'kind'),
str)
result['Schedule'] = C(GetNamedNodeText(dom, 'schedule'), str)
result['Path'] = C(GetNodeNamedAttributeText(dom, 'entry', 'path'), str)
result['Copied From URL'] = C(GetNamedNodeText(dom, 'copy-from-url'), str)
result['Copied From Rev'] = C(GetNamedNodeText(dom, 'copy-from-rev'), str)
return result
def CaptureSVNHeadRevision(url):
"""Get the head revision of a SVN repository.
Returns:
Int head revision
"""
info = CaptureSVN(["info", "--xml", url], os.getcwd())
dom = xml.dom.minidom.parseString(info)
return dom.getElementsByTagName('entry')[0].getAttribute('revision')
def CaptureSVNStatus(files):
"""Returns the svn 1.5 svn status emulated output.
@files can be a string (one file) or a list of files.
Returns an array of (status, file) tuples."""
command = ["status", "--xml"]
if not files:
pass
elif isinstance(files, basestring):
command.append(files)
else:
command.extend(files)
status_letter = {
None: ' ',
'': ' ',
'added': 'A',
'conflicted': 'C',
'deleted': 'D',
'external': 'X',
'ignored': 'I',
'incomplete': '!',
'merged': 'G',
'missing': '!',
'modified': 'M',
'none': ' ',
'normal': ' ',
'obstructed': '~',
'replaced': 'R',
'unversioned': '?',
}
dom = gclient_utils.ParseXML(CaptureSVN(command))
results = []
if dom:
# /status/target/entry/(wc-status|commit|author|date)
for target in dom.getElementsByTagName('target'):
for entry in target.getElementsByTagName('entry'):
file_path = entry.getAttribute('path')
wc_status = entry.getElementsByTagName('wc-status')
assert len(wc_status) == 1
# Emulate svn 1.5 status ouput...
statuses = [' '] * 7
# Col 0
xml_item_status = wc_status[0].getAttribute('item')
if xml_item_status in status_letter:
statuses[0] = status_letter[xml_item_status]
else:
raise Exception('Unknown item status "%s"; please implement me!' %
xml_item_status)
# Col 1
xml_props_status = wc_status[0].getAttribute('props')
if xml_props_status == 'modified':
statuses[1] = 'M'
elif xml_props_status == 'conflicted':
statuses[1] = 'C'
elif (not xml_props_status or xml_props_status == 'none' or
xml_props_status == 'normal'):
pass
else:
raise Exception('Unknown props status "%s"; please implement me!' %
xml_props_status)
# Col 2
if wc_status[0].getAttribute('wc-locked') == 'true':
statuses[2] = 'L'
# Col 3
if wc_status[0].getAttribute('copied') == 'true':
statuses[3] = '+'
# Col 4
if wc_status[0].getAttribute('switched') == 'true':
statuses[4] = 'S'
# TODO(maruel): Col 5 and 6
item = (''.join(statuses), file_path)
results.append(item)
return results

@ -267,6 +267,115 @@ class SVNWrapperTestCase(BaseTestCase):
file_list = []
scm.update(options, self.args, file_list)
def testGetSVNFileInfo(self):
xml_text = r"""<?xml version="1.0"?>
<info>
<entry kind="file" path="%s" revision="14628">
<url>http://src.chromium.org/svn/trunk/src/chrome/app/d</url>
<repository><root>http://src.chromium.org/svn</root></repository>
<wc-info>
<schedule>add</schedule>
<depth>infinity</depth>
<copy-from-url>http://src.chromium.org/svn/trunk/src/chrome/app/DEPS</copy-from-url>
<copy-from-rev>14628</copy-from-rev>
<checksum>369f59057ba0e6d9017e28f8bdfb1f43</checksum>
</wc-info>
</entry>
</info>
""" % self.url
gclient_scm.CaptureSVN(['info', '--xml', self.url],
'.', True).AndReturn(xml_text)
expected = {
'URL': 'http://src.chromium.org/svn/trunk/src/chrome/app/d',
'UUID': None,
'Repository Root': 'http://src.chromium.org/svn',
'Schedule': 'add',
'Copied From URL':
'http://src.chromium.org/svn/trunk/src/chrome/app/DEPS',
'Copied From Rev': '14628',
'Path': self.url,
'Revision': 14628,
'Node Kind': 'file',
}
self.mox.ReplayAll()
file_info = self._CaptureSVNInfo(self.url, '.', True)
self.assertEquals(sorted(file_info.items()), sorted(expected.items()))
def testCaptureSvnInfo(self):
xml_text = """<?xml version="1.0"?>
<info>
<entry
kind="dir"
path="."
revision="35">
<url>%s</url>
<repository>
<root>%s</root>
<uuid>7b9385f5-0452-0410-af26-ad4892b7a1fb</uuid>
</repository>
<wc-info>
<schedule>normal</schedule>
<depth>infinity</depth>
</wc-info>
<commit
revision="35">
<author>maruel</author>
<date>2008-12-04T20:12:19.685120Z</date>
</commit>
</entry>
</info>
""" % (self.url, self.root_dir)
gclient_scm.CaptureSVN(['info', '--xml',
self.url], '.', True).AndReturn(xml_text)
self.mox.ReplayAll()
file_info = self._CaptureSVNInfo(self.url, '.', True)
expected = {
'URL': self.url,
'UUID': '7b9385f5-0452-0410-af26-ad4892b7a1fb',
'Revision': 35,
'Repository Root': self.root_dir,
'Schedule': 'normal',
'Copied From URL': None,
'Copied From Rev': None,
'Path': '.',
'Node Kind': 'dir',
}
self.assertEqual(file_info, expected)
def testRevinfo(self):
options = self.Options(verbose=False)
xml_text = """<?xml version="1.0"?>
<info>
<entry
kind="dir"
path="."
revision="35">
<url>%s</url>
<repository>
<root>%s</root>
<uuid>7b9385f5-0452-0410-af26-ad4892b7a1fb</uuid>
</repository>
<wc-info>
<schedule>normal</schedule>
<depth>infinity</depth>
</wc-info>
<commit
revision="35">
<author>maruel</author>
<date>2008-12-04T20:12:19.685120Z</date>
</commit>
</entry>
</info>
""" % (self.url, self.root_dir)
gclient_scm.os.getcwd().AndReturn('bleh')
gclient_scm.CaptureSVN(['info', '--xml', self.url], 'bleh'
).AndReturn(xml_text)
self.mox.ReplayAll()
scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir,
relpath=self.relpath)
rev_info = scm.revinfo(options, self.args, None)
self.assertEqual(rev_info, '35')
class GitWrapperTestCase(SuperMoxBaseTestBase):
"""This class doesn't use pymox."""
@ -328,7 +437,6 @@ from :3
return self.OptionsObject(self, *args, **kwargs)
def CreateGitRepo(self, git_import, path):
"""Do it for real."""
try:
Popen(['git', 'init'], stdout=PIPE, stderr=STDOUT,
cwd=path).communicate()
@ -504,6 +612,16 @@ from :3
self.assertEquals(rev_info, '069c602044c5388d2d15c3f875b057c852003458')
class RunSVNTestCase(BaseTestCase):
def testRunSVN(self):
self.UnMock(gclient_scm, 'RunSVN')
param2 = 'bleh'
gclient_scm.gclient_utils.SubprocessCall(['svn', 'foo', 'bar'],
param2).AndReturn(None)
self.mox.ReplayAll()
gclient_scm.RunSVN(['foo', 'bar'], param2)
if __name__ == '__main__':
import unittest
unittest.main()

@ -22,8 +22,6 @@ import __builtin__
import StringIO
import gclient
# Temporary due to the "from scm import *" in gclient_scm.
import scm
from super_mox import mox, IsOneOf, SuperMoxTestBase
@ -55,11 +53,6 @@ class GClientBaseTestCase(BaseTestCase):
self.mox.StubOutWithMock(gclient.gclient_scm, 'CaptureSVNStatus')
self.mox.StubOutWithMock(gclient.gclient_scm, 'RunSVN')
self.mox.StubOutWithMock(gclient.gclient_scm, 'RunSVNAndGetFileList')
self.mox.StubOutWithMock(scm, 'CaptureSVN')
self.mox.StubOutWithMock(scm, 'CaptureSVNInfo')
self.mox.StubOutWithMock(scm, 'CaptureSVNStatus')
self.mox.StubOutWithMock(scm, 'RunSVN')
self.mox.StubOutWithMock(scm, 'RunSVNAndGetFileList')
self._gclient_gclient = gclient.GClient
gclient.GClient = self.mox.CreateMockAnything()
self._scm_wrapper = gclient.gclient_scm.CreateSCM
@ -1073,6 +1066,112 @@ deps = {
pass
class SubprocessCallAndFilterTestCase(BaseTestCase):
def setUp(self):
BaseTestCase.setUp(self)
self.mox.StubOutWithMock(gclient.gclient_scm, 'CaptureSVN')
def testSubprocessCallAndFilter(self):
command = ['boo', 'foo', 'bar']
in_directory = 'bleh'
fail_status = None
pattern = 'a(.*)b'
test_string = 'ahah\naccb\nallo\naddb\n'
class Mock(object):
stdout = StringIO.StringIO(test_string)
def wait(self):
pass
kid = Mock()
print("\n________ running 'boo foo bar' in 'bleh'")
for i in test_string:
gclient.sys.stdout.write(i)
gclient.gclient_utils.subprocess.Popen(
command, bufsize=0, cwd=in_directory,
shell=(gclient.sys.platform == 'win32'),
stdout=gclient.gclient_utils.subprocess.PIPE,
stderr=gclient.gclient_utils.subprocess.STDOUT).AndReturn(kid)
self.mox.ReplayAll()
compiled_pattern = gclient.re.compile(pattern)
line_list = []
capture_list = []
def FilterLines(line):
line_list.append(line)
match = compiled_pattern.search(line)
if match:
capture_list.append(match.group(1))
gclient.gclient_utils.SubprocessCallAndFilter(
command, in_directory,
True, True,
fail_status, FilterLines)
self.assertEquals(line_list, ['ahah', 'accb', 'allo', 'addb'])
self.assertEquals(capture_list, ['cc', 'dd'])
def testCaptureSVNStatus(self):
gclient.gclient_scm.CaptureSVN(
['status', '--xml', '.']
).AndReturn(r"""<?xml version="1.0"?>
<status>
<target path=".">
<entry path="unversionned_file.txt">
<wc-status props="none" item="unversioned"></wc-status>
</entry>
<entry path="build\internal\essential.vsprops">
<wc-status props="normal" item="modified" revision="14628">
<commit revision="13818">
<author>ajwong@chromium.org</author>
<date>2009-04-16T00:42:06.872358Z</date>
</commit>
</wc-status>
</entry>
<entry path="chrome\app\d">
<wc-status props="none" copied="true" tree-conflicted="true" item="added">
</wc-status>
</entry>
<entry path="chrome\app\DEPS">
<wc-status props="modified" item="modified" revision="14628">
<commit revision="1279">
<author>brettw@google.com</author>
<date>2008-08-23T17:16:42.090152Z</date>
</commit>
</wc-status>
</entry>
<entry path="scripts\master\factory\gclient_factory.py">
<wc-status props="normal" item="conflicted" revision="14725">
<commit revision="14633">
<author>nsylvain@chromium.org</author>
<date>2009-04-27T19:37:17.977400Z</date>
</commit>
</wc-status>
</entry>
</target>
</status>
""")
self.mox.ReplayAll()
info = gclient.gclient_scm.CaptureSVNStatus('.')
expected = [
('? ', 'unversionned_file.txt'),
('M ', 'build\\internal\\essential.vsprops'),
('A + ', 'chrome\\app\\d'),
('MM ', 'chrome\\app\\DEPS'),
('C ', 'scripts\\master\\factory\\gclient_factory.py'),
]
self.assertEquals(sorted(info), sorted(expected))
def testCaptureSVNStatusEmpty(self):
gclient.gclient_scm.CaptureSVN(
['status', '--xml']
).AndReturn(r"""<?xml version="1.0"?>
<status>
<target
path="perf">
</target>
</status>
""")
self.mox.ReplayAll()
info = gclient.gclient_scm.CaptureSVNStatus(None)
self.assertEquals(info, [])
if __name__ == '__main__':
import unittest
unittest.main()

@ -1,279 +0,0 @@
#!/usr/bin/python
# Copyright (c) 2006-2009 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.
"""Unit tests for scm.py."""
from gclient_test import BaseTestCase
import scm
from super_mox import mox
class BaseSCMTestCase(BaseTestCase):
def setUp(self):
BaseTestCase.setUp(self)
self.mox.StubOutWithMock(scm.gclient_utils, 'SubprocessCall')
self.mox.StubOutWithMock(scm.gclient_utils, 'SubprocessCallAndFilter')
class RootTestCase(BaseSCMTestCase):
def testMembersChanged(self):
self.mox.ReplayAll()
members = [
'CaptureGit', 'CaptureGitStatus', 'GIT_COMMAND',
'CaptureSVN', 'CaptureSVNHeadRevision', 'CaptureSVNInfo',
'CaptureSVNStatus', 'RunSVN', 'RunSVNAndFilterOutput',
'RunSVNAndGetFileList', 'SVN_COMMAND',
'gclient_utils', 'os', 're', 'subprocess', 'sys', 'xml',
]
# If this test fails, you should add the relevant test.
self.compareMembers(scm, members)
class GitWrapperTestCase(BaseSCMTestCase):
sample_git_import = """blob
mark :1
data 6
Hello
blob
mark :2
data 4
Bye
reset refs/heads/master
commit refs/heads/master
mark :3
author Bob <bob@example.com> 1253744361 -0700
committer Bob <bob@example.com> 1253744361 -0700
data 8
A and B
M 100644 :1 a
M 100644 :2 b
blob
mark :4
data 10
Hello
You
blob
mark :5
data 8
Bye
You
commit refs/heads/origin
mark :6
author Alice <alice@example.com> 1253744424 -0700
committer Alice <alice@example.com> 1253744424 -0700
data 13
Personalized
from :3
M 100644 :4 a
M 100644 :5 b
reset refs/heads/master
from :3
"""
def CreateGitRepo(self, git_import, path):
try:
subprocess.Popen(['git', 'init'], stdout=subprocess.PIPE,
stderr=subprocess.STDOUT, cwd=path).communicate()
except WindowsError:
# git is not available, skip this test.
return False
subprocess.Popen(['git', 'fast-import'], stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
cwd=path).communicate(input=git_import)
subprocess.Popen(['git', 'checkout'], stdout=subprocess.PIPE,
stderr=subprocess.STDOUT, cwd=path).communicate()
return True
def setUp(self):
BaseSCMTestCase.setUp(self)
self.args = self.Args()
self.url = 'git://foo'
self.root_dir = tempfile.mkdtemp()
self.relpath = '.'
self.base_path = os.path.join(self.root_dir, self.relpath)
self.enabled = self.CreateGitRepo(self.sample_git_import, self.base_path)
def tearDown(self):
shutil.rmtree(self.root_dir)
gclient_test.BaseTestCase.tearDown(self)
class SVNTestCase(BaseSCMTestCase):
def setUp(self):
BaseSCMTestCase.setUp(self)
self.root_dir = self.Dir()
self.args = self.Args()
self.url = self.Url()
self.relpath = 'asf'
def testGetSVNFileInfo(self):
xml_text = r"""<?xml version="1.0"?>
<info>
<entry kind="file" path="%s" revision="14628">
<url>http://src.chromium.org/svn/trunk/src/chrome/app/d</url>
<repository><root>http://src.chromium.org/svn</root></repository>
<wc-info>
<schedule>add</schedule>
<depth>infinity</depth>
<copy-from-url>http://src.chromium.org/svn/trunk/src/chrome/app/DEPS</copy-from-url>
<copy-from-rev>14628</copy-from-rev>
<checksum>369f59057ba0e6d9017e28f8bdfb1f43</checksum>
</wc-info>
</entry>
</info>
""" % self.url
self.mox.StubOutWithMock(scm, 'CaptureSVN')
scm.CaptureSVN(['info', '--xml', self.url], '.', True).AndReturn(xml_text)
expected = {
'URL': 'http://src.chromium.org/svn/trunk/src/chrome/app/d',
'UUID': None,
'Repository Root': 'http://src.chromium.org/svn',
'Schedule': 'add',
'Copied From URL':
'http://src.chromium.org/svn/trunk/src/chrome/app/DEPS',
'Copied From Rev': '14628',
'Path': self.url,
'Revision': 14628,
'Node Kind': 'file',
}
self.mox.ReplayAll()
file_info = scm.CaptureSVNInfo(self.url, '.', True)
self.assertEquals(sorted(file_info.items()), sorted(expected.items()))
def testCaptureSvnInfo(self):
xml_text = """<?xml version="1.0"?>
<info>
<entry
kind="dir"
path="."
revision="35">
<url>%s</url>
<repository>
<root>%s</root>
<uuid>7b9385f5-0452-0410-af26-ad4892b7a1fb</uuid>
</repository>
<wc-info>
<schedule>normal</schedule>
<depth>infinity</depth>
</wc-info>
<commit
revision="35">
<author>maruel</author>
<date>2008-12-04T20:12:19.685120Z</date>
</commit>
</entry>
</info>
""" % (self.url, self.root_dir)
self.mox.StubOutWithMock(scm, 'CaptureSVN')
scm.CaptureSVN(['info', '--xml', self.url], '.', True).AndReturn(xml_text)
self.mox.ReplayAll()
file_info = scm.CaptureSVNInfo(self.url, '.', True)
expected = {
'URL': self.url,
'UUID': '7b9385f5-0452-0410-af26-ad4892b7a1fb',
'Revision': 35,
'Repository Root': self.root_dir,
'Schedule': 'normal',
'Copied From URL': None,
'Copied From Rev': None,
'Path': '.',
'Node Kind': 'dir',
}
self.assertEqual(file_info, expected)
def testCaptureSVNStatus(self):
text =r"""<?xml version="1.0"?>
<status>
<target path=".">
<entry path="unversionned_file.txt">
<wc-status props="none" item="unversioned"></wc-status>
</entry>
<entry path="build\internal\essential.vsprops">
<wc-status props="normal" item="modified" revision="14628">
<commit revision="13818">
<author>ajwong@chromium.org</author>
<date>2009-04-16T00:42:06.872358Z</date>
</commit>
</wc-status>
</entry>
<entry path="chrome\app\d">
<wc-status props="none" copied="true" tree-conflicted="true" item="added">
</wc-status>
</entry>
<entry path="chrome\app\DEPS">
<wc-status props="modified" item="modified" revision="14628">
<commit revision="1279">
<author>brettw@google.com</author>
<date>2008-08-23T17:16:42.090152Z</date>
</commit>
</wc-status>
</entry>
<entry path="scripts\master\factory\gclient_factory.py">
<wc-status props="normal" item="conflicted" revision="14725">
<commit revision="14633">
<author>nsylvain@chromium.org</author>
<date>2009-04-27T19:37:17.977400Z</date>
</commit>
</wc-status>
</entry>
</target>
</status>
"""
proc = self.mox.CreateMockAnything()
scm.subprocess.Popen(['svn', 'status', '--xml', '.'],
cwd=None,
shell=scm.sys.platform.startswith('win'),
stderr=None,
stdout=scm.subprocess.PIPE).AndReturn(proc)
proc.communicate().AndReturn((text, 0))
self.mox.ReplayAll()
info = scm.CaptureSVNStatus('.')
expected = [
('? ', 'unversionned_file.txt'),
('M ', 'build\\internal\\essential.vsprops'),
('A + ', 'chrome\\app\\d'),
('MM ', 'chrome\\app\\DEPS'),
('C ', 'scripts\\master\\factory\\gclient_factory.py'),
]
self.assertEquals(sorted(info), sorted(expected))
def testRunSVN(self):
param2 = 'bleh'
scm.gclient_utils.SubprocessCall(['svn', 'foo', 'bar'],
param2).AndReturn(None)
self.mox.ReplayAll()
scm.RunSVN(['foo', 'bar'], param2)
def testCaptureSVNStatusEmpty(self):
text = r"""<?xml version="1.0"?>
<status>
<target
path="perf">
</target>
</status>"""
proc = self.mox.CreateMockAnything()
scm.subprocess.Popen(['svn', 'status', '--xml'],
cwd=None,
shell=scm.sys.platform.startswith('win'),
stderr=None,
stdout=scm.subprocess.PIPE).AndReturn(proc)
proc.communicate().AndReturn((text, 0))
self.mox.ReplayAll()
info = scm.CaptureSVNStatus(None)
self.assertEquals(info, [])
if __name__ == '__main__':
import unittest
unittest.main()
# vim: ts=2:sw=2:tw=80:et:
Loading…
Cancel
Save