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
176
177
178
179
180
181
182
183
184
185
186
|
"""
This file demonstrates writing tests using the unittest module. These will pass
when you run "manage.py test".
Replace this with more appropriate tests for your application.
"""
from django.test import TestCase
from bldcontrol.bbcontroller import BitbakeController, BuildSetupException
from bldcontrol.localhostbecontroller import LocalhostBEController
from bldcontrol.sshbecontroller import SSHBEController
from bldcontrol.models import BuildEnvironment, BuildRequest
from bldcontrol.management.commands.runbuilds import Command
import socket
import subprocess
import os
# standard poky data hardcoded for testing
BITBAKE_LAYERS = [type('bitbake_info', (object,), { "giturl": "git://git.yoctoproject.org/poky.git", "dirpath": "", "commit": "HEAD"})]
POKY_LAYERS = [
type('poky_info', (object,), { "name": "meta", "giturl": "git://git.yoctoproject.org/poky.git", "dirpath": "meta", "commit": "HEAD"}),
type('poky_info', (object,), { "name": "meta-yocto", "giturl": "git://git.yoctoproject.org/poky.git", "dirpath": "meta-yocto", "commit": "HEAD"}),
type('poky_info', (object,), { "name": "meta-yocto-bsp", "giturl": "git://git.yoctoproject.org/poky.git", "dirpath": "meta-yocto-bsp", "commit": "HEAD"}),
]
# we have an abstract test class designed to ensure that the controllers use a single interface
# specific controller tests only need to override the _getBuildEnvironment() method
test_sourcedir = os.getenv("TTS_SOURCE_DIR")
test_builddir = os.getenv("TTS_BUILD_DIR")
test_address = os.getenv("TTS_TEST_ADDRESS", "localhost")
if test_sourcedir == None or test_builddir == None or test_address == None:
raise Exception("Please set TTTS_SOURCE_DIR, TTS_BUILD_DIR and TTS_TEST_ADDRESS")
# The bb server will expect a toaster-pre.conf file to exist. If it doesn't exit then we make
# an empty one here.
open(test_builddir + 'conf/toaster-pre.conf', 'a').close()
class BEControllerTests(object):
def _serverForceStop(self, bc):
err = bc._shellcmd("netstat -tapn 2>/dev/null | grep 8200 | awk '{print $7}' | sort -fu | cut -d \"/\" -f 1 | grep -v -- - | tee /dev/fd/2 | xargs -r kill")
self.assertTrue(err == '', "bitbake server pid %s not stopped" % err)
def test_serverStartAndStop(self):
from bldcontrol.sshbecontroller import NotImplementedException
obe = self._getBuildEnvironment()
bc = self._getBEController(obe)
try:
# setting layers, skip any layer info
bc.setLayers(BITBAKE_LAYERS, POKY_LAYERS)
except NotImplementedException, e:
print "Test skipped due to command not implemented yet"
return True
# We are ok with the exception as we're handling the git already exists
except BuildSetupException:
pass
bc.pokydirname = test_sourcedir
bc.islayerset = True
hostname = test_address.split("@")[-1]
# test start server and stop
bc.startBBServer()
self.assertFalse(socket.socket(socket.AF_INET, socket.SOCK_STREAM).connect_ex((hostname, int(bc.be.bbport))), "Server not answering")
self._serverForceStop(bc)
def test_getBBController(self):
from bldcontrol.sshbecontroller import NotImplementedException
obe = self._getBuildEnvironment()
bc = self._getBEController(obe)
layerSet = False
try:
# setting layers, skip any layer info
layerSet = bc.setLayers(BITBAKE_LAYERS, POKY_LAYERS)
except NotImplementedException:
print "Test skipped due to command not implemented yet"
return True
# We are ok with the exception as we're handling the git already exists
except BuildSetupException:
pass
bc.pokydirname = test_sourcedir
bc.islayerset = True
bbc = bc.getBBController()
self.assertTrue(isinstance(bbc, BitbakeController))
self._serverForceStop(bc)
class LocalhostBEControllerTests(TestCase, BEControllerTests):
def __init__(self, *args):
super(LocalhostBEControllerTests, self).__init__(*args)
def _getBuildEnvironment(self):
return BuildEnvironment.objects.create(
lock = BuildEnvironment.LOCK_FREE,
betype = BuildEnvironment.TYPE_LOCAL,
address = test_address,
sourcedir = test_sourcedir,
builddir = test_builddir )
def _getBEController(self, obe):
return LocalhostBEController(obe)
class SSHBEControllerTests(TestCase, BEControllerTests):
def __init__(self, *args):
super(SSHBEControllerTests, self).__init__(*args)
def _getBuildEnvironment(self):
return BuildEnvironment.objects.create(
lock = BuildEnvironment.LOCK_FREE,
betype = BuildEnvironment.TYPE_SSH,
address = test_address,
sourcedir = test_sourcedir,
builddir = test_builddir )
def _getBEController(self, obe):
return SSHBEController(obe)
def test_pathExists(self):
obe = BuildEnvironment.objects.create(betype = BuildEnvironment.TYPE_SSH, address= test_address)
sbc = SSHBEController(obe)
self.assertTrue(sbc._pathexists("/"))
self.assertFalse(sbc._pathexists("/.deadbeef"))
self.assertTrue(sbc._pathexists(sbc._shellcmd("pwd")))
class RunBuildsCommandTests(TestCase):
def test_bec_select(self):
"""
Tests that we can find and lock a build environment
"""
obe = BuildEnvironment.objects.create(lock = BuildEnvironment.LOCK_FREE, betype = BuildEnvironment.TYPE_LOCAL)
command = Command()
bec = command._selectBuildEnvironment()
# make sure we select the object we've just built
self.assertTrue(bec.be.id == obe.id, "Environment is not properly selected")
# we have a locked environment
self.assertTrue(bec.be.lock == BuildEnvironment.LOCK_LOCK, "Environment is not locked")
# no more selections possible here
self.assertRaises(IndexError, command._selectBuildEnvironment)
def test_br_select(self):
from orm.models import Project, Release, BitbakeVersion, Branch
p = Project.objects.create_project("test", Release.objects.get_or_create(name = "HEAD", bitbake_version = BitbakeVersion.objects.get_or_create(name="HEAD", branch=Branch.objects.get_or_create(name="HEAD"))[0])[0])
obr = BuildRequest.objects.create(state = BuildRequest.REQ_QUEUED, project = p)
command = Command()
br = command._selectBuildRequest()
# make sure we select the object we've just built
self.assertTrue(obr.id == br.id, "Request is not properly selected")
# we have a locked environment
self.assertTrue(br.state == BuildRequest.REQ_INPROGRESS, "Request is not updated")
# no more selections possible here
self.assertRaises(IndexError, command._selectBuildRequest)
class UtilityTests(TestCase):
def test_reduce_path(self):
from bldcontrol.management.commands.loadconf import _reduce_canon_path, _get_id_for_sourcetype
self.assertTrue( _reduce_canon_path("/") == "/")
self.assertTrue( _reduce_canon_path("/home/..") == "/")
self.assertTrue( _reduce_canon_path("/home/../ana") == "/ana")
self.assertTrue( _reduce_canon_path("/home/../ana/..") == "/")
self.assertTrue( _reduce_canon_path("/home/ana/mihai/../maria") == "/home/ana/maria")
def test_get_id_for_sorucetype(self):
from bldcontrol.management.commands.loadconf import _reduce_canon_path, _get_id_for_sourcetype
self.assertTrue( _get_id_for_sourcetype("layerindex") == 1)
self.assertTrue( _get_id_for_sourcetype("local") == 0)
self.assertTrue( _get_id_for_sourcetype("imported") == 2)
with self.assertRaises(Exception):
_get_id_for_sourcetype("unknown")
|
