<commit_before>#
#  factory.py:  Automatically generate a (near-)complete new cmdline test
#               from a series of shell commands.
#
#  Subversion is a tool for revision control.
#  See http://subversion.tigris.org for more information.
#
# ====================================================================
#    Licensed to the Apache Software Foundation (ASF) under one
#    or more contributor license agreements.  See the NOTICE file
#    distributed with this work for additional information
#    regarding copyright ownership.  The ASF licenses this file
#    to you under the Apache License, Version 2.0 (the
#    "License"); you may not use this file except in compliance
#    with the License.  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing,
#    software distributed under the License is distributed on an
#    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
#    KIND, either express or implied.  See the License for the
#    specific language governing permissions and limitations
#    under the License.
######################################################################


## HOW TO USE:
#
# (1) Edit the py test script you want to enhance (for example
#     cmdline/basic_tests.py), add a new test header as usual.
#     Insert a call to factory.make() into the empty test:
#
#  def my_new_test(sbox):
#    "my new test modifies iota"
#    svntest.factory.make(sbox, """
#                               echo "foo" > A/D/foo
#                               svn add A/D/foo
#                               svn st
#                               svn ci
#                               """)
#
# (2) Add the test to the tests list at the bottom of the py file.
# [...]
#              some_other_test,
#              my_new_test,
#             ]
#
#
# (3) Run the test, paste the output back into your new test,
#     replacing the factory call.
#
#  $ ./foo_tests.py my_new_test
#  OR
#  $ ./foo_tests.py my_new_test > new_test.snippet
#  OR
#  $ ./foo_tests.py my_new_test >> basic_tests.py
#  Then edit (e.g.) basic_tests.py to put the script in the right place.
#
#     Ensure that the py script (e.g. basic_tests.py) has these imports,
#     so that the composed script that you pasted back finds everything
#     that it uses:
#     import os, shutil
#     from svntest import main, wc, actions, verify
#
#     Be aware that you have to paste the result back to the .py file.
#
#     Be more aware that you have to read every single line and understand
#     that it makes sense. If the current behaviour is wrong, you need to
#     make the changes to expect the correct behaviour and XFail() your test.
#
#     factory.make() just probes the current situation and writes a test that
#     PASSES any success AND ANY FAILURE THAT IT FINDS. The resulting script
#     will never fail anything (if it works correctly), not even a failure.
#
#     ### TODO: some sort of intelligent pasting directly into the
#               right place, like looking for the factory call,
#               inserting the new test there, back-up-ing the old file.
#
#
# TROUBLESHOOTING
# If the result has a problem somewhere along the middle, you can,
# of course, only use the first part of the output result, maybe tweak
# something, and continue with another factory.make() at the end of that.
#
# Or you can first do stuff to your sbox and then call factory on it.
# Factory will notice if the sbox has already been built and calls
# sbox.build() only if you didn't already.
#
# You can also have any number of factory.make() calls scattered
# around "real" test code.
#
# Note that you can pass a prev_status and prev_disk to factory, to make
# the expected_* trees re-use a pre-existing one in the test, entirely
# for code beauty :P (has to match the wc_dir you will be using next).
#
#
# YOU ARE CORDIALLY INVITED to add/tweak/change to your needs.
# If you want to know what's going on, look at the switch()
# funtion of TestFactory below.
#
#
# DETAILS
# =======
#
# The input to factory.make(sbox, input) is not "real" shell-script.
# Factory goes at great lengths to try and understand your script, it
# parses common shell operations during tests and translates them.
#
# All arguments are tokenized similarly to shell, so if you need a space
# in an argument, use quotes.
#   echo "my content" > A/new_file
# Quote char escaping is done like this:
#   echo "my \\" content" > A/new_file
#   echo 'my \\' content' > A/new_file
# If you use r""" echo 'my \' content' > A/new_file """ (triple quotes
# with a leading 'r' character), you don't need to double-escape any
# characters.
#
# You can either supply multiple lines, or separate the lines with ';'.
#   factory.make(sbox, 'echo foo > bar; svn add bar')
#   factory.make(sbox, 'echo foo > bar\n svn add bar')
#   factory.make(sbox, r"""
#                      echo "foo\nbar" > bar
#                      svn add bar
#                      """)
#
#
# WORKING COPY PATHS
# - Factory will automatically build sbox.wc_dir if you didn't do so yet.
#
# - If you supply any path or file name, factory will prepend sbox.wc_dir
#   to it.
#     echo more >> iota
#                      -->   main.file_append(
#                                 os.path.join(sbox.wc_dir, 'iota'),
#                                 "more")
#   You can also do so explicitly.
#     echo more >> wc_dir/iota
#                      -->   main.file_append(
#                                 os.path.join(sbox.wc_dir, 'iota'),
#                                 "more")
#
#   Factory implies the sbox.wc_dir if you fail to supply an explicit
#   working copy dir. If you want to supply one explicitly, you can
#   choose among these wildcards:
#     'wc_dir', 'wcdir', '$WC_DIR', '$WC'  -- all expanded to sbox.wc_dir
#   For example:
#     'svn mkdir wc_dir/A/D/X'
#   But as long as you want to use only the default sbox.wc_dir, you usually
#   don't need to supply any wc_dir-wildcard:
#     'mkdir A/X' creates the directory sbox.wc_dir/A/X
#   (Factory tries to know which arguments of the commands you supplied
#   are eligible to be path arguments. If something goes wrong here, try
#   to fix factory.py to not mistake the arg for something different.
#   You usually just need to tweak some parameters to args2svntest() to
#   achieve correct expansion.)
#
# - If you want to use a second (or Nth) working copy, just supply any
#   working copy wildcard with any made-up suffix, e.g. like this:
#     'svn st wc_dir_2' or 'svn info $WC_2'
#   Factory will detect that you used another wc_dir and will automatically
#   add a corresponding directory to your sbox. The directory will initially
#   be nonexistent, so call 'mkdir', 'svn co' or 'cp' before using:
#     'cp wc_dir wc_dir_other'  -- a copy of the current WC
#     'svn co $URL wc_dir_new'  -- a clean checkout
#     'mkdir wc_dir_empty'      -- an empty directory
#   You can subsequently use any wc-dir wildcard with your suffix added.
#
#     cp wc_dir wc_dir_2
#     echo more >> wc_dir_2/iota
#                      -->   wc_dir_2 = sbox.add_wc_path('2')
#                            shutil.copytrees(wc_dir, wc_dir_2)
#                            main.file_append(
#                                 os.path.join(wc_dir_2, 'iota'),
#                                 "more")
#
#
# URLs
# Factory currently knows only one repository, thus only one repos root.
# The wildcards you can use for it are:
#   'url', '$URL'
# A URL is not inserted automatically like wc_dir, you need to supply a
# URL wildcard.
# Alternatively, you can use '^/' URLs. However, that is in effect a different
# test from an explicit entire URL. The test needs to chdir to the working
# copy in order find which URL '^/' should expand to.
# (currently, factory will chdir to sbox.wc_dir. It will only chdir
#  to another working copy if one of the other arguments involved a WC.
#  ### TODO add a 'cd wc_dir_2' command to select another WC as default.)
# Example:
#   'svn co $URL Y'   -- make a new nested working copy in sbox.wc_dir/Y
#   'svn co $URL wc_dir_2'  -- create a new separate working copy
#   'svn cp ^/A ^/X'        -- do a URL copy, creating $URL/X (branch)
#
#
# SOME EXAMPLES
# These commands should work:
#
# - "svn <subcommand> <options>"
#   Some subcommands are parsed specially, others by a catch-all default
#   parser (cmd_svn()), see switch().
#     'svn commit', 'svn commit --force', 'svn ci wc_dir_2'
#     'svn copy url/A url/X'
#
# - "echo contents > file"     (replace)
#   "echo contents >> file"    (append)
#   Calls main.file_write() / main.file_append().
#     'echo "froogle" >> A/D/G/rho'  -- append to an existing file
#     'echo "bar" > A/quux'          -- create a new file
#     'echo "fool" > wc_dir_2/me'    -- manipulate other working copies
#
# - "mkdir <names> ..."
#   Calls os.makedirs().
#   You probably want 'svn mkdir' instead, or use 'svn add' after this.
#     'mkdir A/D/X'     -- create an unversioned directory
#     'mkdir wc_dir_5'  -- create a new, empty working copy
#
# - "rm <targets>"
#   Calls main.safe_rmtree().
#   You probably want to use 'svn delete' instead.
#     'rm A/D/G'
#     'rm wc_dir_2'
#
# - "mv <source> [<source2> ...] <target>"
#   Calls shutil.move()
#   You probably want to use 'svn move' instead.
#     'mv iota A/D/'    -- move sbox.wc_dir/iota to sbox.wc_dir/A/D/.
#
# - "cp <source> [<source2> ...] <target>"
#   Do a filesystem copy.
#   You probably want to use 'svn copy' instead.
#     'cp wc_dir wc_dir_copy'
#     'cp A/D/G A/X'
#
# IF YOU NEED ANY OTHER COMMANDS:
# - first check if it doesn't work already. If not,
# - add your desired commands to factory.py! :)
# - alternatively, use a number of separate factory calls, doing what
#   you need done in "real" svntest language in-between.
#
# IF YOU REALLY DON'T GROK THIS:
# - ask #svn-dev
# - ask dev@
# - ask neels

