1
# Copyright 2009 Canonical Ltd.
3
# This file is part of desktopcouch.
5
# desktopcouch is free software: you can redistribute it and/or modify
6
# it under the terms of the GNU Lesser General Public License version 3
7
# as published by the Free Software Foundation.
9
# desktopcouch is distributed in the hope that it will be useful,
10
# but WITHOUT ANY WARRANTY; without even the implied warranty of
11
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12
# GNU Lesser General Public License for more details.
14
# You should have received a copy of the GNU Lesser General Public License
15
# along with desktopcouch. If not, see <http://www.gnu.org/licenses/>.
21
import desktopcouch.tests as dctests
23
from desktopcouch.pair.couchdb_pairing import couchdb_io
24
from desktopcouch.records.server import CouchDatabase
25
from desktopcouch.records.record import Record
30
URI = None # use autodiscovery that desktopcouch.tests permits.
32
class TestCouchdbIo(unittest.TestCase):
36
self.mgt_database = CouchDatabase('management', create=True, uri=URI)
37
self.foo_database = CouchDatabase('foo', create=True, uri=URI)
38
#create some records to pull out and test
39
self.foo_database.put_record(Record({
40
"key1_1": "val1_1", "key1_2": "val1_2", "key1_3": "val1_3",
41
"record_type": "test.com"}))
42
self.foo_database.put_record(Record({
43
"key2_1": "val2_1", "key2_2": "val2_2", "key2_3": "val2_3",
44
"record_type": "test.com"}))
45
self.foo_database.put_record(Record({
46
"key13_1": "va31_1", "key3_2": "val3_2", "key3_3": "val3_3",
47
"record_type": "test.com"}))
50
"""tear down each test"""
51
del self.mgt_database._server['management']
52
del self.mgt_database._server['foo']
54
def test_put_static_paired_service(self):
55
service_name = "dummyfortest"
57
"consumer_key": str("abcdef"),
58
"consumer_secret": str("ghighjklm"),
59
"token": str("opqrst"),
60
"token_secret": str("uvwxyz"),
62
couchdb_io.put_static_paired_service(oauth_data, service_name, uri=URI)
63
pairings = list(couchdb_io.get_pairings())
65
def test_put_dynamic_paired_host(self):
66
hostname = "host%d" % (os.getpid(),)
67
remote_uuid = str(uuid.uuid4())
69
"consumer_key": str("abcdef"),
70
"consumer_secret": str("ghighjklm"),
71
"token": str("opqrst"),
72
"token_secret": str("uvwxyz"),
75
couchdb_io.put_dynamic_paired_host(hostname, remote_uuid, oauth_data,
77
couchdb_io.put_dynamic_paired_host(hostname, remote_uuid, oauth_data,
79
couchdb_io.put_dynamic_paired_host(hostname, remote_uuid, oauth_data,
82
pairings = list(couchdb_io.get_pairings())
83
self.assertEqual(3, len(pairings))
84
self.assertEqual(pairings[0].value["oauth"], oauth_data)
85
self.assertEqual(pairings[0].value["server"], hostname)
86
self.assertEqual(pairings[0].value["pairing_identifier"], remote_uuid)
88
for i, row in enumerate(pairings):
89
couchdb_io.remove_pairing(row.id, i == 1)
91
pairings = list(couchdb_io.get_pairings())
92
self.assertEqual(0, len(pairings))
95
def test_get_database_names_replicatable_bad_server(self):
96
# If this resolves, FIRE YOUR DNS PROVIDER.
97
self.assertRaises(httplib2.ServerNotFoundError,
98
couchdb_io.get_database_names_replicatable,
99
uri='http://test.desktopcouch.example.com:9/')
101
def test_get_database_names_replicatable(self):
102
names = couchdb_io.get_database_names_replicatable(uri=URI)
103
self.assertFalse('management' in names)
104
self.assertTrue('foo' in names)
106
def test_get_my_host_unique_id(self):
107
got = couchdb_io.get_my_host_unique_id(uri=URI)
108
again = couchdb_io.get_my_host_unique_id(uri=URI)
109
self.assertEquals(len(got), 1)
110
self.assertEquals(got, again)
112
def Xtest_replication_good(self):
115
def Xtest_replication_no_oauth_remote(self):
118
def Xtest_replication_bad_oauth_remote(self):
121
def Xtest_replication_no_oauth_local(self):
124
def Xtest_replication_bad_oauth_local(self):
128
if __name__ == "__main__":
added
removed
desktopcouch/pair/tests/test_couchdb_io.py
