You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@subversion.apache.org by st...@apache.org on 2020/05/23 14:17:03 UTC
svn commit: r1878061 [8/43] - in /subversion/branches/addremove: ./ build/
build/ac-macros/ build/generator/ build/generator/swig/
build/generator/templates/ build/generator/util/ build/win32/
contrib/client-side/ contrib/client-side/svn_load_dirs/ con...
Modified: subversion/branches/addremove/subversion/bindings/swig/python/tests/client.py
URL: http://svn.apache.org/viewvc/subversion/branches/addremove/subversion/bindings/swig/python/tests/client.py?rev=1878061&r1=1878060&r2=1878061&view=diff
==============================================================================
--- subversion/branches/addremove/subversion/bindings/swig/python/tests/client.py (original)
+++ subversion/branches/addremove/subversion/bindings/swig/python/tests/client.py Sat May 23 14:16:56 2020
@@ -32,10 +32,21 @@ except ImportError:
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 "Test log message"
+ 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(). """
@@ -50,10 +61,10 @@ class SubversionClientTestCase(unittest.
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.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
@@ -84,22 +95,22 @@ class SubversionClientTestCase(unittest.
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.assertEqual(self.client_ctx.log_msg_baton2(), 1)
+ self.assertEqual(weakref_baton()(), 1)
self.client_ctx.log_msg_baton2 = None
- self.assertEquals(self.client_ctx.log_msg_baton2, None)
- self.assertEquals(weakref_baton(), None)
+ self.assertEqual(self.client_ctx.log_msg_baton2, None)
+ self.assertEqual(weakref_baton(), None)
# External objects should retain their current parent pool
- self.assertNotEquals(self.client_ctx._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.assertEquals(self.client_ctx._parent_pool,
+ self.assertEqual(self.client_ctx._parent_pool,
self.client_ctx.notify_func2._parent_pool)
- self.assertEquals(self.client_ctx._parent_pool,
+ self.assertEqual(self.client_ctx._parent_pool,
self.client_ctx.notify_baton2._parent_pool)
def testMethodCalls(self):
@@ -129,8 +140,8 @@ class SubversionClientTestCase(unittest.
# 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"
+ 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.
@@ -186,7 +197,7 @@ class SubversionClientTestCase(unittest.
def test_mkdir_url(self):
"""Test svn_client_mkdir2 on a file:// URL"""
- directory = urljoin(self.repos_uri+"/", "dir1")
+ directory = urljoin(self.repos_uri+b"/", b"dir1")
commit_info = client.mkdir2((directory,), self.client_ctx)
self.assertEqual(commit_info.revision, 13)
@@ -194,28 +205,53 @@ class SubversionClientTestCase(unittest.
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")
+ 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, {'customprop':'value'},
+ 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
+
+ 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+"/", "trunk/dir1")
+ 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, "4:0")
+ 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, "john")
- self.assertEqual(self.log_message, "More directories.")
+ 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 ('/trunk/dir1', '/trunk/dir2', '/trunk/dir3'):
- self.assert_(dir in self.changed_paths)
- self.assertEqual(self.changed_paths[dir].action, 'A')
+ 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."""
@@ -236,17 +272,50 @@ class SubversionClientTestCase(unittest.
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))
+ 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 argument)
+ 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']
+ 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.assert_(isinstance(
+ self.assertTrue(isinstance(
client.uuid_from_url(self.repos_uri, self.client_ctx),
- basestring))
+ bytes))
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)
+ 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
@@ -256,7 +325,7 @@ class SubversionClientTestCase(unittest.
client.checkout2(self.repos_uri, path, rev, rev, True, True,
self.client_ctx)
- self.assertEquals(client.url_from_path(path), self.repos_uri)
+ self.assertEqual(client.url_from_path(path), self.repos_uri)
def test_uuid_from_path(self):
"""Test svn_client_uuid_from_path."""
@@ -270,11 +339,11 @@ class SubversionClientTestCase(unittest.
wc_adm = wc.adm_open3(None, path, False, 0, None)
- self.assertEquals(client.uuid_from_path(path, wc_adm, self.client_ctx),
+ self.assertEqual(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))
+ 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()."""
@@ -297,8 +366,8 @@ class SubversionClientTestCase(unittest.
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
+ 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)
@@ -344,8 +413,8 @@ class SubversionClientTestCase(unittest.
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')
+ 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
@@ -365,25 +434,24 @@ class SubversionClientTestCase(unittest.
# Did it take effect?
readme_path_native = core.svn_dirent_local_style(
- core.svn_dirent_join(trunk_path, 'README.txt')
+ core.svn_dirent_join(trunk_path, b'README.txt')
)
- readme = open(readme_path_native, 'r')
+ readme = open(readme_path_native, 'rb')
readme_text = readme.read()
readme.close()
- self.assertEqual(readme_text, 'This is a test.\n')
+ 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.assert_(isinstance(providers, list))
- self.assert_(not filter(lambda x:
- not isinstance(x, core.svn_auth_provider_object_t),
- providers))
+ self.assertTrue(isinstance(providers, list))
+ self.assert_all_instances_of(providers, core.svn_auth_provider_object_t)
def testGnomeKeyring(self):
- if not hasattr(core, 'svn_auth_set_gnome_keyring_unlock_prompt_func'):
+ if getattr(core, 'svn_auth_set_gnome_keyring_unlock_prompt_func', None) is None:
# gnome-keying not compiled in, do nothing
return
@@ -393,49 +461,49 @@ class SubversionClientTestCase(unittest.
# 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"
+ 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.assertEquals(props['svn:global-ignores'], '*.q\n')
+ 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.assertEquals(iprops[self.proplist_receiver_dir1_key],
- {'svn:global-ignores':'*.q\n'})
+ 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 + '/trunk'
- client.propset_remote('svn:global-ignores', '*.q', trunk_url,
+ 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('svn:global-ignores', trunk_url,
+ props, iprops, rev = client.propget5(b'svn:global-ignores', trunk_url,
head, head, core.svn_depth_infinity,
None, self.client_ctx)
- self.assertEquals(props[trunk_url], '*.q\n')
+ self.assertEqual(props[trunk_url], b'*.q\n')
- dir1_url = trunk_url + '/dir1'
- props, iprops, rev = client.propget5('svn:global-ignores', dir1_url,
+ 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.assertEquals(iprops[trunk_url], {'svn:global-ignores':'*.q\n'})
+ 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.assertEquals(self.proplist_receiver_trunk_calls, 1)
+ 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.assertEquals(self.proplist_receiver_dir1_calls, 1)
+ self.assertEqual(self.proplist_receiver_dir1_calls, 1)
def test_update4(self):
"""Test update and the notify function callbacks"""
@@ -456,6 +524,9 @@ class SubversionClientTestCase(unittest.
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
@@ -464,19 +535,19 @@ class SubversionClientTestCase(unittest.
False, False, self.client_ctx)
expected_paths = [
path,
- os.path.join(path, 'branches'),
- os.path.join(path, 'tags'),
- os.path.join(path, 'trunk'),
+ 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(os.path.sep, '/') for x in expected_paths]
+ expected_paths = [x.replace(PATH_SEPARATOR, b'/') for x in expected_paths]
self.notified_paths.sort()
expected_paths.sort()
- self.assertEquals(self.notified_paths, expected_paths)
+ self.assertEqual(self.notified_paths, expected_paths)
def notify_func2(notify, pool):
self.notified_paths.append(notify.path)
@@ -487,17 +558,98 @@ class SubversionClientTestCase(unittest.
self.notified_paths = []
expected_paths = [
path,
- os.path.join(path, 'trunk', 'README.txt'),
- os.path.join(path, 'trunk'),
+ os.path.join(path, b'trunk', b'README.txt'),
+ os.path.join(path, b'trunk'),
path,
path
]
- expected_paths = [x.replace(os.path.sep, '/') for x in expected_paths]
+ 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.assertEquals(self.notified_paths, expected_paths)
+ 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():
Modified: subversion/branches/addremove/subversion/bindings/swig/python/tests/core.py
URL: http://svn.apache.org/viewvc/subversion/branches/addremove/subversion/bindings/swig/python/tests/core.py?rev=1878061&r1=1878060&r2=1878061&view=diff
==============================================================================
--- subversion/branches/addremove/subversion/bindings/swig/python/tests/core.py (original)
+++ subversion/branches/addremove/subversion/bindings/swig/python/tests/core.py Sat May 23 14:16:56 2020
@@ -19,10 +19,16 @@
#
#
import unittest
+import os
+import tempfile
+import sys
+
+IS_PY3 = sys.version_info[0] >= 3
import svn.core, svn.client
import utils
+
class SubversionCoreTestCase(unittest.TestCase):
"""Test cases for the basic SWIG Subversion core"""
@@ -39,13 +45,13 @@ class SubversionCoreTestCase(unittest.Te
'error message')
def test_mime_type_is_binary(self):
- self.assertEqual(0, svn.core.svn_mime_type_is_binary("text/plain"))
- self.assertEqual(1, svn.core.svn_mime_type_is_binary("image/png"))
+ self.assertEqual(0, svn.core.svn_mime_type_is_binary(b"text/plain"))
+ self.assertEqual(1, svn.core.svn_mime_type_is_binary(b"image/png"))
def test_mime_type_validate(self):
self.assertRaises(svn.core.SubversionException,
- svn.core.svn_mime_type_validate, "this\nis\ninvalid\n")
- svn.core.svn_mime_type_validate("unknown/but-valid; charset=utf8")
+ svn.core.svn_mime_type_validate, b"this\nis\ninvalid\n")
+ svn.core.svn_mime_type_validate(b"unknown/but-valid; charset=utf8")
def test_exception_interoperability(self):
"""Test if SubversionException is correctly converted into svn_error_t
@@ -108,44 +114,44 @@ class SubversionCoreTestCase(unittest.Te
def test_config_enumerate2(self):
cfg = svn.core.svn_config_create(False)
entries = {
- 'one': 'one-value',
- 'two': 'two-value',
- 'three': 'three-value'
+ b'one': b'one-value',
+ b'two': b'two-value',
+ b'three': b'three-value'
}
- for (name, value) in entries.iteritems():
- svn.core.svn_config_set(cfg, "section", name, value)
+ for (name, value) in entries.items():
+ svn.core.svn_config_set(cfg, b"section", name, value)
received_entries = {}
def enumerator(name, value, pool):
received_entries[name] = value
return len(received_entries) < 2
- svn.core.svn_config_enumerate2(cfg, "section", enumerator)
+ svn.core.svn_config_enumerate2(cfg, b"section", enumerator)
self.assertEqual(len(received_entries), 2)
- for (name, value) in received_entries.iteritems():
- self.assert_(name in entries)
+ for (name, value) in received_entries.items():
+ self.assertTrue(name in entries)
self.assertEqual(value, entries[name])
def test_config_enumerate2_exception(self):
cfg = svn.core.svn_config_create(False)
- svn.core.svn_config_set(cfg, "section", "one", "one-value")
- svn.core.svn_config_set(cfg, "section", "two", "two-value")
+ svn.core.svn_config_set(cfg, b"section", b"one", b"one-value")
+ svn.core.svn_config_set(cfg, b"section", b"two", b"two-value")
def enumerator(name, value, pool):
raise Exception
# the exception will be swallowed, but enumeration must be stopped
self.assertEqual(
- svn.core.svn_config_enumerate2(cfg, "section", enumerator), 1)
+ svn.core.svn_config_enumerate2(cfg, b"section", enumerator), 1)
def test_config_enumerate_sections2(self):
cfg = svn.core.svn_config_create(False)
- sections = ['section-one', 'section-two', 'section-three']
+ sections = [b'section-one', b'section-two', b'section-three']
for section in sections:
- svn.core.svn_config_set(cfg, section, "name", "value")
+ svn.core.svn_config_set(cfg, section, b"name", b"value")
received_sections = []
def enumerator(section, pool):
@@ -156,12 +162,12 @@ class SubversionCoreTestCase(unittest.Te
self.assertEqual(len(received_sections), 2)
for section in received_sections:
- self.assert_(section in sections)
+ self.assertTrue(section in sections)
def test_config_enumerate_sections2_exception(self):
cfg = svn.core.svn_config_create(False)
- svn.core.svn_config_set(cfg, "section-one", "name", "value")
- svn.core.svn_config_set(cfg, "section-two", "name", "value")
+ svn.core.svn_config_set(cfg, b"section-one", b"name", b"value")
+ svn.core.svn_config_set(cfg, b"section-two", b"name", b"value")
def enumerator(section, pool):
raise Exception
@@ -170,6 +176,163 @@ class SubversionCoreTestCase(unittest.Te
self.assertEqual(
svn.core.svn_config_enumerate_sections2(cfg, enumerator), 1)
+ def test_stream_from_stringbuf(self):
+ stream = svn.core.svn_stream_from_stringbuf(b'')
+ svn.core.svn_stream_close(stream)
+ with self.assertRaises(TypeError):
+ stream = svn.core.svn_stream_from_stringbuf(b''.decode())
+ svn.core.svn_stream_close(stream)
+
+ def test_stream_read_full(self):
+ in_str = (b'Python\x00'
+ b'\xa4\xd1\xa4\xa4\xa4\xbd\xa4\xf3\r\n'
+ b'Subversion\x00'
+ b'\xa4\xb5\xa4\xd6\xa4\xd0\xa1\xbc\xa4\xb8\xa4\xe7\xa4\xf3\n'
+ b'swig\x00'
+ b'\xa4\xb9\xa4\xa6\xa4\xa3\xa4\xb0\r'
+ b'end')
+ stream = svn.core.svn_stream_from_stringbuf(in_str)
+ self.assertEqual(svn.core.svn_stream_read_full(stream, 4096), in_str)
+ svn.core.svn_stream_seek(stream, None)
+ self.assertEqual(svn.core.svn_stream_read_full(stream, 10), in_str[0:10])
+ svn.core.svn_stream_seek(stream, None)
+ svn.core.svn_stream_skip(stream, 20)
+ self.assertEqual(svn.core.svn_stream_read_full(stream, 4096), in_str[20:])
+ self.assertEqual(svn.core.svn_stream_read_full(stream, 4096), b'')
+ svn.core.svn_stream_close(stream)
+
+ def test_stream_read2(self):
+ # as we can't create non block stream by using swig-py API directly,
+ # we only test svn_stream_read2() behaves just same as
+ # svn_stream_read_full()
+ in_str = (b'Python\x00'
+ b'\xa4\xd1\xa4\xa4\xa4\xbd\xa4\xf3\r\n'
+ b'Subversion\x00'
+ b'\xa4\xb5\xa4\xd6\xa4\xd0\xa1\xbc\xa4\xb8\xa4\xe7\xa4\xf3\n'
+ b'swig\x00'
+ b'\xa4\xb9\xa4\xa6\xa4\xa3\xa4\xb0\r'
+ b'end')
+ stream = svn.core.svn_stream_from_stringbuf(in_str)
+ self.assertEqual(svn.core.svn_stream_read2(stream, 4096), in_str)
+ svn.core.svn_stream_seek(stream, None)
+ self.assertEqual(svn.core.svn_stream_read2(stream, 10), in_str[0:10])
+ svn.core.svn_stream_seek(stream, None)
+ svn.core.svn_stream_skip(stream, 20)
+ self.assertEqual(svn.core.svn_stream_read2(stream, 4096), in_str[20:])
+ self.assertEqual(svn.core.svn_stream_read2(stream, 4096), b'')
+ svn.core.svn_stream_close(stream)
+
+ def test_stream_write_exception(self):
+ stream = svn.core.svn_stream_empty()
+ with self.assertRaises(TypeError):
+ svn.core.svn_stream_write(stream, 16)
+ # Check UnicodeEncodeError
+ # o1_str = b'Python\x00\xa4\xd1\xa4\xa4\xa4\xbd\xa4\xf3\r\n'
+ # ostr_unicode = o1_str.decode('ascii', 'surrogateescape')
+ ostr_unicode = (u'Python\x00'
+ u'\udca4\udcd1\udca4\udca4\udca4\udcbd\udca4\udcf3\r\n')
+ with self.assertRaises(UnicodeEncodeError):
+ svn.core.svn_stream_write(stream, ostr_unicode)
+ svn.core.svn_stream_close(stream)
+
+ # As default codec of Python 2 is 'ascii', conversion from unicode to bytes
+ # will be success only if all characters of target strings are in the range
+ # of \u0000 ~ \u007f.
+ @unittest.skipUnless(IS_PY3, "test for Python 3 only")
+ def test_stream_write_str(self):
+ o1_str = u'Python\x00\u3071\u3044\u305d\83093\r\n'
+ o2_str = u'subVersioN\x00\u3055\u3076\u3070\u30fc\u3058\u3087\u3093'
+ o3_str = u'swig\x00\u3059\u3046\u3043\u3050\rend'
+ out_str = o1_str + o2_str + o3_str
+ rewrite_str = u'Subversion'
+ fd, fname = tempfile.mkstemp()
+ os.close(fd)
+ try:
+ stream = svn.core.svn_stream_from_aprfile2(fname, False)
+ self.assertEqual(svn.core.svn_stream_write(stream, out_str),
+ len(out_str.encode('UTF-8')))
+ svn.core.svn_stream_seek(stream, None)
+ self.assertEqual(svn.core.svn_stream_read_full(stream, 4096),
+ out_str.encode('UTF-8'))
+ svn.core.svn_stream_seek(stream, None)
+ svn.core.svn_stream_skip(stream, len(o1_str.encode('UTF-8')))
+ self.assertEqual(svn.core.svn_stream_write(stream, rewrite_str),
+ len(rewrite_str.encode('UTF-8')))
+ svn.core.svn_stream_seek(stream, None)
+ self.assertEqual(
+ svn.core.svn_stream_read_full(stream, 4096),
+ (o1_str + rewrite_str
+ + o2_str[len(rewrite_str.encode('UTF-8')):]
+ + o3_str ).encode('UTF-8'))
+ svn.core.svn_stream_close(stream)
+ finally:
+ try:
+ os.remove(fname)
+ except OSError:
+ pass
+
+ def test_stream_write_bytes(self):
+ o1_str = b'Python\x00\xa4\xd1\xa4\xa4\xa4\xbd\xa4\xf3\r\n'
+ o2_str = (b'subVersioN\x00'
+ b'\xa4\xb5\xa4\xd6\xa4\xd0\xa1\xbc\xa4\xb8\xa4\xe7\xa4\xf3\n')
+ o3_str = b'swig\x00\xa4\xb9\xa4\xa6\xa4\xa3\xa4\xb0\rend'
+ out_str = o1_str + o2_str + o3_str
+ rewrite_str = b'Subversion'
+ fd, fname = tempfile.mkstemp()
+ fname_bytes = fname if isinstance(fname, bytes) else fname.encode('UTF-8')
+ os.close(fd)
+ try:
+ stream = svn.core.svn_stream_from_aprfile2(fname_bytes, False)
+ self.assertEqual(svn.core.svn_stream_write(stream, out_str),
+ len(out_str))
+ svn.core.svn_stream_seek(stream, None)
+ self.assertEqual(svn.core.svn_stream_read_full(stream, 4096), out_str)
+ svn.core.svn_stream_seek(stream, None)
+ svn.core.svn_stream_skip(stream, len(o1_str))
+ self.assertEqual(svn.core.svn_stream_write(stream, rewrite_str),
+ len(rewrite_str))
+ svn.core.svn_stream_seek(stream, None)
+ self.assertEqual(
+ svn.core.svn_stream_read_full(stream, 4096),
+ o1_str + rewrite_str + o2_str[len(rewrite_str):] + o3_str)
+ svn.core.svn_stream_close(stream)
+ finally:
+ try:
+ os.remove(fname)
+ except OSError:
+ pass
+
+ def test_stream_readline(self):
+ o1_str = b'Python\t\xa4\xd1\xa4\xa4\xa4\xbd\xa4\xf3\r\n'
+ o2_str = (b'Subversion\t'
+ b'\xa4\xb5\xa4\xd6\xa4\xd0\xa1\xbc\xa4\xb8\xa4\xe7\xa4\xf3\n')
+ o3_str = b'swig\t\xa4\xb9\xa4\xa6\xa4\xa3\xa4\xb0\rend'
+ in_str = o1_str + o2_str + o3_str
+ stream = svn.core.svn_stream_from_stringbuf(in_str)
+ self.assertEqual(svn.core.svn_stream_readline(stream, b'\n'),
+ [o1_str[:-1], 0])
+ self.assertEqual(svn.core.svn_stream_readline(stream, b'\n'),
+ [o2_str[:-1], 0])
+ self.assertEqual(svn.core.svn_stream_readline(stream, b'\n'),
+ [o3_str, 1])
+ self.assertEqual(svn.core.svn_stream_readline(stream, b'\n'),
+ [b'', 1])
+ svn.core.svn_stream_seek(stream, None)
+ self.assertEqual(svn.core.svn_stream_readline(stream, b'\r\n'),
+ [o1_str[:-2], 0])
+ self.assertEqual(svn.core.svn_stream_readline(stream, b'\r\n'),
+ [o2_str + o3_str, 1])
+ svn.core.svn_stream_write(stream, b'\r\n')
+ svn.core.svn_stream_seek(stream, None)
+ self.assertEqual(svn.core.svn_stream_readline(stream, b'\r\n'),
+ [o1_str[:-2], 0])
+ self.assertEqual(svn.core.svn_stream_readline(stream, b'\r\n'),
+ [o2_str + o3_str, 0])
+ self.assertEqual(svn.core.svn_stream_readline(stream, b'\r\n'),
+ [b'', 1])
+ svn.core.svn_stream_close(stream)
+
+
def suite():
return unittest.defaultTestLoader.loadTestsFromTestCase(
SubversionCoreTestCase)
Modified: subversion/branches/addremove/subversion/bindings/swig/python/tests/delta.py
URL: http://svn.apache.org/viewvc/subversion/branches/addremove/subversion/bindings/swig/python/tests/delta.py?rev=1878061&r1=1878060&r2=1878061&view=diff
==============================================================================
--- subversion/branches/addremove/subversion/bindings/swig/python/tests/delta.py (original)
+++ subversion/branches/addremove/subversion/bindings/swig/python/tests/delta.py Sat May 23 14:16:56 2020
@@ -19,23 +19,20 @@
#
#
import unittest, setup_path
+import os
+import tempfile
import svn.delta
import svn.core
from sys import version_info # For Python version check
-if version_info[0] >= 3:
- # Python >=3.0
- from io import StringIO
-else:
- # Python <3.0
- from cStringIO import StringIO
+from io import BytesIO
# Test case for svn.delta
class DeltaTestCase(unittest.TestCase):
def testTxWindowHandler(self):
"""Test tx_invoke_window_handler"""
- src_stream = StringIO("hello world")
- target_stream = StringIO("bye world")
+ src_stream = BytesIO("hello world".encode('UTF-8'))
+ target_stream = BytesIO("bye world".encode('UTF-8'))
# Invoke the window_handler using a helper function
window_handler, baton = \
@@ -47,17 +44,72 @@ class DeltaTestCase(unittest.TestCase):
svn.delta.tx_apply(src_stream, target_stream, None)
window_handler(None, baton)
+ def testTxWindowHandler_stream_IF(self):
+ """Test tx_invoke_window_handler, with svn.core.svn_stream_t object"""
+ pool = svn.core.Pool()
+ in_str = b"hello world"
+ src_stream = svn.core.svn_stream_from_stringbuf(in_str)
+ content_str = b"bye world"
+ content_stream = svn.core.svn_stream_from_stringbuf(content_str)
+ fd, fname = tempfile.mkstemp()
+ fname_bytes = fname if isinstance(fname, bytes) else fname.encode('UTF-8')
+ os.close(fd)
+ try:
+ target_stream = svn.core.svn_stream_from_aprfile2(fname_bytes, False)
+ window_handler, baton = \
+ svn.delta.tx_apply(src_stream, target_stream, None)
+ svn.delta.tx_send_stream(content_stream, window_handler, baton, pool)
+ fp = open(fname, 'rb')
+ out_str = fp.read()
+ fp.close()
+ self.assertEqual(content_str, out_str)
+ finally:
+ del pool
+ try:
+ os.remove(fname)
+ except OSError:
+ pass
+
+ def testTxWindowHandler_Stream_IF(self):
+ """Test tx_invoke_window_handler, with svn.core.Stream object"""
+ pool = svn.core.Pool()
+ in_str = b"hello world"
+ src_stream = svn.core.Stream(
+ svn.core.svn_stream_from_stringbuf(in_str))
+ content_str = b"bye world"
+ content_stream = svn.core.Stream(
+ svn.core.svn_stream_from_stringbuf(content_str))
+ fd, fname = tempfile.mkstemp()
+ fname_bytes = fname if isinstance(fname, bytes) else fname.encode('UTF-8')
+ os.close(fd)
+ try:
+ target_stream = svn.core.Stream(
+ svn.core.svn_stream_from_aprfile2(fname_bytes, False))
+ window_handler, baton = \
+ svn.delta.tx_apply(src_stream, target_stream, None)
+ svn.delta.tx_send_stream(content_stream, window_handler, baton, None)
+ fp = open(fname, 'rb')
+ out_str = fp.read()
+ fp.close()
+ self.assertEqual(content_str, out_str)
+ finally:
+ del pool
+ try:
+ os.remove(fname)
+ except OSError:
+ pass
+
def testTxdeltaWindowT(self):
"""Test the svn_txdelta_window_t wrapper."""
- a = StringIO("abc\ndef\n")
- b = StringIO("def\nghi\n")
+ a = BytesIO("abc\ndef\n".encode('UTF-8'))
+ b = BytesIO("def\nghi\n".encode('UTF-8'))
delta_stream = svn.delta.svn_txdelta(a, b)
window = svn.delta.svn_txdelta_next_window(delta_stream)
- self.assert_(window.sview_offset + window.sview_len <= len(a.getvalue()))
- self.assert_(window.tview_len <= len(b.getvalue()))
- self.assert_(len(window.new_data) > 0)
+ self.assertTrue(window.sview_offset + window.sview_len <= len(a.getvalue()))
+ self.assertTrue(window.tview_len <= len(b.getvalue()))
+ self.assertTrue(len(window.new_data) > 0)
self.assertEqual(window.num_ops, len(window.ops))
self.assertEqual(window.src_ops, len([op for op in window.ops
if op.action_code == svn.delta.svn_txdelta_source]))
Modified: subversion/branches/addremove/subversion/bindings/swig/python/tests/mergeinfo.py
URL: http://svn.apache.org/viewvc/subversion/branches/addremove/subversion/bindings/swig/python/tests/mergeinfo.py?rev=1878061&r1=1878060&r2=1878061&view=diff
==============================================================================
--- subversion/branches/addremove/subversion/bindings/swig/python/tests/mergeinfo.py (original)
+++ subversion/branches/addremove/subversion/bindings/swig/python/tests/mergeinfo.py Sat May 23 14:16:56 2020
@@ -34,7 +34,7 @@ def get_svn_merge_range_t_objects():
garbage collector, used for detecting memory leaks."""
return [
o for o in gc.get_objects()
- if hasattr(o, '__class__') and
+ if getattr(o, '__class__', None) is not None and
o.__class__.__name__ == 'svn_merge_range_t'
]
@@ -42,11 +42,11 @@ class SubversionMergeinfoTestCase(unitte
"""Test cases for mergeinfo"""
# Some textual mergeinfo.
- TEXT_MERGEINFO1 = "/trunk:3-9,27,42*"
- TEXT_MERGEINFO2 = "/trunk:27-29,41-43*"
+ TEXT_MERGEINFO1 = b"/trunk:3-9,27,42*"
+ TEXT_MERGEINFO2 = b"/trunk:27-29,41-43*"
# Meta data used in conjunction with this mergeinfo.
- MERGEINFO_SRC = "/trunk"
+ MERGEINFO_SRC = b"/trunk"
MERGEINFO_NBR_REV_RANGES = 3
def setUp(self):
@@ -93,9 +93,9 @@ class SubversionMergeinfoTestCase(unitte
reversed_rl = core.svn_rangelist_reverse(rangelist)
expected_ranges = ((42, 41), (27, 26), (9, 2))
for i in range(0, len(reversed_rl)):
- self.assertEquals(reversed_rl[i].start, expected_ranges[i][0],
+ self.assertEqual(reversed_rl[i].start, expected_ranges[i][0],
"Unexpected range start: %d" % reversed_rl[i].start)
- self.assertEquals(reversed_rl[i].end, expected_ranges[i][1],
+ self.assertEqual(reversed_rl[i].end, expected_ranges[i][1],
"Unexpected range end: %d" % reversed_rl[i].end)
def test_mergeinfo_sort(self):
@@ -113,15 +113,15 @@ class SubversionMergeinfoTestCase(unitte
self.MERGEINFO_NBR_REV_RANGES)
def test_mergeinfo_get(self):
- mergeinfo = repos.fs_get_mergeinfo(self.repos, ['/trunk'], self.rev,
+ mergeinfo = repos.fs_get_mergeinfo(self.repos, [b'/trunk'], self.rev,
core.svn_mergeinfo_inherited,
False, None, None)
expected_mergeinfo = \
- { '/trunk' :
- { '/branches/a' : [RevRange(2, 11)],
- '/branches/b' : [RevRange(9, 13)],
- '/branches/c' : [RevRange(2, 16)],
- '/trunk' : [RevRange(1, 9)], },
+ { b'/trunk' :
+ { b'/branches/a' : [RevRange(2, 11)],
+ b'/branches/b' : [RevRange(9, 13)],
+ b'/branches/c' : [RevRange(2, 16)],
+ b'/trunk' : [RevRange(1, 9)], },
}
self.compare_mergeinfo_catalogs(mergeinfo, expected_mergeinfo)
@@ -131,7 +131,7 @@ class SubversionMergeinfoTestCase(unitte
# When reference counting is working properly, each svn_merge_range_t in
# the returned mergeinfo will have a ref count of 1...
mergeinfo = core.svn_mergeinfo_parse(self.TEXT_MERGEINFO1)
- for (path, rangelist) in mergeinfo.items():
+ for (path, rangelist) in core._as_list(mergeinfo.items()):
# ....and now 2 (incref during iteration of rangelist)
for (i, r) in enumerate(rangelist):
@@ -144,7 +144,7 @@ class SubversionMergeinfoTestCase(unitte
# Note: if path and index are not '/trunk' and 0 respectively, then
# only some of the range objects are leaking, which is, as far as
# leaks go, even more impressive.
- self.assertEquals(refcount, expected, (
+ self.assertEqual(refcount, expected, (
"Memory leak! Expected a ref count of %d for svn_merge_range_t "
"object, but got %d instead (path: %s, index: %d). Probable "
"cause: incorrect Py_INCREF/Py_DECREF usage in libsvn_swig_py/"
@@ -165,7 +165,7 @@ class SubversionMergeinfoTestCase(unitte
del mergeinfo
gc.collect()
lingering = get_svn_merge_range_t_objects()
- self.assertEquals(lingering, list(), (
+ self.assertEqual(lingering, list(), (
"Memory leak! Found lingering svn_merge_range_t objects left over from "
"our call to svn_mergeinfo_parse(), even though we explicitly deleted "
"the returned mergeinfo object. Probable cause: incorrect Py_INCREF/"
@@ -177,16 +177,16 @@ class SubversionMergeinfoTestCase(unitte
self.inspect_rangelist_tuple(rangelist, nbr_rev_ranges)
def inspect_rangelist_tuple(self, rangelist, nbr_rev_ranges):
- self.assert_(rangelist is not None,
+ self.assertTrue(rangelist is not None,
"Rangelist for '%s' not parsed" % self.MERGEINFO_SRC)
- self.assertEquals(len(rangelist), nbr_rev_ranges,
+ self.assertEqual(len(rangelist), nbr_rev_ranges,
"Wrong number of revision ranges parsed")
- self.assertEquals(rangelist[0].inheritable, True,
+ self.assertEqual(rangelist[0].inheritable, True,
"Unexpected revision range 'non-inheritable' flag: %s" %
rangelist[0].inheritable)
- self.assertEquals(rangelist[1].start, 26,
+ self.assertEqual(rangelist[1].start, 26,
"Unexpected revision range end: %d" % rangelist[1].start)
- self.assertEquals(rangelist[2].inheritable, False,
+ self.assertEqual(rangelist[2].inheritable, False,
"Missing revision range 'non-inheritable' flag")
def compare_mergeinfo_catalogs(self, catalog1, catalog2):
@@ -194,7 +194,7 @@ class SubversionMergeinfoTestCase(unitte
keys2 = sorted(catalog2.keys())
self.assertEqual(keys1, keys2)
- for k in catalog1.keys():
+ for k in catalog1:
self.compare_mergeinfos(catalog1[k], catalog2[k])
def compare_mergeinfos(self, mergeinfo1, mergeinfo2):
@@ -202,7 +202,7 @@ class SubversionMergeinfoTestCase(unitte
keys2 = sorted(mergeinfo2.keys())
self.assertEqual(keys1, keys2)
- for k in mergeinfo1.keys():
+ for k in mergeinfo1:
self.compare_rangelists(mergeinfo1[k], mergeinfo2[k])
def compare_rangelists(self, rangelist1, rangelist2):
Modified: subversion/branches/addremove/subversion/bindings/swig/python/tests/ra.py
URL: http://svn.apache.org/viewvc/subversion/branches/addremove/subversion/bindings/swig/python/tests/ra.py?rev=1878061&r1=1878060&r2=1878061&view=diff
==============================================================================
--- subversion/branches/addremove/subversion/bindings/swig/python/tests/ra.py (original)
+++ subversion/branches/addremove/subversion/bindings/swig/python/tests/ra.py Sat May 23 14:16:56 2020
@@ -22,12 +22,7 @@ import unittest, setup_path
from svn import core, repos, fs, delta, ra
from sys import version_info # For Python version check
-if version_info[0] >= 3:
- # Python >=3.0
- from io import StringIO
-else:
- # Python <3.0
- from StringIO import StringIO
+from io import BytesIO
import utils
@@ -58,16 +53,16 @@ class SubversionRepositoryAccessTestCase
def test_get_file(self):
# Test getting the properties of a file
fs_revnum = fs.youngest_rev(self.fs)
- rev, properties = ra.get_file(self.ra_ctx, "trunk/README2.txt",
+ rev, properties = ra.get_file(self.ra_ctx, b"trunk/README2.txt",
core.SVN_INVALID_REVNUM, None)
self.assertEqual(rev, fs_revnum)
- self.assertEqual(properties["svn:mime-type"], "text/plain")
+ self.assertEqual(properties[b"svn:mime-type"], b"text/plain")
# Test getting the contents of a file
- filestream = StringIO()
- rev, properties = ra.get_file(self.ra_ctx, "trunk/README2.txt",
+ filestream = BytesIO()
+ rev, properties = ra.get_file(self.ra_ctx, b"trunk/README2.txt",
fs_revnum, filestream)
- self.assertEqual("A test.\n", filestream.getvalue())
+ self.assertEqual(b"A test.\n", filestream.getvalue())
def test_get_repos_root(self):
root = ra.get_repos_root(self.ra_ctx)
@@ -84,53 +79,53 @@ class SubversionRepositoryAccessTestCase
self.assertEqual(ra_revnum, fs_revnum)
def test_get_dir2(self):
- (dirents, _, props) = ra.get_dir2(self.ra_ctx, '', 1, core.SVN_DIRENT_KIND)
- self.assert_('trunk' in dirents)
- self.assert_('branches' in dirents)
- self.assert_('tags' in dirents)
- self.assertEqual(dirents['trunk'].kind, core.svn_node_dir)
- self.assertEqual(dirents['branches'].kind, core.svn_node_dir)
- self.assertEqual(dirents['tags'].kind, core.svn_node_dir)
- self.assert_(core.SVN_PROP_ENTRY_UUID in props)
- self.assert_(core.SVN_PROP_ENTRY_LAST_AUTHOR in props)
+ (dirents, _, props) = ra.get_dir2(self.ra_ctx, b'', 1, core.SVN_DIRENT_KIND)
+ self.assertTrue(b'trunk' in dirents)
+ self.assertTrue(b'branches' in dirents)
+ self.assertTrue(b'tags' in dirents)
+ self.assertEqual(dirents[b'trunk'].kind, core.svn_node_dir)
+ self.assertEqual(dirents[b'branches'].kind, core.svn_node_dir)
+ self.assertEqual(dirents[b'tags'].kind, core.svn_node_dir)
+ self.assertTrue(core.SVN_PROP_ENTRY_UUID in props)
+ self.assertTrue(core.SVN_PROP_ENTRY_LAST_AUTHOR in props)
- (dirents, _, _) = ra.get_dir2(self.ra_ctx, 'trunk', 1, core.SVN_DIRENT_KIND)
+ (dirents, _, _) = ra.get_dir2(self.ra_ctx, b'trunk', 1, core.SVN_DIRENT_KIND)
self.assertEqual(dirents, {})
- (dirents, _, _) = ra.get_dir2(self.ra_ctx, 'trunk', 10,
+ (dirents, _, _) = ra.get_dir2(self.ra_ctx, b'trunk', 10,
core.SVN_DIRENT_KIND)
- self.assert_('README2.txt' in dirents)
- self.assertEqual(dirents['README2.txt'].kind, core.svn_node_file)
+ self.assertTrue(b'README2.txt' in dirents)
+ self.assertEqual(dirents[b'README2.txt'].kind, core.svn_node_file)
def test_commit3(self):
commit_info = []
def my_callback(info, pool):
commit_info.append(info)
- revprops = {"svn:log": "foobar", "testprop": ""}
+ revprops = {b"svn:log": b"foobar", b"testprop": b""}
editor, edit_baton = ra.get_commit_editor3(self.ra_ctx, revprops, my_callback, None, False)
root = editor.open_root(edit_baton, 4)
self.assertNotEqual(root, None)
- child = editor.add_directory("bla3", root, None, 0)
+ child = editor.add_directory(b"bla3", root, None, 0)
self.assertNotEqual(child, None)
editor.close_edit(edit_baton)
info = commit_info[0]
self.assertEqual(info.revision, fs.youngest_rev(self.fs))
- revprops['svn:author'] = info.author
- revprops['svn:date'] = info.date
+ revprops[b'svn:author'] = info.author
+ revprops[b'svn:date'] = info.date
self.assertEqual(ra.rev_proplist(self.ra_ctx, info.revision), revprops)
def test_commit2(self):
def my_callback(info, pool):
self.assertEqual(info.revision, fs.youngest_rev(self.fs))
- editor, edit_baton = ra.get_commit_editor2(self.ra_ctx, "foobar", my_callback, None, False)
+ editor, edit_baton = ra.get_commit_editor2(self.ra_ctx, b"foobar", my_callback, None, False)
root = editor.open_root(edit_baton, 4)
self.assertNotEqual(root, None)
- child = editor.add_directory("bla", root, None, 0)
+ child = editor.add_directory(b"bla", root, None, 0)
self.assertNotEqual(child, None)
editor.close_edit(edit_baton)
@@ -138,18 +133,18 @@ class SubversionRepositoryAccessTestCase
def my_callback(revision, date, author):
self.assertEqual(revision, fs.youngest_rev(self.fs))
- editor, edit_baton = ra.get_commit_editor(self.ra_ctx, "foobar", my_callback, None, False)
+ editor, edit_baton = ra.get_commit_editor(self.ra_ctx, b"foobar", my_callback, None, False)
root = editor.open_root(edit_baton, 4)
- child = editor.add_directory("blah", root, None, 0)
+ child = editor.add_directory(b"blah", root, None, 0)
editor.close_edit(edit_baton)
def test_delta_driver_commit(self):
# Setup paths we'll commit in this test.
- to_delete = ['trunk/README.txt', 'trunk/dir1/dir2']
- to_mkdir = ['test_delta_driver_commit.d', 'test_delta_driver_commit2.d']
- to_add = ['test_delta_driver_commit', 'test_delta_driver_commit2']
- to_dir_prop = ['trunk/dir1/dir3', 'test_delta_driver_commit2.d']
- to_file_prop = ['trunk/README2.txt', 'test_delta_driver_commit2']
+ to_delete = [b'trunk/README.txt', b'trunk/dir1/dir2']
+ to_mkdir = [b'test_delta_driver_commit.d', b'test_delta_driver_commit2.d']
+ to_add = [b'test_delta_driver_commit', b'test_delta_driver_commit2']
+ to_dir_prop = [b'trunk/dir1/dir3', b'test_delta_driver_commit2.d']
+ to_file_prop = [b'trunk/README2.txt', b'test_delta_driver_commit2']
all_paths = {}
for i in to_delete + to_mkdir + to_add + to_dir_prop + to_file_prop:
all_paths[i] = True
@@ -159,12 +154,12 @@ class SubversionRepositoryAccessTestCase
commit_info = []
def commit_cb(info, pool):
commit_info.append(info)
- revprops = {"svn:log": "foobar", "testprop": ""}
+ revprops = {b"svn:log": b"foobar", b"testprop": b""}
(editor, edit_baton) = ra.get_commit_editor3(self.ra_ctx, revprops,
commit_cb, None, False)
try:
def driver_cb(parent, path, pool):
- self.assert_(path in all_paths)
+ self.assertTrue(path in all_paths)
dir_baton = file_baton = None
if path in to_delete:
# Leave dir_baton alone, as it must be None for delete.
@@ -182,16 +177,16 @@ class SubversionRepositoryAccessTestCase
if dir_baton is None:
dir_baton = editor.open_directory(path, parent, revision, pool)
editor.change_dir_prop(dir_baton,
- 'test_delta_driver_commit', 'foo', pool)
+ b'test_delta_driver_commit', b'foo', pool)
elif path in to_file_prop:
if file_baton is None:
file_baton = editor.open_file(path, parent, revision, pool)
editor.change_file_prop(file_baton,
- 'test_delta_driver_commit', 'foo', pool)
+ b'test_delta_driver_commit', b'foo', pool)
if file_baton is not None:
editor.close_file(file_baton, None, pool)
return dir_baton
- delta.path_driver(editor, edit_baton, -1, list(all_paths.keys()), driver_cb)
+ delta.path_driver(editor, edit_baton, -1, core._as_list(all_paths.keys()), driver_cb)
editor.close_edit(edit_baton)
except:
try:
@@ -204,8 +199,8 @@ class SubversionRepositoryAccessTestCase
info = commit_info[0]
if info.author is not None:
- revprops['svn:author'] = info.author
- revprops['svn:date'] = info.date
+ revprops[b'svn:author'] = info.author
+ revprops[b'svn:date'] = info.date
self.assertEqual(ra.rev_proplist(self.ra_ctx, info.revision), revprops)
receiver_called = [False]
@@ -214,22 +209,22 @@ class SubversionRepositoryAccessTestCase
self.assertEqual(revision, info.revision)
self.assertEqual(author, info.author)
self.assertEqual(date, info.date)
- self.assertEqual(message, revprops['svn:log'])
- for (path, change) in changed_paths.items():
- path = path.lstrip('/')
- self.assert_(path in all_paths)
+ self.assertEqual(message, revprops[b'svn:log'])
+ for (path, change) in core._as_list(changed_paths.items()):
+ path = path.lstrip(b'/')
+ self.assertTrue(path in all_paths)
if path in to_delete:
- self.assertEqual(change.action, 'D')
+ self.assertEqual(change.action, b'D')
elif path in to_mkdir or path in to_add:
- self.assertEqual(change.action, 'A')
+ self.assertEqual(change.action, b'A')
elif path in to_dir_prop or path in to_file_prop:
- self.assertEqual(change.action, 'M')
- ra.get_log(self.ra_ctx, [''], info.revision, info.revision,
+ self.assertEqual(change.action, b'M')
+ ra.get_log(self.ra_ctx, [b''], info.revision, info.revision,
0, # limit
True, # discover_changed_paths
True, # strict_node_history
receiver)
- self.assert_(receiver_called[0])
+ self.assertTrue(receiver_called[0])
def test_do_diff2(self):
@@ -251,26 +246,26 @@ class SubversionRepositoryAccessTestCase
sess_url = ra.get_session_url(self.ra_ctx)
try:
- ra.reparent(self.ra_ctx, self.repos_uri+"/trunk")
+ ra.reparent(self.ra_ctx, self.repos_uri+b"/trunk")
reporter, reporter_baton = ra.do_diff2(self.ra_ctx, fs_revnum,
- "README.txt", 0, 0, 1,
+ b"README.txt", 0, 0, 1,
self.repos_uri
- +"/trunk/README.txt",
+ +b"/trunk/README.txt",
e_ptr, e_baton)
- reporter.set_path(reporter_baton, "", 0, True, None)
+ reporter.set_path(reporter_baton, b"", 0, True, None)
reporter.finish_report(reporter_baton)
finally:
ra.reparent(self.ra_ctx, sess_url)
- self.assertEqual("A test.\n", editor.textdeltas[0].new_data)
+ self.assertEqual(b"A test.\n", editor.textdeltas[0].new_data)
self.assertEqual(1, len(editor.textdeltas))
def test_get_locations(self):
- locations = ra.get_locations(self.ra_ctx, "trunk/README.txt", 2, list(range(1, 5)))
+ locations = ra.get_locations(self.ra_ctx, b"trunk/README.txt", 2, list(range(1, 5)))
self.assertEqual(locations, {
- 2: '/trunk/README.txt',
- 3: '/trunk/README.txt',
- 4: '/trunk/README.txt'})
+ 2: b'/trunk/README.txt',
+ 3: b'/trunk/README.txt',
+ 4: b'/trunk/README.txt'})
def test_has_capability(self):
self.assertEqual(True, ra.has_capability(self.ra_ctx,
@@ -278,24 +273,24 @@ class SubversionRepositoryAccessTestCase
def test_get_file_revs(self):
def rev_handler(path, rev, rev_props, prop_diffs, pool):
- self.assert_(rev == 2 or rev == 3)
- self.assertEqual(path, "/trunk/README.txt")
+ self.assertTrue(rev == 2 or rev == 3)
+ self.assertEqual(path, b"/trunk/README.txt")
if rev == 2:
self.assertEqual(rev_props, {
- 'svn:log': 'Added README.',
- 'svn:author': 'john',
- 'svn:date': '2005-04-01T13:12:18.216267Z'
+ b'svn:log': b'Added README.',
+ b'svn:author': b'john',
+ b'svn:date': b'2005-04-01T13:12:18.216267Z'
})
self.assertEqual(prop_diffs, {})
elif rev == 3:
self.assertEqual(rev_props, {
- 'svn:log': 'Fixed README.\n',
- 'svn:author': 'kate',
- 'svn:date': '2005-04-01T13:24:58.234643Z'
+ b'svn:log': b'Fixed README.\n',
+ b'svn:author': b'kate',
+ b'svn:date': b'2005-04-01T13:24:58.234643Z'
})
- self.assertEqual(prop_diffs, {'svn:mime-type': 'text/plain', 'svn:eol-style': 'native'})
+ self.assertEqual(prop_diffs, {b'svn:mime-type': b'text/plain', b'svn:eol-style': b'native'})
- ra.get_file_revs(self.ra_ctx, "trunk/README.txt", 0, 10, rev_handler)
+ ra.get_file_revs(self.ra_ctx, b"trunk/README.txt", 0, 10, rev_handler)
def test_lock(self):
@@ -304,12 +299,12 @@ class SubversionRepositoryAccessTestCase
self.errors = 0
def callback(path, do_lock, lock, ra_err, pool):
self.calls += 1
- self.assertEqual(path, "trunk/README2.txt")
+ self.assertEqual(path, b"trunk/README2.txt")
if lock:
- self.assertEqual(lock.owner, "jrandom")
+ self.assertEqual(lock.owner, b"jrandom")
self.locks += 1
if ra_err:
- self.assert_(ra_err.apr_err == core.SVN_ERR_FS_PATH_ALREADY_LOCKED
+ self.assertTrue(ra_err.apr_err == core.SVN_ERR_FS_PATH_ALREADY_LOCKED
or ra_err.apr_err == core.SVN_ERR_FS_NO_SUCH_LOCK)
self.errors += 1
@@ -317,50 +312,50 @@ class SubversionRepositoryAccessTestCase
self.callbacks.auth_baton = core.svn_auth_open(providers)
core.svn_auth_set_parameter(self.callbacks.auth_baton,
core.SVN_AUTH_PARAM_DEFAULT_USERNAME,
- "jrandom")
+ b"jrandom")
self.ra_ctx = ra.open2(self.repos_uri, self.callbacks, {})
rev = fs.youngest_rev(self.fs)
- ra.lock(self.ra_ctx, {"trunk/README2.txt":rev}, "sleutel", False, callback)
+ ra.lock(self.ra_ctx, {b"trunk/README2.txt":rev}, b"sleutel", False, callback)
self.assertEqual(self.calls, 1)
self.assertEqual(self.locks, 1)
self.assertEqual(self.errors, 0)
self.calls = 0
self.locks = 0
- ra.lock(self.ra_ctx, {"trunk/README2.txt":rev}, "sleutel", False, callback)
+ ra.lock(self.ra_ctx, {b"trunk/README2.txt":rev}, b"sleutel", False, callback)
self.assertEqual(self.calls, 1)
self.assertEqual(self.locks, 0)
self.assertEqual(self.errors, 1)
self.calls = 0
self.errors = 0
- the_lock = fs.get_lock(self.fs, "/trunk/README2.txt")
- ra.unlock(self.ra_ctx, {"trunk/README2.txt":the_lock.token}, False, callback)
+ the_lock = fs.get_lock(self.fs, b"/trunk/README2.txt")
+ ra.unlock(self.ra_ctx, {b"trunk/README2.txt":the_lock.token}, False, callback)
self.assertEqual(self.calls, 1)
self.assertEqual(self.locks, 0)
self.assertEqual(self.errors, 0)
self.calls = 0
- ra.unlock(self.ra_ctx, {"trunk/README2.txt":the_lock.token}, False, callback)
+ ra.unlock(self.ra_ctx, {b"trunk/README2.txt":the_lock.token}, False, callback)
self.assertEqual(self.calls, 1)
self.assertEqual(self.locks, 0)
self.assertEqual(self.errors, 1)
def test_get_log2(self):
- # Get an interesting commmit.
+ # Get an interesting commit.
self.test_commit3()
rev = fs.youngest_rev(self.fs)
revprops = ra.rev_proplist(self.ra_ctx, rev)
- self.assert_("svn:log" in revprops)
- self.assert_("testprop" in revprops)
+ self.assertTrue(b"svn:log" in revprops)
+ self.assertTrue(b"testprop" in revprops)
def receiver(log_entry, pool):
called[0] = True
self.assertEqual(log_entry.revision, rev)
if discover_changed_paths:
- self.assertEqual(list(log_entry.changed_paths.keys()), ['/bla3'])
- changed_path = log_entry.changed_paths['/bla3']
- self.assert_(changed_path.action in ['A', 'D', 'R', 'M'])
+ self.assertEqual(core._as_list(log_entry.changed_paths.keys()), [b'/bla3'])
+ changed_path = log_entry.changed_paths[b'/bla3']
+ self.assertTrue(changed_path.action in [b'A', b'D', b'R', b'M'])
self.assertEqual(changed_path.copyfrom_path, None)
self.assertEqual(changed_path.copyfrom_rev, -1)
else:
@@ -368,7 +363,7 @@ class SubversionRepositoryAccessTestCase
if log_revprops is None:
self.assertEqual(log_entry.revprops, revprops)
elif len(log_revprops) == 0:
- self.assert_(log_entry.revprops == None or len(log_entry.revprops) == 0)
+ self.assertTrue(log_entry.revprops == None or len(log_entry.revprops) == 0)
else:
revprop_names = sorted(log_entry.revprops.keys())
log_revprops.sort()
@@ -381,9 +376,9 @@ class SubversionRepositoryAccessTestCase
for log_revprops in (
# Retrieve the standard three.
- ["svn:author", "svn:date", "svn:log"],
+ [b"svn:author", b"svn:date", b"svn:log"],
# Retrieve just testprop.
- ["testprop"],
+ [b"testprop"],
# Retrieve all.
None,
# Retrieve none.
@@ -391,14 +386,14 @@ class SubversionRepositoryAccessTestCase
):
for discover_changed_paths in [True, False]:
called = [False]
- ra.get_log2(self.ra_ctx, [""],
+ ra.get_log2(self.ra_ctx, [b""],
rev, rev, # start, end
1, # limit
discover_changed_paths,
True, # strict_node_history
False, # include_merged_revisions
log_revprops, receiver)
- self.assert_(called[0])
+ self.assertTrue(called[0])
def test_update(self):
class TestEditor(delta.Editor):
@@ -408,23 +403,23 @@ class SubversionRepositoryAccessTestCase
e_ptr, e_baton = delta.make_editor(editor)
- reporter, reporter_baton = ra.do_update(self.ra_ctx, 10, "", True, e_ptr, e_baton)
+ reporter, reporter_baton = ra.do_update(self.ra_ctx, 10, b"", True, e_ptr, e_baton)
- reporter.set_path(reporter_baton, "", 0, True, None)
+ reporter.set_path(reporter_baton, b"", 0, True, None)
reporter.finish_report(reporter_baton)
def test_namestring(self):
# Only ra-{svn,serf} support this right now.
uri = self.repos_uri
- if uri.startswith('http') or uri.startswith('svn'):
+ if uri.startswith(b'http') or uri.startswith(b'svn'):
called = [False]
def cb(pool):
called[0] = True
- return 'namestring_test'
+ return b'namestring_test'
self.callbacks.get_client_string = cb
- ra.stat(self.ra_ctx, "", 1)
- self.assert_(called[0])
+ ra.stat(self.ra_ctx, b"", 1)
+ self.assertTrue(called[0])
def suite():
return unittest.defaultTestLoader.loadTestsFromTestCase(
Modified: subversion/branches/addremove/subversion/bindings/swig/python/tests/repository.py
URL: http://svn.apache.org/viewvc/subversion/branches/addremove/subversion/bindings/swig/python/tests/repository.py?rev=1878061&r1=1878060&r2=1878061&view=diff
==============================================================================
--- subversion/branches/addremove/subversion/bindings/swig/python/tests/repository.py (original)
+++ subversion/branches/addremove/subversion/bindings/swig/python/tests/repository.py Sat May 23 14:16:56 2020
@@ -20,12 +20,7 @@
#
import unittest, setup_path, os, sys
from sys import version_info # For Python version check
-if version_info[0] >= 3:
- # Python >=3.0
- from io import StringIO
-else:
- # Python <3.0
- from StringIO import StringIO
+from io import BytesIO
from svn import core, repos, fs, delta
from svn.core import SubversionException
import utils
@@ -49,34 +44,34 @@ class DumpStreamParser(repos.ParseFns3):
repos.ParseFns3.__init__(self)
self.ops = []
def magic_header_record(self, version, pool=None):
- self.ops.append(("magic-header", version))
+ self.ops.append((b"magic-header", version))
def uuid_record(self, uuid, pool=None):
- self.ops.append(("uuid", uuid))
+ self.ops.append((b"uuid", uuid))
def new_revision_record(self, headers, pool=None):
rev = int(headers[repos.DUMPFILE_REVISION_NUMBER])
- self.ops.append(("new-revision", rev))
+ self.ops.append((b"new-revision", rev))
return rev
def close_revision(self, revision_baton):
- self.ops.append(("close-revision", revision_baton))
+ self.ops.append((b"close-revision", revision_baton))
def new_node_record(self, headers, revision_baton, pool=None):
node = headers[repos.DUMPFILE_NODE_PATH]
- self.ops.append(("new-node", revision_baton, node))
+ self.ops.append((b"new-node", revision_baton, node))
return (revision_baton, node)
def close_node(self, node_baton):
- self.ops.append(("close-node", node_baton[0], node_baton[1]))
+ self.ops.append((b"close-node", node_baton[0], node_baton[1]))
def set_revision_property(self, revision_baton, name, value):
- self.ops.append(("set-revision-prop", revision_baton, name, value))
+ self.ops.append((b"set-revision-prop", revision_baton, name, value))
def set_node_property(self, node_baton, name, value):
- self.ops.append(("set-node-prop", node_baton[0], node_baton[1], name, value))
+ self.ops.append((b"set-node-prop", node_baton[0], node_baton[1], name, value))
def remove_node_props(self, node_baton):
- self.ops.append(("remove-node-props", node_baton[0], node_baton[1]))
+ self.ops.append((b"remove-node-props", node_baton[0], node_baton[1]))
def delete_node_property(self, node_baton, name):
- self.ops.append(("delete-node-prop", node_baton[0], node_baton[1], name))
+ self.ops.append((b"delete-node-prop", node_baton[0], node_baton[1], name))
def apply_textdelta(self, node_baton):
- self.ops.append(("apply-textdelta", node_baton[0], node_baton[1]))
+ self.ops.append((b"apply-textdelta", node_baton[0], node_baton[1]))
return None
def set_fulltext(self, node_baton):
- self.ops.append(("set-fulltext", node_baton[0], node_baton[1]))
+ self.ops.append((b"set-fulltext", node_baton[0], node_baton[1]))
return None
@@ -109,17 +104,17 @@ class SubversionRepositoryTestCase(unitt
raise core.SubversionException(apr_err=core.SVN_ERR_CEASE_INVOCATION,
message="Hi from history_lookup")
- repos.history2(self.fs, '/trunk/README2.txt', history_lookup, None, 0,
+ repos.history2(self.fs, b'/trunk/README2.txt', history_lookup, None, 0,
self.rev, True)
self.assertEqual(len(revs), 1)
def test_create(self):
"""Make sure that repos.create doesn't segfault when we set fs-type
using a config hash"""
- fs_config = { "fs-type": "fsfs" }
+ fs_config = { b"fs-type": b"fsfs" }
for i in range(5):
path = self.temper.alloc_empty_dir(suffix='-repository-create%d' % i)
- repos.create(path, "", "", None, fs_config)
+ repos.create(path, b"", b"", None, fs_config)
def test_dump_fs2(self):
"""Test the dump_fs2 function"""
@@ -130,18 +125,18 @@ class SubversionRepositoryTestCase(unitt
self.callback_calls += 1
return None
- dumpstream = StringIO()
- feedbackstream = StringIO()
+ dumpstream = BytesIO()
+ feedbackstream = BytesIO()
repos.dump_fs2(self.repos, dumpstream, feedbackstream, 0, self.rev, 0, 0,
is_cancelled)
# Check that we can dump stuff
dump = dumpstream.getvalue()
feedback = feedbackstream.getvalue()
- expected_feedback = "* Dumped revision " + str(self.rev)
- self.assertEquals(dump.count("Node-path: trunk/README.txt"), 2)
- self.assertEquals(feedback.count(expected_feedback), 1)
- self.assertEquals(self.callback_calls, 13)
+ expected_feedback = b"* Dumped revision " + str(self.rev).encode('utf-8')
+ self.assertEqual(dump.count(b"Node-path: trunk/README.txt"), 2)
+ self.assertEqual(feedback.count(expected_feedback), 1)
+ self.assertEqual(self.callback_calls, 13)
# Check that the dump can be cancelled
self.assertRaises(SubversionException, repos.dump_fs2,
@@ -154,22 +149,22 @@ class SubversionRepositoryTestCase(unitt
self.assertRaises(ValueError, repos.dump_fs2,
self.repos, dumpstream, feedbackstream, 0, self.rev, 0, 0, None)
- dumpstream = StringIO()
- feedbackstream = StringIO()
+ dumpstream = BytesIO()
+ feedbackstream = BytesIO()
# Check that we can grab the feedback stream, but not the dumpstream
repos.dump_fs2(self.repos, None, feedbackstream, 0, self.rev, 0, 0, None)
feedback = feedbackstream.getvalue()
- self.assertEquals(feedback.count(expected_feedback), 1)
+ self.assertEqual(feedback.count(expected_feedback), 1)
# Check that we can grab the dumpstream, but not the feedbackstream
repos.dump_fs2(self.repos, dumpstream, None, 0, self.rev, 0, 0, None)
dump = dumpstream.getvalue()
- self.assertEquals(dump.count("Node-path: trunk/README.txt"), 2)
+ self.assertEqual(dump.count(b"Node-path: trunk/README.txt"), 2)
# Check that we can ignore both the dumpstream and the feedbackstream
repos.dump_fs2(self.repos, dumpstream, None, 0, self.rev, 0, 0, None)
- self.assertEquals(feedback.count(expected_feedback), 1)
+ self.assertEqual(feedback.count(expected_feedback), 1)
# FIXME: The Python bindings don't check for 'NULL' values for
# svn_repos_t objects, so the following call segfaults
@@ -182,55 +177,70 @@ class SubversionRepositoryTestCase(unitt
return None
dump_path = os.path.join(os.path.dirname(sys.argv[0]),
"trac/versioncontrol/tests/svnrepos.dump")
- stream = open(dump_path)
+ stream = open(dump_path, 'rb')
dsp = DumpStreamParser()
ptr, baton = repos.make_parse_fns3(dsp)
repos.parse_dumpstream3(stream, ptr, baton, False, is_cancelled)
stream.close()
self.assertEqual(self.cancel_calls, 76)
expected_list = [
- ("magic-header", 2),
- ('uuid', '92ea810a-adf3-0310-b540-bef912dcf5ba'),
- ('new-revision', 0),
- ('set-revision-prop', 0, 'svn:date', '2005-04-01T09:57:41.312767Z'),
- ('close-revision', 0),
- ('new-revision', 1),
- ('set-revision-prop', 1, 'svn:log', 'Initial directory layout.'),
- ('set-revision-prop', 1, 'svn:author', 'john'),
- ('set-revision-prop', 1, 'svn:date', '2005-04-01T10:00:52.353248Z'),
- ('new-node', 1, 'branches'),
- ('remove-node-props', 1, 'branches'),
- ('close-node', 1, 'branches'),
- ('new-node', 1, 'tags'),
- ('remove-node-props', 1, 'tags'),
- ('close-node', 1, 'tags'),
- ('new-node', 1, 'trunk'),
- ('remove-node-props', 1, 'trunk'),
- ('close-node', 1, 'trunk'),
- ('close-revision', 1),
- ('new-revision', 2),
- ('set-revision-prop', 2, 'svn:log', 'Added README.'),
- ('set-revision-prop', 2, 'svn:author', 'john'),
- ('set-revision-prop', 2, 'svn:date', '2005-04-01T13:12:18.216267Z'),
- ('new-node', 2, 'trunk/README.txt'),
- ('remove-node-props', 2, 'trunk/README.txt'),
- ('set-fulltext', 2, 'trunk/README.txt'),
- ('close-node', 2, 'trunk/README.txt'),
- ('close-revision', 2), ('new-revision', 3),
- ('set-revision-prop', 3, 'svn:log', 'Fixed README.\n'),
- ('set-revision-prop', 3, 'svn:author', 'kate'),
- ('set-revision-prop', 3, 'svn:date', '2005-04-01T13:24:58.234643Z'),
- ('new-node', 3, 'trunk/README.txt'),
- ('remove-node-props', 3, 'trunk/README.txt'),
- ('set-node-prop', 3, 'trunk/README.txt', 'svn:mime-type', 'text/plain'),
- ('set-node-prop', 3, 'trunk/README.txt', 'svn:eol-style', 'native'),
- ('set-fulltext', 3, 'trunk/README.txt'),
- ('close-node', 3, 'trunk/README.txt'), ('close-revision', 3),
+ (b"magic-header", 2),
+ (b'uuid', b'92ea810a-adf3-0310-b540-bef912dcf5ba'),
+ (b'new-revision', 0),
+ (b'set-revision-prop', 0, b'svn:date', b'2005-04-01T09:57:41.312767Z'),
+ (b'close-revision', 0),
+ (b'new-revision', 1),
+ (b'set-revision-prop', 1, b'svn:log', b'Initial directory layout.'),
+ (b'set-revision-prop', 1, b'svn:author', b'john'),
+ (b'set-revision-prop', 1, b'svn:date', b'2005-04-01T10:00:52.353248Z'),
+ (b'new-node', 1, b'branches'),
+ (b'remove-node-props', 1, b'branches'),
+ (b'close-node', 1, b'branches'),
+ (b'new-node', 1, b'tags'),
+ (b'remove-node-props', 1, b'tags'),
+ (b'close-node', 1, b'tags'),
+ (b'new-node', 1, b'trunk'),
+ (b'remove-node-props', 1, b'trunk'),
+ (b'close-node', 1, b'trunk'),
+ (b'close-revision', 1),
+ (b'new-revision', 2),
+ (b'set-revision-prop', 2, b'svn:log', b'Added README.'),
+ (b'set-revision-prop', 2, b'svn:author', b'john'),
+ (b'set-revision-prop', 2, b'svn:date', b'2005-04-01T13:12:18.216267Z'),
+ (b'new-node', 2, b'trunk/README.txt'),
+ (b'remove-node-props', 2, b'trunk/README.txt'),
+ (b'set-fulltext', 2, b'trunk/README.txt'),
+ (b'close-node', 2, b'trunk/README.txt'),
+ (b'close-revision', 2), (b'new-revision', 3),
+ (b'set-revision-prop', 3, b'svn:log', b'Fixed README.\n'),
+ (b'set-revision-prop', 3, b'svn:author', b'kate'),
+ (b'set-revision-prop', 3, b'svn:date', b'2005-04-01T13:24:58.234643Z'),
+ (b'new-node', 3, b'trunk/README.txt'),
+ (b'remove-node-props', 3, b'trunk/README.txt'),
+ (b'set-node-prop', 3, b'trunk/README.txt', b'svn:mime-type', b'text/plain'),
+ (b'set-node-prop', 3, b'trunk/README.txt', b'svn:eol-style', b'native'),
+ (b'set-fulltext', 3, b'trunk/README.txt'),
+ (b'close-node', 3, b'trunk/README.txt'), (b'close-revision', 3),
]
# Compare only the first X nodes described in the expected list - otherwise
# the comparison list gets too long.
self.assertEqual(dsp.ops[:len(expected_list)], expected_list)
+ def test_parse_fns3_invalid_set_fulltext(self):
+ class DumpStreamParserSubclass(DumpStreamParser):
+ def set_fulltext(self, node_baton):
+ DumpStreamParser.set_fulltext(self, node_baton)
+ return 42
+ stream = open(os.path.join(os.path.dirname(sys.argv[0]),
+ "trac/versioncontrol/tests/svnrepos.dump"), "rb")
+ try:
+ dsp = DumpStreamParserSubclass()
+ ptr, baton = repos.make_parse_fns3(dsp)
+ self.assertRaises(TypeError, repos.parse_dumpstream3,
+ stream, ptr, baton, False, None)
+ finally:
+ stream.close()
+
def test_get_logs(self):
"""Test scope of get_logs callbacks"""
logs = []
@@ -239,16 +249,16 @@ class SubversionRepositoryTestCase(unitt
logs.append(paths)
# Run get_logs
- repos.get_logs(self.repos, ['/'], self.rev, 0, True, 0, addLog)
+ repos.get_logs(self.repos, [b'/'], self.rev, 0, True, 0, addLog)
# Count and verify changes
change_count = 0
for log in logs:
- for path_changed in log.values():
+ for path_changed in core._as_list(log.values()):
change_count += 1
path_changed.assert_valid()
- self.assertEqual(logs[2]["/tags/v1.1"].action, "A")
- self.assertEqual(logs[2]["/tags/v1.1"].copyfrom_path, "/branches/v1x")
+ self.assertEqual(logs[2][b"/tags/v1.1"].action, b"A")
+ self.assertEqual(logs[2][b"/tags/v1.1"].copyfrom_path, b"/branches/v1x")
self.assertEqual(len(logs), 12)
self.assertEqual(change_count, 19)
@@ -259,14 +269,14 @@ class SubversionRepositoryTestCase(unitt
prev_root = fs.revision_root(self.fs, self.rev-1)
editor = ChangeReceiver(this_root, prev_root)
e_ptr, e_baton = delta.make_editor(editor)
- repos.dir_delta(prev_root, '', '', this_root, '', e_ptr, e_baton,
+ repos.dir_delta(prev_root, b'', b'', this_root, b'', e_ptr, e_baton,
_authz_callback, 1, 1, 0, 0)
# Check results.
# Ignore the order in which the editor delivers the two sibling files.
self.assertEqual(set([editor.textdeltas[0].new_data,
editor.textdeltas[1].new_data]),
- set(["This is a test.\n", "A test.\n"]))
+ set([b"This is a test.\n", b"A test.\n"]))
self.assertEqual(len(editor.textdeltas), 2)
def test_unnamed_editor(self):
@@ -277,29 +287,29 @@ class SubversionRepositoryTestCase(unitt
this_root = fs.revision_root(self.fs, self.rev)
prev_root = fs.revision_root(self.fs, self.rev-1)
e_ptr, e_baton = delta.make_editor(ChangeReceiver(this_root, prev_root))
- repos.dir_delta(prev_root, '', '', this_root, '', e_ptr, e_baton,
+ repos.dir_delta(prev_root, b'', b'', this_root, b'', e_ptr, e_baton,
_authz_callback, 1, 1, 0, 0)
def test_retrieve_and_change_rev_prop(self):
"""Test playing with revprops"""
- self.assertEqual(repos.fs_revision_prop(self.repos, self.rev, "svn:log",
+ self.assertEqual(repos.fs_revision_prop(self.repos, self.rev, b"svn:log",
_authz_callback),
- "''(a few years later)'' Argh... v1.1 was buggy, "
- "after all")
+ b"''(a few years later)'' Argh... v1.1 was buggy, "
+ b"after all")
# We expect this to complain because we have no pre-revprop-change
# hook script for the repository.
self.assertRaises(SubversionException, repos.fs_change_rev_prop3,
- self.repos, self.rev, "jrandom", "svn:log",
- "Youngest revision", True, True, _authz_callback)
+ self.repos, self.rev, b"jrandom", b"svn:log",
+ b"Youngest revision", True, True, _authz_callback)
- repos.fs_change_rev_prop3(self.repos, self.rev, "jrandom", "svn:log",
- "Youngest revision", False, False,
+ repos.fs_change_rev_prop3(self.repos, self.rev, b"jrandom", b"svn:log",
+ b"Youngest revision", False, False,
_authz_callback)
- self.assertEqual(repos.fs_revision_prop(self.repos, self.rev, "svn:log",
+ self.assertEqual(repos.fs_revision_prop(self.repos, self.rev, b"svn:log",
_authz_callback),
- "Youngest revision")
+ b"Youngest revision")
def freeze_body(self, pool):
self.freeze_invoked += 1
@@ -314,52 +324,52 @@ class SubversionRepositoryTestCase(unitt
def test_lock_unlock(self):
"""Basic lock/unlock"""
- access = fs.create_access('jrandom')
+ access = fs.create_access(b'jrandom')
fs.set_access(self.fs, access)
- fs.lock(self.fs, '/trunk/README.txt', None, None, 0, 0, self.rev, False)
+ fs.lock(self.fs, b'/trunk/README.txt', None, None, 0, 0, self.rev, False)
try:
- fs.lock(self.fs, '/trunk/README.txt', None, None, 0, 0, self.rev, False)
+ fs.lock(self.fs, b'/trunk/README.txt', None, None, 0, 0, self.rev, False)
except core.SubversionException as exc:
self.assertEqual(exc.apr_err, core.SVN_ERR_FS_PATH_ALREADY_LOCKED)
- fs.lock(self.fs, '/trunk/README.txt', None, None, 0, 0, self.rev, True)
+ fs.lock(self.fs, b'/trunk/README.txt', None, None, 0, 0, self.rev, True)
self.calls = 0
self.errors = 0
def unlock_callback(path, lock, err, pool):
- self.assertEqual(path, '/trunk/README.txt')
+ self.assertEqual(path, b'/trunk/README.txt')
self.assertEqual(lock, None)
self.calls += 1
if err != None:
self.assertEqual(err.apr_err, core.SVN_ERR_FS_NO_SUCH_LOCK)
self.errors += 1
- the_lock = fs.get_lock(self.fs, '/trunk/README.txt')
- fs.unlock_many(self.fs, {'/trunk/README.txt':the_lock.token}, False,
+ the_lock = fs.get_lock(self.fs, b'/trunk/README.txt')
+ fs.unlock_many(self.fs, {b'/trunk/README.txt':the_lock.token}, False,
unlock_callback)
self.assertEqual(self.calls, 1)
self.assertEqual(self.errors, 0)
self.calls = 0
- fs.unlock_many(self.fs, {'/trunk/README.txt':the_lock.token}, False,
+ fs.unlock_many(self.fs, {b'/trunk/README.txt':the_lock.token}, False,
unlock_callback)
self.assertEqual(self.calls, 1)
self.assertEqual(self.errors, 1)
self.locks = 0
def lock_callback(path, lock, err, pool):
- self.assertEqual(path, '/trunk/README.txt')
+ self.assertEqual(path, b'/trunk/README.txt')
if lock != None:
- self.assertEqual(lock.owner, 'jrandom')
+ self.assertEqual(lock.owner, b'jrandom')
self.locks += 1
self.calls += 1
if err != None:
self.assertEqual(err.apr_err, core.SVN_ERR_FS_PATH_ALREADY_LOCKED)
self.errors += 1
-
+
self.calls = 0
self.errors = 0
target = fs.lock_target_create(None, self.rev)
- fs.lock_many(self.fs, {'trunk/README.txt':target},
+ fs.lock_many(self.fs, {b'trunk/README.txt':target},
None, False, 0, False, lock_callback)
self.assertEqual(self.calls, 1)
self.assertEqual(self.locks, 1)
@@ -367,7 +377,7 @@ class SubversionRepositoryTestCase(unitt
self.calls = 0
self.locks = 0
- fs.lock_many(self.fs, {'trunk/README.txt':target},
+ fs.lock_many(self.fs, {b'trunk/README.txt':target},
None, False, 0, False, lock_callback)
self.assertEqual(self.calls, 1)
self.assertEqual(self.locks, 0)
@@ -375,21 +385,21 @@ class SubversionRepositoryTestCase(unitt
self.calls = 0
self.errors = 0
- the_lock = fs.get_lock(self.fs, '/trunk/README.txt')
- repos.fs_unlock_many(self.repos, {'trunk/README.txt':the_lock.token},
+ the_lock = fs.get_lock(self.fs, b'/trunk/README.txt')
+ repos.fs_unlock_many(self.repos, {b'trunk/README.txt':the_lock.token},
False, unlock_callback)
self.assertEqual(self.calls, 1)
self.assertEqual(self.errors, 0)
self.calls = 0
- repos.fs_unlock_many(self.repos, {'trunk/README.txt':the_lock.token},
+ repos.fs_unlock_many(self.repos, {b'trunk/README.txt':the_lock.token},
False, unlock_callback)
self.assertEqual(self.calls, 1)
self.assertEqual(self.errors, 1)
self.calls = 0
self.errors = 0
- repos.fs_lock_many(self.repos, {'trunk/README.txt':target},
+ repos.fs_lock_many(self.repos, {b'trunk/README.txt':target},
None, False, 0, False, lock_callback)
self.assertEqual(self.calls, 1)
self.assertEqual(self.locks, 1)
@@ -397,7 +407,7 @@ class SubversionRepositoryTestCase(unitt
self.calls = 0
self.locks = 0
- repos.fs_lock_many(self.repos, {'trunk/README.txt':target},
+ repos.fs_lock_many(self.repos, {b'trunk/README.txt':target},
None, False, 0, False, lock_callback)
self.assertEqual(self.calls, 1)
self.assertEqual(self.locks, 0)
Modified: subversion/branches/addremove/subversion/bindings/swig/python/tests/run_all.py
URL: http://svn.apache.org/viewvc/subversion/branches/addremove/subversion/bindings/swig/python/tests/run_all.py?rev=1878061&r1=1878060&r2=1878061&view=diff
==============================================================================
--- subversion/branches/addremove/subversion/bindings/swig/python/tests/run_all.py (original)
+++ subversion/branches/addremove/subversion/bindings/swig/python/tests/run_all.py Sat May 23 14:16:56 2020
@@ -18,9 +18,10 @@
# under the License.
#
#
+import sys
import unittest, setup_path
-import mergeinfo, core, client, delta, checksum, pool, ra, wc, repository, \
- auth, trac.versioncontrol.tests
+import mergeinfo, core, client, delta, checksum, pool, fs, ra, wc, repository, \
+ auth, trac.versioncontrol.tests, typemap
from svn.core import svn_cache_config_get, svn_cache_config_set
# Run all tests
@@ -28,7 +29,10 @@ from svn.core import svn_cache_config_ge
def suite():
"""Run all tests"""
settings = svn_cache_config_get()
- settings.cache_size = long(1024*1024*32) ### Need explicit long
+ if sys.hexversion < 0x3000000:
+ settings.cache_size = long(1024*1024*32) ### Need explicit long
+ else:
+ settings.cache_size = 1024*1024*32
svn_cache_config_set(settings)
s = unittest.TestSuite()
s.addTest(core.suite())
@@ -37,11 +41,13 @@ def suite():
s.addTest(client.suite())
s.addTest(delta.suite())
s.addTest(pool.suite())
+ s.addTest(fs.suite())
s.addTest(ra.suite())
s.addTest(wc.suite())
s.addTest(repository.suite())
s.addTest(auth.suite())
s.addTest(trac.versioncontrol.tests.suite())
+ s.addTest(typemap.suite())
return s
if __name__ == '__main__':
Modified: subversion/branches/addremove/subversion/bindings/swig/python/tests/trac/versioncontrol/main.py
URL: http://svn.apache.org/viewvc/subversion/branches/addremove/subversion/bindings/swig/python/tests/trac/versioncontrol/main.py?rev=1878061&r1=1878060&r2=1878061&view=diff
==============================================================================
--- subversion/branches/addremove/subversion/bindings/swig/python/tests/trac/versioncontrol/main.py (original)
+++ subversion/branches/addremove/subversion/bindings/swig/python/tests/trac/versioncontrol/main.py Sat May 23 14:16:56 2020
@@ -154,12 +154,12 @@ class Node(object):
Represents a directory or file in the repository.
"""
- DIRECTORY = "dir"
- FILE = "file"
+ DIRECTORY = b"dir"
+ FILE = b"file"
def __init__(self, path, rev, kind):
assert kind in (Node.DIRECTORY, Node.FILE), "Unknown node kind %s" % kind
- self.path = str(path)
+ self.path = path
self.rev = rev
self.kind = kind
@@ -217,7 +217,7 @@ class Node(object):
content_type = property(lambda x: x.get_content_type())
def get_name(self):
- return self.path.split('/')[-1]
+ return self.path.split(b'/')[-1]
name = property(lambda x: x.get_name())
def get_last_modified(self):
@@ -233,11 +233,11 @@ class Changeset(object):
Represents a set of changes of a repository.
"""
- ADD = 'add'
- COPY = 'copy'
- DELETE = 'delete'
- EDIT = 'edit'
- MOVE = 'move'
+ ADD = b'add'
+ COPY = b'copy'
+ DELETE = b'delete'
+ EDIT = b'edit'
+ MOVE = b'move'
def __init__(self, rev, message, author, date):
self.rev = rev