import sys, re, os, shutil, bisect, textwrap, shlex

import svntest
from svntest import main, actions, tree
from svntest import Failure

if sys.version_info[0] >= 3:
  # Python >=3.0
  from io import StringIO
else:
  # Python <3.0
  from StringIO import StringIO

def make(wc_dir, commands, prev_status=None, prev_disk=None, verbose=True):
  """The Factory Invocation Function. This is typically the only one
  called from outside this file. See top comment in factory.py.
  Prints the resulting py script to stdout when verbose is True and
  returns the resulting line-list containing items as:
    [ ['pseudo-shell input line #1', '  translation\n  to\n  py #1'], ...]"""
  fac = TestFactory(wc_dir, prev_status, prev_disk)
  fac.make(commands)
  fac.print_script()
  return fac.lines



class TestFactory:
  """This class keeps all state around a factory.make() call."""

  def __init__(self, sbox, prev_status=None, prev_disk=None):
    self.sbox = sbox

    # The input lines and their translations.
    # Each translation usually has multiple output lines ('\n' characters).
    self.lines = []       #  [ ['in1', 'out1'], ['in2', 'out'], ...

    # Any expected_status still there from a previous verification
    self.prev_status = None
    if prev_status:
      self.prev_status = [None, prev_status]  # svntest.wc.State

    # Any expected_disk still there from a previous verification
    self.prev_disk = None
    if prev_disk:
      self.prev_disk = [None, prev_disk]      # svntest.wc.State

    # Those command line options that expect an argument following
    # which is not a path. (don't expand args following these)
    self.keep_args_of = ['--depth', '--encoding', '-r',
                         '--changelist', '-m', '--message']

    # A stack of $PWDs, to be able to chdir back after a chdir.
    self.prevdirs = []

    # The python variables we want to be declared at the beginning.
    # These are path variables like "A_D = os.path.join(wc_dir, 'A', 'D')".
    # The original wc_dir and url vars are not kept here.
    self.vars = {}

    # An optimized list kept up-to-date by variable additions
    self.sorted_vars_by_pathlen = []

    # Wether we ever used the variables 'wc_dir' and 'url' (tiny tweak)
    self.used_wc_dir = False
    self.used_url = False

    # The alternate working copy directories created that need to be
    # registered with sbox (are not inside another working copy).
    self.other_wc_dirs = {}


  def make(self, commands):
    "internal main function, delegates everything except final output."

    # keep a spacer for init
    self.add_line(None, None)

    init = ""
    if not self.sbox.is_built():
      self.sbox.build()
      init += "sbox.build()\n"


    try:
      # split input args
      input_lines = commands.replace(';','\n').splitlines()
      for str in input_lines:
        if len(str.strip()) > 0:
          self.add_line(str)

      for i in range(len(self.lines)):
        if self.lines[i][0] is not None:
          # This is where everything happens:
          self.lines[i][1] = self.switch(self.lines[i][0])

      # We're done. Add a final greeting.
      self.add_line(
        None,
        "Remember, this only saves you typing. Doublecheck everything.")

      # -- Insert variable defs in the first line --
      # main wc_dir and url
      if self.used_wc_dir:
        init += 'wc_dir = sbox.wc_dir\n'
      if self.used_url:
        init += 'url = sbox.repo_url\n'

      # registration of new WC dirs
      sorted_names = self.get_sorted_other_wc_dir_names()
      for name in sorted_names:
        init += name + ' = ' + self.other_wc_dirs[name][0] + '\n'

      if len(init) > 0:
        init += '\n'

      # general variable definitions
      sorted_names = self.get_sorted_var_names()
      for name in sorted_names:
        init += name + ' = ' + self.vars[name][0] + '\n'

      # Insert at the first line, being the spacer from above
      if len(init) > 0:
        self.lines[0][1] = init

      # This usually goes to make() below (outside this class)
      return self.lines
    except:
      for line in self.lines:
        if line[1] is not None:
          print(line[1])
      raise


  def print_script(self, stream=sys.stdout):
    "Output the resulting script of the preceding make() call"
    if self.lines is not None:
      for line in self.lines:
        if line[1] is None:
          # fall back to just that line as it was in the source
          stripped = line[0].strip()
          if not stripped.startswith('#'):
            # for comments, don't say this:
            stream.write("  # don't know how to handle:\n")
          stream.write("  " + line[0].strip() + '\n')
        else:
          if line[0] is not None:
            stream.write( wrap_each_line(line[0].strip(),
                                         "  # ", "  # ", True) + '\n')
          stream.write(wrap_each_line(line[1], "  ", "    ", False) + '\n\n')
    else:
      stream.write("  # empty.\n")
    stream.flush()


  # End of public functions.



  # "Shell" command handlers:

  def switch(self, line):
    "Given one input line, delegates to the appropriate sub-functions."
    args = shlex.split(line)
    if len(args) < 1:
      return ""
    first = args[0]

    # This is just an if-cascade. Feel free to change that.

    if first == 'svn':
      second = args[1]

      if second == 'add':
        return self.cmd_svn(args[1:], False, self.keep_args_of)

      if second in ['changelist', 'cl']:
        keep_count = 2
        if '--remove' in args:
          keep_count = 1
        return self.cmd_svn(args[1:], False, self.keep_args_of, keep_count)

      if second in ['status','stat','st']:
        return self.cmd_svn_status(args[2:])

      if second in ['commit','ci']:
        return self.cmd_svn_commit(args[2:])

      if second in ['update','up']:
        return self.cmd_svn_update(args[2:])

      if second in ['switch','sw']:
        return self.cmd_svn_switch(args[2:])

      if second in ['copy', 'cp',
                    'move', 'mv', 'rename', 'ren']:
        return self.cmd_svn_copy_move(args[1:])

      if second in ['checkout', 'co']:
        return self.cmd_svn_checkout(args[2:])

      if second in ['propset','pset','ps']:
        return self.cmd_svn(args[1:], False,
          self.keep_args_of, 3)

      if second in ['delete','del','remove', 'rm']:
        return self.cmd_svn(args[1:], False,
          self.keep_args_of + ['--with-revprop'])

      # NOTE that not all commands need to be listed here, since
      # some are already adequately handled by self.cmd_svn().
      # If you find yours is not, add another self.cmd_svn_xxx().
      return self.cmd_svn(args[1:], False, self.keep_args_of)

    if first == 'echo':
      return self.cmd_echo(args[1:])

    if first == 'mkdir':
      return self.cmd_mkdir(args[1:])

    if first == 'rm':
      return self.cmd_rm(args[1:])

    if first == 'mv':
      return self.cmd_mv(args[1:])

    if first == 'cp':
      return self.cmd_cp(args[1:])

    # if all fails, take the line verbatim
    return None


  def cmd_svn_standard_run(self, pyargs, runargs, do_chdir, wc):
    "The generic invocation of svn, helper function."
    pychdir = self.chdir(do_chdir, wc)

    code, out, err = main.run_svn("Maybe", *runargs)

    if code == 0 and len(err) < 1:
      # write a test that expects success
      pylist = self.strlist2py(out)
      if len(out) <= 1:
        py = "expected_stdout = " + pylist + "\n\n"
      else:
        py = "expected_stdout = verify.UnorderedOutput(" + pylist + ")\n\n"
      py += pychdir
      py += "actions.run_and_verify_svn2('OUTPUT', expected_stdout, [], 0"
    else:
      # write a test that expects failure
      pylist = self.strlist2py(err)
      if len(err) <= 1:
        py = "expected_stderr = " + pylist + "\n\n"
      else:
        py = "expected_stderr = verify.UnorderedOutput(" + pylist + ")\n\n"
      py += pychdir
      py += ("actions.run_and_verify_svn2('OUTPUT', " +
             "[], expected_stderr, " + str(code))

    if len(pyargs) > 0:
      py += ", " + ", ".join(pyargs)
    py += ")\n"
    py += self.chdir_back(do_chdir)
    return py


  def cmd_svn(self, svnargs, append_wc_dir_if_missing = False,
              keep_args_of = [], keep_first_count = 1,
              drop_with_arg = []):
    "Handles all svn calls not handled by more specific functions."

    pyargs, runargs, do_chdir, targets = self.args2svntest(svnargs,
                        append_wc_dir_if_missing, keep_args_of,
                        keep_first_count, drop_with_arg)

    return self.cmd_svn_standard_run(pyargs, runargs, do_chdir,
                                     self.get_first_wc(targets))


  def cmd_svn_status(self, status_args):
    "Runs svn status, looks what happened and writes the script for it."
    pyargs, runargs, do_chdir, targets = self.args2svntest(
                                  status_args, True, self.keep_args_of, 0)

    py = ""

    for target in targets:
      if not target.wc:
        py += '# SKIPPING NON-WC ' + target.runarg + '\n'
        continue

      if '-q' in status_args:
        pystatus = self.get_current_status(target.wc, True)
        py += (pystatus +
               "actions.run_and_verify_status(" + target.wc.py +
               ", expected_status)\n")
      else:
        pystatus = self.get_current_status(target.wc, False)
        py += (pystatus +
               "actions.run_and_verify_unquiet_status(" + target.wc.py +
               ", expected_status)\n")
    return py


  def cmd_svn_commit(self, commit_args):
    "Runs svn commit, looks what happened and writes the script for it."
    # these are the options that are followed by something that should not
    # be parsed as a filename in the WC.
    commit_arg_opts = [
      "--depth",
      "--with-revprop",
      "--changelist",
      # "-F", "--file", these take a file argument, don't list here.
      # "-m", "--message", treated separately
      ]

    pyargs, runargs, do_chdir, targets = self.args2svntest(
      commit_args, True, commit_arg_opts, 0, ['-m', '--message'])

    wc = self.get_first_wc(targets)
    pychdir = self.chdir(do_chdir, wc)

    code, output, err = main.run_svn("Maybe", 'ci',
                                  '-m', 'log msg',
                                  *runargs)

    if code == 0 and len(err) < 1:
      # write a test that expects success

      output = actions.process_output_for_commit(output)
      actual_out = tree.build_tree_from_commit(output)
      py = ("expected_output = " +
            self.tree2py(actual_out, wc) + "\n\n")

      pystatus = self.get_current_status(wc)
      py += pystatus

      py += pychdir
      py += ("actions.run_and_verify_commit(" + wc.py + ", " +
             "expected_output, expected_status, " +
             "None")
    else:
      # write a test that expects error
      py = "expected_error = " + self.strlist2py(err) + "\n\n"
      py += pychdir
      py += ("actions.run_and_verify_commit(" + wc.py + ", " +
             "None, None, expected_error")

    if len(pyargs) > 0:
      py += ', ' + ', '.join(pyargs)
    py += ")"
    py += self.chdir_back(do_chdir)
    return py


  def cmd_svn_update(self, update_args):
    "Runs svn update, looks what happened and writes the script for it."

    pyargs, runargs, do_chdir, targets = self.args2svntest(
                                  update_args, True, self.keep_args_of, 0)

    wc = self.get_first_wc(targets)
    pychdir = self.chdir(do_chdir, wc)

    code, output, err = main.run_svn('Maybe', 'up', *runargs)

    if code == 0 and len(err) < 1:
      # write a test that expects success

      actual_out = svntest.wc.State.from_checkout(output).old_tree()
      py = ("expected_output = " +
            self.tree2py(actual_out, wc) + "\n\n")

      pydisk = self.get_current_disk(wc)
      py += pydisk

      pystatus = self.get_current_status(wc)
      py += pystatus

      py += pychdir
      py += ("actions.run_and_verify_update(" + wc.py + ", " +
             "expected_output, expected_disk, expected_status, " +
             "None, None, None, None, None, False")
    else:
      # write a test that expects error
      py = "expected_error = " + self.strlist2py(err) + "\n\n"
      py += pychdir
      py += ("actions.run_and_verify_update(" + wc.py + ", None, None, " +
             "None, expected_error, None, None, None, None, False")

    if len(pyargs) > 0:
      py += ', ' + ', '.join(pyargs)
    py += ")"
    py += self.chdir_back(do_chdir)
    return py


  def cmd_svn_switch(self, switch_args):
    "Runs svn switch, looks what happened and writes the script for it."

    pyargs, runargs, do_chdir, targets = self.args2svntest(
                                  switch_args, True, self.keep_args_of, 0)

    # Sort out the targets. We need one URL and one wc node, in that order.
    if len(targets) < 2:
      raise Failure("Sorry, I'm currently enforcing two targets for svn " +
                    "switch. If you want to supply less, remove this " +
                    "check and implement whatever seems appropriate.")

    wc_arg = targets[1]
    del pyargs[wc_arg.argnr]
    del runargs[wc_arg.argnr]
    url_arg = targets[0]
    del pyargs[url_arg.argnr]
    del runargs[url_arg.argnr]

    wc = wc_arg.wc

    pychdir = self.chdir(do_chdir, wc)

    #if '--force' in runargs:
    #  self.really_safe_rmtree(wc_arg.runarg)

    code, output, err = main.run_svn('Maybe', 'sw',
                                     url_arg.runarg, wc_arg.runarg,
                                     *runargs)

    py = ""

    if code == 0 and len(err) < 1:
      # write a test that expects success

      actual_out = tree.build_tree_from_checkout(output)
      py = ("expected_output = " +
            self.tree2py(actual_out, wc) + "\n\n")

      pydisk = self.get_current_disk(wc)
      py += pydisk

      pystatus = self.get_current_status(wc)
      py += pystatus

      py += pychdir
      py += ("actions.run_and_verify_switch(" + wc.py + ", " +
             wc_arg.pyarg + ", " + url_arg.pyarg + ", " +
             "expected_output, expected_disk, expected_status, " +<commit_msg>* subversion/tests/cmdline/svntest/factory.py: Add an error check.<commit_after>#
#  factory.py:  Automatically generate a (near-)complete new cmdline test
#               from a series of shell commands.
#
#  Subversion is a tool for revision control.
#  See http://subversion.tigris.org for more information.
#
# ====================================================================
#    Licensed to the Apache Software Foundation (ASF) under one
#    or more contributor license agreements.  See the NOTICE file
#    distributed with this work for additional information
#    regarding copyright ownership.  The ASF licenses this file
#    to you under the Apache License, Version 2.0 (the
#    "License"); you may not use this file except in compliance
#    with the License.  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing,
#    software distributed under the License is distributed on an
#    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
#    KIND, either express or implied.  See the License for the
#    specific language governing permissions and limitations
#    under the License.
######################################################################


## HOW TO USE:
#
# (1) Edit the py test script you want to enhance (for example
#     cmdline/basic_tests.py), add a new test header as usual.
#     Insert a call to factory.make() into the empty test:
#
#  def my_new_test(sbox):
#    "my new test modifies iota"
#    svntest.factory.make(sbox, """
#                               echo "foo" > A/D/foo
#                               svn add A/D/foo
#                               svn st
#                               svn ci
#                               """)
#
# (2) Add the test to the tests list at the bottom of the py file.
# [...]
#              some_other_test,
#              my_new_test,
#             ]
#
#
# (3) Run the test, paste the output back into your new test,
#     replacing the factory call.
#
#  $ ./foo_tests.py my_new_test
#  OR
#  $ ./foo_tests.py my_new_test > new_test.snippet
#  OR
#  $ ./foo_tests.py my_new_test >> basic_tests.py
#  Then edit (e.g.) basic_tests.py to put the script in the right place.
#
#     Ensure that the py script (e.g. basic_tests.py) has these imports,
#     so that the composed script that you pasted back finds everything
#     that it uses:
#     import os, shutil
#     from svntest import main, wc, actions, verify
#
#     Be aware that you have to paste the result back to the .py file.
#
#     Be more aware that you have to read every single line and understand
#     that it makes sense. If the current behaviour is wrong, you need to
#     make the changes to expect the correct behaviour and XFail() your test.
#
#     factory.make() just probes the current situation and writes a test that
#     PASSES any success AND ANY FAILURE THAT IT FINDS. The resulting script
#     will never fail anything (if it works correctly), not even a failure.
#
#     ### TODO: some sort of intelligent pasting directly into the
#               right place, like looking for the factory call,
#               inserting the new test there, back-up-ing the old file.
#
#
# TROUBLESHOOTING
# If the result has a problem somewhere along the middle, you can,
# of course, only use the first part of the output result, maybe tweak
# something, and continue with another factory.make() at the end of that.
#
# Or you can first do stuff to your sbox and then call factory on it.
# Factory will notice if the sbox has already been built and calls
# sbox.build() only if you didn't already.
#
# You can also have any number of factory.make() calls scattered
# around "real" test code.
#
# Note that you can pass a prev_status and prev_disk to factory, to make
# the expected_* trees re-use a pre-existing one in the test, entirely
# for code beauty :P (has to match the wc_dir you will be using next).
#
#
# YOU ARE CORDIALLY INVITED to add/tweak/change to your needs.
# If you want to know what's going on, look at the switch()
# funtion of TestFactory below.
#
#
# DETAILS
# =======
#
# The input to factory.make(sbox, input) is not "real" shell-script.
# Factory goes at great lengths to try and understand your script, it
# parses common shell operations during tests and translates them.
#
# All arguments are tokenized similarly to shell, so if you need a space
# in an argument, use quotes.
#   echo "my content" > A/new_file
# Quote char escaping is done like this:
#   echo "my \\" content" > A/new_file
#   echo 'my \\' content' > A/new_file
# If you use r""" echo 'my \' content' > A/new_file """ (triple quotes
# with a leading 'r' character), you don't need to double-escape any
# characters.
#
# You can either supply multiple lines, or separate the lines with ';'.
#   factory.make(sbox, 'echo foo > bar; svn add bar')
#   factory.make(sbox, 'echo foo > bar\n svn add bar')
#   factory.make(sbox, r"""
#                      echo "foo\nbar" > bar
#                      svn add bar
#                      """)
#
#
# WORKING COPY PATHS
# - Factory will automatically build sbox.wc_dir if you didn't do so yet.
#
# - If you supply any path or file name, factory will prepend sbox.wc_dir
#   to it.
#     echo more >> iota
#                      -->   main.file_append(
#                                 os.path.join(sbox.wc_dir, 'iota'),
#                                 "more")
#   You can also do so explicitly.
#     echo more >> wc_dir/iota
#                      -->   main.file_append(
#                                 os.path.join(sbox.wc_dir, 'iota'),
#                                 "more")
#
#   Factory implies the sbox.wc_dir if you fail to supply an explicit
#   working copy dir. If you want to supply one explicitly, you can
#   choose among these wildcards:
#     'wc_dir', 'wcdir', '$WC_DIR', '$WC'  -- all expanded to sbox.wc_dir
#   For example:
#     'svn mkdir wc_dir/A/D/X'
#   But as long as you want to use only the default sbox.wc_dir, you usually
#   don't need to supply any wc_dir-wildcard:
#     'mkdir A/X' creates the directory sbox.wc_dir/A/X
#   (Factory tries to know which arguments of the commands you supplied
#   are eligible to be path arguments. If something goes wrong here, try
#   to fix factory.py to not mistake the arg for something different.
#   You usually just need to tweak some parameters to args2svntest() to
#   achieve correct expansion.)
#
# - If you want to use a second (or Nth) working copy, just supply any
#   working copy wildcard with any made-up suffix, e.g. like this:
#     'svn st wc_dir_2' or 'svn info $WC_2'
#   Factory will detect that you used another wc_dir and will automatically
#   add a corresponding directory to your sbox. The directory will initially
#   be nonexistent, so call 'mkdir', 'svn co' or 'cp' before using:
#     'cp wc_dir wc_dir_other'  -- a copy of the current WC
#     'svn co $URL wc_dir_new'  -- a clean checkout
#     'mkdir wc_dir_empty'      -- an empty directory
#   You can subsequently use any wc-dir wildcard with your suffix added.
#
#     cp wc_dir wc_dir_2
#     echo more >> wc_dir_2/iota
#                      -->   wc_dir_2 = sbox.add_wc_path('2')
#                            shutil.copytrees(wc_dir, wc_dir_2)
#                            main.file_append(
#                                 os.path.join(wc_dir_2, 'iota'),
#                                 "more")
#
#
# URLs
# Factory currently knows only one repository, thus only one repos root.
# The wildcards you can use for it are:
#   'url', '$URL'
# A URL is not inserted automatically like wc_dir, you need to supply a
# URL wildcard.
# Alternatively, you can use '^/' URLs. However, that is in effect a different
# test from an explicit entire URL. The test needs to chdir to the working
# copy in order find which URL '^/' should expand to.
# (currently, factory will chdir to sbox.wc_dir. It will only chdir
#  to another working copy if one of the other arguments involved a WC.
#  ### TODO add a 'cd wc_dir_2' command to select another WC as default.)
# Example:
#   'svn co $URL Y'   -- make a new nested working copy in sbox.wc_dir/Y
#   'svn co $URL wc_dir_2'  -- create a new separate working copy
#   'svn cp ^/A ^/X'        -- do a URL copy, creating $URL/X (branch)
#
#
# SOME EXAMPLES
# These commands should work:
#
# - "svn <subcommand> <options>"
#   Some subcommands are parsed specially, others by a catch-all default
#   parser (cmd_svn()), see switch().
#     'svn commit', 'svn commit --force', 'svn ci wc_dir_2'
#     'svn copy url/A url/X'
#
# - "echo contents > file"     (replace)
#   "echo contents >> file"    (append)
#   Calls main.file_write() / main.file_append().
#     'echo "froogle" >> A/D/G/rho'  -- append to an existing file
#     'echo "bar" > A/quux'          -- create a new file
#     'echo "fool" > wc_dir_2/me'    -- manipulate other working copies
#
# - "mkdir <names> ..."
#   Calls os.makedirs().
#   You probably want 'svn mkdir' instead, or use 'svn add' after this.
#     'mkdir A/D/X'     -- create an unversioned directory
#     'mkdir wc_dir_5'  -- create a new, empty working copy
#
# - "rm <targets>"
#   Calls main.safe_rmtree().
#   You probably want to use 'svn delete' instead.
#     'rm A/D/G'
#     'rm wc_dir_2'
#
# - "mv <source> [<source2> ...] <target>"
#   Calls shutil.move()
#   You probably want to use 'svn move' instead.
#     'mv iota A/D/'    -- move sbox.wc_dir/iota to sbox.wc_dir/A/D/.
#
# - "cp <source> [<source2> ...] <target>"
#   Do a filesystem copy.
#   You probably want to use 'svn copy' instead.
#     'cp wc_dir wc_dir_copy'
#     'cp A/D/G A/X'
#
# IF YOU NEED ANY OTHER COMMANDS:
# - first check if it doesn't work already. If not,
# - add your desired commands to factory.py! :)
# - alternatively, use a number of separate factory calls, doing what
#   you need done in "real" svntest language in-between.
#
# IF YOU REALLY DON'T GROK THIS:
# - ask #svn-dev
# - ask dev@
# - ask neels

import sys, re, os, shutil, bisect, textwrap, shlex

import svntest
from svntest import main, actions, tree
from svntest import Failure

if sys.version_info[0] >= 3:
  # Python >=3.0
  from io import StringIO
else:
  # Python <3.0
  from StringIO import StringIO

def make(wc_dir, commands, prev_status=None, prev_disk=None, verbose=True):
  """The Factory Invocation Function. This is typically the only one
  called from outside this file. See top comment in factory.py.
  Prints the resulting py script to stdout when verbose is True and
  returns the resulting line-list containing items as:
    [ ['pseudo-shell input line #1', '  translation\n  to\n  py #1'], ...]"""
  fac = TestFactory(wc_dir, prev_status, prev_disk)
  fac.make(commands)
  fac.print_script()
  return fac.lines



class TestFactory:
  """This class keeps all state around a factory.make() call."""

  def __init__(self, sbox, prev_status=None, prev_disk=None):
    self.sbox = sbox

    # The input lines and their translations.
    # Each translation usually has multiple output lines ('\n' characters).
    self.lines = []       #  [ ['in1', 'out1'], ['in2', 'out'], ...

    # Any expected_status still there from a previous verification
    self.prev_status = None
    if prev_status:
      self.prev_status = [None, prev_status]  # svntest.wc.State

    # Any expected_disk still there from a previous verification
    self.prev_disk = None
    if prev_disk:
      self.prev_disk = [None, prev_disk]      # svntest.wc.State

    # Those command line options that expect an argument following
    # which is not a path. (don't expand args following these)
    self.keep_args_of = ['--depth', '--encoding', '-r',
                         '--changelist', '-m', '--message']

    # A stack of $PWDs, to be able to chdir back after a chdir.
    self.prevdirs = []

    # The python variables we want to be declared at the beginning.
    # These are path variables like "A_D = os.path.join(wc_dir, 'A', 'D')".
    # The original wc_dir and url vars are not kept here.
    self.vars = {}

    # An optimized list kept up-to-date by variable additions
    self.sorted_vars_by_pathlen = []

    # Wether we ever used the variables 'wc_dir' and 'url' (tiny tweak)
    self.used_wc_dir = False
    self.used_url = False

    # The alternate working copy directories created that need to be
    # registered with sbox (are not inside another working copy).
    self.other_wc_dirs = {}


  def make(self, commands):
    "internal main function, delegates everything except final output."

    # keep a spacer for init
    self.add_line(None, None)

    init = ""
    if not self.sbox.is_built():
      self.sbox.build()
      init += "sbox.build()\n"


    try:
      # split input args
      input_lines = commands.replace(';','\n').splitlines()
      for str in input_lines:
        if len(str.strip()) > 0:
          self.add_line(str)

      for i in range(len(self.lines)):
        if self.lines[i][0] is not None:
          # This is where everything happens:
          self.lines[i][1] = self.switch(self.lines[i][0])

      # We're done. Add a final greeting.
      self.add_line(
        None,
        "Remember, this only saves you typing. Doublecheck everything.")

      # -- Insert variable defs in the first line --
      # main wc_dir and url
      if self.used_wc_dir:
        init += 'wc_dir = sbox.wc_dir\n'
      if self.used_url:
        init += 'url = sbox.repo_url\n'

      # registration of new WC dirs
      sorted_names = self.get_sorted_other_wc_dir_names()
      for name in sorted_names:
        init += name + ' = ' + self.other_wc_dirs[name][0] + '\n'

      if len(init) > 0:
        init += '\n'

      # general variable definitions
      sorted_names = self.get_sorted_var_names()
      for name in sorted_names:
        init += name + ' = ' + self.vars[name][0] + '\n'

      # Insert at the first line, being the spacer from above
      if len(init) > 0:
        self.lines[0][1] = init

      # This usually goes to make() below (outside this class)
      return self.lines
    except:
      for line in self.lines:
        if line[1] is not None:
          print(line[1])
      raise


  def print_script(self, stream=sys.stdout):
    "Output the resulting script of the preceding make() call"
    if self.lines is not None:
      for line in self.lines:
        if line[1] is None:
          # fall back to just that line as it was in the source
          stripped = line[0].strip()
          if not stripped.startswith('#'):
            # for comments, don't say this:
            stream.write("  # don't know how to handle:\n")
          stream.write("  " + line[0].strip() + '\n')
        else:
          if line[0] is not None:
            stream.write( wrap_each_line(line[0].strip(),
                                         "  # ", "  # ", True) + '\n')
          stream.write(wrap_each_line(line[1], "  ", "    ", False) + '\n\n')
    else:
      stream.write("  # empty.\n")
    stream.flush()


  # End of public functions.



  # "Shell" command handlers:

  def switch(self, line):
    "Given one input line, delegates to the appropriate sub-functions."
    args = shlex.split(line)
    if len(args) < 1:
      return ""
    first = args[0]

    # This is just an if-cascade. Feel free to change that.

    if first == 'svn':
      second = args[1]

      if second == 'add':
        return self.cmd_svn(args[1:], False, self.keep_args_of)

      if second in ['changelist', 'cl']:
        keep_count = 2
        if '--remove' in args:
          keep_count = 1
        return self.cmd_svn(args[1:], False, self.keep_args_of, keep_count)

      if second in ['status','stat','st']:
        return self.cmd_svn_status(args[2:])

      if second in ['commit','ci']:
        return self.cmd_svn_commit(args[2:])

      if second in ['update','up']:
        return self.cmd_svn_update(args[2:])

      if second in ['switch','sw']:
        return self.cmd_svn_switch(args[2:])

      if second in ['copy', 'cp',
                    'move', 'mv', 'rename', 'ren']:
        return self.cmd_svn_copy_move(args[1:])

      if second in ['checkout', 'co']:
        return self.cmd_svn_checkout(args[2:])

      if second in ['propset','pset','ps']:
        return self.cmd_svn(args[1:], False,
          self.keep_args_of, 3)

      if second in ['delete','del','remove', 'rm']:
        return self.cmd_svn(args[1:], False,
          self.keep_args_of + ['--with-revprop'])

      # NOTE that not all commands need to be listed here, since
      # some are already adequately handled by self.cmd_svn().
      # If you find yours is not, add another self.cmd_svn_xxx().
      return self.cmd_svn(args[1:], False, self.keep_args_of)

    if first == 'echo':
      return self.cmd_echo(args[1:])

    if first == 'mkdir':
      return self.cmd_mkdir(args[1:])

    if first == 'rm':
      return self.cmd_rm(args[1:])

    if first == 'mv':
      return self.cmd_mv(args[1:])

    if first == 'cp':
      return self.cmd_cp(args[1:])

    # if all fails, take the line verbatim
    return None


  def cmd_svn_standard_run(self, pyargs, runargs, do_chdir, wc):
    "The generic invocation of svn, helper function."
    pychdir = self.chdir(do_chdir, wc)

    code, out, err = main.run_svn("Maybe", *runargs)

    if code == 0 and len(err) < 1:
      # write a test that expects success
      pylist = self.strlist2py(out)
      if len(out) <= 1:
        py = "expected_stdout = " + pylist + "\n\n"
      else:
        py = "expected_stdout = verify.UnorderedOutput(" + pylist + ")\n\n"
      py += pychdir
      py += "actions.run_and_verify_svn2('OUTPUT', expected_stdout, [], 0"
    else:
      # write a test that expects failure
      pylist = self.strlist2py(err)
      if len(err) <= 1:
        py = "expected_stderr = " + pylist + "\n\n"
      else:
        py = "expected_stderr = verify.UnorderedOutput(" + pylist + ")\n\n"
      py += pychdir
      py += ("actions.run_and_verify_svn2('OUTPUT', " +
             "[], expected_stderr, " + str(code))

    if len(pyargs) > 0:
      py += ", " + ", ".join(pyargs)
    py += ")\n"
    py += self.chdir_back(do_chdir)
    return py


  def cmd_svn(self, svnargs, append_wc_dir_if_missing = False,
              keep_args_of = [], keep_first_count = 1,
              drop_with_arg = []):
    "Handles all svn calls not handled by more specific functions."

    pyargs, runargs, do_chdir, targets = self.args2svntest(svnargs,
                        append_wc_dir_if_missing, keep_args_of,
                        keep_first_count, drop_with_arg)

    return self.cmd_svn_standard_run(pyargs, runargs, do_chdir,
                                     self.get_first_wc(targets))


  def cmd_svn_status(self, status_args):
    "Runs svn status, looks what happened and writes the script for it."
    pyargs, runargs, do_chdir, targets = self.args2svntest(
                                  status_args, True, self.keep_args_of, 0)

    py = ""

    for target in targets:
      if not target.wc:
        py += '# SKIPPING NON-WC ' + target.runarg + '\n'
        continue

      if '-q' in status_args:
        pystatus = self.get_current_status(target.wc, True)
        py += (pystatus +
               "actions.run_and_verify_status(" + target.wc.py +
               ", expected_status)\n")
      else:
        pystatus = self.get_current_status(target.wc, False)
        py += (pystatus +
               "actions.run_and_verify_unquiet_status(" + target.wc.py +
               ", expected_status)\n")
    return py


  def cmd_svn_commit(self, commit_args):
    "Runs svn commit, looks what happened and writes the script for it."
    # these are the options that are followed by something that should not
    # be parsed as a filename in the WC.
    commit_arg_opts = [
      "--depth",
      "--with-revprop",
      "--changelist",
      # "-F", "--file", these take a file argument, don't list here.
      # "-m", "--message", treated separately
      ]

    pyargs, runargs, do_chdir, targets = self.args2svntest(
      commit_args, True, commit_arg_opts, 0, ['-m', '--message'])

    wc = self.get_first_wc(targets)
    pychdir = self.chdir(do_chdir, wc)

    code, output, err = main.run_svn("Maybe", 'ci',
                                  '-m', 'log msg',
                                  *runargs)

    if code == 0 and len(err) < 1:
      # write a test that expects success

      output = actions.process_output_for_commit(output)
      actual_out = tree.build_tree_from_commit(output)
      py = ("expected_output = " +
            self.tree2py(actual_out, wc) + "\n\n")

      pystatus = self.get_current_status(wc)
      py += pystatus

      py += pychdir
      py += ("actions.run_and_verify_commit(" + wc.py + ", " +
             "expected_output, expected_status, " +
             "None")
    else:
      # write a test that expects error
      py = "expected_error = " + self.strlist2py(err) + "\n\n"
      py += pychdir
      py += ("actions.run_and_verify_commit(" + wc.py + ", " +
             "None, None, expected_error")

    if len(pyargs) > 0:
      py += ', ' + ', '.join(pyargs)
    py += ")"
    py += self.chdir_back(do_chdir)
    return py


  def cmd_svn_update(self, update_args):
    "Runs svn update, looks what happened and writes the script for it."

    pyargs, runargs, do_chdir, targets = self.args2svntest(
                                  update_args, True, self.keep_args_of, 0)

    wc = self.get_first_wc(targets)
    pychdir = self.chdir(do_chdir, wc)

    code, output, err = main.run_svn('Maybe', 'up', *runargs)

    if code == 0 and len(err) < 1:
      # write a test that expects success

      actual_out = svntest.wc.State.from_checkout(output).old_tree()
      py = ("expected_output = " +
            self.tree2py(actual_out, wc) + "\n\n")

      pydisk = self.get_current_disk(wc)
      py += pydisk

      pystatus = self.get_current_status(wc)
      py += pystatus

      py += pychdir
      py += ("actions.run_and_verify_update(" + wc.py + ", " +
             "expected_output, expected_disk, expected_status, " +
             "None, None, None, None, None, False")
    else:
      # write a test that expects error
      py = "expected_error = " + self.strlist2py(err) + "\n\n"
      py += pychdir
      py += ("actions.run_and_verify_update(" + wc.py + ", None, None, " +
             "None, expected_error, None, None, None, None, False")

    if len(pyargs) > 0:
      py += ', ' + ', '.join(pyargs)
    py += ")"
    py += self.chdir_back(do_chdir)
    return py


  def cmd_svn_switch(self, switch_args):
    "Runs svn switch, looks what happened and writes the script for it."

    pyargs, runargs, do_chdir, targets = self.args2svntest(
                                  switch_args, True, self.keep_args_of, 0)

    # Sort out the targets. We need one URL and one wc node, in that order.
    if len(targets) < 2:
      raise Failure("Sorry, I'm currently enforcing two targets for svn " +
                    "switch. If you want to supply less, remove this " +
                    "check and implement whatever seems appropriate.")

    wc_arg = targets[1]
    del pyargs[wc_arg.argnr]
    del runargs[wc_arg.argnr]
    url_arg = targets[0]
    del pyargs[url_arg.argnr]
    del runargs[url_arg.argnr]

    wc = wc_arg.wc
    if not wc:
      raise Failure("Unexpected argument ordering to factory's 'svn switch'?")

    pychdir = self.chdir(do_chdir, wc)

    #if '--force' in runargs:
    #  self.really_safe_rmtree(wc_arg.runarg)

    code, output, err = main.run_svn('Maybe', 'sw',
                                     url_arg.runarg, wc_arg.runarg,
                                     *runargs)

    py = ""

    if code == 0 and len(err) < 1:
      # write a test that expects success

      actual_out = tree.build_tree_from_checkout(output)
      py = ("expected_output = " +
            self.tree2py(actual_out, wc) + "\n\n")

      pydisk = self.get_current_disk(wc)
      py += pydisk

      pystatus = self.get_current_status(wc)
      py += pystatus

      py += pychdir
      py += ("actions.run_and_verify_switch(" + wc.py + ", " +
             wc_arg.pyarg + ", " + url_arg.pyarg + ", " +
             "expected_output, expected_disk, expected_status, " +
{'fraction_non_alphanumeric': 0.11798504849970139, 'fraction_numerical': 0.004571946372304714, 'mean_word_length': 2.3732546022924628, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 93, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 34, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4253683', 'n_tokens_mistral': 15740, 'n_tokens_neox': 15198, 'n_words': 5146}