path: root/subversion/bindings/swig/python/tests/client.py

#
#
# 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 log_message_func(self, items, pool):
    """ Simple log message provider for unit tests. """
    self.log_message_func_calls += 1
    return "Test log message"

  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.assertEquals(self.client_ctx.log_msg_baton2, None)
    self.assertEquals(self.client_ctx.log_msg_func2, None)
    self.assertEquals(self.client_ctx.log_msg_baton3, None)
    self.assertEquals(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.assertEquals(self.client_ctx.log_msg_baton2(), 1)
    self.assertEquals(weakref_baton()(), 1)
    self.client_ctx.log_msg_baton2 = None
    self.assertEquals(self.client_ctx.log_msg_baton2, None)
    self.assertEquals(weakref_baton(), None)

    # External objects should retain their current parent pool
    self.assertNotEquals(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.assertEquals(self.client_ctx._parent_pool,
                      self.client_ctx.notify_func2._parent_pool)
    self.assertEquals(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: "message 1"
    test_object2 = lambda *args: "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')

    self.assertRaises(ValueError, 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+"/", "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+"/", "some/deep/subdir")

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

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

    start = core.svn_opt_revision_t()
    end = core.svn_opt_revision_t()
    core.svn_opt_parse_revision(start, end, "4:0")
    client.log3((directory,), start, start, end, 1, True, False,
        self.log_receiver, self.client_ctx)
    self.assertEqual(self.change_author, "john")
    self.assertEqual(self.log_message, "More directories.")
    self.assertEqual(len(self.changed_paths), 3)
    for dir in ('/trunk/dir1', '/trunk/dir2', '/trunk/dir3'):
      self.assert_(dir in self.changed_paths)
      self.assertEqual(self.changed_paths[dir].action, '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, range(0, 5))

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

  def test_url_from_path(self):
    """Test svn_client_url_from_path for a file:// URL"""
    self.assertEquals(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.assertEquals(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.assertEquals(client.uuid_from_path(path, wc_adm, self.client_ctx),
                      client.uuid_from_url(self.repos_uri, self.client_ctx))

    self.assert_(isinstance(client.uuid_from_path(path, wc_adm,
                            self.client_ctx), basestring))

  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 = '%s/trunk/README.txt' % wc_path
      readme_url = '%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, 'trunk')
    v1x_path = core.svn_dirent_join(wc_path, '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, 'README.txt')
    )

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

    self.assertEqual(readme_text, 'This is a test.\n')

  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.assert_(isinstance(providers, list))
    self.assert_(not filter(lambda x:
                             not isinstance(x, core.svn_auth_provider_object_t),
                            providers))

  def testGnomeKeyring(self):
    if not hasattr(core, 'svn_auth_set_gnome_keyring_unlock_prompt_func'):
      # 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 "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.assertEquals(props['svn:global-ignores'], '*.q\n')
    self.proplist_receiver_trunk_calls += 1

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

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

    trunk_url = self.repos_uri + '/trunk'
    client.propset_remote('svn:global-ignores', '*.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('svn:global-ignores', trunk_url,
                                         head, head, core.svn_depth_infinity,
                                         None, self.client_ctx)
    self.assertEquals(props[trunk_url], '*.q\n')

    dir1_url = trunk_url + '/dir1'
    props, iprops, rev = client.propget5('svn:global-ignores', dir1_url,
                                         head, head, core.svn_depth_infinity,
                                         None, self.client_ctx)
    self.assertEquals(iprops[trunk_url], {'svn:global-ignores':'*.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.assertEquals(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.assertEquals(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')

    self.assertRaises(ValueError, 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)

    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, 'branches'),
        os.path.join(path, 'tags'),
        os.path.join(path, '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(os.path.sep, '/') for x in expected_paths]
    self.notified_paths.sort()
    expected_paths.sort()

    self.assertEquals(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, 'trunk', 'README.txt'),
        os.path.join(path, 'trunk'),
        path,
        path
    ]
    expected_paths = [x.replace(os.path.sep, '/') 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.assertEquals(self.notified_paths, expected_paths)


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

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