mercurial: comparison tests/test-verify-repo-operations.py

equal deleted inserted replaced

-:57875cf423c9
+:2372284d9457
 import os
 import subprocess
 import sys
 # Only run if slow tests are allowed
-if subprocess.call(['python', '%s/hghave' % os.environ['TESTDIR'],
+if subprocess.call(['python', '%s/hghave' % os.environ['TESTDIR'], 'slow']):
-'slow']):
 sys.exit(80)
 # These tests require Hypothesis and pytz to be installed.
 # Running 'pip install hypothesis pytz' will achieve that.
 # Note: This won't work if you're running Python < 2.7.
 # If you are running an old version of pip you may find that the enum34
 # backport is not installed automatically. If so 'pip install enum34' will
 # fix this problem.
 try:
 import enum
 assert enum  # Silence pyflakes
 except ImportError:
 sys.stderr.write("skipped: enum34 not installed" + os.linesep)
 sys.exit(80)
 import silenttestrunner
 import subprocess
 from hypothesis.errors import HypothesisException
 from hypothesis.stateful import (
-rule, RuleBasedStateMachine, Bundle, precondition)
+rule,
+RuleBasedStateMachine,
+Bundle,
+precondition,
+)
 from hypothesis import settings, note, strategies as st
 from hypothesis.configuration import set_hypothesis_home_dir
 from hypothesis.database import ExampleDatabase
 testdir = os.path.abspath(os.environ["TESTDIR"])
 # parallel we use atomic file create to ensure that we always get a unique
 # name.
 file_index = 0
 while True:
 file_index += 1
-savefile = os.path.join(generatedtests, "test-generated-%d.t" % (
+savefile = os.path.join(
-file_index,
+generatedtests, "test-generated-%d.t" % (file_index,)
-))
+)
 try:
 os.close(os.open(savefile, os.O_CREAT | os.O_EXCL | os.O_WRONLY))
 break
 except OSError as e:
 if e.errno != errno.EEXIST:
 filecharacters = (
 "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
 "[]^_`;=@{}~ !#$%&'()+,-"
 )
-files = st.text(filecharacters, min_size=1).map(lambda x: x.strip()).filter(
+files = (
-bool).map(lambda s: s.encode('ascii'))
+st.text(filecharacters, min_size=1)
+.map(lambda x: x.strip())
-safetext = st.text(st.characters(
+.filter(bool)
-min_codepoint=1, max_codepoint=127,
+.map(lambda s: s.encode('ascii'))
-blacklist_categories=('Cc', 'Cs')), min_size=1).map(
-lambda s: s.encode('utf-8')
 )
+safetext = st.text(
+st.characters(
+min_codepoint=1, max_codepoint=127, blacklist_categories=('Cc', 'Cs')
+),
+min_size=1,
+).map(lambda s: s.encode('utf-8'))
 extensions = st.sampled_from(('shelve', 'mq', 'blackbox',))
 @contextmanager
 def acceptableerrors(*args):
 """Sometimes we know an operation we're about to perform might fail, and
 we're OK with some of the failures. In those cases this may be used as a
 except subprocess.CalledProcessError as e:
 if not any(a in e.output for a in args):
 note(e.output)
 raise
 reponames = st.text("abcdefghijklmnopqrstuvwxyz01234556789", min_size=1).map(
 lambda s: s.encode('ascii')
 )
 class verifyingstatemachine(RuleBasedStateMachine):
 """This defines the set of acceptable operations on a Mercurial repository
 using Hypothesis's RuleBasedStateMachine.
 path = os.path.join(testtmp, "test-generated.t")
 with open(path, 'w') as o:
 o.write(ttest + os.linesep)
 with open(os.devnull, "w") as devnull:
 rewriter = subprocess.Popen(
-[runtests, "--local", "-i", path], stdin=subprocess.PIPE,
+[runtests, "--local", "-i", path],
-stdout=devnull, stderr=devnull,
+stdin=subprocess.PIPE,
+stdout=devnull,
+stderr=devnull,
 )
 rewriter.communicate("yes")
 with open(path, 'r') as i:
 ttest = i.read()
 e = None
 if not self.failed:
 try:
-output = subprocess.check_output([
+output = subprocess.check_output(
-runtests, path, "--local", "--pure"
+[runtests, path, "--local", "--pure"],
-], stderr=subprocess.STDOUT)
+stderr=subprocess.STDOUT,
+)
 assert "Ran 1 test" in output, output
-for ext in (
+for ext in self.all_extensions - self.non_skippable_extensions:
-self.all_extensions - self.non_skippable_extensions
+tf = os.path.join(
-):
+testtmp, "test-generated-no-%s.t" % (ext,)
-tf = os.path.join(testtmp, "test-generated-no-%s.t" % (
+)
-ext,
-))
 with open(tf, 'w') as o:
 for l in ttest.splitlines():
 if l.startswith("  $ hg"):
 l = l.replace(
-"--config %s=" % (
+"--config %s=" % (extensionconfigkey(ext),),
-extensionconfigkey(ext),), "")
+"",
+)
 o.write(l + os.linesep)
 with open(tf, 'r') as r:
 t = r.read()
 assert ext not in t, t
