diff options
Diffstat (limited to 'subversion/bindings/swig/python/tests/client.py')
| -rw-r--r-- | subversion/bindings/swig/python/tests/client.py | 284 |
1 files changed, 218 insertions, 66 deletions
diff --git a/subversion/bindings/swig/python/tests/client.py b/subversion/bindings/swig/python/tests/client.py index 07fb773..422dc4f 100644 --- a/subversion/bindings/swig/python/tests/client.py +++ b/subversion/bindings/swig/python/tests/client.py @@ -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.TestCase): 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.TestCase): 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.TestCase): # 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.TestCase): 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.TestCase): 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.TestCase): 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 argment) + client.log5((directory,), start, (rev_range,), 1, True, False, False, + ('svn:author', b'svn:log'), + log_entry_receiver_whole, self.client_ctx) + self.assertEqual(len(self.received_log_entries), 1) + revprops = self.received_log_entries[0].revprops + self.assertEqual(revprops[b'svn:log'], b"More directories.") + self.assertEqual(revprops[b'svn:author'], b"john") + with self.assertRaises(KeyError): + commit_date = revprops['svn:date'] + 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.TestCase): 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.TestCase): 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.TestCase): 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.TestCase): 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.TestCase): # 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.TestCase): # 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.TestCase): 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.TestCase): 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.TestCase): 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(): |