← Back to branch summary

~launchpad-pqm/launchpad/devel

« back to all changes in this revision

Viewing changes to lib/canonical/librarian/tests/test_upload.py

  • Committer: Launchpad Patch Queue Manager
  • Date: 2011-12-30 10:09:30 UTC
  • mfrom: (14606.2.8 apocalibrarian)
  • Revision ID: launchpad@pqm.canonical.com-20111230100930-kp3e0l6wakissewm
[r=wgrant][no-qa] Move canonical.librarian to
        lp.services.librarian(server).

Show diffs side-by-side

added added

removed removed

Lines of Context:
lib/canonical/librarian/tests/test_upload.py
1
 
# Copyright 2009 Canonical Ltd.  This software is licensed under the
2
 
# GNU Affero General Public License version 3 (see the file LICENSE).
3
 
 
4
 
__metaclass__ = type
5
 
 
6
 
from canonical.librarian.libraryprotocol import FileUploadProtocol
7
 
from canonical.librarian.storage import WrongDatabaseError
8
 
from lp.testing.systemdocs import LayeredDocFileSuite
9
 
 
10
 
 
11
 
class MockTransport:
12
 
    disconnecting = False
13
 
 
14
 
    bytesWritten = ''
15
 
    connectionLost = False
16
 
 
17
 
    def write(self, bytes):
18
 
        self.bytesWritten += bytes
19
 
 
20
 
    def loseConnection(self):
21
 
        self.connectionLost = True
22
 
        self.disconnecting = True
23
 
 
24
 
 
25
 
class MockLibrary:
26
 
    file = None
27
 
    def startAddFile(self, name, size):
28
 
        self.file = MockFile(name)
29
 
        return self.file
30
 
 
31
 
 
32
 
class MockFile:
33
 
    bytes = ''
34
 
    stored = False
35
 
    databaseName = None
36
 
    debugID = None
37
 
    debugLog = ()
38
 
 
39
 
    def __init__(self, name):
40
 
        self.name = name
41
 
 
42
 
    def append(self, bytes):
43
 
        self.bytes += bytes
44
 
 
45
 
    def store(self):
46
 
        databaseName = self.databaseName
47
 
        if databaseName is not None and databaseName != 'right_database':
48
 
            raise WrongDatabaseError(databaseName, 'right_database')
49
 
        self.stored = True
50
 
        return (987, 654)
51
 
 
52
 
 
53
 
def upload_request(request):
54
 
    """Librarian upload server test helper, process a request and report what
55
 
    happens.
56
 
 
57
 
    Hands a request to a librarian file upload protocol, and prints the reply
58
 
    from the server, a summary of the file uploaded, and whether the connection
59
 
    closed, e.g.::
60
 
 
61
 
        reply: '200'
62
 
        file u'foo.txt' stored as text/plain, contents: 'Foo!'
63
 
 
64
 
    or::
65
 
 
66
 
        reply: '400 STORE command expects the filename to be in UTF-8'
67
 
        connection closed
68
 
 
69
 
    Note that the Librarian itself except for the protocol logic is stubbed out
70
 
    by this function; it's intended to be used to unit test the protocol
71
 
    implementation, not end-to-end test the Librarian.
72
 
    """
73
 
    # Send tracebacks from Twisted to stderr, if they occur, to make debugging
74
 
    # test failures easier.
75
 
    import sys
76
 
    def log_observer(x):
77
 
        print >> sys.stderr, x
78
 
        if 'failure' in x:
79
 
            x['failure'].printTraceback(file=sys.stderr)
80
 
    from twisted.python import log
81
 
    log.addObserver(log_observer)
82
 
 
83
 
    # Create a FileUploadProtocol, and instrument it for testing:
84
 
    server = FileUploadProtocol()
85
 
 
86
 
    #  * hook _storeFile to dispatch straight to newFile.store without
87
 
    #    spawning a thread.
88
 
    from twisted.internet import defer
89
 
    server._storeFile = lambda: defer.maybeDeferred(server.newFile.store)
90
 
 
91
 
    #  * give it a fake transport
92
 
    server.transport = MockTransport()
93
 
    server.connectionMade()
94
 
 
95
 
    #  * give it a fake factory (itself!), and a fake library.
96
 
    server.factory = server
97
 
    server.fileLibrary = MockLibrary()
98
 
 
99
 
    # Feed in the request
100
 
    server.dataReceived(request.replace('\n', '\r\n'))
101
 
 
102
 
    # Report on what happened
103
 
    print "reply: %r" % server.transport.bytesWritten.rstrip('\r\n')
104
 
 
105
 
    if server.transport.connectionLost:
106
 
        print 'connection closed'
107
 
 
108
 
    mockFile = server.fileLibrary.file
109
 
    if mockFile is not None and mockFile.stored:
110
 
        print "file %r stored as %s, contents: %r" % (
111
 
                mockFile.name, mockFile.mimetype, mockFile.bytes)
112
 
 
113
 
    # Cleanup: remove the observer.
114
 
    log.removeObserver(log_observer)
115
 
 
116
 
 
117
 
def test_suite():
118
 
    return LayeredDocFileSuite(
119
 
        'test_upload.txt', globs={'upload_request': upload_request},
120
 
        stdout_logging=False)
121