summaryrefslogtreecommitdiff
path: root/tests/test_couch.py
blob: 2337be9b864d0fc50f05f4c5de6693886b59248b (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
"""Test ObjectStore backend bits.

For these tests to run, a couch server has to be running on (default) port
5984.
"""

import sys
import copy
import testtools
import testscenarios
from leap.soledad.backends import couch
from leap.soledad.tests import u1db_tests as tests
from leap.soledad.tests.u1db_tests.test_backends import (
  TestAlternativeDocument,
  AllDatabaseTests,
  LocalDatabaseTests,
  LocalDatabaseValidateGenNTransIdTests,
  LocalDatabaseValidateSourceGenTests,
  LocalDatabaseWithConflictsTests,
  DatabaseIndexTests,
)


#-----------------------------------------------------------------------------
# The following tests come from `u1db.tests.test_common_backends`.
#-----------------------------------------------------------------------------

class TestCouchBackendImpl(tests.TestCase):

    def test__allocate_doc_id(self):
        db = couch.CouchDatabase('http://localhost:5984', 'u1db_tests')
        doc_id1 = db._allocate_doc_id()
        self.assertTrue(doc_id1.startswith('D-'))
        self.assertEqual(34, len(doc_id1))
        int(doc_id1[len('D-'):], 16)
        self.assertNotEqual(doc_id1, db._allocate_doc_id())


#-----------------------------------------------------------------------------
# The following tests come from `u1db.tests.test_backends`.
#-----------------------------------------------------------------------------

def make_couch_database_for_test(test, replica_uid):
    return couch.CouchDatabase('http://localhost:5984', 'u1db_tests',
                               replica_uid=replica_uid)

def copy_couch_database_for_test(test, db):
    new_db = couch.CouchDatabase('http://localhost:5984', 'u1db_tests_2',
                                 replica_uid=db.replica_uid)
    new_db._transaction_log = copy.deepcopy(db._transaction_log)
    new_db._sync_log = copy.deepcopy(db._sync_log)
    gen, docs = db.get_all_docs(include_deleted=True)
    for doc in docs:
        new_db._put_doc(doc)
    new_db._ensure_u1db_data()
    return new_db


COUCH_SCENARIOS = [
        ('couch', {'make_database_for_test': make_couch_database_for_test,
                  'copy_database_for_test': copy_couch_database_for_test,
                  'make_document_for_test': tests.make_document_for_test,}),
        ]


class CouchTests(AllDatabaseTests):

    scenarios = COUCH_SCENARIOS

    def tearDown(self):
        self.db.delete_database()
        super(CouchTests, self).tearDown()


class CouchDatabaseTests(LocalDatabaseTests):

    scenarios = COUCH_SCENARIOS

    def tearDown(self):
        self.db.delete_database()
        super(CouchDatabaseTests, self).tearDown()


class CouchValidateGenNTransIdTests(LocalDatabaseValidateGenNTransIdTests):

    scenarios = COUCH_SCENARIOS

    def tearDown(self):
        self.db.delete_database()
        super(CouchValidateGenNTransIdTests, self).tearDown()


class CouchValidateSourceGenTests(LocalDatabaseValidateSourceGenTests):

    scenarios = COUCH_SCENARIOS

    def tearDown(self):
        self.db.delete_database()
        super(CouchValidateSourceGenTests, self).tearDown()


class CouchWithConflictsTests(LocalDatabaseWithConflictsTests):

    scenarios = COUCH_SCENARIOS

    def tearDown(self):
        self.db.delete_database()
        super(CouchWithConflictsTests, self).tearDown()


# Notice: the CouchDB backend is currently used for storing encrypted data in
# the server, so indexing makes no sense. Thus, we ignore index testing for
# now.

#class CouchIndexTests(DatabaseIndexTests):
#
#    scenarios = COUCH_SCENARIOS
#
#    def tearDown(self):
#        self.db.delete_database()
#        super(CouchIndexTests, self).tearDown()
#

load_tests = tests.load_with_scenarios