-output = subprocess.check_output([
+output = subprocess.check_output(
-runtests, tf, "--local",
+[runtests, tf, "--local",], stderr=subprocess.STDOUT
-], stderr=subprocess.STDOUT)
+)
 assert "Ran 1 test" in output, output
 except subprocess.CalledProcessError as e:
 note(e.output)
 if self.failed or e is not None:
 with open(savefile, "wb") as o:
 # Section: Basic commands.
 def mkdirp(self, path):
 if os.path.exists(path):
 return
 self.log.append(
-"$ mkdir -p -- %s" % (pipes.quote(os.path.relpath(path)),))
+"$ mkdir -p -- %s" % (pipes.quote(os.path.relpath(path)),)
+)
 os.makedirs(path)
 def cd(self, path):
 path = os.path.relpath(path)
 if path == ".":
 # Section: Set up basic data
 # This section has no side effects but generates data that we will want
 # to use later.
 @rule(
 target=paths,
-source=st.lists(files, min_size=1).map(lambda l: os.path.join(*l)))
+source=st.lists(files, min_size=1).map(lambda l: os.path.join(*l)),
+)
 def genpath(self, source):
 return source
 @rule(
 target=committimes,
-when=datetimes(min_year=1970, max_year=2038) | st.none())
+when=datetimes(min_year=1970, max_year=2038) | st.none(),
+)
 def gentime(self, when):
 return when
 @rule(
 target=contents,
 content=st.one_of(
-st.binary(),
+st.binary(), st.text().map(lambda x: x.encode('utf-8'))
-st.text().map(lambda x: x.encode('utf-8'))
+),
-))
+)
 def gencontent(self, content):
 return content
 @rule(
-target=branches,
+target=branches, name=safetext,
-name=safetext,
 )
 def genbranch(self, name):
 return name
 @rule(target=paths, source=paths)
 # of the current path. This will cause mkdirp to fail with this
 # error. We just turn this into a no-op in that case.
 return
 with open(path, 'wb') as o:
 o.write(content)
-self.log.append((
+self.log.append(
-"$ python -c 'import binascii; "
+(
-"print(binascii.unhexlify(\"%s\"))' > %s") % (
+"$ python -c 'import binascii; "
-binascii.hexlify(content),
+"print(binascii.unhexlify(\"%s\"))' > %s"
-pipes.quote(path),
+)
-))
+% (binascii.hexlify(content), pipes.quote(path),)
+)
 @rule(path=paths)
 def addpath(self, path):
 if os.path.exists(path):
 self.hg("add", "--", path)
 @rule(path=paths)
 def forgetpath(self, path):
 if os.path.exists(path):
