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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
|
# Copyright 2015-2016 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
__metaclass__ = type
import os
import shutil
import tempfile
from testtools import TestCase
from lpbuildd.snap import (
SnapBuildManager,
SnapBuildState,
)
from lpbuildd.tests.fakeslave import FakeSlave
class MockBuildManager(SnapBuildManager):
def __init__(self, *args, **kwargs):
super(MockBuildManager, self).__init__(*args, **kwargs)
self.commands = []
self.iterators = []
def runSubProcess(self, path, command, iterate=None, env=None):
self.commands.append([path] + command)
if iterate is None:
iterate = self.iterate
self.iterators.append(iterate)
return 0
class TestSnapBuildManagerIteration(TestCase):
"""Run SnapBuildManager through its iteration steps."""
def setUp(self):
super(TestSnapBuildManagerIteration, self).setUp()
self.working_dir = tempfile.mkdtemp()
self.addCleanup(lambda: shutil.rmtree(self.working_dir))
slave_dir = os.path.join(self.working_dir, "slave")
home_dir = os.path.join(self.working_dir, "home")
for dir in (slave_dir, home_dir):
os.mkdir(dir)
self.slave = FakeSlave(slave_dir)
self.buildid = "123"
self.buildmanager = MockBuildManager(self.slave, self.buildid)
self.buildmanager.home = home_dir
self.buildmanager._cachepath = self.slave._cachepath
self.build_dir = os.path.join(
home_dir, "build-%s" % self.buildid, "chroot-autobuild", "build")
def getState(self):
"""Retrieve build manager's state."""
return self.buildmanager._state
def startBuild(self):
# The build manager's iterate() kicks off the consecutive states
# after INIT.
extra_args = {
"arch_tag": "i386",
"name": "test-snap",
"git_repository": "https://git.launchpad.dev/~example/+git/snap",
"git_path": "master",
}
self.buildmanager.initiate({}, "chroot.tar.gz", extra_args)
# Skip states that are done in DebianBuildManager to the state
# directly before BUILD_SNAP.
self.buildmanager._state = SnapBuildState.UPDATE
# BUILD_SNAP: Run the slave's payload to build the snap package.
self.buildmanager.iterate(0)
self.assertEqual(SnapBuildState.BUILD_SNAP, self.getState())
expected_command = [
"sharepath/slavebin/buildsnap", "buildsnap",
"--build-id", self.buildid, "--arch", "i386",
"--git-repository", "https://git.launchpad.dev/~example/+git/snap",
"--git-path", "master",
"test-snap",
]
self.assertEqual(expected_command, self.buildmanager.commands[-1])
self.assertEqual(
self.buildmanager.iterate, self.buildmanager.iterators[-1])
self.assertFalse(self.slave.wasCalled("chrootFail"))
def test_status(self):
# The build manager returns saved status information on request.
self.assertEqual({}, self.buildmanager.status())
status_path = os.path.join(
self.working_dir, "home", "build-%s" % self.buildid, "status")
os.makedirs(os.path.dirname(status_path))
with open(status_path, "w") as status_file:
status_file.write('{"revision_id": "dummy"}')
self.assertEqual({"revision_id": "dummy"}, self.buildmanager.status())
def test_iterate(self):
# The build manager iterates a normal build from start to finish.
self.startBuild()
log_path = os.path.join(self.buildmanager._cachepath, "buildlog")
log = open(log_path, "w")
log.write("I am a build log.")
log.close()
output_dir = os.path.join(self.build_dir, "test-snap")
os.makedirs(output_dir)
snap_path = os.path.join(output_dir, "test-snap_0_all.snap")
with open(snap_path, "w") as snap:
snap.write("I am a snap package.")
# After building the package, reap processes.
self.buildmanager.iterate(0)
expected_command = [
"sharepath/slavebin/scan-for-processes", "scan-for-processes",
self.buildid,
]
self.assertEqual(SnapBuildState.BUILD_SNAP, self.getState())
self.assertEqual(expected_command, self.buildmanager.commands[-1])
self.assertNotEqual(
self.buildmanager.iterate, self.buildmanager.iterators[-1])
self.assertFalse(self.slave.wasCalled("buildFail"))
self.assertEqual([((snap_path,), {})], self.slave.addWaitingFile.calls)
# Control returns to the DebianBuildManager in the UMOUNT state.
self.buildmanager.iterateReap(self.getState(), 0)
expected_command = [
"sharepath/slavebin/umount-chroot", "umount-chroot", self.buildid]
self.assertEqual(SnapBuildState.UMOUNT, self.getState())
self.assertEqual(expected_command, self.buildmanager.commands[-1])
self.assertEqual(
self.buildmanager.iterate, self.buildmanager.iterators[-1])
self.assertFalse(self.slave.wasCalled("buildFail"))
def test_iterate_with_manifest(self):
# The build manager iterates a build that uploads a manifest from
# start to finish.
self.startBuild()
log_path = os.path.join(self.buildmanager._cachepath, "buildlog")
log = open(log_path, "w")
log.write("I am a build log.")
log.close()
output_dir = os.path.join(self.build_dir, "test-snap")
os.makedirs(output_dir)
snap_path = os.path.join(output_dir, "test-snap_0_all.snap")
with open(snap_path, "w") as snap:
snap.write("I am a snap package.")
manifest_path = os.path.join(output_dir, "test-snap_0_all.manifest")
with open(manifest_path, "w") as manifest:
manifest.write("I am a manifest.")
# After building the package, reap processes.
self.buildmanager.iterate(0)
expected_command = [
"sharepath/slavebin/scan-for-processes", "scan-for-processes",
self.buildid,
]
self.assertEqual(SnapBuildState.BUILD_SNAP, self.getState())
self.assertEqual(expected_command, self.buildmanager.commands[-1])
self.assertNotEqual(
self.buildmanager.iterate, self.buildmanager.iterators[-1])
self.assertFalse(self.slave.wasCalled("buildFail"))
self.assertEqual(
[((manifest_path,), {}), ((snap_path,), {})],
self.slave.addWaitingFile.calls)
# Control returns to the DebianBuildManager in the UMOUNT state.
self.buildmanager.iterateReap(self.getState(), 0)
expected_command = [
"sharepath/slavebin/umount-chroot", "umount-chroot", self.buildid]
self.assertEqual(SnapBuildState.UMOUNT, self.getState())
self.assertEqual(expected_command, self.buildmanager.commands[-1])
self.assertEqual(
self.buildmanager.iterate, self.buildmanager.iterators[-1])
self.assertFalse(self.slave.wasCalled("buildFail"))
|