1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
|
# Copyright 2011 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Tests for Twisted Poppy FTP."""
__metaclass__ = type
import os
from testtools.deferredruntest import AsynchronousDeferredRunTest
import transaction
from twisted.protocols import ftp
from zope.component import getUtility
from lp.poppy.twistedftp import PoppyFileWriter
from lp.registry.interfaces.gpg import (
GPGKeyAlgorithm,
IGPGKeySet,
)
from lp.services.config import config
from lp.services.database.isolation import check_no_transaction
from lp.services.gpg.interfaces import IGPGHandler
from lp.testing import TestCaseWithFactory
from lp.testing.gpgkeys import gpgkeysdir
from lp.testing.keyserver import KeyServerTac
from lp.testing.layers import ZopelessDatabaseLayer
class TestPoppyFileWriter(TestCaseWithFactory):
layer = ZopelessDatabaseLayer
run_tests_with = AsynchronousDeferredRunTest.make_factory(timeout=20)
def setUp(self):
TestCaseWithFactory.setUp(self)
# Start the test keyserver. Starting up and tearing this down
# for each test is expensive, but we don't have a way of having
# cross-test persistent fixtures yet. See bug 724349.
self.tac = KeyServerTac()
self.tac.setUp()
self.addCleanup(self.tac.tearDown)
# Load a key.
gpg_handler = getUtility(IGPGHandler)
key_path = os.path.join(gpgkeysdir, 'ftpmaster@canonical.com.pub')
key_data = open(key_path).read()
key = gpg_handler.importPublicKey(key_data)
assert key is not None
# Make a new user and add the above key to it.
user = self.factory.makePerson()
key_set = getUtility(IGPGKeySet)
user_key = key_set.new(
ownerID=user.id, keyid=key.keyid, fingerprint=key.fingerprint,
algorithm=GPGKeyAlgorithm.items[key.algorithm],
keysize=key.keysize, can_encrypt=key.can_encrypt,
active=True)
# validateGPG runs in its own transaction.
transaction.commit()
# Locate the directory with test files.
self.test_files_dir = os.path.join(
config.root, "lib/lp/soyuz/scripts/tests/upload_test_files/")
def test_changes_file_with_valid_GPG(self):
valid_changes_file = os.path.join(
self.test_files_dir, "etherwake_1.08-1_source.changes")
def callback(result):
self.assertIs(None, result)
with open(valid_changes_file) as opened_file:
file_writer = PoppyFileWriter(opened_file)
d = file_writer.close()
d.addBoth(callback)
return d
def test_changes_file_with_invalid_GPG(self):
invalid_changes_file = os.path.join(
self.test_files_dir, "broken_source.changes")
def error_callback(failure):
self.assertTrue(failure.check, ftp.PermissionDeniedError)
self.assertIn(
"Changes file must be signed with a valid GPG signature",
failure.getErrorMessage())
def success_callback(result):
self.fail("Success when there should have been failure.")
with open(invalid_changes_file) as opened_file:
file_writer = PoppyFileWriter(opened_file)
d = file_writer.close()
d.addCallbacks(success_callback, error_callback)
return d
def test_aborts_transaction(self):
valid_changes_file = os.path.join(
self.test_files_dir, "etherwake_1.08-1_source.changes")
def callback(result):
check_no_transaction()
with open(valid_changes_file) as opened_file:
file_writer = PoppyFileWriter(opened_file)
d = file_writer.close()
d.addBoth(callback)
return d
|