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/>.
22
import desktopcouch.application.tests as test_environment
24
from desktopcouch.application.pair.couchdb_pairing import couchdb_io
25
from desktopcouch.application.server import DesktopDatabase
26
from desktopcouch.records import Record
27
from desktopcouch.application.replication_services import ubuntuone
32
URI = None # use autodiscovery that desktopcouch.tests permits.
35
class TestCouchdbIo(unittest.TestCase):
36
"""Test case for Couchdb IO"""
40
self.mgt_database = DesktopDatabase('management', create=True, uri=URI,
41
ctx=test_environment.test_context)
42
self.foo_database = DesktopDatabase('foo', create=True, uri=URI,
43
ctx=test_environment.test_context)
44
#create some records to pull out and test
45
self.foo_database = DesktopDatabase('bar', create=True, uri=URI,
46
ctx=test_environment.test_context)
47
#create some records to pull out and test
48
self.foo_database.put_record(Record({
49
"key1_1": "val1_1", "key1_2": "val1_2", "key1_3": "val1_3",
50
"record_type": "test.com"}))
51
self.foo_database.put_record(Record({
52
"key2_1": "val2_1", "key2_2": "val2_2", "key2_3": "val2_3",
53
"record_type": "test.com"}))
54
self.foo_database.put_record(Record({
55
"key13_1": "va31_1", "key3_2": "val3_2", "key3_3": "val3_3",
56
"record_type": "test.com"}))
58
couchdb_io.put_static_paired_service({
59
"consumer_key": str("abcdef"),
60
"consumer_secret": str("ghighjklm"),
61
"token": str("opqrst"),
62
"token_secret": str("uvwxyz")}, "ubuntuone", uri=URI,
63
ctx=test_environment.test_context)
64
excl = ubuntuone.ReplicationExclusion(
65
ctx=test_environment.test_context)
69
"""tear down each test"""
70
del self.mgt_database._server['management'] # pylint: disable=W0212
71
del self.mgt_database._server['foo'] # pylint: disable=W0212
73
def test_obsfuscation(self):
74
"""Test the obfuscation of sensitive data."""
77
'https://couchdb.one.ubuntu.com/u%2Fb2%2Fc8%2F276%2Ftest',
80
'consumer_secret': 'SeCrEtSe',
81
'token': '3XRjQrWX92TTTJFDTWJJ',
82
'consumer_key': 'ubuntuone',
84
'jBmSeCrEtawkefwklefliwuregqwlkeh347wq87w4fiuq4fyu3q4f'
85
'iqwu4fqwfiqufM6xjsPwSeCrEt4'}}}
86
cleaned_t = couchdb_io.obsfuscate(test_record)
88
"SeCrEt" in str(cleaned_t), {
90
'https://couchdb.one.ubuntu.com/u%2Fb2%2Fc8%2F276%2Ftest',
92
'consumer_secret': 'HiddenHidd',
93
'token': '3XRjQrWX92TTTJFDTWJJ',
94
'consumer_key': 'ubuntuone',
96
'HiddenHiddenHiddenHiddenHiddenHiddenHiddenHiddenHidde'
97
'nHiddenHiddenHiddenHiddenHi'}}})
98
self.assertEqual(couchdb_io.obsfuscate(""), "")
99
self.assertEqual(couchdb_io.obsfuscate({}), {})
100
self.assertEqual(couchdb_io.obsfuscate({1: {}}), {1: {}})
101
self.assertEqual(couchdb_io.obsfuscate({1: 1}), {1: 1})
103
def test_put_static_paired_service(self): # pylint: disable=R0201
104
"""Test putting a static paired service."""
105
service_name = "dummyfortest"
107
"consumer_key": str("abcdef"),
108
"consumer_secret": str("ghighjklm"),
109
"token": str("opqrst"),
110
"token_secret": str("uvwxyz"),
112
couchdb_io.put_static_paired_service(
113
oauth_data, service_name, uri=URI,
114
ctx=test_environment.test_context)
115
couchdb_io.get_pairings(ctx=test_environment.test_context)
118
def test_put_dynamic_paired_host(self):
119
"""Test putting a dynamically paired host."""
120
hostname = "host%d" % (os.getpid(),)
121
remote_uuid = str(uuid.uuid4())
123
"consumer_key": str("abcdef"),
124
"consumer_secret": str("ghighjklm"),
125
"token": str("opqrst"),
126
"token_secret": str("uvwxyz"),
129
couchdb_io.put_dynamic_paired_host(hostname, remote_uuid, oauth_data,
130
uri=URI, ctx=test_environment.test_context)
131
couchdb_io.put_dynamic_paired_host(hostname, remote_uuid, oauth_data,
132
uri=URI, ctx=test_environment.test_context)
133
couchdb_io.put_dynamic_paired_host(hostname, remote_uuid, oauth_data,
134
uri=URI, ctx=test_environment.test_context)
136
pairings = list(couchdb_io.get_pairings(
137
ctx=test_environment.test_context))
138
# 3, plus 1 from setUp()
139
self.assertEqual(4, len(pairings))
140
self.assertEqual(pairings[0].value["oauth"], oauth_data)
141
self.assertEqual(pairings[0].value["server"], hostname)
142
self.assertEqual(pairings[0].value["pairing_identifier"], remote_uuid)
144
for i, row in enumerate(pairings):
145
couchdb_io.remove_pairing(
146
row.id, i == 1, ctx=test_environment.test_context)
149
couchdb_io.get_pairings(ctx=test_environment.test_context))
150
self.assertEqual(0, len(pairings))
152
def test_get_database_names_replicatable(self):
153
"""Test get the names of replicatable databases."""
154
names = couchdb_io.get_database_names_replicatable(
155
uri=URI, ctx=test_environment.test_context)
156
self.assertFalse('management' in names)
157
self.assertTrue('foo' in names)
158
self.assertFalse('bar' in names, names) # is excluded
160
def test_get_my_host_unique_id(self):
161
"""Test get the unique id of the host record."""
162
got = couchdb_io.get_my_host_unique_id(
163
uri=URI, ctx=test_environment.test_context)
164
again = couchdb_io.get_my_host_unique_id(
165
uri=URI, ctx=test_environment.test_context)
166
self.assertEquals(len(got), 1)
167
self.assertEquals(got, again)
169
def test_mkuri(self):
170
"""Test creating a URI."""
171
uri = couchdb_io.mkuri(
172
'fnord.org', 55241, has_ssl=True, path='a/b/c',
173
auth_pair=('f o o', 'b=a=r'))
175
'https://f%20o%20o:b%3Da%3Dr@fnord.org:55241/a/b/c', uri)
added
removed
desktopcouch/application/pair/tests/test_couchdb_io.py
