← Back to branch summary

~ubuntu-branches/debian/squeeze/desktopcouch/squeeze

« back to all changes in this revision

Viewing changes to desktopcouch/pair/tests/test_couchdb_io.py

  • Committer: Bazaar Package Importer
  • Author(s): David Paleino
  • Date: 2010-03-04 19:11:48 UTC
  • mfrom: (1.5.3 upstream)
  • Revision ID: james.westby@ubuntu.com-20100304191148-zcwp3rffruwkalvc
Tags: 0.6.2-1
* New upstream version
* debian/patches/01-use_simplejson.patch removed, better handled
  upstream (simplejson for py2.5 and json for py2.6)
* debian/rules: chmod desktop-stop to 755, since it's intended as
  an executable script
* debian/control: bump debhelper dependency to >= 7.0.50~ to use
  overridden rules in debian/rules

Show diffs side-by-side

added added

removed removed

Lines of Context:
desktopcouch/pair/tests/test_couchdb_io.py
18
18
import pygtk
19
19
pygtk.require('2.0')
20
20
 
21
 
import desktopcouch.tests as dctests
 
21
import desktopcouch.tests as test_environment
22
22
 
23
23
from desktopcouch.pair.couchdb_pairing import couchdb_io
24
24
from desktopcouch.records.server import CouchDatabase
25
25
from desktopcouch.records.record import Record
26
 
import unittest
 
26
from twisted.trial import unittest
27
27
import uuid
28
28
import os
29
29
import httplib2
 
30
import socket
30
31
URI = None  # use autodiscovery that desktopcouch.tests permits.
31
32
 
32
33
class TestCouchdbIo(unittest.TestCase):
33
34
 
34
35
    def setUp(self):
35
36
        """setup each test"""
36
 
        self.mgt_database = CouchDatabase('management', create=True, uri=URI)
37
 
        self.foo_database = CouchDatabase('foo', create=True, uri=URI)
 
37
        self.mgt_database = CouchDatabase('management', create=True, uri=URI,
 
38
                ctx=test_environment.test_context)
 
39
        self.foo_database = CouchDatabase('foo', create=True, uri=URI,
 
40
                ctx=test_environment.test_context)
38
41
        #create some records to pull out and test
