diff options
Diffstat (limited to 'u1db/tests/commandline/test_client.py')
| -rw-r--r-- | u1db/tests/commandline/test_client.py | 916 |
1 files changed, 0 insertions, 916 deletions
diff --git a/u1db/tests/commandline/test_client.py b/u1db/tests/commandline/test_client.py deleted file mode 100644 index 78ca21eb..00000000 --- a/u1db/tests/commandline/test_client.py +++ /dev/null @@ -1,916 +0,0 @@ -# Copyright 2011 Canonical Ltd. -# -# This file is part of u1db. -# -# u1db is free software: you can redistribute it and/or modify -# it under the terms of the GNU Lesser General Public License version 3 -# as published by the Free Software Foundation. -# -# u1db is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Lesser General Public License for more details. -# -# You should have received a copy of the GNU Lesser General Public License -# along with u1db. If not, see <http://www.gnu.org/licenses/>. - -import cStringIO -import os -import sys -try: - import simplejson as json -except ImportError: - import json # noqa -import subprocess - -from u1db import ( - errors, - open as u1db_open, - tests, - vectorclock, - ) -from u1db.commandline import ( - client, - serve, - ) -from u1db.tests.commandline import safe_close -from u1db.tests import test_remote_sync_target - - -class TestArgs(tests.TestCase): - """These tests are meant to test just the argument parsing. - - Each Command should have at least one test, possibly more if it allows - optional arguments, etc. - """ - - def setUp(self): - super(TestArgs, self).setUp() - self.parser = client.client_commands.make_argparser() - - def parse_args(self, args): - # ArgumentParser.parse_args doesn't play very nicely with a test suite, - # so we trap SystemExit in case something is wrong with the args we're - # parsing. - try: - return self.parser.parse_args(args) - except SystemExit: - raise AssertionError('got SystemExit') - - def test_create(self): - args = self.parse_args(['create', 'test.db']) - self.assertEqual(client.CmdCreate, args.subcommand) - self.assertEqual('test.db', args.database) - self.assertEqual(None, args.doc_id) - self.assertEqual(None, args.infile) - - def test_create_custom_doc_id(self): - args = self.parse_args(['create', '--id', 'xyz', 'test.db']) - self.assertEqual(client.CmdCreate, args.subcommand) - self.assertEqual('test.db', args.database) - self.assertEqual('xyz', args.doc_id) - self.assertEqual(None, args.infile) - - def test_delete(self): - args = self.parse_args(['delete', 'test.db', 'doc-id', 'doc-rev']) - self.assertEqual(client.CmdDelete, args.subcommand) - self.assertEqual('test.db', args.database) - self.assertEqual('doc-id', args.doc_id) - self.assertEqual('doc-rev', args.doc_rev) - - def test_get(self): - args = self.parse_args(['get', 'test.db', 'doc-id']) - self.assertEqual(client.CmdGet, args.subcommand) - self.assertEqual('test.db', args.database) - self.assertEqual('doc-id', args.doc_id) - self.assertEqual(None, args.outfile) - - def test_get_dash(self): - args = self.parse_args(['get', 'test.db', 'doc-id', '-']) - self.assertEqual(client.CmdGet, args.subcommand) - self.assertEqual('test.db', args.database) - self.assertEqual('doc-id', args.doc_id) - self.assertEqual(sys.stdout, args.outfile) - - def test_init_db(self): - args = self.parse_args( - ['init-db', 'test.db', '--replica-uid=replica-uid']) - self.assertEqual(client.CmdInitDB, args.subcommand) - self.assertEqual('test.db', args.database) - self.assertEqual('replica-uid', args.replica_uid) - - def test_init_db_no_replica(self): - args = self.parse_args(['init-db', 'test.db']) - self.assertEqual(client.CmdInitDB, args.subcommand) - self.assertEqual('test.db', args.database) - self.assertIs(None, args.replica_uid) - - def test_put(self): - args = self.parse_args(['put', 'test.db', 'doc-id', 'old-doc-rev']) - self.assertEqual(client.CmdPut, args.subcommand) - self.assertEqual('test.db', args.database) - self.assertEqual('doc-id', args.doc_id) - self.assertEqual('old-doc-rev', args.doc_rev) - self.assertEqual(None, args.infile) - - def test_sync(self): - args = self.parse_args(['sync', 'source', 'target']) - self.assertEqual(client.CmdSync, args.subcommand) - self.assertEqual('source', args.source) - self.assertEqual('target', args.target) - - def test_create_index(self): - args = self.parse_args(['create-index', 'db', 'index', 'expression']) - self.assertEqual(client.CmdCreateIndex, args.subcommand) - self.assertEqual('db', args.database) - self.assertEqual('index', args.index) - self.assertEqual(['expression'], args.expression) - - def test_create_index_multi_expression(self): - args = self.parse_args(['create-index', 'db', 'index', 'e1', 'e2']) - self.assertEqual(client.CmdCreateIndex, args.subcommand) - self.assertEqual('db', args.database) - self.assertEqual('index', args.index) - self.assertEqual(['e1', 'e2'], args.expression) - - def test_list_indexes(self): - args = self.parse_args(['list-indexes', 'db']) - self.assertEqual(client.CmdListIndexes, args.subcommand) - self.assertEqual('db', args.database) - - def test_delete_index(self): - args = self.parse_args(['delete-index', 'db', 'index']) - self.assertEqual(client.CmdDeleteIndex, args.subcommand) - self.assertEqual('db', args.database) - self.assertEqual('index', args.index) - - def test_get_index_keys(self): - args = self.parse_args(['get-index-keys', 'db', 'index']) - self.assertEqual(client.CmdGetIndexKeys, args.subcommand) - self.assertEqual('db', args.database) - self.assertEqual('index', args.index) - - def test_get_from_index(self): - args = self.parse_args(['get-from-index', 'db', 'index', 'foo']) - self.assertEqual(client.CmdGetFromIndex, args.subcommand) - self.assertEqual('db', args.database) - self.assertEqual('index', args.index) - self.assertEqual(['foo'], args.values) - - def test_get_doc_conflicts(self): - args = self.parse_args(['get-doc-conflicts', 'db', 'doc-id']) - self.assertEqual(client.CmdGetDocConflicts, args.subcommand) - self.assertEqual('db', args.database) - self.assertEqual('doc-id', args.doc_id) - - def test_resolve(self): - args = self.parse_args( - ['resolve-doc', 'db', 'doc-id', 'rev:1', 'other:1']) - self.assertEqual(client.CmdResolve, args.subcommand) - self.assertEqual('db', args.database) - self.assertEqual('doc-id', args.doc_id) - self.assertEqual(['rev:1', 'other:1'], args.doc_revs) - self.assertEqual(None, args.infile) - - -class TestCaseWithDB(tests.TestCase): - """These next tests are meant to have one class per Command. - - It is meant to test the inner workings of each command. The detailed - testing should happen in these classes. Stuff like how it handles errors, - etc. should be done here. - """ - - def setUp(self): - super(TestCaseWithDB, self).setUp() - self.working_dir = self.createTempDir() - self.db_path = self.working_dir + '/test.db' - self.db = u1db_open(self.db_path, create=True) - self.db._set_replica_uid('test') - self.addCleanup(self.db.close) - - def make_command(self, cls, stdin_content=''): - inf = cStringIO.StringIO(stdin_content) - out = cStringIO.StringIO() - err = cStringIO.StringIO() - return cls(inf, out, err) - - -class TestCmdCreate(TestCaseWithDB): - - def test_create(self): - cmd = self.make_command(client.CmdCreate) - inf = cStringIO.StringIO(tests.simple_doc) - cmd.run(self.db_path, inf, 'test-id') - doc = self.db.get_doc('test-id') - self.assertEqual(tests.simple_doc, doc.get_json()) - self.assertFalse(doc.has_conflicts) - self.assertEqual('', cmd.stdout.getvalue()) - self.assertEqual('id: test-id\nrev: %s\n' % (doc.rev,), - cmd.stderr.getvalue()) - - -class TestCmdDelete(TestCaseWithDB): - - def test_delete(self): - doc = self.db.create_doc_from_json(tests.simple_doc) - cmd = self.make_command(client.CmdDelete) - cmd.run(self.db_path, doc.doc_id, doc.rev) - doc2 = self.db.get_doc(doc.doc_id, include_deleted=True) - self.assertEqual(doc.doc_id, doc2.doc_id) - self.assertNotEqual(doc.rev, doc2.rev) - self.assertIs(None, doc2.get_json()) - self.assertEqual('', cmd.stdout.getvalue()) - self.assertEqual('rev: %s\n' % (doc2.rev,), cmd.stderr.getvalue()) - - def test_delete_fails_if_nonexistent(self): - doc = self.db.create_doc_from_json(tests.simple_doc) - db2_path = self.db_path + '.typo' - cmd = self.make_command(client.CmdDelete) - # TODO: We should really not be showing a traceback here. But we need - # to teach the commandline infrastructure how to handle - # exceptions. - # However, we *do* want to test that the db doesn't get created - # by accident. - self.assertRaises(errors.DatabaseDoesNotExist, - cmd.run, db2_path, doc.doc_id, doc.rev) - self.assertFalse(os.path.exists(db2_path)) - - def test_delete_no_such_doc(self): - cmd = self.make_command(client.CmdDelete) - # TODO: We should really not be showing a traceback here. But we need - # to teach the commandline infrastructure how to handle - # exceptions. - self.assertRaises(errors.DocumentDoesNotExist, - cmd.run, self.db_path, 'no-doc-id', 'no-rev') - - def test_delete_bad_rev(self): - doc = self.db.create_doc_from_json(tests.simple_doc) - cmd = self.make_command(client.CmdDelete) - self.assertRaises(errors.RevisionConflict, - cmd.run, self.db_path, doc.doc_id, 'not-the-actual-doc-rev:1') - # TODO: Test that we get a pretty output. - - -class TestCmdGet(TestCaseWithDB): - - def setUp(self): - super(TestCmdGet, self).setUp() - self.doc = self.db.create_doc_from_json( - tests.simple_doc, doc_id='my-test-doc') - - def test_get_simple(self): - cmd = self.make_command(client.CmdGet) - cmd.run(self.db_path, 'my-test-doc', None) - self.assertEqual(tests.simple_doc + "\n", cmd.stdout.getvalue()) - self.assertEqual('rev: %s\n' % (self.doc.rev,), - cmd.stderr.getvalue()) - - def test_get_conflict(self): - doc = self.make_document('my-test-doc', 'other:1', '{}', False) - self.db._put_doc_if_newer( - doc, save_conflict=True, replica_uid='r', replica_gen=1, - replica_trans_id='foo') - cmd = self.make_command(client.CmdGet) - cmd.run(self.db_path, 'my-test-doc', None) - self.assertEqual('{}\n', cmd.stdout.getvalue()) - self.assertEqual('rev: %s\nDocument has conflicts.\n' % (doc.rev,), - cmd.stderr.getvalue()) - - def test_get_fail(self): - cmd = self.make_command(client.CmdGet) - result = cmd.run(self.db_path, 'doc-not-there', None) - self.assertEqual(1, result) - self.assertEqual("", cmd.stdout.getvalue()) - self.assertTrue("not found" in cmd.stderr.getvalue()) - - def test_get_no_database(self): - cmd = self.make_command(client.CmdGet) - retval = cmd.run(self.db_path + "__DOES_NOT_EXIST", "my-doc", None) - self.assertEqual(retval, 1) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual(cmd.stderr.getvalue(), 'Database does not exist.\n') - - -class TestCmdGetDocConflicts(TestCaseWithDB): - - def setUp(self): - super(TestCmdGetDocConflicts, self).setUp() - self.doc1 = self.db.create_doc_from_json( - tests.simple_doc, doc_id='my-doc') - self.doc2 = self.make_document('my-doc', 'other:1', '{}', False) - self.db._put_doc_if_newer( - self.doc2, save_conflict=True, replica_uid='r', replica_gen=1, - replica_trans_id='foo') - - def test_get_doc_conflicts_none(self): - self.db.create_doc_from_json(tests.simple_doc, doc_id='a-doc') - cmd = self.make_command(client.CmdGetDocConflicts) - cmd.run(self.db_path, 'a-doc') - self.assertEqual([], json.loads(cmd.stdout.getvalue())) - self.assertEqual('', cmd.stderr.getvalue()) - - def test_get_doc_conflicts_simple(self): - cmd = self.make_command(client.CmdGetDocConflicts) - cmd.run(self.db_path, 'my-doc') - self.assertEqual( - [dict(rev=self.doc2.rev, content=self.doc2.content), - dict(rev=self.doc1.rev, content=self.doc1.content)], - json.loads(cmd.stdout.getvalue())) - self.assertEqual('', cmd.stderr.getvalue()) - - def test_get_doc_conflicts_no_db(self): - cmd = self.make_command(client.CmdGetDocConflicts) - retval = cmd.run(self.db_path + "__DOES_NOT_EXIST", "my-doc") - self.assertEqual(retval, 1) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual(cmd.stderr.getvalue(), 'Database does not exist.\n') - - def test_get_doc_conflicts_no_doc(self): - cmd = self.make_command(client.CmdGetDocConflicts) - retval = cmd.run(self.db_path, "some-doc") - self.assertEqual(retval, 1) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual(cmd.stderr.getvalue(), 'Document does not exist.\n') - - -class TestCmdInit(TestCaseWithDB): - - def test_init_new(self): - path = self.working_dir + '/test2.db' - self.assertFalse(os.path.exists(path)) - cmd = self.make_command(client.CmdInitDB) - cmd.run(path, 'test-uid') - self.assertTrue(os.path.exists(path)) - db = u1db_open(path, create=False) - self.assertEqual('test-uid', db._replica_uid) - - def test_init_no_uid(self): - path = self.working_dir + '/test2.db' - cmd = self.make_command(client.CmdInitDB) - cmd.run(path, None) - self.assertTrue(os.path.exists(path)) - db = u1db_open(path, create=False) - self.assertIsNot(None, db._replica_uid) - - -class TestCmdPut(TestCaseWithDB): - - def setUp(self): - super(TestCmdPut, self).setUp() - self.doc = self.db.create_doc_from_json( - tests.simple_doc, doc_id='my-test-doc') - - def test_put_simple(self): - cmd = self.make_command(client.CmdPut) - inf = cStringIO.StringIO(tests.nested_doc) - cmd.run(self.db_path, 'my-test-doc', self.doc.rev, inf) - doc = self.db.get_doc('my-test-doc') - self.assertNotEqual(self.doc.rev, doc.rev) - self.assertGetDoc(self.db, 'my-test-doc', doc.rev, - tests.nested_doc, False) - self.assertEqual('', cmd.stdout.getvalue()) - self.assertEqual('rev: %s\n' % (doc.rev,), - cmd.stderr.getvalue()) - - def test_put_no_db(self): - cmd = self.make_command(client.CmdPut) - inf = cStringIO.StringIO(tests.nested_doc) - retval = cmd.run(self.db_path + "__DOES_NOT_EXIST", - 'my-test-doc', self.doc.rev, inf) - self.assertEqual(retval, 1) - self.assertEqual('', cmd.stdout.getvalue()) - self.assertEqual('Database does not exist.\n', cmd.stderr.getvalue()) - - def test_put_no_doc(self): - cmd = self.make_command(client.CmdPut) - inf = cStringIO.StringIO(tests.nested_doc) - retval = cmd.run(self.db_path, 'no-such-doc', 'wut:1', inf) - self.assertEqual(1, retval) - self.assertEqual('', cmd.stdout.getvalue()) - self.assertEqual('Document does not exist.\n', cmd.stderr.getvalue()) - - def test_put_doc_old_rev(self): - rev = self.doc.rev - doc = self.make_document('my-test-doc', rev, '{}', False) - self.db.put_doc(doc) - cmd = self.make_command(client.CmdPut) - inf = cStringIO.StringIO(tests.nested_doc) - retval = cmd.run(self.db_path, 'my-test-doc', rev, inf) - self.assertEqual(1, retval) - self.assertEqual('', cmd.stdout.getvalue()) - self.assertEqual('Given revision is not current.\n', - cmd.stderr.getvalue()) - - def test_put_doc_w_conflicts(self): - doc = self.make_document('my-test-doc', 'other:1', '{}', False) - self.db._put_doc_if_newer( - doc, save_conflict=True, replica_uid='r', replica_gen=1, - replica_trans_id='foo') - cmd = self.make_command(client.CmdPut) - inf = cStringIO.StringIO(tests.nested_doc) - retval = cmd.run(self.db_path, 'my-test-doc', 'other:1', inf) - self.assertEqual(1, retval) - self.assertEqual('', cmd.stdout.getvalue()) - self.assertEqual('Document has conflicts.\n' - 'Inspect with get-doc-conflicts, then resolve.\n', - cmd.stderr.getvalue()) - - -class TestCmdResolve(TestCaseWithDB): - - def setUp(self): - super(TestCmdResolve, self).setUp() - self.doc1 = self.db.create_doc_from_json( - tests.simple_doc, doc_id='my-doc') - self.doc2 = self.make_document('my-doc', 'other:1', '{}', False) - self.db._put_doc_if_newer( - self.doc2, save_conflict=True, replica_uid='r', replica_gen=1, - replica_trans_id='foo') - - def test_resolve_simple(self): - self.assertTrue(self.db.get_doc('my-doc').has_conflicts) - cmd = self.make_command(client.CmdResolve) - inf = cStringIO.StringIO(tests.nested_doc) - cmd.run(self.db_path, 'my-doc', [self.doc1.rev, self.doc2.rev], inf) - doc = self.db.get_doc('my-doc') - vec = vectorclock.VectorClockRev(doc.rev) - self.assertTrue( - vec.is_newer(vectorclock.VectorClockRev(self.doc1.rev))) - self.assertTrue( - vec.is_newer(vectorclock.VectorClockRev(self.doc2.rev))) - self.assertGetDoc(self.db, 'my-doc', doc.rev, tests.nested_doc, False) - self.assertEqual('', cmd.stdout.getvalue()) - self.assertEqual('rev: %s\n' % (doc.rev,), - cmd.stderr.getvalue()) - - def test_resolve_double(self): - moar = '{"x": 42}' - doc3 = self.make_document('my-doc', 'third:1', moar, False) - self.db._put_doc_if_newer( - doc3, save_conflict=True, replica_uid='r', replica_gen=1, - replica_trans_id='foo') - cmd = self.make_command(client.CmdResolve) - inf = cStringIO.StringIO(tests.nested_doc) - cmd.run(self.db_path, 'my-doc', [self.doc1.rev, self.doc2.rev], inf) - doc = self.db.get_doc('my-doc') - self.assertGetDoc(self.db, 'my-doc', doc.rev, moar, True) - self.assertEqual('', cmd.stdout.getvalue()) - self.assertEqual( - 'rev: %s\nDocument still has conflicts.\n' % (doc.rev,), - cmd.stderr.getvalue()) - - def test_resolve_no_db(self): - cmd = self.make_command(client.CmdResolve) - retval = cmd.run(self.db_path + "__DOES_NOT_EXIST", "my-doc", [], None) - self.assertEqual(retval, 1) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual(cmd.stderr.getvalue(), 'Database does not exist.\n') - - def test_resolve_no_doc(self): - cmd = self.make_command(client.CmdResolve) - retval = cmd.run(self.db_path, "foo", [], None) - self.assertEqual(retval, 1) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual(cmd.stderr.getvalue(), 'Document does not exist.\n') - - -class TestCmdSync(TestCaseWithDB): - - def setUp(self): - super(TestCmdSync, self).setUp() - self.db2_path = self.working_dir + '/test2.db' - self.db2 = u1db_open(self.db2_path, create=True) - self.addCleanup(self.db2.close) - self.db2._set_replica_uid('test2') - self.doc = self.db.create_doc_from_json( - tests.simple_doc, doc_id='test-id') - self.doc2 = self.db2.create_doc_from_json( - tests.nested_doc, doc_id='my-test-id') - - def test_sync(self): - cmd = self.make_command(client.CmdSync) - cmd.run(self.db_path, self.db2_path) - self.assertGetDoc(self.db2, 'test-id', self.doc.rev, tests.simple_doc, - False) - self.assertGetDoc(self.db, 'my-test-id', self.doc2.rev, - tests.nested_doc, False) - - -class TestCmdSyncRemote(tests.TestCaseWithServer, TestCaseWithDB): - - make_app_with_state = \ - staticmethod(test_remote_sync_target.make_http_app) - - def setUp(self): - super(TestCmdSyncRemote, self).setUp() - self.startServer() - self.db2 = self.request_state._create_database('test2.db') - - def test_sync_remote(self): - doc1 = self.db.create_doc_from_json(tests.simple_doc) - doc2 = self.db2.create_doc_from_json(tests.nested_doc) - db2_url = self.getURL('test2.db') - self.assertTrue(db2_url.startswith('http://')) - self.assertTrue(db2_url.endswith('/test2.db')) - cmd = self.make_command(client.CmdSync) - cmd.run(self.db_path, db2_url) - self.assertGetDoc(self.db2, doc1.doc_id, doc1.rev, tests.simple_doc, - False) - self.assertGetDoc(self.db, doc2.doc_id, doc2.rev, tests.nested_doc, - False) - - -class TestCmdCreateIndex(TestCaseWithDB): - - def test_create_index(self): - cmd = self.make_command(client.CmdCreateIndex) - retval = cmd.run(self.db_path, "foo", ["bar", "baz"]) - self.assertEqual(self.db.list_indexes(), [('foo', ['bar', "baz"])]) - self.assertEqual(retval, None) # conveniently mapped to 0 - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual(cmd.stderr.getvalue(), '') - - def test_create_index_no_db(self): - cmd = self.make_command(client.CmdCreateIndex) - retval = cmd.run(self.db_path + "__DOES_NOT_EXIST", "foo", ["bar"]) - self.assertEqual(retval, 1) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual(cmd.stderr.getvalue(), 'Database does not exist.\n') - - def test_create_dupe_index(self): - self.db.create_index("foo", "bar") - cmd = self.make_command(client.CmdCreateIndex) - retval = cmd.run(self.db_path, "foo", ["bar"]) - self.assertEqual(retval, None) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual(cmd.stderr.getvalue(), '') - - def test_create_dupe_index_different_expression(self): - self.db.create_index("foo", "bar") - cmd = self.make_command(client.CmdCreateIndex) - retval = cmd.run(self.db_path, "foo", ["baz"]) - self.assertEqual(retval, 1) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual(cmd.stderr.getvalue(), - "There is already a different index named 'foo'.\n") - - def test_create_index_bad_expression(self): - cmd = self.make_command(client.CmdCreateIndex) - retval = cmd.run(self.db_path, "foo", ["WAT()"]) - self.assertEqual(retval, 1) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual(cmd.stderr.getvalue(), - 'Bad index expression.\n') - - -class TestCmdListIndexes(TestCaseWithDB): - - def test_list_no_indexes(self): - cmd = self.make_command(client.CmdListIndexes) - retval = cmd.run(self.db_path) - self.assertEqual(retval, None) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual(cmd.stderr.getvalue(), '') - - def test_list_indexes(self): - self.db.create_index("foo", "bar", "baz") - cmd = self.make_command(client.CmdListIndexes) - retval = cmd.run(self.db_path) - self.assertEqual(retval, None) - self.assertEqual(cmd.stdout.getvalue(), 'foo: bar, baz\n') - self.assertEqual(cmd.stderr.getvalue(), '') - - def test_list_several_indexes(self): - self.db.create_index("foo", "bar", "baz") - self.db.create_index("bar", "baz", "foo") - self.db.create_index("baz", "foo", "bar") - cmd = self.make_command(client.CmdListIndexes) - retval = cmd.run(self.db_path) - self.assertEqual(retval, None) - self.assertEqual(cmd.stdout.getvalue(), - 'bar: baz, foo\n' - 'baz: foo, bar\n' - 'foo: bar, baz\n' - ) - self.assertEqual(cmd.stderr.getvalue(), '') - - def test_list_indexes_no_db(self): - cmd = self.make_command(client.CmdListIndexes) - retval = cmd.run(self.db_path + "__DOES_NOT_EXIST") - self.assertEqual(retval, 1) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual(cmd.stderr.getvalue(), 'Database does not exist.\n') - - -class TestCmdDeleteIndex(TestCaseWithDB): - - def test_delete_index(self): - self.db.create_index("foo", "bar", "baz") - cmd = self.make_command(client.CmdDeleteIndex) - retval = cmd.run(self.db_path, "foo") - self.assertEqual(retval, None) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual(cmd.stderr.getvalue(), '') - self.assertEqual([], self.db.list_indexes()) - - def test_delete_index_no_db(self): - cmd = self.make_command(client.CmdDeleteIndex) - retval = cmd.run(self.db_path + "__DOES_NOT_EXIST", "foo") - self.assertEqual(retval, 1) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual(cmd.stderr.getvalue(), 'Database does not exist.\n') - - def test_delete_index_no_index(self): - cmd = self.make_command(client.CmdDeleteIndex) - retval = cmd.run(self.db_path, "foo") - self.assertEqual(retval, None) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual(cmd.stderr.getvalue(), '') - - -class TestCmdGetIndexKeys(TestCaseWithDB): - - def test_get_index_keys(self): - self.db.create_index("foo", "bar") - self.db.create_doc_from_json('{"bar": 42}') - cmd = self.make_command(client.CmdGetIndexKeys) - retval = cmd.run(self.db_path, "foo") - self.assertEqual(retval, None) - self.assertEqual(cmd.stdout.getvalue(), '42\n') - self.assertEqual(cmd.stderr.getvalue(), '') - - def test_get_index_keys_nonascii(self): - self.db.create_index("foo", "bar") - self.db.create_doc_from_json('{"bar": "\u00a4"}') - cmd = self.make_command(client.CmdGetIndexKeys) - retval = cmd.run(self.db_path, "foo") - self.assertEqual(retval, None) - self.assertEqual(cmd.stdout.getvalue(), '\xc2\xa4\n') - self.assertEqual(cmd.stderr.getvalue(), '') - - def test_get_index_keys_empty(self): - self.db.create_index("foo", "bar") - cmd = self.make_command(client.CmdGetIndexKeys) - retval = cmd.run(self.db_path, "foo") - self.assertEqual(retval, None) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual(cmd.stderr.getvalue(), '') - - def test_get_index_keys_no_db(self): - cmd = self.make_command(client.CmdGetIndexKeys) - retval = cmd.run(self.db_path + "__DOES_NOT_EXIST", "foo") - self.assertEqual(retval, 1) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual(cmd.stderr.getvalue(), 'Database does not exist.\n') - - def test_get_index_keys_no_index(self): - cmd = self.make_command(client.CmdGetIndexKeys) - retval = cmd.run(self.db_path, "foo") - self.assertEqual(retval, 1) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual(cmd.stderr.getvalue(), 'Index does not exist.\n') - - -class TestCmdGetFromIndex(TestCaseWithDB): - - def test_get_from_index(self): - self.db.create_index("index", "key") - doc1 = self.db.create_doc_from_json(tests.simple_doc) - doc2 = self.db.create_doc_from_json(tests.nested_doc) - cmd = self.make_command(client.CmdGetFromIndex) - retval = cmd.run(self.db_path, "index", ["value"]) - self.assertEqual(retval, None) - self.assertEqual(sorted(json.loads(cmd.stdout.getvalue())), - sorted([dict(id=doc1.doc_id, - rev=doc1.rev, - content=doc1.content), - dict(id=doc2.doc_id, - rev=doc2.rev, - content=doc2.content), - ])) - self.assertEqual(cmd.stderr.getvalue(), '') - - def test_get_from_index_empty(self): - self.db.create_index("index", "key") - cmd = self.make_command(client.CmdGetFromIndex) - retval = cmd.run(self.db_path, "index", ["value"]) - self.assertEqual(retval, None) - self.assertEqual(cmd.stdout.getvalue(), '[]\n') - self.assertEqual(cmd.stderr.getvalue(), '') - - def test_get_from_index_no_db(self): - cmd = self.make_command(client.CmdGetFromIndex) - retval = cmd.run(self.db_path + "__DOES_NOT_EXIST", "foo", []) - self.assertEqual(retval, 1) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual(cmd.stderr.getvalue(), 'Database does not exist.\n') - - def test_get_from_index_no_index(self): - cmd = self.make_command(client.CmdGetFromIndex) - retval = cmd.run(self.db_path, "foo", []) - self.assertEqual(retval, 1) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual(cmd.stderr.getvalue(), 'Index does not exist.\n') - - def test_get_from_index_two_expr_instead_of_one(self): - self.db.create_index("index", "key1") - cmd = self.make_command(client.CmdGetFromIndex) - cmd.argv = ["XX", "YY"] - retval = cmd.run(self.db_path, "index", ["value1", "value2"]) - self.assertEqual(retval, 1) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual("Invalid query: index 'index' requires" - " 1 query expression, not 2.\n" - "For example, the following would be valid:\n" - " XX YY %r 'index' 'value1'\n" - % self.db_path, cmd.stderr.getvalue()) - - def test_get_from_index_three_expr_instead_of_two(self): - self.db.create_index("index", "key1", "key2") - cmd = self.make_command(client.CmdGetFromIndex) - cmd.argv = ["XX", "YY"] - retval = cmd.run(self.db_path, "index", ["value1", "value2", "value3"]) - self.assertEqual(retval, 1) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual("Invalid query: index 'index' requires" - " 2 query expressions, not 3.\n" - "For example, the following would be valid:\n" - " XX YY %r 'index' 'value1' 'value2'\n" - % self.db_path, cmd.stderr.getvalue()) - - def test_get_from_index_one_expr_instead_of_two(self): - self.db.create_index("index", "key1", "key2") - cmd = self.make_command(client.CmdGetFromIndex) - cmd.argv = ["XX", "YY"] - retval = cmd.run(self.db_path, "index", ["value1"]) - self.assertEqual(retval, 1) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual("Invalid query: index 'index' requires" - " 2 query expressions, not 1.\n" - "For example, the following would be valid:\n" - " XX YY %r 'index' 'value1' '*'\n" - % self.db_path, cmd.stderr.getvalue()) - - def test_get_from_index_cant_bad_glob(self): - self.db.create_index("index", "key1", "key2") - cmd = self.make_command(client.CmdGetFromIndex) - cmd.argv = ["XX", "YY"] - retval = cmd.run(self.db_path, "index", ["value1*", "value2"]) - self.assertEqual(retval, 1) - self.assertEqual(cmd.stdout.getvalue(), '') - self.assertEqual("Invalid query:" - " a star can only be followed by stars.\n" - "For example, the following would be valid:\n" - " XX YY %r 'index' 'value1*' '*'\n" - % self.db_path, cmd.stderr.getvalue()) - - -class RunMainHelper(object): - - def run_main(self, args, stdin=None): - if stdin is not None: - self.patch(sys, 'stdin', cStringIO.StringIO(stdin)) - stdout = cStringIO.StringIO() - stderr = cStringIO.StringIO() - self.patch(sys, 'stdout', stdout) - self.patch(sys, 'stderr', stderr) - try: - ret = client.main(args) - except SystemExit, e: - self.fail("Intercepted SystemExit: %s" % (e,)) - if ret is None: - ret = 0 - return ret, stdout.getvalue(), stderr.getvalue() - - -class TestCommandLine(TestCaseWithDB, RunMainHelper): - """These are meant to test that the infrastructure is fully connected. - - Each command is likely to only have one test here. Something that ensures - 'main()' knows about and can run the command correctly. Most logic-level - testing of the Command should go into its own test class above. - """ - - def _get_u1db_client_path(self): - from u1db import __path__ as u1db_path - u1db_parent_dir = os.path.dirname(u1db_path[0]) - return os.path.join(u1db_parent_dir, 'u1db-client') - - def runU1DBClient(self, args): - command = [sys.executable, self._get_u1db_client_path()] - command.extend(args) - p = subprocess.Popen(command, stdin=subprocess.PIPE, - stdout=subprocess.PIPE, stderr=subprocess.PIPE) - self.addCleanup(safe_close, p) - return p - - def test_create_subprocess(self): - p = self.runU1DBClient(['create', '--id', 'test-id', self.db_path]) - stdout, stderr = p.communicate(tests.simple_doc) - self.assertEqual(0, p.returncode) - self.assertEqual('', stdout) - doc = self.db.get_doc('test-id') - self.assertEqual(tests.simple_doc, doc.get_json()) - self.assertFalse(doc.has_conflicts) - expected = 'id: test-id\nrev: %s\n' % (doc.rev,) - stripped = stderr.replace('\r\n', '\n') - if expected != stripped: - # When run under python-dbg, it prints out the refs after the - # actual content, so match it if we need to. - expected_re = expected + '\[\d+ refs\]\n' - self.assertRegexpMatches(stripped, expected_re) - - def test_get(self): - doc = self.db.create_doc_from_json(tests.simple_doc, doc_id='test-id') - ret, stdout, stderr = self.run_main(['get', self.db_path, 'test-id']) - self.assertEqual(0, ret) - self.assertEqual(tests.simple_doc + "\n", stdout) - self.assertEqual('rev: %s\n' % (doc.rev,), stderr) - ret, stdout, stderr = self.run_main(['get', self.db_path, 'not-there']) - self.assertEqual(1, ret) - - def test_delete(self): - doc = self.db.create_doc_from_json(tests.simple_doc, doc_id='test-id') - ret, stdout, stderr = self.run_main( - ['delete', self.db_path, 'test-id', doc.rev]) - doc = self.db.get_doc('test-id', include_deleted=True) - self.assertEqual(0, ret) - self.assertEqual('', stdout) - self.assertEqual('rev: %s\n' % (doc.rev,), stderr) - - def test_init_db(self): - path = self.working_dir + '/test2.db' - ret, stdout, stderr = self.run_main(['init-db', path]) - u1db_open(path, create=False) - - def test_put(self): - doc = self.db.create_doc_from_json(tests.simple_doc, doc_id='test-id') - ret, stdout, stderr = self.run_main( - ['put', self.db_path, 'test-id', doc.rev], - stdin=tests.nested_doc) - doc = self.db.get_doc('test-id') - self.assertFalse(doc.has_conflicts) - self.assertEqual(tests.nested_doc, doc.get_json()) - self.assertEqual(0, ret) - self.assertEqual('', stdout) - self.assertEqual('rev: %s\n' % (doc.rev,), stderr) - - def test_sync(self): - doc = self.db.create_doc_from_json(tests.simple_doc, doc_id='test-id') - self.db2_path = self.working_dir + '/test2.db' - self.db2 = u1db_open(self.db2_path, create=True) - self.addCleanup(self.db2.close) - ret, stdout, stderr = self.run_main( - ['sync', self.db_path, self.db2_path]) - self.assertEqual(0, ret) - self.assertEqual('', stdout) - self.assertEqual('', stderr) - self.assertGetDoc( - self.db2, 'test-id', doc.rev, tests.simple_doc, False) - - -class TestHTTPIntegration(tests.TestCaseWithServer, RunMainHelper): - """Meant to test the cases where commands operate over http.""" - - def server_def(self): - def make_server(host_port, _application): - return serve.make_server(host_port[0], host_port[1], - self.working_dir) - return make_server, "shutdown", "http" - - def setUp(self): - super(TestHTTPIntegration, self).setUp() - self.working_dir = self.createTempDir(prefix='u1db-http-server-') - self.startServer() - - def getPath(self, dbname): - return os.path.join(self.working_dir, dbname) - - def test_init_db(self): - url = self.getURL('new.db') - ret, stdout, stderr = self.run_main(['init-db', url]) - u1db_open(self.getPath('new.db'), create=False) - - def test_create_get_put_delete(self): - db = u1db_open(self.getPath('test.db'), create=True) - url = self.getURL('test.db') - doc_id = '%abcd' - ret, stdout, stderr = self.run_main(['create', url, '--id', doc_id], - stdin=tests.simple_doc) - self.assertEqual(0, ret) - ret, stdout, stderr = self.run_main(['get', url, doc_id]) - self.assertEqual(0, ret) - self.assertTrue(stderr.startswith('rev: ')) - doc_rev = stderr[len('rev: '):].rstrip() - ret, stdout, stderr = self.run_main(['put', url, doc_id, doc_rev], - stdin=tests.nested_doc) - self.assertEqual(0, ret) - self.assertTrue(stderr.startswith('rev: ')) - doc_rev1 = stderr[len('rev: '):].rstrip() - self.assertGetDoc(db, doc_id, doc_rev1, tests.nested_doc, False) - ret, stdout, stderr = self.run_main(['delete', url, doc_id, doc_rev1]) - self.assertEqual(0, ret) - self.assertTrue(stderr.startswith('rev: ')) - doc_rev2 = stderr[len('rev: '):].rstrip() - self.assertGetDocIncludeDeleted(db, doc_id, doc_rev2, None, False) |