-with acceptableerrors(
+with acceptableerrors("file is already untracked",):
-"file is already untracked",
-):
 self.hg("forget", "--", path)
 @rule(s=st.none() | st.integers(0, 100))
 def addremove(self, s):
 args = ["addremove"]
 if when is not None:
 if when.year == 1970:
 errors.append('negative date value')
 if when.year == 2038:
 errors.append('exceeds 32 bits')
-command.append("--date=%s" % (
+command.append(
-when.strftime('%Y-%m-%d %H:%M:%S %z'),))
+"--date=%s" % (when.strftime('%Y-%m-%d %H:%M:%S %z'),)
+)
 with acceptableerrors(*errors):
 self.hg(*command)
 # Section: Repository management
 @property
 def config(self):
 return self.configperrepo.setdefault(self.currentrepo, {})
 @rule(
-target=repos,
+target=repos, source=repos, name=reponames,
-source=repos,
-name=reponames,
 )
 def clone(self, source, name):
 if not os.path.exists(os.path.join("..", name)):
 self.cd("..")
 self.hg("clone", source, name)
 self.cd(name)
 return name
 @rule(
-target=repos,
+target=repos, name=reponames,
-name=reponames,
 )
 def fresh(self, name):
 if not os.path.exists(os.path.join("..", name)):
 self.cd("..")
 self.mkdirp(name)
 return "repo1"
 @rule()
 def pull(self, repo=repos):
 with acceptableerrors(
-"repository default not found",
+"repository default not found", "repository is unrelated",
-"repository is unrelated",
 ):
 self.hg("pull")
 @rule(newbranch=st.booleans())
 def push(self, newbranch):
 with acceptableerrors(
-"default repository not configured",
+"default repository not configured", "no changes found",
-"no changes found",
 ):
 if newbranch:
 self.hg("push", "--new-branch")
 else:
-with acceptableerrors(
+with acceptableerrors("creates new branches"):
-"creates new branches"
-):
 self.hg("push")
 # Section: Simple side effect free "check" operations
 @rule()
 def log(self):
 self.hg("branch", "--", branch)
 @rule(branch=branches, clean=st.booleans())
 def update(self, branch, clean):
 with acceptableerrors(
-'unknown revision',
+'unknown revision', 'parse error',
-'parse error',
 ):
 if clean:
 self.hg("update", "-C", "--", branch)
 else:
 self.hg("update", "--", branch)
 def unshelve(self):
 self.commandused("shelve")
 with acceptableerrors("no shelved changes to apply"):
 self.hg("unshelve")
 class writeonlydatabase(ExampleDatabase):
 def __init__(self, underlying):
 super(ExampleDatabase, self).__init__()
 self.underlying = underlying
 self.underlying.delete(key, value)
 def close(self):
 self.underlying.close()
 def extensionconfigkey(extension):
 return "extensions." + extension
 settings.register_profile(
-'default',  settings(
+'default', settings(timeout=300, stateful_step_count=50, max_examples=10,)
-timeout=300,
-stateful_step_count=50,
-max_examples=10,
-)
 )
 settings.register_profile(
-'fast',  settings(
+'fast',
+settings(
 timeout=10,
 stateful_step_count=20,
 max_examples=5,
 min_satisfying_examples=1,
 max_shrinks=0,
-)
+),
 )
 settings.register_profile(
-'continuous', settings(
+'continuous',
+settings(
 timeout=-1,
 stateful_step_count=1000,
 max_examples=10 ** 8,
 max_iterations=10 ** 8,
-database=writeonlydatabase(settings.default.database)
+database=writeonlydatabase(settings.default.database),
-)
+),
 )
 settings.load_profile(os.getenv('HYPOTHESIS_PROFILE', 'default'))
 verifyingtest = verifyingstatemachine.TestCase

changeset 43076	2372284d9457
parent 28499	8b90367c4cf3
child 45942	89a2afe31e82