← Back to branch summary

~lool/ubuntuone-storage-protocol/unix-line-endings-for-certs

« back to all changes in this revision

Viewing changes to tests/test_putcontent.py

  • Committer: Tarmac
  • Author(s): Guillermo Gonzalez
  • Date: 2011-02-10 19:45:50 UTC
  • mfrom: (123.2.4 upload-id)
  • Revision ID: tarmac-20110210194550-835792nh0r4nlyzu
Add upload_id support to PutContent (BEGIN_CONTENT and PUT_CONTENT messages)

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/test_putcontent.py
1
1
# ubuntuone.storageprotocol.tests.test_upload_offset -
2
 
#     tests for the offset argument of BeginContent messages of upload requests
 
2
#     tests for the PutContent request
3
3
#
4
4
# Author: John R. Lenton <john.lenton@canonical.com>
 
5
#         Guillermo Gonzalez <guillermo.gonzalez@canonical.com>
5
6
#
6
 
# Copyright (C) 2009 Canonical
 
7
# Copyright (C) 2009-2011 Canonical
7
8
#
8
9
# This program is free software: you can redistribute it and/or modify it
9
10
# under the terms of the GNU Affero General Public License version 3,
16
17
#
17
18
# You should have received a copy of the GNU Affero General Public License
18
19
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
19
 
"""Tests for begin_content.offset"""
 
20
"""Tests for PutContent request"""
20
21
 
21
22
import unittest
 
23
 
 
24
from StringIO import StringIO
 
25
 
 
26
from twisted.test.proto_helpers import StringTransport
 
27
 
22
28
from ubuntuone.storageprotocol.request import RequestHandler
23
29
from ubuntuone.storageprotocol.client import PutContent
24
30
from ubuntuone.storageprotocol import protocol_pb2
78
84
        message.type = protocol_pb2.Message.BEGIN_CONTENT
79
85
        pc.start()
80
86
        pc.processMessage(message)
 
87
 
 
88
 
 
89
class TestUploadId(unittest.TestCase):
 
90
    """Tests for BEGIN_CONTENT and PUT_CONTENT upload_id attribute."""
 
91
 
 
92
    def setUp(self):
 
93
        unittest.TestCase.setUp(self)
 
94
        transport = StringTransport()
 
95
        self.protocol = RequestHandler()
 
96
        self.protocol.transport = transport
 
97
 
 
98
    def test_server_upload_id(self):
 
99
        """Test that, if the server specify an upload_id, we save it."""
 
100
        upload_id = "foo"
 
101
        called = []
 
102
        pc = PutContent(self.protocol, 'share', 'node', '', '',
 
103
                        0, 0, 0, StringIO(''), upload_id_cb=called.append)
 
104
        message = protocol_pb2.Message()
 
105
        message.type = protocol_pb2.Message.BEGIN_CONTENT
 
106
        message.begin_content.upload_id = upload_id
 
107
        pc.start()
 
108
        pc.processMessage(message)
 
109
        self.assertEqual(len(called), 1)
 
110
        self.assertEqual(called[0], 'foo')
 
111
 
 
112
    def test_server_upload_id_no_cb(self):
 
113
        """Test that, if the server specify an upload_id, we save it.
 
114
 
 
115
        Only if we have the upload_id_cb defined.
 
116
        """
 
117
        offset = 23
 
118
        # just to check something
 
119
        mocker = Mocker()
 
120
        fd = mocker.mock()
 
121
        fd.seek(offset)  # this is really all I care about
 
122
        fd.read(ANY)
 
123
        mocker.result('')
 
124
        mocker.replay()
 
125
 
 
126
        upload_id = "foo"
 
127
        pc = PutContent(self.protocol, 'share', 'node', '', '',
 
128
                        0, 0, 0, fd)
 
129
        message = protocol_pb2.Message()
 
130
        message.type = protocol_pb2.Message.BEGIN_CONTENT
 
131
        message.begin_content.upload_id = upload_id
 
132
        message.begin_content.offset = offset
 
133
        pc.start()
 
134
        pc.processMessage(message)
 
135
 
 
136
    def test_server_upload_id_none(self):
 
137
        """Test that if there is no upload_id we ignore it."""
 
138
        called = []
 
139
        pc = PutContent(self.protocol, 'share', 'node', '', '',
 
140
                        0, 0, 0, StringIO(''), upload_id_cb=called.append)
 
141
        message = protocol_pb2.Message()
 
142
        message.type = protocol_pb2.Message.BEGIN_CONTENT
 
143
        pc.start()
 
144
        pc.processMessage(message)
 
145
        self.assertEqual(len(called), 0)
 
146
 
 
147
    def test_client_upload_id(self):
 
148
        """Test that, we send the upload id to the server."""
 
149
        pc = PutContent(self.protocol, 'share', 'node', '', '',
 
150
                        0, 0, 0, StringIO(''), upload_id='foo')
 
151
        pc.start()
 
152
        pc_msg = protocol_pb2.Message()
 
153
        pc_msg.ParseFromString(self.protocol.transport.value())
 
154
        self.assertEqual(pc_msg.put_content.upload_id, 'foo')
 
155