39
42
        self.foo_database.put_record(Record({
40
43
            "key1_1": "val1_1", "key1_2": "val1_2", "key1_3": "val1_3",
51
54
        del self.mgt_database._server['management']
52
55
        del self.mgt_database._server['foo']
53
56
 
 
57
 
 
58
    def test_obsfuscation(self):
 
59
        t = {'url': 'https://couchdb.one.ubuntu.com/u%2Fb2%2Fc8%2F276%2Ftest', 'auth': {'oauth': {'consumer_secret': 'SeCrEtSe', 'token': '3XRjQrWX92TTTJFDTWJJ', 'consumer_key': 'ubuntuone', 'token_secret': 'jBmSeCrEtawkefwklefliwuregqwlkeh347wq87w4fiuq4fyu3q4fiqwu4fqwfiqufM6xjsPwSeCrEt4'}}}
 
60
        cleaned_t = couchdb_io.obsfuscate(t)
 
61
        self.failIf("SeCrEt" in str(cleaned_t), {'url': 'https://couchdb.one.ubuntu.com/u%2Fb2%2Fc8%2F276%2Ftest', 'auth': {'oauth': {'consumer_secret': 'HiddenHidd', 'token': '3XRjQrWX92TTTJFDTWJJ', 'consumer_key': 'ubuntuone', 'token_secret': 'HiddenHiddenHiddenHiddenHiddenHiddenHiddenHiddenHiddenHiddenHiddenHiddenHiddenHi'}}})
 
62
 
 
63
        self.assertEqual(couchdb_io.obsfuscate(""), "")
 
64
        self.assertEqual(couchdb_io.obsfuscate({}), {})
 
65
        self.assertEqual(couchdb_io.obsfuscate({1:{}}), {1:{}})
 
66
        self.assertEqual(couchdb_io.obsfuscate({1:1}), {1:1})
 
67
 
 
68
 
54
69
    def test_put_static_paired_service(self):
55
70
        service_name = "dummyfortest"
56
71
        oauth_data = {
59
74
                "token": str("opqrst"),
60
75
                "token_secret": str("uvwxyz"),
61
76
            }
62
 
        couchdb_io.put_static_paired_service(oauth_data, service_name, uri=URI)
63
 
        pairings = list(couchdb_io.get_pairings())
 
77
        couchdb_io.put_static_paired_service(oauth_data, service_name, uri=URI, ctx=test_environment.test_context)
 
78
        pairings = list(couchdb_io.get_pairings(ctx=test_environment.test_context))
64
79
 
65
80
    def test_put_dynamic_paired_host(self):
66
81
        hostname = "host%d" % (os.getpid(),)
73
88
            }
74
89
 
75
90
        couchdb_io.put_dynamic_paired_host(hostname, remote_uuid, oauth_data,
76
 
                uri=URI)
77
 
        couchdb_io.put_dynamic_paired_host(hostname, remote_uuid, oauth_data,
78
 
                uri=URI)
79
 
        couchdb_io.put_dynamic_paired_host(hostname, remote_uuid, oauth_data,
80
 
                uri=URI)
 
91
                uri=URI, ctx=test_environment.test_context)
 
92
        couchdb_io.put_dynamic_paired_host(hostname, remote_uuid, oauth_data,
 
93
                uri=URI, ctx=test_environment.test_context)
 
94
        couchdb_io.put_dynamic_paired_host(hostname, remote_uuid, oauth_data,
 
95
                uri=URI, ctx=test_environment.test_context)
81
96
 
82
 
        pairings = list(couchdb_io.get_pairings())
 
97
        pairings = list(couchdb_io.get_pairings(ctx=test_environment.test_context))
83
98
        self.assertEqual(3, len(pairings))
84
99
        self.assertEqual(pairings[0].value["oauth"], oauth_data)
85
100
        self.assertEqual(pairings[0].value["server"], hostname)
86
101
        self.assertEqual(pairings[0].value["pairing_identifier"], remote_uuid)
87
102
 
88
103
        for i, row in enumerate(pairings):
89
 
            couchdb_io.remove_pairing(row.id, i == 1)
 
104
            couchdb_io.remove_pairing(row.id, i == 1, ctx=test_environment.test_context)
90
105
 
91
 
        pairings = list(couchdb_io.get_pairings())
 
106
        pairings = list(couchdb_io.get_pairings(ctx=test_environment.test_context))
92
107
        self.assertEqual(0, len(pairings))
93
108
 
94
109
 
95
110
    def test_get_database_names_replicatable_bad_server(self):
96
 
        # If this resolves, FIRE YOUR DNS PROVIDER.
 
111
        hostname = "test.desktopcouch.example.com"
 
112
        try:
 
113
            socket.gethostbyname(hostname)
 
114
            raise unittest.SkipTest("nxdomain hijacked")
 
115
        except socket.gaierror:
 
116
            pass
97
117
 
98
118
        try:
99
119
            names = couchdb_io.get_database_names_replicatable(
100
 
                uri='http://test.desktopcouch.example.com:9/')
 
120
                uri='http://' + hostname + ':9/')
101
121
            self.assertEqual(set(), names)
102
122
        except httplib2.ServerNotFoundError:
103
123
            pass
104
124
 
105
125
    def test_get_database_names_replicatable(self):
106
 
        names = couchdb_io.get_database_names_replicatable(uri=URI)
 
126
        names = couchdb_io.get_database_names_replicatable(uri=URI, ctx=test_environment.test_context)
107
127
        self.assertFalse('management' in names)
108
128
        self.assertTrue('foo' in names)
109
129
 
110
130
    def test_get_my_host_unique_id(self):
111
 
        got = couchdb_io.get_my_host_unique_id(uri=URI)
112
 
        again = couchdb_io.get_my_host_unique_id(uri=URI)
 
131
        got = couchdb_io.get_my_host_unique_id(uri=URI, ctx=test_environment.test_context)
 
132
        again = couchdb_io.get_my_host_unique_id(uri=URI, ctx=test_environment.test_context)
113
133
        self.assertEquals(len(got), 1)
114
134
        self.assertEquals(got, again)
115
135