cq/scripts/slave/step/fuchsia_emulator_step.py

"""Step class for managing a Fuchsia emulator instance using ffx emu."""

import logging
import os
import shutil
import socket
import subprocess
import time
from unittest import mock

from slave import base_step


_STEP_DATA_ENV_PID = 'step_data_env_pid'


class FuchsiaEmulatorStartStep(base_step.BaseStep):
  """Step for starting local Fuchsia assets."""

  def __init__(self, chromium_root, out_dir, fuchsia_workdir, **kwargs):
    """Creates a FuchsiaEmulatorCreateStep instance.

    Args:
      kwargs: Passed to FuchsiaEmulatorBaseStep.
    """
    base_step.BaseStep.__init__(self, name='Start fuchsia test env', **kwargs)
    self._chromium_root = chromium_root
    self._out_dir = out_dir
    self._fuchsia_workdir = fuchsia_workdir


  def run(self):
    daemon_dir = os.path.join(self._fuchsia_workdir, 'daemon')
    os.makedirs(daemon_dir, exist_ok=True)
    self.add_step_data('step_data_daemon_dir', daemon_dir)
    proc = self.start_subprocess(
            None,
            subprocess.PIPE,
            subprocess.PIPE, [
              os.path.join(self._chromium_root,
                           'build/fuchsia/test/test_env_setup.py'),
              '--logs-dir',
              self.get_gcs_dir(),
            ],
            env={**os.environ,
                 'FFX_ISOLATE_DIR': daemon_dir,
                 'CHROME_HEADLESS': '1'})[0]
    pid_file = os.path.join(self.get_gcs_dir(),
                            'test_env_setup.' + str(proc.pid) + '.pid')
    while not os.path.isfile(pid_file):
        proc.poll()
        if proc.returncode:
            return False
        time.sleep(1)
    self.add_step_data(_STEP_DATA_ENV_PID, proc.pid)
    return True


class FuchsiaEmulatorTeardownStep(base_step.BaseStep):
  """Step for cleaning up after the FuchsiaEmulatorStartStep."""

  def __init__(self, **kwargs):
    """Creates a FuchsiaEmulatorTeardownStep instance."""
    base_step.BaseStep.__init__(self,
                                name='Teardown fuchsia test env',
                                **kwargs)

  def run(self):
    """Cleanup after FuchsiaEmulatorTeardownStep."""
    self.exec_subprocess(['kill', str(self.get_step_data(_STEP_DATA_ENV_PID))])
    return True