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
132
133
134
|
#
# Copyright (c) 2014 ThoughtWorks, Inc.
#
# Pixelated is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Pixelated 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Pixelated. If not, see <http://www.gnu.org/licenses/>.
import json
import shutil
from leap.exceptions import SRPAuthenticationError
from leap.mail.imap.account import IMAPAccount
from mockito import mock, when, any as ANY
from twisted.cred import portal
from twisted.cred.checkers import AllowAnonymousAccess
from twisted.internet import defer
from leap.auth import SRPAuth
from pixelated.adapter.mailstore.leap_attachment_store import LeapAttachmentStore
from pixelated.adapter.services.feedback_service import FeedbackService
from pixelated.application import UserAgentMode, ServicesFactory
from pixelated.adapter.mailstore import LeapMailStore
from pixelated.adapter.mailstore.searchable_mailstore import SearchableMailStore
from pixelated.adapter.search import SearchEngine
from pixelated.adapter.services.draft_service import DraftService
from pixelated.bitmask_libraries.session import LeapSession, LeapSessionFactory
from pixelated.config import services as config_services
# from pixelated.config.services import Services
from pixelated.resources.auth import LeapPasswordChecker, SessionChecker, PixelatedRealm, PixelatedAuthSessionWrapper
from pixelated.resources.login_resource import LoginResource
from pixelated.resources.root_resource import RootResource
from test.support.integration import AppTestClient
from test.support.integration.app_test_client import initialize_soledad
from test.support.test_helper import request_mock
class MultiUserClient(AppTestClient):
@defer.inlineCallbacks
def start_client(self):
self.soledad_test_folder = self._generate_soledad_test_folder_name()
SearchEngine.DEFAULT_INDEX_HOME = self.soledad_test_folder
self.cleanup = lambda: shutil.rmtree(self.soledad_test_folder)
self.soledad = yield initialize_soledad(tempdir=self.soledad_test_folder)
self.service_factory = ServicesFactory(UserAgentMode(is_single_user=False))
root_resource = RootResource(self.service_factory)
anonymous_resource = LoginResource(self.service_factory)
leap_provider = mock()
checker = LeapPasswordChecker(leap_provider)
session_checker = SessionChecker()
realm = PixelatedRealm(root_resource, anonymous_resource)
_portal = portal.Portal(realm, [checker, session_checker, AllowAnonymousAccess()])
protected_resource = PixelatedAuthSessionWrapper(_portal, root_resource, anonymous_resource, [])
anonymous_resource.set_portal(_portal)
root_resource.initialize(_portal)
self.resource = protected_resource
@defer.inlineCallbacks
def login(self, username='username', password='password'):
leap_session = mock(LeapSession)
user_auth = mock()
user_auth.uuid = 'some_user_uuid'
leap_session.user_auth = user_auth
config = mock()
config.leap_home = 'some_folder'
leap_session.config = config
leap_session.fresh_account = False
self._set_leap_srp_auth(username, password)
when(LeapSessionFactory).create(username, password).thenReturn(leap_session)
_services = yield self.generate_services()
when(config_services).Services(leap_session).thenReturn(_services)
# when(Services).setup().thenReturn(defer.succeed('mocked so irrelevant'))
request = request_mock(path='/login', method="POST", body={'username': username, 'password': password})
defer.returnValue(self._render(request, as_json=False))
def _set_leap_srp_auth(self, username, password):
auth_dict = {'username': 'password'}
if auth_dict[username] == password:
when(SRPAuth).authenticate(username, password).thenReturn(True)
else:
when(SRPAuth).authenticate(username, password).thenRaise(SRPAuthenticationError())
def get(self, path, get_args='', as_json=True, from_request=None):
request = request_mock(path)
request.args = get_args
if from_request:
session = from_request.getSession()
request.session = session
return self._render(request, as_json)
@defer.inlineCallbacks
def generate_services(self):
search_engine = SearchEngine(self.INDEX_KEY, user_home=self.soledad_test_folder)
self.mail_sender = self._create_mail_sender()
self.mail_store = SearchableMailStore(LeapMailStore(self.soledad), search_engine)
self.attachment_store = LeapAttachmentStore(self.soledad)
account_ready_cb = defer.Deferred()
self.account = IMAPAccount(self.ACCOUNT, self.soledad, account_ready_cb)
yield account_ready_cb
self.leap_session = mock()
mail_service = self._create_mail_service(self.mail_sender, self.mail_store, search_engine, self.attachment_store)
mails = yield mail_service.all_mails()
search_engine.index_mails(mails)
services = mock()
services.keymanager = mock()
services.mail_service = mail_service
services.draft_service = DraftService(self.mail_store)
services.search_engine = search_engine
services.feedback_service = FeedbackService(self.leap_session)
defer.returnValue(services)
|
