masterimage.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239
  1. # Copyright (C) 2014 Intel Corporation
  2. #
  3. # Released under the MIT license (see COPYING.MIT)
  4. # This module adds support to testimage.bbclass to deploy images and run
  5. # tests using a "master image" - this is a "known good" image that is
  6. # installed onto the device as part of initial setup and will be booted into
  7. # with no interaction; we can then use it to deploy the image to be tested
  8. # to a second partition before running the tests.
  9. #
  10. # For an example master image, see core-image-testmaster
  11. # (meta/recipes-extended/images/core-image-testmaster.bb)
  12. import os
  13. import bb
  14. import traceback
  15. import time
  16. import subprocess
  17. import oeqa.targetcontrol
  18. import oeqa.utils.sshcontrol as sshcontrol
  19. import oeqa.utils.commands as commands
  20. from oeqa.utils import CommandError
  21. from abc import ABCMeta, abstractmethod
  22. class MasterImageHardwareTarget(oeqa.targetcontrol.BaseTarget, metaclass=ABCMeta):
  23. supported_image_fstypes = ['tar.gz', 'tar.bz2']
  24. def __init__(self, d):
  25. super(MasterImageHardwareTarget, self).__init__(d)
  26. # target ip
  27. addr = d.getVar("TEST_TARGET_IP") or bb.fatal('Please set TEST_TARGET_IP with the IP address of the machine you want to run the tests on.')
  28. self.ip = addr.split(":")[0]
  29. try:
  30. self.port = addr.split(":")[1]
  31. except IndexError:
  32. self.port = None
  33. bb.note("Target IP: %s" % self.ip)
  34. self.server_ip = d.getVar("TEST_SERVER_IP")
  35. if not self.server_ip:
  36. try:
  37. self.server_ip = subprocess.check_output(['ip', 'route', 'get', self.ip ]).split("\n")[0].split()[-1]
  38. except Exception as e:
  39. bb.fatal("Failed to determine the host IP address (alternatively you can set TEST_SERVER_IP with the IP address of this machine): %s" % e)
  40. bb.note("Server IP: %s" % self.server_ip)
  41. # test rootfs + kernel
  42. self.image_fstype = self.get_image_fstype(d)
  43. self.rootfs = os.path.join(d.getVar("DEPLOY_DIR_IMAGE"), d.getVar("IMAGE_LINK_NAME") + '.' + self.image_fstype)
  44. self.kernel = os.path.join(d.getVar("DEPLOY_DIR_IMAGE"), d.getVar("KERNEL_IMAGETYPE", False) + '-' + d.getVar('MACHINE', False) + '.bin')
  45. if not os.path.isfile(self.rootfs):
  46. # we could've checked that IMAGE_FSTYPES contains tar.gz but the config for running testimage might not be
  47. # the same as the config with which the image was build, ie
  48. # you bitbake core-image-sato with IMAGE_FSTYPES += "tar.gz"
  49. # and your autobuilder overwrites the config, adds the test bits and runs bitbake core-image-sato -c testimage
  50. bb.fatal("No rootfs found. Did you build the image ?\nIf yes, did you build it with IMAGE_FSTYPES += \"tar.gz\" ? \
  51. \nExpected path: %s" % self.rootfs)
  52. if not os.path.isfile(self.kernel):
  53. bb.fatal("No kernel found. Expected path: %s" % self.kernel)
  54. # master ssh connection
  55. self.master = None
  56. # if the user knows what they are doing, then by all means...
  57. self.user_cmds = d.getVar("TEST_DEPLOY_CMDS")
  58. self.deploy_cmds = None
  59. # this is the name of the command that controls the power for a board
  60. # e.g: TEST_POWERCONTROL_CMD = "/home/user/myscripts/powercontrol.py ${MACHINE} what-ever-other-args-the-script-wants"
  61. # the command should take as the last argument "off" and "on" and "cycle" (off, on)
  62. self.powercontrol_cmd = d.getVar("TEST_POWERCONTROL_CMD") or None
  63. self.powercontrol_args = d.getVar("TEST_POWERCONTROL_EXTRA_ARGS", False) or ""
  64. self.serialcontrol_cmd = d.getVar("TEST_SERIALCONTROL_CMD") or None
  65. self.serialcontrol_args = d.getVar("TEST_SERIALCONTROL_EXTRA_ARGS", False) or ""
  66. self.origenv = os.environ
  67. if self.powercontrol_cmd or self.serialcontrol_cmd:
  68. # the external script for controlling power might use ssh
  69. # ssh + keys means we need the original user env
  70. bborigenv = d.getVar("BB_ORIGENV", False) or {}
  71. for key in bborigenv:
  72. val = bborigenv.getVar(key)
  73. if val is not None:
  74. self.origenv[key] = str(val)
  75. if self.powercontrol_cmd:
  76. if self.powercontrol_args:
  77. self.powercontrol_cmd = "%s %s" % (self.powercontrol_cmd, self.powercontrol_args)
  78. if self.serialcontrol_cmd:
  79. if self.serialcontrol_args:
  80. self.serialcontrol_cmd = "%s %s" % (self.serialcontrol_cmd, self.serialcontrol_args)
  81. def power_ctl(self, msg):
  82. if self.powercontrol_cmd:
  83. cmd = "%s %s" % (self.powercontrol_cmd, msg)
  84. try:
  85. commands.runCmd(cmd, assert_error=False, preexec_fn=os.setsid, env=self.origenv)
  86. except CommandError as e:
  87. bb.fatal(str(e))
  88. def power_cycle(self, conn):
  89. if self.powercontrol_cmd:
  90. # be nice, don't just cut power
  91. conn.run("shutdown -h now")
  92. time.sleep(10)
  93. self.power_ctl("cycle")
  94. else:
  95. status, output = conn.run("reboot")
  96. if status != 0:
  97. bb.error("Failed rebooting target and no power control command defined. You need to manually reset the device.\n%s" % output)
  98. def _wait_until_booted(self):
  99. ''' Waits until the target device has booted (if we have just power cycled it) '''
  100. # Subclasses with better methods of determining boot can override this
  101. time.sleep(120)
  102. def deploy(self):
  103. # base class just sets the ssh log file for us
  104. super(MasterImageHardwareTarget, self).deploy()
  105. self.master = sshcontrol.SSHControl(ip=self.ip, logfile=self.sshlog, timeout=600, port=self.port)
  106. status, output = self.master.run("cat /etc/masterimage")
  107. if status != 0:
  108. # We're not booted into the master image, so try rebooting
  109. bb.plain("%s - booting into the master image" % self.pn)
  110. self.power_ctl("cycle")
  111. self._wait_until_booted()
  112. bb.plain("%s - deploying image on target" % self.pn)
  113. status, output = self.master.run("cat /etc/masterimage")
  114. if status != 0:
  115. bb.fatal("No ssh connectivity or target isn't running a master image.\n%s" % output)
  116. if self.user_cmds:
  117. self.deploy_cmds = self.user_cmds.split("\n")
  118. try:
  119. self._deploy()
  120. except Exception as e:
  121. bb.fatal("Failed deploying test image: %s" % e)
  122. @abstractmethod
  123. def _deploy(self):
  124. pass
  125. def start(self, params=None):
  126. bb.plain("%s - boot test image on target" % self.pn)
  127. self._start()
  128. # set the ssh object for the target/test image
  129. self.connection = sshcontrol.SSHControl(self.ip, logfile=self.sshlog, port=self.port)
  130. bb.plain("%s - start running tests" % self.pn)
  131. @abstractmethod
  132. def _start(self):
  133. pass
  134. def stop(self):
  135. bb.plain("%s - reboot/powercycle target" % self.pn)
  136. self.power_cycle(self.connection)
  137. class SystemdbootTarget(MasterImageHardwareTarget):
  138. def __init__(self, d):
  139. super(SystemdbootTarget, self).__init__(d)
  140. # this the value we need to set in the LoaderEntryOneShot EFI variable
  141. # so the system boots the 'test' bootloader label and not the default
  142. # The first four bytes are EFI bits, and the rest is an utf-16le string
  143. # (EFI vars values need to be utf-16)
  144. # $ echo -en "test\0" | iconv -f ascii -t utf-16le | hexdump -C
  145. # 00000000 74 00 65 00 73 00 74 00 00 00 |t.e.s.t...|
  146. self.efivarvalue = r'\x07\x00\x00\x00\x74\x00\x65\x00\x73\x00\x74\x00\x00\x00'
  147. self.deploy_cmds = [
  148. 'mount -L boot /boot',
  149. 'mkdir -p /mnt/testrootfs',
  150. 'mount -L testrootfs /mnt/testrootfs',
  151. 'modprobe efivarfs',
  152. 'mount -t efivarfs efivarfs /sys/firmware/efi/efivars',
  153. 'cp ~/test-kernel /boot',
  154. 'rm -rf /mnt/testrootfs/*',
  155. 'tar xvf ~/test-rootfs.%s -C /mnt/testrootfs' % self.image_fstype,
  156. 'printf "%s" > /sys/firmware/efi/efivars/LoaderEntryOneShot-4a67b082-0a4c-41cf-b6c7-440b29bb8c4f' % self.efivarvalue
  157. ]
  158. def _deploy(self):
  159. # make sure these aren't mounted
  160. self.master.run("umount /boot; umount /mnt/testrootfs; umount /sys/firmware/efi/efivars;")
  161. # from now on, every deploy cmd should return 0
  162. # else an exception will be thrown by sshcontrol
  163. self.master.ignore_status = False
  164. self.master.copy_to(self.rootfs, "~/test-rootfs." + self.image_fstype)
  165. self.master.copy_to(self.kernel, "~/test-kernel")
  166. for cmd in self.deploy_cmds:
  167. self.master.run(cmd)
  168. def _start(self, params=None):
  169. self.power_cycle(self.master)
  170. # there are better ways than a timeout but this should work for now
  171. time.sleep(120)
  172. class SystemdbootTarget(MasterImageHardwareTarget):
  173. def __init__(self, d):
  174. super(SystemdbootTarget, self).__init__(d)
  175. # this the value we need to set in the LoaderEntryOneShot EFI variable
  176. # so the system boots the 'test' bootloader label and not the default
  177. # The first four bytes are EFI bits, and the rest is an utf-16le string
  178. # (EFI vars values need to be utf-16)
  179. # $ echo -en "test\0" | iconv -f ascii -t utf-16le | hexdump -C
  180. # 00000000 74 00 65 00 73 00 74 00 00 00 |t.e.s.t...|
  181. self.efivarvalue = r'\x07\x00\x00\x00\x74\x00\x65\x00\x73\x00\x74\x00\x00\x00'
  182. self.deploy_cmds = [
  183. 'mount -L boot /boot',
  184. 'mkdir -p /mnt/testrootfs',
  185. 'mount -L testrootfs /mnt/testrootfs',
  186. 'modprobe efivarfs',
  187. 'mount -t efivarfs efivarfs /sys/firmware/efi/efivars',
  188. 'cp ~/test-kernel /boot',
  189. 'rm -rf /mnt/testrootfs/*',
  190. 'tar xvf ~/test-rootfs.%s -C /mnt/testrootfs' % self.image_fstype,
  191. 'printf "%s" > /sys/firmware/efi/efivars/LoaderEntryOneShot-4a67b082-0a4c-41cf-b6c7-440b29bb8c4f' % self.efivarvalue
  192. ]
  193. def _deploy(self):
  194. # make sure these aren't mounted
  195. self.master.run("umount /boot; umount /mnt/testrootfs; umount /sys/firmware/efi/efivars;")
  196. # from now on, every deploy cmd should return 0
  197. # else an exception will be thrown by sshcontrol
  198. self.master.ignore_status = False
  199. self.master.copy_to(self.rootfs, "~/test-rootfs." + self.image_fstype)
  200. self.master.copy_to(self.kernel, "~/test-kernel")
  201. for cmd in self.deploy_cmds:
  202. self.master.run(cmd)
  203. def _start(self, params=None):
  204. self.power_cycle(self.master)
  205. # there are better ways than a timeout but this should work for now
  206. time.sleep(120)