summaryrefslogtreecommitdiff
path: root/src/leap/mail/imap/tests/utils.py
blob: 83c3f298381619c9510b750e619ceb6291badca5 (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
125
126
127
128
129
130
131
# -*- coding: utf-8 -*-
# utils.py
# Copyright (C) 2014, 2015 LEAP
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
"""
Common utilities for testing Soledad IMAP Server.
"""
from email import parser

from mock import Mock
from twisted.mail import imap4
from twisted.internet import defer
from twisted.protocols import loopback
from twisted.python import log

from leap.mail.adaptors import soledad as soledad_adaptor
from leap.mail.imap.account import IMAPAccount
from leap.mail.imap.server import LEAPIMAPServer
from leap.mail.tests.common import SoledadTestMixin

TEST_USER = "testuser@leap.se"
TEST_PASSWD = "1234"


#
# Simple IMAP4 Client for testing
#

class SimpleClient(imap4.IMAP4Client):
    """
    A Simple IMAP4 Client to test our
    Soledad-LEAPServer
    """

    def __init__(self, deferred, contextFactory=None):
        imap4.IMAP4Client.__init__(self, contextFactory)
        self.deferred = deferred
        self.events = []

    def serverGreeting(self, caps):
        self.deferred.callback(None)

    def modeChanged(self, writeable):
        self.events.append(['modeChanged', writeable])
        self.transport.loseConnection()

    def flagsChanged(self, newFlags):
        self.events.append(['flagsChanged', newFlags])
        self.transport.loseConnection()

    def newMessages(self, exists, recent):
        self.events.append(['newMessages', exists, recent])
        self.transport.loseConnection()


class IMAP4HelperMixin(SoledadTestMixin):
    """
    MixIn containing several utilities to be shared across
    different TestCases
    """
    serverCTX = None
    clientCTX = None

    def setUp(self):

        soledad_adaptor.cleanup_deferred_locks()

        UUID = 'deadbeef',
        USERID = TEST_USER

        def setup_server(account):
            self.server = LEAPIMAPServer(
                uuid=UUID, userid=USERID,
                contextFactory=self.serverCTX,
                soledad=self._soledad)
            self.server.theAccount = account

            d_server_ready = defer.Deferred()
            self.client = SimpleClient(
                d_server_ready, contextFactory=self.clientCTX)
            self.connected = d_server_ready

        def setup_account(_):
            self.parser = parser.Parser()

            # XXX this should be fixed in soledad.
            # Soledad sync makes trial block forever. The sync it's mocked to
            # fix this problem. _mock_soledad_get_from_index can be used from
            # the tests to provide documents.
            # TODO see here, possibly related? -- http://www.pythoneye.com/83_20424875/
            self._soledad.sync = Mock()

            d = defer.Deferred()
            self.acc = IMAPAccount(USERID, self._soledad, d=d)
            return d

        d = super(IMAP4HelperMixin, self).setUp()
        d.addCallback(setup_account)
        d.addCallback(setup_server)
        return d

    def tearDown(self):
        SoledadTestMixin.tearDown(self)
        del self._soledad
        del self.client
        del self.server
        del self.connected

    def _cbStopClient(self, ignore):
        self.client.transport.loseConnection()

    def _ebGeneral(self, failure):
        self.client.transport.loseConnection()
        self.server.transport.loseConnection()
        if hasattr(self, 'function'):
            log.err(failure, "Problem with %r" % (self.function,))

    def loopback(self):
        return loopback.loopbackAsync(self.server, self.client)