You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@subversion.apache.org by ju...@apache.org on 2022/01/14 14:01:51 UTC
svn commit: r1897034 [9/37] - in /subversion/branches/multi-wc-format: ./ build/ build/ac-macros/ build/generator/ build/generator/swig/ build/generator/templates/ contrib/client-side/ contrib/client-side/svn_load_dirs/ contrib/hook-scripts/ contrib/se...
Modified: subversion/branches/multi-wc-format/subversion/bindings/swig/python/tests/auth.py
URL: http://svn.apache.org/viewvc/subversion/branches/multi-wc-format/subversion/bindings/swig/python/tests/auth.py?rev=1897034&r1=1897033&r2=1897034&view=diff
==============================================================================
--- subversion/branches/multi-wc-format/subversion/bindings/swig/python/tests/auth.py (original)
+++ subversion/branches/multi-wc-format/subversion/bindings/swig/python/tests/auth.py Fri Jan 14 14:01:45 2022
@@ -27,87 +27,87 @@ class SubversionAuthTestCase(unittest.Te
def test_open(self):
baton = core.svn_auth_open([])
- self.assert_(baton is not None)
+ self.assertTrue(baton is not None)
def test_set_parameter(self):
baton = core.svn_auth_open([])
- core.svn_auth_set_parameter(baton, "name", "somedata")
- core.svn_auth_set_parameter(baton, "name", None)
- core.svn_auth_set_parameter(baton, "name", 2)
- core.svn_auth_set_parameter(baton, "name",
+ core.svn_auth_set_parameter(baton, b"name", b"somedata")
+ core.svn_auth_set_parameter(baton, b"name", None)
+ core.svn_auth_set_parameter(baton, b"name", 2)
+ core.svn_auth_set_parameter(baton, b"name",
core.svn_auth_ssl_server_cert_info_t())
def test_invalid_cred_kind(self):
baton = core.svn_auth_open([])
self.assertRaises(core.SubversionException,
lambda: core.svn_auth_first_credentials(
- "unknown", "somerealm", baton))
+ b"unknown", b"somerealm", baton))
def test_credentials_get_username(self):
def myfunc(realm, maysave, pool):
- self.assertEquals("somerealm", realm)
+ self.assertEqual(b"somerealm", realm)
username_cred = core.svn_auth_cred_username_t()
- username_cred.username = "bar"
+ username_cred.username = b"bar"
username_cred.may_save = False
return username_cred
baton = core.svn_auth_open([core.svn_auth_get_username_prompt_provider(myfunc, 1)])
creds = core.svn_auth_first_credentials(
- core.SVN_AUTH_CRED_USERNAME, "somerealm", baton)
- self.assert_(creds is not None)
+ core.SVN_AUTH_CRED_USERNAME, b"somerealm", baton)
+ self.assertTrue(creds is not None)
def test_credentials_get_simple(self):
def myfunc(realm, username, may_save, pool):
- self.assertEquals("somerealm", realm)
+ self.assertEqual(b"somerealm", realm)
simple_cred = core.svn_auth_cred_simple_t()
- simple_cred.username = "mijnnaam"
- simple_cred.password = "geheim"
+ simple_cred.username = b"mijnnaam"
+ simple_cred.password = b"geheim"
simple_cred.may_save = False
return simple_cred
baton = core.svn_auth_open([core.svn_auth_get_simple_prompt_provider(myfunc, 1)])
creds = core.svn_auth_first_credentials(
- core.SVN_AUTH_CRED_SIMPLE, "somerealm", baton)
- self.assert_(creds is not None)
+ core.SVN_AUTH_CRED_SIMPLE, b"somerealm", baton)
+ self.assertTrue(creds is not None)
def test_credentials_get_ssl_client_cert(self):
def myfunc(realm, may_save, pool):
- self.assertEquals("somerealm", realm)
+ self.assertEqual(b"somerealm", realm)
ssl_cred = core.svn_auth_cred_ssl_client_cert_t()
- ssl_cred.cert_file = "my-certs-file"
+ ssl_cred.cert_file = b"my-certs-file"
ssl_cred.may_save = False
return ssl_cred
baton = core.svn_auth_open([core.svn_auth_get_ssl_client_cert_prompt_provider(myfunc, 1)])
creds = core.svn_auth_first_credentials(
- core.SVN_AUTH_CRED_SSL_CLIENT_CERT, "somerealm", baton)
- self.assert_(creds is not None)
+ core.SVN_AUTH_CRED_SSL_CLIENT_CERT, b"somerealm", baton)
+ self.assertTrue(creds is not None)
def test_credentials_get_ssl_client_cert_pw(self):
def myfunc(realm, may_save, pool):
- self.assertEquals("somerealm", realm)
+ self.assertEqual(b"somerealm", realm)
ssl_cred_pw = core.svn_auth_cred_ssl_client_cert_pw_t()
- ssl_cred_pw.password = "supergeheim"
+ ssl_cred_pw.password = b"supergeheim"
ssl_cred_pw.may_save = False
return ssl_cred_pw
baton = core.svn_auth_open([core.svn_auth_get_ssl_client_cert_pw_prompt_provider(myfunc, 1)])
creds = core.svn_auth_first_credentials(
- core.SVN_AUTH_CRED_SSL_CLIENT_CERT_PW, "somerealm", baton)
- self.assert_(creds is not None)
+ core.SVN_AUTH_CRED_SSL_CLIENT_CERT_PW, b"somerealm", baton)
+ self.assertTrue(creds is not None)
def test_credentials_get_ssl_server_trust(self):
def myfunc(realm, failures, cert_info, may_save, pool):
- self.assertEquals("somerealm", realm)
+ self.assertEqual(b"somerealm", realm)
ssl_trust = core.svn_auth_cred_ssl_server_trust_t()
ssl_trust.accepted_failures = 0
ssl_trust.may_save = False
return ssl_trust
baton = core.svn_auth_open([core.svn_auth_get_ssl_server_trust_prompt_provider(myfunc)])
core.svn_auth_set_parameter(baton, core.SVN_AUTH_PARAM_SSL_SERVER_FAILURES,
- "2")
+ b"2")
cert_info = core.svn_auth_ssl_server_cert_info_t()
core.svn_auth_set_parameter(baton, core.SVN_AUTH_PARAM_SSL_SERVER_CERT_INFO,
cert_info)
creds = core.svn_auth_first_credentials(
- core.SVN_AUTH_CRED_SSL_SERVER_TRUST, "somerealm", baton)
- self.assert_(creds is not None)
+ core.SVN_AUTH_CRED_SSL_SERVER_TRUST, b"somerealm", baton)
+ self.assertTrue(creds is not None)
def suite():
return unittest.defaultTestLoader.loadTestsFromTestCase(
Modified: subversion/branches/multi-wc-format/subversion/bindings/swig/python/tests/checksum.py
URL: http://svn.apache.org/viewvc/subversion/branches/multi-wc-format/subversion/bindings/swig/python/tests/checksum.py?rev=1897034&r1=1897033&r2=1897034&view=diff
==============================================================================
--- subversion/branches/multi-wc-format/subversion/bindings/swig/python/tests/checksum.py (original)
+++ subversion/branches/multi-wc-format/subversion/bindings/swig/python/tests/checksum.py Fri Jan 14 14:01:45 2022
@@ -29,7 +29,7 @@ class ChecksumTestCases(unittest.TestCas
val = svn.core.svn_checksum_create(kind)
check_val = svn.core.svn_checksum_to_cstring_display(val)
- self.assertTrue(isinstance(check_val, str),
+ self.assertTrue(isinstance(check_val, bytes),
"Type of digest not string")
self.assertEqual(len(check_val), 2*expected_length,
"Length of digest does not match kind")
Modified: subversion/branches/multi-wc-format/subversion/bindings/swig/python/tests/client.py
URL: http://svn.apache.org/viewvc/subversion/branches/multi-wc-format/subversion/bindings/swig/python/tests/client.py?rev=1897034&r1=1897033&r2=1897034&view=diff
==============================================================================
--- subversion/branches/multi-wc-format/subversion/bindings/swig/python/tests/client.py (original)
+++ subversion/branches/multi-wc-format/subversion/bindings/swig/python/tests/client.py Fri Jan 14 14:01:45 2022
@@ -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,61 @@ 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, {'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_accept_unicode(self):
+ """Test svn_client_get_commit_log3_t callback wrapper accept unicode as return value"""
+ directory = urljoin(self.repos_uri+b"/", b"dir1")
+ # override callback function which returns commit log as unicode
+ unicode_log_message_func = self.make_log_message_func(u"Test log message")
+ self.client_ctx.log_msg_baton3 = unicode_log_message_func
+
+ commit_info = client.mkdir3((directory,), 1, {b'customprop':b'value'},
+ self.client_ctx)
+ self.assertEqual(commit_info.revision, 13)
+ self.assertEqual(self.log_message_func_calls, 1)
+
+ def test_get_commit_log3_callback_unicode_error(self):
+ """Test svn_client_get_commit_log3_t callback wrapper handles UnicodeEncodeError correctly"""
+ directory = urljoin(self.repos_uri+b"/", b"dir1")
+ # override callback function which returns commit log as unicode
+ # which contains surrogate escaped character
+ bogus_log_message_func = self.make_log_message_func(u"Test \udc6cog"
+ u" message")
+ self.client_ctx.log_msg_baton3 = bogus_log_message_func
+
+ if not utils.IS_PY3 and utils.is_defaultencoding_utf8():
+ # 'utf-8' codecs on Python 2 does not raise UnicodeEncodeError
+ # on surrogate code point U+dc00 - U+dcff, however it causes
+ # Subversion error on property validation of svn:log
+ with self.assertRaises(core.SubversionException):
+ commit_info = client.mkdir3((directory,), 1, {b'customprop':b'value'},
+ self.client_ctx)
+ else:
+ with self.assertRaises(UnicodeEncodeError):
+ commit_info = client.mkdir3((directory,), 1, {b'customprop':b'value'},
+ self.client_ctx)
+
def test_log3_url(self):
"""Test svn_client_log3 on a file:// URL"""
- directory = urljoin(self.repos_uri+"/", "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 +280,54 @@ 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 = []
+
+ # (Pass tuple of bytes and str(unicode) mixture as revprops argument)
+ client.log5((directory,), start, (rev_range,), 1, True, False, False,
+ (u'svn:author', b'svn:log'),
+ log_entry_receiver_whole, self.client_ctx)
+ self.assertEqual(len(self.received_log_entries), 1)
+ revprops = self.received_log_entries[0].revprops
+ self.assertEqual(revprops[b'svn:log'], b"More directories.")
+ self.assertEqual(revprops[b'svn:author'], b"john")
+ with self.assertRaises(KeyError):
+ commit_date = revprops['svn:date']
+ if utils.IS_PY3 or not utils.is_defaultencoding_utf8():
+ # 'utf-8' codecs on Python 2 does not raise UnicodeEncodeError
+ # on surrogate code point U+dc00 - U+dcff. So we need to skip
+ # below in such a case.
+ with self.assertRaises(UnicodeEncodeError):
+ client.log5((directory,), start, (rev_range,), 1, True, False, False,
+ (u'svn:\udc61uthor', b'svn:log'),
+ log_entry_receiver_whole, self.client_ctx)
def test_uuid_from_url(self):
"""Test svn_client_uuid_from_url on a file:// URL"""
- self.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 +337,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 +351,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 +378,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 +425,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 +446,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 +473,102 @@ 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_propset_local(self):
+ """Test svn_client_propset_local.
+(also, testing const svn_string_t * input)"""
+
+ head = core.svn_opt_revision_t()
+ head.kind = core.svn_opt_revision_head
+ unspecified = core.svn_opt_revision_t()
+ unspecified.kind = core.svn_opt_revision_working
+
+ path = self.temper.alloc_empty_dir('-propset_local')
+
+ target_path = core.svn_dirent_join(path, b'trunk/README.txt')
+ target_prop = b'local_prop_test'
+ prop_val1 = b'foo'
+
+ co_rev = client.checkout3(self.repos_uri, path, head, head,
+ core.svn_depth_infinity, True, True,
+ self.client_ctx)
+
+ client.propset_local(target_prop, prop_val1, [target_path],
+ core.svn_depth_empty, False, None, self.client_ctx)
+ props, iprops, prop_rev = client.propget5(target_prop, target_path,
+ unspecified, unspecified,
+ core.svn_depth_empty,
+ None, self.client_ctx)
+ self.assertFalse(iprops)
+ self.assertEqual(prop_rev, co_rev)
+ self.assertEqual(props, { target_path : prop_val1 })
+
+ # Using str(unicode) to specify property value.
+ prop_val2 = b'bar'
+ client.propset_local(target_prop, prop_val2.decode('utf-8'), [target_path],
+ core.svn_depth_empty, False, None, self.client_ctx)
+ props, iprops, prop_rev = client.propget5(target_prop, target_path,
+ unspecified, unspecified,
+ core.svn_depth_empty,
+ None, self.client_ctx)
+ self.assertEqual(props, { target_path : prop_val2 })
+
+ # Using str(unicode) and check if it uses 'utf-8' codecs on Python 3
+ # (or Python 2, only if its default encoding is 'utf-8')
+ if utils.IS_PY3 or utils.is_defaultencoding_utf8():
+ # prop_val3 = '(checkmark)UNICODE'
+ prop_val3_str = (u'\u2705\U0001F1FA\U0001F1F3\U0001F1EE'
+ u'\U0001F1E8\U0001F1F4\U0001F1E9\U0001F1EA')
+ client.propset_local(target_prop, prop_val3_str, [target_path],
+ core.svn_depth_empty, False, None, self.client_ctx)
+ props, iprops, prop_rev = client.propget5(target_prop, target_path,
+ unspecified, unspecified,
+ core.svn_depth_empty,
+ None, self.client_ctx)
+ self.assertEqual(props, { target_path : prop_val3_str.encode('utf-8') })
def test_update4(self):
"""Test update and the notify function callbacks"""
@@ -456,6 +589,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 +600,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 +623,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/multi-wc-format/subversion/bindings/swig/python/tests/core.py
URL: http://svn.apache.org/viewvc/subversion/branches/multi-wc-format/subversion/bindings/swig/python/tests/core.py?rev=1897034&r1=1897033&r2=1897034&view=diff
==============================================================================
--- subversion/branches/multi-wc-format/subversion/bindings/swig/python/tests/core.py (original)
+++ subversion/branches/multi-wc-format/subversion/bindings/swig/python/tests/core.py Fri Jan 14 14:01:45 2022
@@ -19,10 +19,14 @@
#
#
import unittest
+import os
+import tempfile
+import sys
import svn.core, svn.client
import utils
+
class SubversionCoreTestCase(unittest.TestCase):
"""Test cases for the basic SWIG Subversion core"""
@@ -39,13 +43,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 +112,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 +160,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 +174,181 @@ 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)
+ stream = svn.core.svn_stream_from_stringbuf(b''.decode())
+ svn.core.svn_stream_close(stream)
+ stream = svn.core.svn_stream_from_stringbuf(None)
+ 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)
+
+ @unittest.skipIf(not utils.IS_PY3 and utils.is_defaultencoding_utf8(),
+ "'utf-8' codecs of Python 2 accepts any unicode strings")
+ 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(utils.IS_PY3 or utils.is_defaultencoding_utf8(),
+ "test ony for Python 3 or Python 2 'utf-8' codecs")
+ def test_stream_write_str(self):
+ o1_str = u'Python\x00\u3071\u3044\u305d\u3093\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)
+
+ @unittest.skipUnless(utils.IS_PY3 or utils.is_defaultencoding_utf8(),
+ "test ony for Python 3 or Python 2 'utf-8' codecs")
+ def test_stream_from_stringbuf_unicode(self):
+ "Check svn_stream_from_stringbuf() handle str on Python 3 correctly."
+ # instr_inicode = '(checkmark)UNICODE'
+ in_str_unicode = (u'\u2705\U0001F1FA\U0001F1F3\U0001F1EE'
+ u'\U0001F1E8\U0001F1F4\U0001F1E9\U0001F1EA')
+ stream = svn.core.svn_stream_from_stringbuf(in_str_unicode)
+ try:
+ self.assertEqual(svn.core.svn_stream_read_full(stream, 4096),
+ in_str_unicode.encode('utf-8'))
+ finally:
+ svn.core.svn_stream_close(stream)
+
+
def suite():
return unittest.defaultTestLoader.loadTestsFromTestCase(
SubversionCoreTestCase)
Modified: subversion/branches/multi-wc-format/subversion/bindings/swig/python/tests/delta.py
URL: http://svn.apache.org/viewvc/subversion/branches/multi-wc-format/subversion/bindings/swig/python/tests/delta.py?rev=1897034&r1=1897033&r2=1897034&view=diff
==============================================================================
--- subversion/branches/multi-wc-format/subversion/bindings/swig/python/tests/delta.py (original)
+++ subversion/branches/multi-wc-format/subversion/bindings/swig/python/tests/delta.py Fri Jan 14 14:01:45 2022
@@ -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]))