summaryrefslogtreecommitdiffstats
path: root/meta/lib/oeqa
diff options
context:
space:
mode:
Diffstat (limited to 'meta/lib/oeqa')
-rw-r--r--meta/lib/oeqa/selftest/__init__.py2
-rw-r--r--meta/lib/oeqa/selftest/base.py98
-rw-r--r--meta/lib/oeqa/utils/commands.py137
-rw-r--r--meta/lib/oeqa/utils/ftools.py27
4 files changed, 264 insertions, 0 deletions
diff --git a/meta/lib/oeqa/selftest/__init__.py b/meta/lib/oeqa/selftest/__init__.py
new file mode 100644
index 0000000000..3ad9513f40
--- /dev/null
+++ b/meta/lib/oeqa/selftest/__init__.py
@@ -0,0 +1,2 @@
1from pkgutil import extend_path
2__path__ = extend_path(__path__, __name__)
diff --git a/meta/lib/oeqa/selftest/base.py b/meta/lib/oeqa/selftest/base.py
new file mode 100644
index 0000000000..30a71e886f
--- /dev/null
+++ b/meta/lib/oeqa/selftest/base.py
@@ -0,0 +1,98 @@
1# Copyright (c) 2013 Intel Corporation
2#
3# Released under the MIT license (see COPYING.MIT)
4
5
6# DESCRIPTION
7# Base class inherited by test classes in meta/lib/selftest
8
9import unittest
10import os
11import sys
12import logging
13import errno
14
15import oeqa.utils.ftools as ftools
16
17
18class oeSelfTest(unittest.TestCase):
19
20 log = logging.getLogger("selftest.base")
21 longMessage = True
22
23 def __init__(self, methodName="runTest"):
24 self.builddir = os.environ.get("BUILDDIR")
25 self.localconf_path = os.path.join(self.builddir, "conf/local.conf")
26 self.testinc_path = os.path.join(self.builddir, "conf/selftest.inc")
27 self.testlayer_path = oeSelfTest.testlayer_path
28 super(oeSelfTest, self).__init__(methodName)
29
30 def setUp(self):
31 os.chdir(self.builddir)
32 # we don't know what the previous test left around in config or inc files
33 # if it failed so we need a fresh start
34 try:
35 os.remove(self.testinc_path)
36 except OSError as e:
37 if e.errno != errno.ENOENT:
38 raise
39 for root, _, files in os.walk(self.testlayer_path):
40 for f in files:
41 if f == 'test_recipe.inc':
42 os.remove(os.path.join(root, f))
43 # tests might need their own setup
44 # but if they overwrite this one they have to call
45 # super each time, so let's give them an alternative
46 self.setUpLocal()
47
48 def setUpLocal(self):
49 pass
50
51 def tearDown(self):
52 self.tearDownLocal()
53
54 def tearDownLocal(self):
55 pass
56
57 # write to <builddir>/conf/selftest.inc
58 def write_config(self, data):
59 self.log.debug("Writing to: %s\n%s\n" % (self.testinc_path, data))
60 ftools.write_file(self.testinc_path, data)
61
62 # append to <builddir>/conf/selftest.inc
63 def append_config(self, data):
64 self.log.debug("Appending to: %s\n%s\n" % (self.testinc_path, data))
65 ftools.append_file(self.testinc_path, data)
66
67 # remove data from <builddir>/conf/selftest.inc
68 def remove_config(self, data):
69 self.log.debug("Removing from: %s\n\%s\n" % (self.testinc_path, data))
70 ftools.remove_from_file(self.testinc_path, data)
71
72 # write to meta-sefltest/recipes-test/<recipe>/test_recipe.inc
73 def write_recipeinc(self, recipe, data):
74 inc_file = os.path.join(self.testlayer_path, 'recipes-test', recipe, 'test_recipe.inc')
75 self.log.debug("Writing to: %s\n%s\n" % (inc_file, data))
76 ftools.write_file(inc_file, data)
77
78 # append data to meta-sefltest/recipes-test/<recipe>/test_recipe.inc
79 def append_recipeinc(self, recipe, data):
80 inc_file = os.path.join(self.testlayer_path, 'recipes-test', recipe, 'test_recipe.inc')
81 self.log.debug("Appending to: %s\n%s\n" % (inc_file, data))
82 ftools.append_file(inc_file, data)
83
84 # remove data from meta-sefltest/recipes-test/<recipe>/test_recipe.inc
85 def remove_recipeinc(self, recipe, data):
86 inc_file = os.path.join(self.testlayer_path, 'recipes-test', recipe, 'test_recipe.inc')
87 self.log.debug("Removing from: %s\n%s\n" % (inc_file, data))
88 ftools.remove_from_file(inc_file, data)
89
90 # delete meta-sefltest/recipes-test/<recipe>/test_recipe.inc file
91 def delete_recipeinc(self, recipe):
92 inc_file = os.path.join(self.testlayer_path, 'recipes-test', recipe, 'test_recipe.inc')
93 self.log.debug("Deleting file: %s" % inc_file)
94 try:
95 os.remove(self.testinc_path)
96 except OSError as e:
97 if e.errno != errno.ENOENT:
98 raise
diff --git a/meta/lib/oeqa/utils/commands.py b/meta/lib/oeqa/utils/commands.py
new file mode 100644
index 0000000000..9b42620610
--- /dev/null
+++ b/meta/lib/oeqa/utils/commands.py
@@ -0,0 +1,137 @@
1# Copyright (c) 2013 Intel Corporation
2#
3# Released under the MIT license (see COPYING.MIT)
4
5# DESCRIPTION
6# This module is mainly used by scripts/oe-selftest and modules under meta/oeqa/selftest
7# It provides a class and methods for running commands on the host in a convienent way for tests.
8
9
10
11import os
12import sys
13import signal
14import subprocess
15import threading
16import logging
17
18class Command(object):
19 def __init__(self, command, bg=False, timeout=None, data=None, **options):
20
21 self.defaultopts = {
22 "stdout": subprocess.PIPE,
23 "stderr": subprocess.STDOUT,
24 "stdin": None,
25 "shell": False,
26 "bufsize": -1,
27 }
28
29 self.cmd = command
30 self.bg = bg
31 self.timeout = timeout
32 self.data = data
33
34 self.options = dict(self.defaultopts)
35 if isinstance(self.cmd, basestring):
36 self.options["shell"] = True
37 if self.data:
38 self.options['stdin'] = subprocess.PIPE
39 self.options.update(options)
40
41 self.status = None
42 self.output = None
43 self.error = None
44 self.thread = None
45
46 self.log = logging.getLogger("utils.commands")
47
48 def run(self):
49 self.process = subprocess.Popen(self.cmd, **self.options)
50
51 def commThread():
52 self.output, self.error = self.process.communicate(self.data)
53
54 self.thread = threading.Thread(target=commThread)
55 self.thread.start()
56
57 self.log.debug("Running command '%s'" % self.cmd)
58
59 if not self.bg:
60 self.thread.join(self.timeout)
61 self.stop()
62
63 def stop(self):
64 if self.thread.isAlive():
65 self.process.terminate()
66 # let's give it more time to terminate gracefully before killing it
67 self.thread.join(5)
68 if self.thread.isAlive():
69 self.process.kill()
70 self.thread.join()
71
72 self.output = self.output.rstrip()
73 self.status = self.process.poll()
74
75 self.log.debug("Command '%s' returned %d as exit code." % (self.cmd, self.status))
76 # logging the complete output is insane
77 # bitbake -e output is really big
78 # and makes the log file useless
79 if self.status:
80 lout = "\n".join(self.output.splitlines()[-20:])
81 self.log.debug("Last 20 lines:\n%s" % lout)
82
83
84class Result(object):
85 pass
86
87def runCmd(command, ignore_status=False, timeout=None, **options):
88
89 result = Result()
90
91 cmd = Command(command, timeout=timeout, **options)
92 cmd.run()
93
94 result.command = command
95 result.status = cmd.status
96 result.output = cmd.output
97 result.pid = cmd.process.pid
98
99 if result.status and not ignore_status:
100 raise AssertionError("Command '%s' returned non-zero exit status %d:\n%s" % (command, result.status, result.output))
101
102 return result
103
104
105def bitbake(command, ignore_status=False, timeout=None, **options):
106 if isinstance(command, basestring):
107 cmd = "bitbake " + command
108 else:
109 cmd = [ "bitbake" ] + command
110
111 return runCmd(cmd, ignore_status, timeout, **options)
112
113
114def get_bb_env(target=None):
115 if target:
116 return runCmd("bitbake -e %s" % target).output
117 else:
118 return runCmd("bitbake -e").output
119
120def get_bb_var(var, target=None):
121 val = None
122 bbenv = get_bb_env(target)
123 for line in bbenv.splitlines():
124 if line.startswith(var + "="):
125 val = line.split('=')[1]
126 val = val.replace('\"','')
127 break
128 return val
129
130def get_test_layer():
131 layers = get_bb_var("BBLAYERS").split()
132 testlayer = None
133 for l in layers:
134 if "/meta-selftest" in l and os.path.isdir(l):
135 testlayer = l
136 break
137 return testlayer
diff --git a/meta/lib/oeqa/utils/ftools.py b/meta/lib/oeqa/utils/ftools.py
new file mode 100644
index 0000000000..64ebe3d217
--- /dev/null
+++ b/meta/lib/oeqa/utils/ftools.py
@@ -0,0 +1,27 @@
1import os
2import re
3
4def write_file(path, data):
5 wdata = data.rstrip() + "\n"
6 with open(path, "w") as f:
7 f.write(wdata)
8
9def append_file(path, data):
10 wdata = data.rstrip() + "\n"
11 with open(path, "a") as f:
12 f.write(wdata)
13
14def read_file(path):
15 data = None
16 with open(path) as f:
17 data = f.read()
18 return data
19
20def remove_from_file(path, data):
21 lines = read_file(path).splitlines()
22 rmdata = data.strip().splitlines()
23 for l in rmdata:
24 for c in range(0, lines.count(l)):
25 i = lines.index(l)
26 del(lines[i])
27 write_file(path, "\n".join(lines))