diff options
author | drebs <drebs@riseup.net> | 2017-09-17 12:08:25 -0300 |
---|---|---|
committer | drebs <drebs@riseup.net> | 2017-09-17 15:50:55 -0300 |
commit | cfff46ff9becdbe5cf48816870e625ed253ecc57 (patch) | |
tree | 8d239e4499f559d86ed17ea3632008303b25d485 /tests/server/test_session.py | |
parent | f29abe28bd778838626d12fcabe3980a8ce4fa8c (diff) |
[refactor] move tests to root of repository
Tests entrypoint was in a testing/ subfolder in the root of the
repository. This was made mainly because we had some common files for
tests and we didn't want to ship them (files in testing/test_soledad,
which is itself a python package. This sometimes causes errors when
loading tests (it seems setuptools is confused with having one python
package in a subdirectory of another).
This commit moves the tests entrypoint to the root of the repository.
Closes: #8952
Diffstat (limited to 'tests/server/test_session.py')
-rw-r--r-- | tests/server/test_session.py | 195 |
1 files changed, 195 insertions, 0 deletions
diff --git a/tests/server/test_session.py b/tests/server/test_session.py new file mode 100644 index 00000000..3dbd2740 --- /dev/null +++ b/tests/server/test_session.py @@ -0,0 +1,195 @@ +# -*- coding: utf-8 -*- +# test_session.py +# Copyright (C) 2017 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/>. +""" +Tests for server session entrypoint. +""" +from twisted.trial import unittest + +from twisted.cred import portal +from twisted.cred.checkers import InMemoryUsernamePasswordDatabaseDontUse +from twisted.cred.credentials import IUsernamePassword +from twisted.web.resource import getChildForRequest +from twisted.web.static import Data +from twisted.web.test.requesthelper import DummyRequest +from twisted.web.test.test_httpauth import b64encode +from twisted.web.test.test_httpauth import Realm +from twisted.web._auth.wrapper import UnauthorizedResource + +from leap.soledad.server.session import SoledadSession + + +class SoledadSessionTestCase(unittest.TestCase): + """ + Tests adapted from for + L{twisted.web.test.test_httpauth.HTTPAuthSessionWrapper}. + """ + + def makeRequest(self, *args, **kwargs): + request = DummyRequest(*args, **kwargs) + request.path = '/' + return request + + def setUp(self): + self.username = b'foo bar' + self.password = b'bar baz' + self.avatarContent = b"contents of the avatar resource itself" + self.childName = b"foo-child" + self.childContent = b"contents of the foo child of the avatar" + self.checker = InMemoryUsernamePasswordDatabaseDontUse() + self.checker.addUser(self.username, self.password) + self.avatar = Data(self.avatarContent, 'text/plain') + self.avatar.putChild( + self.childName, Data(self.childContent, 'text/plain')) + self.avatars = {self.username: self.avatar} + self.realm = Realm(self.avatars.get) + self.portal = portal.Portal(self.realm, [self.checker]) + self.wrapper = SoledadSession(self.portal) + + def _authorizedTokenLogin(self, request): + authorization = b64encode( + self.username + b':' + self.password) + request.requestHeaders.addRawHeader(b'authorization', + b'Token ' + authorization) + return getChildForRequest(self.wrapper, request) + + def test_getChildWithDefault(self): + request = self.makeRequest([self.childName]) + child = getChildForRequest(self.wrapper, request) + d = request.notifyFinish() + + def cbFinished(result): + self.assertEqual(request.responseCode, 401) + + d.addCallback(cbFinished) + request.render(child) + return d + + def _invalidAuthorizationTest(self, response): + request = self.makeRequest([self.childName]) + request.requestHeaders.addRawHeader(b'authorization', response) + child = getChildForRequest(self.wrapper, request) + d = request.notifyFinish() + + def cbFinished(result): + self.assertEqual(request.responseCode, 401) + + d.addCallback(cbFinished) + request.render(child) + return d + + def test_getChildWithDefaultUnauthorizedUser(self): + return self._invalidAuthorizationTest( + b'Basic ' + b64encode(b'foo:bar')) + + def test_getChildWithDefaultUnauthorizedPassword(self): + return self._invalidAuthorizationTest( + b'Basic ' + b64encode(self.username + b':bar')) + + def test_getChildWithDefaultUnrecognizedScheme(self): + return self._invalidAuthorizationTest(b'Quux foo bar baz') + + def test_getChildWithDefaultAuthorized(self): + request = self.makeRequest([self.childName]) + child = self._authorizedTokenLogin(request) + d = request.notifyFinish() + + def cbFinished(ignored): + self.assertEqual(request.written, [self.childContent]) + + d.addCallback(cbFinished) + request.render(child) + return d + + def test_renderAuthorized(self): + # Request it exactly, not any of its children. + request = self.makeRequest([]) + child = self._authorizedTokenLogin(request) + d = request.notifyFinish() + + def cbFinished(ignored): + self.assertEqual(request.written, [self.avatarContent]) + + d.addCallback(cbFinished) + request.render(child) + return d + + def test_decodeRaises(self): + request = self.makeRequest([self.childName]) + request.requestHeaders.addRawHeader(b'authorization', + b'Token decode should fail') + child = getChildForRequest(self.wrapper, request) + self.assertIsInstance(child, UnauthorizedResource) + + def test_parseResponse(self): + basicAuthorization = b'Basic abcdef123456' + self.assertEqual( + self.wrapper._parseHeader(basicAuthorization), + None) + tokenAuthorization = b'Token abcdef123456' + self.assertEqual( + self.wrapper._parseHeader(tokenAuthorization), + b'abcdef123456') + + def test_unexpectedDecodeError(self): + + class UnexpectedException(Exception): + pass + + class BadFactory(object): + scheme = b'bad' + + def getChallenge(self, client): + return {} + + def decode(self, response, request): + print("decode raised") + raise UnexpectedException() + + self.wrapper._credentialFactory = BadFactory() + request = self.makeRequest([self.childName]) + request.requestHeaders.addRawHeader(b'authorization', b'Bad abc') + child = getChildForRequest(self.wrapper, request) + request.render(child) + self.assertEqual(request.responseCode, 500) + errors = self.flushLoggedErrors(UnexpectedException) + self.assertEqual(len(errors), 1) + + def test_unexpectedLoginError(self): + class UnexpectedException(Exception): + pass + + class BrokenChecker(object): + credentialInterfaces = (IUsernamePassword,) + + def requestAvatarId(self, credentials): + raise UnexpectedException() + + self.portal.registerChecker(BrokenChecker()) + request = self.makeRequest([self.childName]) + child = self._authorizedTokenLogin(request) + request.render(child) + self.assertEqual(request.responseCode, 500) + self.assertEqual(len(self.flushLoggedErrors(UnexpectedException)), 1) + + def test_cantAccessOtherUserPathByDefault(self): + request = self.makeRequest([]) + # valid url_mapper path, but for another user + request.path = '/blobs/another-user/' + child = self._authorizedTokenLogin(request) + + request.render(child) + self.assertEqual(request.responseCode, 500) |