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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
|
# Copyright 2009-2010 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Tests for deathrow class."""
__metaclass__ = type
import os
import shutil
import tempfile
from zope.component import getUtility
from lp.archivepublisher.deathrow import DeathRow
from lp.archivepublisher.diskpool import DiskPool
from lp.registry.interfaces.distribution import IDistributionSet
from lp.services.log.logger import BufferLogger
from lp.soyuz.interfaces.component import IComponentSet
from lp.soyuz.tests.test_publishing import SoyuzTestPublisher
from lp.testing import TestCase
from lp.testing.layers import LaunchpadZopelessLayer
class TestDeathRow(TestCase):
layer = LaunchpadZopelessLayer
def getTestPublisher(self, distroseries):
"""Return an `SoyuzTestPublisher`instance."""
stp = SoyuzTestPublisher()
stp.addFakeChroots(distroseries)
stp.setUpDefaultDistroSeries(distroseries)
return stp
def getDeathRow(self, archive):
"""Return an `DeathRow` for the given archive.
Created the temporary 'pool' and 'temp' directories and register
a 'cleanup' to purge them after the test runs.
"""
pool_path = tempfile.mkdtemp('-pool')
temp_path = tempfile.mkdtemp('-pool-tmp')
def clean_pool(pool_path, temp_path):
shutil.rmtree(pool_path)
shutil.rmtree(temp_path)
self.addCleanup(clean_pool, pool_path, temp_path)
logger = BufferLogger()
diskpool = DiskPool(pool_path, temp_path, logger)
return DeathRow(archive, diskpool, logger)
def getDiskPoolPath(self, pub_file, diskpool):
"""Return the absolute path to a published file in the disk pool/."""
return diskpool.pathFor(
pub_file.componentname.encode('utf-8'),
pub_file.sourcepackagename.encode('utf8'),
pub_file.libraryfilealiasfilename.encode('utf-8'))
def assertIsFile(self, path):
"""Assert the path exists and is a regular file."""
self.assertTrue(
os.path.exists(path),
"File %s does not exist" % os.path.basename(path))
self.assertFalse(
os.path.islink(path),
"File %s is a symbolic link" % os.path.basename(path))
def assertIsLink(self, path):
"""Assert the path exists and is a symbolic link."""
self.assertTrue(
os.path.exists(path),
"File %s does not exist" % os.path.basename(path))
self.assertTrue(
os.path.islink(path),
"File %s is a not symbolic link" % os.path.basename(path))
def assertDoesNotExist(self, path):
"""Assert the path does not exit."""
self.assertFalse(
os.path.exists(path),
"File %s exists" % os.path.basename(path))
def test_MissingSymLinkInPool(self):
# When a publication is promoted from 'universe' to 'main' and
# the symbolic links expected in 'universe' are not present,
# a `MissingSymlinkInPool` error is generated and immediately
# ignored by the `DeathRow` processor. Even in this adverse
# circumstances the database record (removal candidate) is
# updated to match the disk status.
# Setup an `SoyuzTestPublisher` and a `DeathRow` instance.
ubuntu = getUtility(IDistributionSet).getByName('ubuntu')
hoary = ubuntu.getSeries('hoary')
stp = self.getTestPublisher(hoary)
deathrow = self.getDeathRow(hoary.main_archive)
# Create a source publication with a since file (DSC) in
# 'universe' and promote it to 'main'.
source_universe = stp.getPubSource(component='universe')
source_main = source_universe.changeOverride(
new_component=getUtility(IComponentSet)['main'])
test_publications = (source_universe, source_main)
# Commit for exposing the just-created librarian files.
self.layer.commit()
# Publish the testing publication on disk, the file for the
# 'universe' component will be a symbolic link to the one
# in 'main'.
for pub in test_publications:
pub.publish(deathrow.diskpool, deathrow.logger)
[main_dsc_path] = [
self.getDiskPoolPath(pub_file, deathrow.diskpool)
for pub_file in source_main.files]
[universe_dsc_path] = [
self.getDiskPoolPath(pub_file, deathrow.diskpool)
for pub_file in source_universe.files]
self.assertIsFile(main_dsc_path)
self.assertIsLink(universe_dsc_path)
# Remove the symbolic link to emulate MissingSymlinkInPool scenario.
os.remove(universe_dsc_path)
# Remove the testing publications.
for pub in test_publications:
pub.requestObsolescence()
# Commit for exposing the just-created removal candidates.
self.layer.commit()
# Due to the MissingSymlinkInPool scenario, it takes 2 iteration to
# remove both references to the shared file in pool/.
deathrow.reap()
deathrow.reap()
for pub in test_publications:
self.assertTrue(
pub.dateremoved is not None,
'%s (%s) is not marked as removed.'
% (pub.displayname, pub.component.name))
self.assertDoesNotExist(main_dsc_path)
self.assertDoesNotExist(universe_dsc_path)
|