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]))
