← Back to branch summary

~launchpad-pqm/launchpad/devel

« back to all changes in this revision

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