#
#
# 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.
#
#
import unittest, os, weakref, setup_path, utils

from svn import core, client, wc

try:
  # Python >=3.0
  from urllib.parse import urljoin
except ImportError:
  # Python <3.0
  from urlparse import urljoin

class SubversionClientTestCase(unittest.TestCase):
  """Test cases for the basic SWIG Subversion client layer"""

  def assert_all_instances_of(self, iterable, instancetype):
    """Asserts that all object from iterable are an instance of instancetype."""

    self.assertTrue(not [x for x in iterable if not isinstance(x, instancetype)])

  def log_message_func(self, items, pool):
    """ Simple log message provider for unit tests. """
    self.log_message_func_calls += 1
    return b"Test log message"

  def make_log_message_func(self, message):
    def generic_log_message_func(items, pool):
      self.log_message_func_calls += 1
      return message
    return generic_log_message_func

  def log_receiver(self, changed_paths, revision, author, date, message, pool):
    """ Function to receive log messages retrieved by client.log3(). """
    self.log_message = message
    self.change_author = author
    self.changed_paths = changed_paths

  def log_entry_receiver(self, log_entry, pool):
    """An implementation of svn_log_entry_receiver_t."""
    self.received_revisions.append(log_entry.revision)

  def setUp(self):
    """Set up authentication and client context"""
    self.client_ctx = client.svn_client_create_context()
    self.assertEqual(self.client_ctx.log_msg_baton2, None)
    self.assertEqual(self.client_ctx.log_msg_func2, None)
    self.assertEqual(self.client_ctx.log_msg_baton3, None)
    self.assertEqual(self.client_ctx.log_msg_func3, None)
    self.client_ctx.log_msg_func3 = client.svn_swig_py_get_commit_log_func
    self.client_ctx.log_msg_baton3 = self.log_message_func
    self.log_message_func_calls = 0
    self.log_message = None
    self.changed_paths = None
    self.change_author = None

    providers = [
       client.svn_client_get_simple_provider(),
       client.svn_client_get_username_provider(),
    ]

    self.client_ctx.auth_baton = core.svn_auth_open(providers)

    self.temper = utils.Temper()
    (_, self.repos_path, self.repos_uri) = self.temper.alloc_known_repo(
      'trac/versioncontrol/tests/svnrepos.dump', suffix='-client')

  def tearDown(self):
    # We have to free client_ctx first, since it may be holding handles
    # to WC DBs
    del self.client_ctx
    self.temper.cleanup()

  def testBatonPlay(self):
    """Test playing with C batons"""
    baton = lambda: 1
    weakref_baton = weakref.ref(baton)
    self.client_ctx.log_msg_baton2 = baton
    baton = None
    self.assertEqual(self.client_ctx.log_msg_baton2(), 1)
    self.assertEqual(weakref_baton()(), 1)
    self.client_ctx.log_msg_baton2 = None
    self.assertEqual(self.client_ctx.log_msg_baton2, None)
    self.assertEqual(weakref_baton(), None)

    # External objects should retain their current parent pool
    self.assertNotEqual(self.client_ctx._parent_pool,
                         self.client_ctx.auth_baton._parent_pool)

    # notify_func2 and notify_baton2 were generated by
    # svn_client_create_context, so they should have
    # the same pool as the context
    self.assertEqual(self.client_ctx._parent_pool,
                      self.client_ctx.notify_func2._parent_pool)
    self.assertEqual(self.client_ctx._parent_pool,
                      self.client_ctx.notify_baton2._parent_pool)

  def testMethodCalls(self):
    """Test direct method calls to callbacks"""

    # Directly invoking the msg_baton should work
    self.client_ctx.log_msg_baton3(None, None)
    b = self.client_ctx.log_msg_baton3
    b(None, None)
    self.assertEqual(self.log_message_func_calls, 2)

    # You can also invoke the log_msg_func3. It'd be
    # nice if we could get log_msg_func3 function
    # to invoke the baton function, but, in order to do that,
    # we'd need to supply a value for the first parameter.
    self.client_ctx.log_msg_func3(None, self.client_ctx.log_msg_baton3)

  def info_receiver(self, path, info, pool):
    """Squirrel away the output from 'svn info' so that the unit tests
       can get at them."""
    self.path = path
    self.info = info

  def test_client_ctx_baton_lifetime(self):
    pool = core.Pool()
    temp_client_ctx = client.svn_client_create_context(pool)

    # We keep track of these objects in separate variables here
    # because you can't get a PyObject back out of a PY_AS_VOID field
    test_object1 = lambda *args: b"message 1"
    test_object2 = lambda *args: b"message 2"

    # Verify that the refcount of a Python object is incremented when
    # you insert it into a PY_AS_VOID field.
    temp_client_ctx.log_msg_baton2 = test_object1
    test_object1 = weakref.ref(test_object1)
    self.assertNotEqual(test_object1(), None)

    # Verify that the refcount of the previous Python object is decremented
    # when a PY_AS_VOID field is replaced.
    temp_client_ctx.log_msg_baton2 = test_object2
    self.assertEqual(test_object1(), None)

    # Verify that the reference count of the new Python object (which
    # replaced test_object1) was incremented.
    test_object2 = weakref.ref(test_object2)
    self.assertNotEqual(test_object2(), None)

    # Verify that the reference count of test_object2 is decremented when
    # test_client_ctx is destroyed.
    temp_client_ctx = None
    self.assertEqual(test_object2(), None)

  def test_checkout(self):
    """Test svn_client_checkout2."""

    rev = core.svn_opt_revision_t()
    rev.kind = core.svn_opt_revision_head

    path = self.temper.alloc_empty_dir('-checkout')

    # TypeError is raised since SWIG 4.3.0
    self.assertRaises((ValueError, TypeError), r'Received a NULL pointer',
                      client.checkout2,
                      self.repos_uri, path, None, None, True, True,
                      self.client_ctx)

    client.checkout2(self.repos_uri, path, rev, rev, True, True,
            self.client_ctx)

  def test_info(self):
    """Test svn_client_info on an empty repository"""

    # Run info
    revt = core.svn_opt_revision_t()
    revt.kind = core.svn_opt_revision_head
    client.info(self.repos_uri, revt, revt, self.info_receiver,
                False, self.client_ctx)

    # Check output from running info. This also serves to verify that
    # the internal 'info' object is still valid
    self.assertEqual(self.path, os.path.basename(self.repos_path))
    self.info.assert_valid()
    self.assertEqual(self.info.URL, self.repos_uri)
    self.assertEqual(self.info.repos_root_URL, self.repos_uri)

  def test_mkdir_url(self):
    """Test svn_client_mkdir2 on a file:// URL"""
    directory = urljoin(self.repos_uri+b"/", b"dir1")

    commit_info = client.mkdir2((directory,), self.client_ctx)
    self.assertEqual(commit_info.revision, 13)
    self.assertEqual(self.log_message_func_calls, 1)

  def test_mkdir_url_with_revprops(self):
    """Test svn_client_mkdir3 on a file:// URL, with added revprops"""
    directory = urljoin(self.repos_uri+b"/", b"some/deep/subdir")

    commit_info = client.mkdir3((directory,), 1, {b'customprop':b'value'},
                                self.client_ctx)
    self.assertEqual(commit_info.revision, 13)
    self.assertEqual(self.log_message_func_calls, 1)

  def test_get_commit_log3_callback_accept_unicode(self):
    """Test svn_client_get_commit_log3_t callback wrapper accept unicode as return value"""
    directory = urljoin(self.repos_uri+b"/", b"dir1")
    # override callback function which returns commit log as unicode
    unicode_log_message_func = self.make_log_message_func(u"Test log message")
    self.client_ctx.log_msg_baton3 = unicode_log_message_func

    commit_info = client.mkdir3((directory,), 1, {b'customprop':b'value'},
                                self.client_ctx)
    self.assertEqual(commit_info.revision, 13)
    self.assertEqual(self.log_message_func_calls, 1)

  def test_get_commit_log3_callback_unicode_error(self):
    """Test svn_client_get_commit_log3_t callback wrapper handles UnicodeEncodeError correctly"""
    directory = urljoin(self.repos_uri+b"/", b"dir1")
    # override callback function which returns commit log as unicode
    # which contains surrogate escaped character
    bogus_log_message_func = self.make_log_message_func(u"Test \udc6cog"
                                                        u" message")
    self.client_ctx.log_msg_baton3 = bogus_log_message_func

    if not utils.IS_PY3 and utils.is_defaultencoding_utf8():
      # 'utf-8' codecs on Python 2 does not raise UnicodeEncodeError
      # on surrogate code point U+dc00 - U+dcff, however it causes
      # Subversion error on property validation of svn:log
      with self.assertRaises(core.SubversionException):
        commit_info = client.mkdir3((directory,), 1, {b'customprop':b'value'},
                                    self.client_ctx)
    else:
      with self.assertRaises(UnicodeEncodeError):
        commit_info = client.mkdir3((directory,), 1, {b'customprop':b'value'},
                                    self.client_ctx)

  def test_log3_url(self):
    """Test svn_client_log3 on a file:// URL"""
    directory = urljoin(self.repos_uri+b"/", b"trunk/dir1")

    start = core.svn_opt_revision_t()
    end = core.svn_opt_revision_t()
    core.svn_opt_parse_revision(start, end, b"4:0")
    client.log3((directory,), start, start, end, 1, True, False,
        self.log_receiver, self.client_ctx)
    self.assertEqual(self.change_author, b"john")
    self.assertEqual(self.log_message, b"More directories.")
    self.assertEqual(len(self.changed_paths), 3)
    for dir in (b'/trunk/dir1', b'/trunk/dir2', b'/trunk/dir3'):
      self.assertTrue(dir in self.changed_paths)
      self.assertEqual(self.changed_paths[dir].action, b'A')

  def test_log5(self):
    """Test svn_client_log5."""
    start = core.svn_opt_revision_t()
    start.kind = core.svn_opt_revision_number
    start.value.number = 0

    end = core.svn_opt_revision_t()
    end.kind = core.svn_opt_revision_number
    end.value.number = 4

    rev_range = core.svn_opt_revision_range_t()
    rev_range.start = start
    rev_range.end = end

    self.received_revisions = []

    client.log5((self.repos_uri,), end, (rev_range,), 0, False, True, False, (),
        self.log_entry_receiver, self.client_ctx)

    self.assertEqual(self.received_revisions, list(range(0, 5)))

  def test_log5_revprops(self):
    """Test svn_client_log5 revprops (for typemap(in) apr_array_t *STRINGLIST)"""
    directory = urljoin(self.repos_uri+b"/", b"trunk/dir1")
    start = core.svn_opt_revision_t()
    end = core.svn_opt_revision_t()
    core.svn_opt_parse_revision(start, end, b"4:0")
    rev_range = core.svn_opt_revision_range_t()
    rev_range.start = start
    rev_range.end = end
    entry_pool = core.Pool()

    def log_entry_receiver_whole(log_entry, pool):
      """An implementation of svn_log_entry_receiver_t, holds whole log entries."""
      self.received_log_entries.append(core.svn_log_entry_dup(log_entry,
                                                              entry_pool))

    self.received_log_entries = []

    # (Python 3: pass tuple of bytes and str mixture as revprops argment)
    client.log5((directory,), start, (rev_range,), 1, True, False, False,
                ('svn:author', b'svn:log'),
                log_entry_receiver_whole, self.client_ctx)
    self.assertEqual(len(self.received_log_entries), 1)
    revprops = self.received_log_entries[0].revprops
    self.assertEqual(revprops[b'svn:log'], b"More directories.")
    self.assertEqual(revprops[b'svn:author'], b"john")
    with self.assertRaises(KeyError):
      commit_date = revprops['svn:date']
    if utils.IS_PY3 or not utils.is_defaultencoding_utf8():
      # 'utf-8' codecs on Python 2 does not raise UnicodeEncodeError
      # on surrogate code point U+dc00 - U+dcff. So we need to skip
      # below in such a case.
      with self.assertRaises(UnicodeEncodeError):
        client.log5((directory,), start, (rev_range,), 1, True, False, False,
                    (u'svn:\udc61uthor', b'svn:log'),
                    log_entry_receiver_whole, self.client_ctx)

  def test_uuid_from_url(self):
    """Test svn_client_uuid_from_url on a file:// URL"""
    self.assertTrue(isinstance(
                 client.uuid_from_url(self.repos_uri, self.client_ctx),
                 bytes))

  def test_url_from_path(self):
    """Test svn_client_url_from_path for a file:// URL"""
    self.assertEqual(client.url_from_path(self.repos_uri), self.repos_uri)

    rev = core.svn_opt_revision_t()
    rev.kind = core.svn_opt_revision_head

    path = self.temper.alloc_empty_dir('-url_from_path')

    client.checkout2(self.repos_uri, path, rev, rev, True, True,
                     self.client_ctx)

    self.assertEqual(client.url_from_path(path), self.repos_uri)

  def test_uuid_from_path(self):
    """Test svn_client_uuid_from_path."""
    rev = core.svn_opt_revision_t()
    rev.kind = core.svn_opt_revision_head

    path = self.temper.alloc_empty_dir('-uuid_from_path')

    client.checkout2(self.repos_uri, path, rev, rev, True, True,
                     self.client_ctx)

    wc_adm = wc.adm_open3(None, path, False, 0, None)

    self.assertEqual(client.uuid_from_path(path, wc_adm, self.client_ctx),
                      client.uuid_from_url(self.repos_uri, self.client_ctx))

    self.assertTrue(isinstance(client.uuid_from_path(path, wc_adm,
                            self.client_ctx), bytes))

  def test_open_ra_session(self):
      """Test svn_client_open_ra_session()."""
      client.open_ra_session(self.repos_uri, self.client_ctx)


  def test_info_file(self):
    """Test svn_client_info on working copy file and remote files."""

    # This test requires a file /trunk/README.txt of size 8 bytes
    # in the repository.
    rev = core.svn_opt_revision_t()
    rev.kind = core.svn_opt_revision_head
    wc_path = self.temper.alloc_empty_dir('-info_file')

    client.checkout2(self.repos_uri, wc_path, rev, rev, True, True,
                     self.client_ctx)
    adm_access = wc.adm_open3(None, wc_path, True, -1, None)

    try:
      # Test 1: Run info -r BASE. We expect the size value to be filled in.
      rev.kind = core.svn_opt_revision_base
      readme_path = b'%s/trunk/README.txt' % wc_path
      readme_url = b'%s/trunk/README.txt' % self.repos_uri
      client.info(readme_path, rev, rev, self.info_receiver,
                  False, self.client_ctx)

      self.assertEqual(self.path, os.path.basename(readme_path))
      self.info.assert_valid()
      self.assertEqual(self.info.working_size, client.SWIG_SVN_INFO_SIZE_UNKNOWN)
      self.assertEqual(self.info.size, 8)

      # Test 2: Run info (revision unspecified). We expect the working_size value
      # to be filled in.
      rev.kind = core.svn_opt_revision_unspecified
      client.info(readme_path, rev, rev, self.info_receiver,
                  False, self.client_ctx)

      self.assertEqual(self.path, readme_path)
      self.info.assert_valid()
      self.assertEqual(self.info.size, client.SWIG_SVN_INFO_SIZE_UNKNOWN)
      # README.txt contains one EOL char, so on Windows it will be expanded from
      # LF to CRLF hence the working_size will be 9 instead of 8.
      if os.name == 'nt':
        self.assertEqual(self.info.working_size, 9)
      else:
        self.assertEqual(self.info.working_size, 8)

      # Test 3: Run info on the repository URL of README.txt. We expect the size
      # value to be filled in.
      rev.kind = core.svn_opt_revision_head
      client.info(readme_url, rev, rev, self.info_receiver,
                  False, self.client_ctx)
      self.info.assert_valid()
      self.assertEqual(self.info.working_size, client.SWIG_SVN_INFO_SIZE_UNKNOWN)
      self.assertEqual(self.info.size, 8)
    finally:
      wc.adm_close(adm_access)

  def test_merge_peg3(self):
    """Test svn_client_merge_peg3."""
    head = core.svn_opt_revision_t()
    head.kind = core.svn_opt_revision_head
    wc_path = self.temper.alloc_empty_dir('-merge_peg3')

    client.checkout3(self.repos_uri, wc_path, head, head, core.svn_depth_infinity,
                     True, False, self.client_ctx)

    # Let's try to backport a change from the v1x branch
    trunk_path = core.svn_dirent_join(wc_path, b'trunk')
    v1x_path = core.svn_dirent_join(wc_path, b'branches/v1x')

    start = core.svn_opt_revision_t()
    start.kind = core.svn_opt_revision_number
    start.value.number = 8

    end = core.svn_opt_revision_t()
    end.kind = core.svn_opt_revision_number
    end.value.number = 9

    rrange = core.svn_opt_revision_range_t()
    rrange.start = start
    rrange.end = end

    client.merge_peg3(v1x_path, (rrange,), end, trunk_path,
                      core.svn_depth_infinity, False, False, False, False,
                      None, self.client_ctx)

    # Did it take effect?
    readme_path_native = core.svn_dirent_local_style(
      core.svn_dirent_join(trunk_path, b'README.txt')
    )

    readme = open(readme_path_native, 'rb')
    readme_text = readme.read()
    readme.close()

    self.assertEqual(readme_text,
                     b'This is a test.' + os.linesep.encode('UTF-8'))

  def test_platform_providers(self):
    providers = core.svn_auth_get_platform_specific_client_providers(None, None)
    # Not much more we can test in this minimal environment.
    self.assertTrue(isinstance(providers, list))
    self.assert_all_instances_of(providers, core.svn_auth_provider_object_t)

  def testGnomeKeyring(self):
    if getattr(core, 'svn_auth_set_gnome_keyring_unlock_prompt_func', None) is None:
      # gnome-keying not compiled in, do nothing
      return

    # This tests setting the gnome-keyring unlock prompt function as an
    # auth baton parameter. It doesn't actually call gnome-keyring
    # stuff, since that would require having a gnome-keyring running. We
    # just test if this doesn't error out, there's not even a return
    # value to test.
    def prompt_func(realm_string, pool):
      return b"Foo"

    core.svn_auth_set_gnome_keyring_unlock_prompt_func(self.client_ctx.auth_baton, prompt_func)

  def proplist_receiver_trunk(self, path, props, iprops, pool):
    self.assertEqual(props[b'svn:global-ignores'], b'*.q\n')
    self.proplist_receiver_trunk_calls += 1

  def proplist_receiver_dir1(self, path, props, iprops, pool):
    self.assertEqual(iprops[self.proplist_receiver_dir1_key],
                      {b'svn:global-ignores':b'*.q\n'})
    self.proplist_receiver_dir1_calls += 1

  def test_inherited_props(self):
    """Test inherited props"""

    trunk_url = self.repos_uri + b'/trunk'
    client.propset_remote(b'svn:global-ignores', b'*.q', trunk_url,
                          False, 12, {}, None, self.client_ctx)

    head = core.svn_opt_revision_t()
    head.kind = core.svn_opt_revision_head
    props, iprops, rev = client.propget5(b'svn:global-ignores', trunk_url,
                                         head, head, core.svn_depth_infinity,
                                         None, self.client_ctx)
    self.assertEqual(props[trunk_url], b'*.q\n')

    dir1_url = trunk_url + b'/dir1'
    props, iprops, rev = client.propget5(b'svn:global-ignores', dir1_url,
                                         head, head, core.svn_depth_infinity,
                                         None, self.client_ctx)
    self.assertEqual(iprops[trunk_url], {b'svn:global-ignores':b'*.q\n'})

    self.proplist_receiver_trunk_calls = 0
    client.proplist4(trunk_url, head, head, core.svn_depth_empty, None, True,
                     self.proplist_receiver_trunk, self.client_ctx)
    self.assertEqual(self.proplist_receiver_trunk_calls, 1)

    self.proplist_receiver_dir1_calls = 0
    self.proplist_receiver_dir1_key = trunk_url
    client.proplist4(dir1_url, head, head, core.svn_depth_empty, None, True,
                     self.proplist_receiver_dir1, self.client_ctx)
    self.assertEqual(self.proplist_receiver_dir1_calls, 1)

  def test_update4(self):
    """Test update and the notify function callbacks"""

    rev = core.svn_opt_revision_t()
    rev.kind = core.svn_opt_revision_number
    rev.value.number = 0

    path = self.temper.alloc_empty_dir('-update')

    # TypeError is raised since SWIG 4.3.0
    self.assertRaises((ValueError, TypeError), r'Received a NULL pointer',
                      client.checkout2,
                      self.repos_uri, path, None, None, True, True,
                      self.client_ctx)

    client.checkout2(self.repos_uri, path, rev, rev, True, True,
            self.client_ctx)

    def notify_func(path, action, kind, mime_type, content_state, prop_state, rev):
        self.notified_paths.append(path)

    PATH_SEPARATOR = os.path.sep
    if not isinstance(PATH_SEPARATOR, bytes):
        PATH_SEPARATOR = PATH_SEPARATOR.encode('UTF-8')
    self.client_ctx.notify_func = client.svn_swig_py_notify_func
    self.client_ctx.notify_baton = notify_func
    rev.value.number = 1
    self.notified_paths = []
    client.update4((path,), rev, core.svn_depth_unknown, True, False, False,
                   False, False, self.client_ctx)
    expected_paths = [
        path,
        os.path.join(path, b'branches'),
        os.path.join(path, b'tags'),
        os.path.join(path, b'trunk'),
        path,
        path
    ]
    # All normal subversion apis process paths in Subversion's canonical format,
    # which isn't the platform specific format
    expected_paths = [x.replace(PATH_SEPARATOR, b'/') for x in expected_paths]
    self.notified_paths.sort()
    expected_paths.sort()

    self.assertEqual(self.notified_paths, expected_paths)

    def notify_func2(notify, pool):
        self.notified_paths.append(notify.path)

    self.client_ctx.notify_func2 = client.svn_swig_py_notify_func2
    self.client_ctx.notify_baton2 = notify_func2
    rev.value.number = 2
    self.notified_paths = []
    expected_paths = [
        path,
        os.path.join(path, b'trunk', b'README.txt'),
        os.path.join(path, b'trunk'),
        path,
        path
    ]
    expected_paths = [x.replace(PATH_SEPARATOR, b'/') for x in expected_paths]
    client.update4((path,), rev, core.svn_depth_unknown, True, False, False,
                   False, False, self.client_ctx)
    self.notified_paths.sort()
    expected_paths.sort()
    self.assertEqual(self.notified_paths, expected_paths)

  def test_conflict(self):
    """Test conflict api."""

    rev = core.svn_opt_revision_t()
    rev.kind = core.svn_opt_revision_number
    rev.value.number = 0

    path = self.temper.alloc_empty_dir('-conflict')

    client.checkout2(self.repos_uri, path, rev, rev, True, True,
            self.client_ctx)

    trunk_path = core.svn_dirent_join(path, b'trunk')

    # Create a conflicting path
    os.mkdir(core.svn_dirent_local_style(trunk_path))

    rev.value.number = 2

    client.update4((path,), rev, core.svn_depth_unknown, True, False, False,
                   False, False, self.client_ctx)

    pool = core.Pool()
    conflict = client.conflict_get(trunk_path, self.client_ctx, pool)

    self.assertTrue(isinstance(conflict, client.svn_client_conflict_t))

    conflict_opts = client.conflict_tree_get_resolution_options(conflict, self.client_ctx)

    self.assertTrue(isinstance(conflict_opts, list))
    self.assert_all_instances_of(conflict_opts, client.svn_client_conflict_option_t)

    pool.clear()

  @unittest.skip("experimental API, not currently exposed")
  def test_shelf(self):
    """Test shelf api."""

    rev = core.svn_opt_revision_t()
    rev.kind = core.svn_opt_revision_number
    rev.value.number = 2

    path = self.temper.alloc_empty_dir('-shelf')


    client.checkout2(self.repos_uri, path, rev, rev, True, True,
            self.client_ctx)

    pool = core.Pool()
    shelf = client._shelf_open_or_create(b"test1", path, self.client_ctx, pool)

    self.assertTrue(isinstance(shelf, client.svn_client__shelf_t))

    new_subpath = core.svn_relpath_join(b'trunk', b'new-shelf-test.txt')
    new_path = core.svn_dirent_join(path, new_subpath)

    with open(core.svn_dirent_local_style(new_path), "wb") as fp:
      fp.write("A new text file\n".encode('utf8'))

    client.add5(new_path, core.svn_depth_unknown, False, False, False, True, self.client_ctx, pool)

    statused_paths = []
    def shelf_status(path, status, pool):
      statused_paths.append(path)

    shelf_version = client._shelf_save_new_version3(shelf, (new_path, ), core.svn_depth_unknown,
                                                    None, shelf_status, None, pool)

    self.assertTrue(isinstance(shelf_version, client.svn_client__shelf_version_t))

    all_versions = client._shelf_get_all_versions(shelf, pool, pool)

    self.assertEqual(1, len(all_versions))
    self.assertTrue(isinstance(all_versions[0], client.svn_client__shelf_version_t))
    self.assertEqual(shelf_version.version_number, all_versions[0].version_number)
    self.assertIn(new_subpath, statused_paths)

    client._shelf_close(shelf, pool)

    pool.clear()


def suite():
    return unittest.defaultTestLoader.loadTestsFromTestCase(
      SubversionClientTestCase)

if __name__ == '__main__':
    runner = unittest.TextTestRunner()
    runner.run(suite())

