← Back to branch summary

~hadware/magicicada-server/trusty-support

« back to all changes in this revision

Viewing changes to src/server/tests/test_validation.py

  • Committer: Facundo Batista
  • Date: 2015-08-05 13:10:02 UTC
  • Revision ID: facundo@taniquetil.com.ar-20150805131002-he7b7k704d8o7js6
First released version.

Show diffs side-by-side

added added

removed removed

Lines of Context:
src/server/tests/test_validation.py
 
1
# -*- coding: utf-8 -*-
 
2
 
 
3
# Copyright 2008-2015 Canonical
 
4
#
 
5
# This program is free software: you can redistribute it and/or modify
 
6
# it under the terms of the GNU Affero General Public License as
 
7
# published by the Free Software Foundation, either version 3 of the
 
8
# License, or (at your option) any later version.
 
9
#
 
10
# This program is distributed in the hope that it will be useful,
 
11
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
12
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
13
# GNU Affero General Public License for more details.
 
14
#
 
15
# You should have received a copy of the GNU Affero General Public License
 
16
# along with this program. If not, see <http://www.gnu.org/licenses/>.
 
17
#
 
18
# For further info, check  http://launchpad.net/filesync-server
 
19
 
 
20
"""Check the message validation step."""
 
21
 
 
22
import re
 
23
 
 
24
from ubuntuone.storageprotocol import request
 
25
from ubuntuone.storage.server.testing.testcase import TestWithDatabase
 
26
 
 
27
 
 
28
def is_protocol_error(failure):
 
29
    """
 
30
    Returns whether the failure is a PROTOCOL_ERROR
 
31
    """
 
32
    return (failure.check(request.StorageProtocolProtocolError) or
 
33
            failure.getErrorMessage() == 'PROTOCOL_ERROR'
 
34
            or re.search(r'^\s*type: PROTOCOL_ERROR\s*$',
 
35
                         failure.getErrorMessage(),
 
36
                         re.MULTILINE))
 
37
 
 
38
 
 
39
def fail_check(client):
 
40
    """
 
41
    Returns a closure to be used as an errback that checks that the
 
42
    failure is a PROTOCOL_ERROR, and passes the test if so.
 
43
    """
 
44
    def fail_check_cb(f):
 
45
        """
 
46
        Check the failure is a PROTOCOL_ERROR
 
47
        """
 
48
        if is_protocol_error(f):
 
49
            return client.test_done('ok')
 
50
        else:
 
51
            return f
 
52
    return fail_check_cb
 
53
 
 
54
 
 
55
class TestValidation(TestWithDatabase):
 
56
    """Tests for message validation."""
 
57
 
 
58
    def test_make_invalid_share(self):
 
59
        """
 
60
        Test validation for invalid MAKE_DIR works
 
61
        """
 
62
        def _worker(client):
 
63
            """
 
64
            async test
 
65
            """
 
66
            d = client.dummy_authenticate("open sesame")
 
67
            d.addCallbacks(lambda r: client.get_root(), client.test_fail)
 
68
            d.addCallbacks(
 
69
                lambda r: client.make_dir('invalid share', r, "hola"),
 
70
                client.test_fail)
 
71
            d.addCallbacks(client.test_fail, fail_check(client))
 
72
            return d
 
73
        return self.callback_test(_worker)
 
74
 
 
75
    def test_make_invalid_parent(self):
 
76
        """
 
77
        Test validation for invalid MAKE_DIR works
 
78
        """
 
79
        def _worker(client):
 
80
            """
 
81
            async test
 
82
            """
 
83
            d = client.dummy_authenticate("open sesame")
 
84
            d.addCallbacks(lambda r: client.get_root(), client.test_fail)
 
85
            d.addCallbacks(
 
86
                lambda r: client.make_dir('', 'invalid parent', "hola"),
 
87
                client.test_fail)
 
88
            d.addCallbacks(client.test_fail, fail_check(client))
 
89
            return d
 
90
        return self.callback_test(_worker)
 
91
 
 
92
    def test_move_invalid_new_parent_node(self):
 
93
        """
 
94
        Test we fail moves to invalid nodes
 
95
        """
 
96
        def _worker(client):
 
97
            """
 
98
            async test
 
99
            """
 
100
            d = client.dummy_authenticate("open sesame")
 
101
            d.addCallbacks(lambda r: client.get_root(), client.test_fail)
 
102
            d.addCallbacks(
 
103
                lambda r: client.make_file('', r, "hola"),
 
104
                client.test_fail)
 
105
            d.addCallbacks(lambda mk: client.move('', mk.new_id,
 
106
                                                  'invalid new parent node',
 
107
                                                  'chau'),
 
108
                           client.test_fail)
 
109
            d.addCallbacks(client.test_fail, fail_check(client))
 
110
            return d
 
111
        return self.callback_test(_worker)
 
112
 
 
113
    def test_accept_share_invalid_share_id(self):
 
114
        """
 
115
        Test we fail accept_shares with invalid share_ids
 
116
        """
 
117
        def _worker(client):
 
118
            """
 
119
            async test
 
120
            """
 
121
            d = client.dummy_authenticate("open sesame")
 
122
            d.addCallbacks(lambda r: client.accept_share('invalid share id',
 
123
                                                         'Yes'),
 
124
                           client.test_fail)
 
125
            d.addCallbacks(client.test_fail, fail_check(client))
 
126
            return d
 
127
        return self.callback_test(_worker)