| # |
| # |
| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| # |
| # |
| import unittest, setup_path, os, sys, weakref |
| from sys import version_info # For Python version check |
| from io import BytesIO |
| from svn import core, repos, fs, delta |
| from svn.core import SubversionException, Pool |
| import utils |
| |
| class ChangeReceiver(delta.Editor): |
| """A delta editor which saves textdeltas for later use""" |
| |
| def __init__(self, src_root, tgt_root): |
| self.src_root = src_root |
| self.tgt_root = tgt_root |
| self.textdeltas = [] |
| |
| def apply_textdelta(self, file_baton, base_checksum, pool=None): |
| def textdelta_handler(textdelta): |
| if textdelta is not None: |
| self.textdeltas.append(textdelta) |
| return textdelta_handler |
| |
| class DumpStreamParser(repos.ParseFns3): |
| def __init__(self, stream=None, pool=None): |
| repos.ParseFns3.__init__(self) |
| self.stream = stream |
| self.ops = [] |
| # for leak checking only. If the parse_fns3 object holds some proxy |
| # object allocated from 'pool' or the 'pool' itself, the 'pool' is not |
| # destroyed until the parse_fns3 object is removed. |
| self.pool = pool |
| def _close_dumpstream(self): |
| if self.stream: |
| self.stream.close() |
| self.stream = None |
| if self.pool: |
| self.pool = None |
| def magic_header_record(self, version, pool=None): |
| self.ops.append((b"magic-header", version)) |
| def uuid_record(self, uuid, pool=None): |
| self.ops.append((b"uuid", uuid)) |
| def new_revision_record(self, headers, pool=None): |
| rev = int(headers[repos.DUMPFILE_REVISION_NUMBER]) |
| self.ops.append((b"new-revision", rev)) |
| return rev |
| def close_revision(self, revision_baton): |
| self.ops.append((b"close-revision", revision_baton)) |
| def new_node_record(self, headers, revision_baton, pool=None): |
| node = headers[repos.DUMPFILE_NODE_PATH] |
| self.ops.append((b"new-node", revision_baton, node)) |
| return (revision_baton, node) |
| def close_node(self, node_baton): |
| self.ops.append((b"close-node", node_baton[0], node_baton[1])) |
| def set_revision_property(self, revision_baton, name, value): |
| self.ops.append((b"set-revision-prop", revision_baton, name, value)) |
| def set_node_property(self, node_baton, name, value): |
| self.ops.append((b"set-node-prop", node_baton[0], node_baton[1], name, value)) |
| def remove_node_props(self, node_baton): |
| self.ops.append((b"remove-node-props", node_baton[0], node_baton[1])) |
| def delete_node_property(self, node_baton, name): |
| self.ops.append((b"delete-node-prop", node_baton[0], node_baton[1], name)) |
| def apply_textdelta(self, node_baton): |
| self.ops.append((b"apply-textdelta", node_baton[0], node_baton[1])) |
| return None |
| def set_fulltext(self, node_baton): |
| self.ops.append((b"set-fulltext", node_baton[0], node_baton[1])) |
| return None |
| |
| class BatonCollector(repos.ChangeCollector): |
| """A ChangeCollector with collecting batons, too""" |
| def __init__(self, fs_ptr, root, pool=None, notify_cb=None): |
| repos.ChangeCollector.__init__(self, fs_ptr, root, pool, notify_cb) |
| self.batons = [] |
| self.close_called = False |
| self.abort_called = False |
| |
| def open_root(self, base_revision, dir_pool=None): |
| bt = repos.ChangeCollector.open_root(self, base_revision, dir_pool) |
| self.batons.append((b'dir baton', b'', bt, sys.getrefcount(bt))) |
| return bt |
| |
| def add_directory(self, path, parent_baton, |
| copyfrom_path, copyfrom_revision, dir_pool=None): |
| bt = repos.ChangeCollector.add_directory(self, path, parent_baton, |
| copyfrom_path, |
| copyfrom_revision, |
| dir_pool) |
| self.batons.append((b'dir baton', path, bt, sys.getrefcount(bt))) |
| return bt |
| |
| def open_directory(self, path, parent_baton, base_revision, |
| dir_pool=None): |
| bt = repos.ChangeCollector.open_directory(self, path, parent_baton, |
| base_revision, dir_pool) |
| self.batons.append((b'dir baton', path, bt, sys.getrefcount(bt))) |
| return bt |
| |
| def add_file(self, path, parent_baton, |
| copyfrom_path, copyfrom_revision, file_pool=None): |
| bt = repos.ChangeCollector.add_file(self, path, parent_baton, |
| copyfrom_path, copyfrom_revision, |
| file_pool) |
| self.batons.append((b'file baton', path, bt, sys.getrefcount(bt))) |
| return bt |
| |
| def open_file(self, path, parent_baton, base_revision, file_pool=None): |
| bt = repos.ChangeCollector.open_file(self, path, parent_baton, |
| base_revision, file_pool) |
| self.batons.append((b'file baton', path, bt, sys.getrefcount(bt))) |
| return bt |
| |
| def close_edit(self, pool=None): |
| self.close_called = True |
| return |
| |
| def abort_edit(self, pool=None): |
| self.abort_called = True |
| return |
| |
| class BatonCollectorErrorOnClose(BatonCollector): |
| """Same as BatonCollector, but raises an Exception when close the |
| file/dir specfied by error_path""" |
| def __init__(self, fs_ptr, root, pool=None, notify_cb=None, error_path=b''): |
| BatonCollector.__init__(self, fs_ptr, root, pool, notify_cb) |
| self.error_path = error_path |
| |
| def close_directory(self, dir_baton): |
| if dir_baton[0] == self.error_path: |
| raise SubversionException('A Dummy Exception!', core.SVN_ERR_BASE) |
| else: |
| BatonCollector.close_directory(self, dir_baton) |
| |
| def close_file(self, file_baton, text_checksum): |
| if file_baton[0] == self.error_path: |
| raise SubversionException('A Dummy Exception!', core.SVN_ERR_BASE) |
| else: |
| return BatonCollector.close_file(self, file_baton, text_checksum) |
| |
| |
| def _authz_callback(root, path, pool): |
| "A dummy authz callback which always returns success." |
| return 1 |
| |
| class SubversionRepositoryTestCase(unittest.TestCase): |
| """Test cases for the Subversion repository layer""" |
| |
| def setUp(self): |
| """Load a Subversion repository""" |
| self.temper = utils.Temper() |
| (self.repos, self.repos_path, _) = self.temper.alloc_known_repo( |
| 'trac/versioncontrol/tests/svnrepos.dump', suffix='-repository') |
| self.fs = repos.fs(self.repos) |
| self.rev = fs.youngest_rev(self.fs) |
| |
| def tearDown(self): |
| self.fs = None |
| self.repos = None |
| self.temper.cleanup() |
| |
| def test_cease_invocation(self): |
| """Test returning SVN_ERR_CEASE_INVOCATION from a callback""" |
| |
| revs = [] |
| def history_lookup(path, rev, pool): |
| revs.append(rev) |
| raise core.SubversionException(apr_err=core.SVN_ERR_CEASE_INVOCATION, |
| message="Hi from history_lookup") |
| |
| repos.history2(self.fs, b'/trunk/README2.txt', history_lookup, None, 0, |
| self.rev, True) |
| self.assertEqual(len(revs), 1) |
| |
| def test_create(self): |
| """Make sure that repos.create doesn't segfault when we set fs-type |
| using a config hash""" |
| fs_config = { b"fs-type": b"fsfs" } |
| for i in range(5): |
| path = self.temper.alloc_empty_dir(suffix='-repository-create%d' % i) |
| repos.create(path, b"", b"", None, fs_config) |
| |
| def test_dump_fs2(self): |
| """Test the dump_fs2 function""" |
| |
| self.callback_calls = 0 |
| |
| def is_cancelled(): |
| self.callback_calls += 1 |
| return None |
| |
| dumpstream = BytesIO() |
| feedbackstream = BytesIO() |
| repos.dump_fs2(self.repos, dumpstream, feedbackstream, 0, self.rev, 0, 0, |
| is_cancelled) |
| |
| # Check that we can dump stuff |
| dump = dumpstream.getvalue() |
| feedback = feedbackstream.getvalue() |
| expected_feedback = b"* Dumped revision " + str(self.rev).encode('utf-8') |
| self.assertEqual(dump.count(b"Node-path: trunk/README.txt"), 2) |
| self.assertEqual(feedback.count(expected_feedback), 1) |
| self.assertEqual(self.callback_calls, 13) |
| |
| # Check that the dump can be cancelled |
| self.assertRaises(SubversionException, repos.dump_fs2, |
| self.repos, dumpstream, feedbackstream, 0, self.rev, 0, 0, lambda: 1) |
| |
| dumpstream.close() |
| feedbackstream.close() |
| |
| # Check that the dump fails when the dumpstream is closed |
| self.assertRaises(ValueError, repos.dump_fs2, |
| self.repos, dumpstream, feedbackstream, 0, self.rev, 0, 0, None) |
| |
| dumpstream = BytesIO() |
| feedbackstream = BytesIO() |
| |
| # Check that we can grab the feedback stream, but not the dumpstream |
| repos.dump_fs2(self.repos, None, feedbackstream, 0, self.rev, 0, 0, None) |
| feedback = feedbackstream.getvalue() |
| self.assertEqual(feedback.count(expected_feedback), 1) |
| |
| # Check that we can grab the dumpstream, but not the feedbackstream |
| repos.dump_fs2(self.repos, dumpstream, None, 0, self.rev, 0, 0, None) |
| dump = dumpstream.getvalue() |
| self.assertEqual(dump.count(b"Node-path: trunk/README.txt"), 2) |
| |
| # Check that we can ignore both the dumpstream and the feedbackstream |
| repos.dump_fs2(self.repos, dumpstream, None, 0, self.rev, 0, 0, None) |
| self.assertEqual(feedback.count(expected_feedback), 1) |
| |
| # FIXME: The Python bindings don't check for 'NULL' values for |
| # svn_repos_t objects, so the following call segfaults |
| #repos.dump_fs2(None, None, None, 0, self.rev, 0, 0, None) |
| |
| def test_parse_fns3(self): |
| self.cancel_calls = 0 |
| def is_cancelled(): |
| self.cancel_calls += 1 |
| return None |
| pool = Pool() |
| subpool = Pool(pool) |
| dump_path = os.path.join(os.path.dirname(sys.argv[0]), |
| "trac/versioncontrol/tests/svnrepos.dump") |
| stream = open(dump_path, 'rb') |
| dsp = DumpStreamParser(stream, subpool) |
| dsp_ref = weakref.ref(dsp) |
| ptr, baton = repos.make_parse_fns3(dsp, subpool) |
| repos.parse_dumpstream3(stream, ptr, baton, False, is_cancelled) |
| self.assertEqual(self.cancel_calls, 76) |
| expected_list = [ |
| (b"magic-header", 2), |
| (b'uuid', b'92ea810a-adf3-0310-b540-bef912dcf5ba'), |
| (b'new-revision', 0), |
| (b'set-revision-prop', 0, b'svn:date', b'2005-04-01T09:57:41.312767Z'), |
| (b'close-revision', 0), |
| (b'new-revision', 1), |
| (b'set-revision-prop', 1, b'svn:log', b'Initial directory layout.'), |
| (b'set-revision-prop', 1, b'svn:author', b'john'), |
| (b'set-revision-prop', 1, b'svn:date', b'2005-04-01T10:00:52.353248Z'), |
| (b'new-node', 1, b'branches'), |
| (b'remove-node-props', 1, b'branches'), |
| (b'close-node', 1, b'branches'), |
| (b'new-node', 1, b'tags'), |
| (b'remove-node-props', 1, b'tags'), |
| (b'close-node', 1, b'tags'), |
| (b'new-node', 1, b'trunk'), |
| (b'remove-node-props', 1, b'trunk'), |
| (b'close-node', 1, b'trunk'), |
| (b'close-revision', 1), |
| (b'new-revision', 2), |
| (b'set-revision-prop', 2, b'svn:log', b'Added README.'), |
| (b'set-revision-prop', 2, b'svn:author', b'john'), |
| (b'set-revision-prop', 2, b'svn:date', b'2005-04-01T13:12:18.216267Z'), |
| (b'new-node', 2, b'trunk/README.txt'), |
| (b'remove-node-props', 2, b'trunk/README.txt'), |
| (b'set-fulltext', 2, b'trunk/README.txt'), |
| (b'close-node', 2, b'trunk/README.txt'), |
| (b'close-revision', 2), (b'new-revision', 3), |
| (b'set-revision-prop', 3, b'svn:log', b'Fixed README.\n'), |
| (b'set-revision-prop', 3, b'svn:author', b'kate'), |
| (b'set-revision-prop', 3, b'svn:date', b'2005-04-01T13:24:58.234643Z'), |
| (b'new-node', 3, b'trunk/README.txt'), |
| (b'remove-node-props', 3, b'trunk/README.txt'), |
| (b'set-node-prop', 3, b'trunk/README.txt', b'svn:mime-type', b'text/plain'), |
| (b'set-node-prop', 3, b'trunk/README.txt', b'svn:eol-style', b'native'), |
| (b'set-fulltext', 3, b'trunk/README.txt'), |
| (b'close-node', 3, b'trunk/README.txt'), (b'close-revision', 3), |
| ] |
| # Compare only the first X nodes described in the expected list - otherwise |
| # the comparison list gets too long. |
| self.assertEqual(dsp.ops[:len(expected_list)], expected_list) |
| |
| # _close_dumpstream should be invoked after 'baton' is removed. |
| self.assertEqual(False, stream.closed) |
| del ptr, baton, subpool, dsp |
| self.assertEqual(True, stream.closed) |
| # Issue SVN-4918 |
| self.assertEqual(None, dsp_ref()) |
| |
| def test_parse_fns3_invalid_set_fulltext(self): |
| class DumpStreamParserSubclass(DumpStreamParser): |
| def set_fulltext(self, node_baton): |
| DumpStreamParser.set_fulltext(self, node_baton) |
| return 42 |
| stream = open(os.path.join(os.path.dirname(sys.argv[0]), |
| "trac/versioncontrol/tests/svnrepos.dump"), "rb") |
| try: |
| dsp = DumpStreamParserSubclass() |
| ptr, baton = repos.make_parse_fns3(dsp) |
| self.assertRaises(TypeError, repos.parse_dumpstream3, |
| stream, ptr, baton, False, None) |
| finally: |
| stream.close() |
| |
| def test_get_logs(self): |
| """Test scope of get_logs callbacks""" |
| logs = [] |
| def addLog(paths, revision, author, date, message, pool): |
| if paths is not None: |
| logs.append(paths) |
| |
| # Run get_logs |
| repos.get_logs(self.repos, [b'/'], self.rev, 0, True, 0, addLog) |
| |
| # Count and verify changes |
| change_count = 0 |
| for log in logs: |
| for path_changed in core._as_list(log.values()): |
| change_count += 1 |
| path_changed.assert_valid() |
| self.assertEqual(logs[2][b"/tags/v1.1"].action, b"A") |
| self.assertEqual(logs[2][b"/tags/v1.1"].copyfrom_path, b"/branches/v1x") |
| self.assertEqual(len(logs), 12) |
| self.assertEqual(change_count, 19) |
| |
| def test_dir_delta(self): |
| """Test scope of dir_delta callbacks""" |
| # Run dir_delta |
| this_root = fs.revision_root(self.fs, self.rev) |
| prev_root = fs.revision_root(self.fs, self.rev-1) |
| editor = ChangeReceiver(this_root, prev_root) |
| e_ptr, e_baton = delta.make_editor(editor) |
| repos.dir_delta(prev_root, b'', b'', this_root, b'', e_ptr, e_baton, |
| _authz_callback, 1, 1, 0, 0) |
| |
| # Check results. |
| # Ignore the order in which the editor delivers the two sibling files. |
| self.assertEqual(set([editor.textdeltas[0].new_data, |
| editor.textdeltas[1].new_data]), |
| set([b"This is a test.\n", b"A test.\n"])) |
| self.assertEqual(len(editor.textdeltas), 2) |
| |
| def test_unnamed_editor(self): |
| """Test editor object without reference from interpreter""" |
| # Check that the delta.Editor object has proper lifetime. Without |
| # increment of the refcount in make_baton, the object was destroyed |
| # immediately because the interpreter does not hold a reference to it. |
| this_root = fs.revision_root(self.fs, self.rev) |
| prev_root = fs.revision_root(self.fs, self.rev-1) |
| e_ptr, e_baton = delta.make_editor(ChangeReceiver(this_root, prev_root)) |
| repos.dir_delta(prev_root, b'', b'', this_root, b'', e_ptr, e_baton, |
| _authz_callback, 1, 1, 0, 0) |
| |
| def test_delta_editor_leak_with_change_collector(self): |
| pool = Pool() |
| subpool = Pool(pool) |
| root = fs.revision_root(self.fs, self.rev, subpool) |
| editor = repos.ChangeCollector(self.fs, root, subpool) |
| editor_ref = weakref.ref(editor) |
| e_ptr, e_baton = delta.make_editor(editor, subpool) |
| repos.replay(root, e_ptr, e_baton, subpool) |
| |
| fs.close_root(root) |
| del root |
| self.assertNotEqual(None, editor_ref()) |
| |
| del e_ptr, e_baton, editor |
| del subpool |
| self.assertEqual(None, editor_ref()) |
| |
| def test_replay_batons_refcounts(self): |
| """Issue SVN-4917: check ref-count of batons created and used in call backs""" |
| root = fs.revision_root(self.fs, self.rev) |
| editor = BatonCollector(self.fs, root) |
| e_ptr, e_baton = delta.make_editor(editor) |
| repos.replay(root, e_ptr, e_baton) |
| for baton in editor.batons: |
| self.assertEqual(sys.getrefcount(baton[2]), 2, |
| "leak on baton %s after replay without errors" |
| % repr(baton)) |
| del e_baton |
| self.assertEqual(sys.getrefcount(e_ptr), 2, |
| "leak on editor baton after replay without errors") |
| |
| editor = BatonCollectorErrorOnClose(self.fs, root, |
| error_path=b'branches/v1x') |
| e_ptr, e_baton = delta.make_editor(editor) |
| self.assertRaises(SubversionException, repos.replay, root, e_ptr, e_baton) |
| batons= editor.batons |
| # As svn_repos_replay calls neigher close_edit callback nor abort_edit |
| # if an error has occured during processing, references of Python objects |
| # in decendant batons may live until e_baton is deleted. |
| del e_baton |
| for baton in batons: |
| self.assertEqual(sys.getrefcount(baton[2]), 2, |
| "leak on baton %s after replay with an error" |
| % repr(baton)) |
| self.assertEqual(sys.getrefcount(e_ptr), 2, |
| "leak on editor baton after replay with an error") |
| |
| def test_retrieve_and_change_rev_prop(self): |
| """Test playing with revprops""" |
| self.assertEqual(repos.fs_revision_prop(self.repos, self.rev, b"svn:log", |
| _authz_callback), |
| b"''(a few years later)'' Argh... v1.1 was buggy, " |
| b"after all") |
| |
| # We expect this to complain because we have no pre-revprop-change |
| # hook script for the repository. |
| self.assertRaises(SubversionException, repos.fs_change_rev_prop3, |
| self.repos, self.rev, b"jrandom", b"svn:log", |
| b"Youngest revision", True, True, _authz_callback) |
| |
| repos.fs_change_rev_prop3(self.repos, self.rev, b"jrandom", b"svn:log", |
| b"Youngest revision", False, False, |
| _authz_callback) |
| |
| self.assertEqual(repos.fs_revision_prop(self.repos, self.rev, b"svn:log", |
| _authz_callback), |
| b"Youngest revision") |
| |
| def freeze_body(self, pool): |
| self.freeze_invoked += 1 |
| |
| def test_freeze(self): |
| """Test repository freeze""" |
| |
| self.freeze_invoked = 0 |
| repos.freeze([self.repos_path], self.freeze_body) |
| self.assertEqual(self.freeze_invoked, 1) |
| |
| def test_lock_unlock(self): |
| """Basic lock/unlock""" |
| |
| access = fs.create_access(b'jrandom') |
| fs.set_access(self.fs, access) |
| fs.lock(self.fs, b'/trunk/README.txt', None, None, 0, 0, self.rev, False) |
| try: |
| fs.lock(self.fs, b'/trunk/README.txt', None, None, 0, 0, self.rev, False) |
| except core.SubversionException as exc: |
| self.assertEqual(exc.apr_err, core.SVN_ERR_FS_PATH_ALREADY_LOCKED) |
| fs.lock(self.fs, b'/trunk/README.txt', None, None, 0, 0, self.rev, True) |
| |
| self.calls = 0 |
| self.errors = 0 |
| def unlock_callback(path, lock, err, pool): |
| self.assertEqual(path, b'/trunk/README.txt') |
| self.assertEqual(lock, None) |
| self.calls += 1 |
| if err != None: |
| self.assertEqual(err.apr_err, core.SVN_ERR_FS_NO_SUCH_LOCK) |
| self.errors += 1 |
| |
| the_lock = fs.get_lock(self.fs, b'/trunk/README.txt') |
| fs.unlock_many(self.fs, {b'/trunk/README.txt':the_lock.token}, False, |
| unlock_callback) |
| self.assertEqual(self.calls, 1) |
| self.assertEqual(self.errors, 0) |
| |
| self.calls = 0 |
| fs.unlock_many(self.fs, {b'/trunk/README.txt':the_lock.token}, False, |
| unlock_callback) |
| self.assertEqual(self.calls, 1) |
| self.assertEqual(self.errors, 1) |
| |
| self.locks = 0 |
| def lock_callback(path, lock, err, pool): |
| self.assertEqual(path, b'/trunk/README.txt') |
| if lock != None: |
| self.assertEqual(lock.owner, b'jrandom') |
| self.locks += 1 |
| self.calls += 1 |
| if err != None: |
| self.assertEqual(err.apr_err, core.SVN_ERR_FS_PATH_ALREADY_LOCKED) |
| self.errors += 1 |
| |
| self.calls = 0 |
| self.errors = 0 |
| target = fs.lock_target_create(None, self.rev) |
| fs.lock_many(self.fs, {b'trunk/README.txt':target}, |
| None, False, 0, False, lock_callback) |
| self.assertEqual(self.calls, 1) |
| self.assertEqual(self.locks, 1) |
| self.assertEqual(self.errors, 0) |
| |
| self.calls = 0 |
| self.locks = 0 |
| fs.lock_many(self.fs, {b'trunk/README.txt':target}, |
| None, False, 0, False, lock_callback) |
| self.assertEqual(self.calls, 1) |
| self.assertEqual(self.locks, 0) |
| self.assertEqual(self.errors, 1) |
| |
| self.calls = 0 |
| self.errors = 0 |
| the_lock = fs.get_lock(self.fs, b'/trunk/README.txt') |
| repos.fs_unlock_many(self.repos, {b'trunk/README.txt':the_lock.token}, |
| False, unlock_callback) |
| self.assertEqual(self.calls, 1) |
| self.assertEqual(self.errors, 0) |
| |
| self.calls = 0 |
| repos.fs_unlock_many(self.repos, {b'trunk/README.txt':the_lock.token}, |
| False, unlock_callback) |
| self.assertEqual(self.calls, 1) |
| self.assertEqual(self.errors, 1) |
| |
| self.calls = 0 |
| self.errors = 0 |
| repos.fs_lock_many(self.repos, {b'trunk/README.txt':target}, |
| None, False, 0, False, lock_callback) |
| self.assertEqual(self.calls, 1) |
| self.assertEqual(self.locks, 1) |
| self.assertEqual(self.errors, 0) |
| |
| self.calls = 0 |
| self.locks = 0 |
| repos.fs_lock_many(self.repos, {b'trunk/README.txt':target}, |
| None, False, 0, False, lock_callback) |
| self.assertEqual(self.calls, 1) |
| self.assertEqual(self.locks, 0) |
| self.assertEqual(self.errors, 1) |
| |
| def suite(): |
| return unittest.defaultTestLoader.loadTestsFromTestCase( |
| SubversionRepositoryTestCase) |
| |
| if __name__ == '__main__': |
| runner = unittest.TextTestRunner() |
| runner.run(suite()) |
