How to use sarge - 10 common examples

# libvirt wants the memory in MB
    memory *= 1024
    vm_name = '{}-{}'.format(rhci.deployment_id, vm_suffix)
    # The following is mostly copied out of the deploy from iso script,
    # and should probably be generalized in rhci_common (or something)
    print 'Provisioning RHCI VM {} via libvirt'.format(vm_name)

    shell_args = {
        'vm_name': vm_name,
        'memory': memory,
        'cpus': cpu_count,
        'disk_size': disk_size,
        'private_net_config': 'bridge:br1',
        'vnc_password': vnc_password
    }
    cmd = sarge.shell_format('virt-install'
        ' -n {vm_name}'
        ' --os-variant=rhel7'
        ' --ram {memory}'
        ' --pxe'
        ' --vcpus {cpus}'
        ' --disk bus="virtio,size={disk_size}"'
        ' --network {private_net_config}'
        ' --boot net,hd'
        ' --graphics "vnc,listen=0.0.0.0,password={vnc_password}"'
        ' --noautoconsole',
        **shell_args)

    proc = sarge.capture_both(cmd)
    if proc.returncode != 0:
        print >>sys.stderr, 'virt-install failed'
        print >>sys.stderr, proc.stderr.read()

# virt-install the ISO we just downloaded
print 'Provisioning RHCI VM {} via libvirt'.format(vm_name)

shell_args = {
    'vm_name': vm_name,
    'memory': 12288,
    'cpus': 4,
    'libvirt_pool': libvirt_storage_pool,
    'disk_size': 100,
    'public_net_config': public_net_config,
    'private_net_config': private_net_config,
    'image_path': image_path,
    'vnc_password': vnc_password
}
cmd = sarge.shell_format('virt-install'
    ' -n {vm_name}'
    ' --os-variant=rhel7'
    ' --ram {memory}'
    ' --vcpus {cpus}'
    ' --disk bus="virtio,pool={libvirt_pool},size={disk_size}"'
    ' --cdrom {image_path}'
    ' --network {public_net_config}'
    ' --network {private_net_config}'
    ' --boot "hd,cdrom"'
    ' --graphics "vnc,listen=0.0.0.0,password={vnc_password}"'
    ' --noautoconsole',
    **shell_args)

# this will block while virt-install runs
proc = sarge.capture_both(cmd)
if proc.returncode != 0:

env.update(extra_env)
    env['PORT'] = '10180'
    env['SITE_ORIGIN'] = 'localhost:10180'
    env['CONTENT_ORIGIN'] = 'localhost:10180'
    env['GA_ID'] = ''
    env['NODE_ENV'] = 'production'
    env['LOG_QUERY_LIMIT'] = '0'
    env['ENABLE_WATCHDOG'] = 'false'
    server_out = Capture()
    env_print = ['%s=%s' % (name, value) for name, value in sorted(extra_env.items())]
    if env_print:
        env_print = ' (%s)' % ' '.join(env_print)
    else:
        env_print = ''
    print('  Starting server%s' % env_print)
    server = run('./bin/run-server --no-auto', cwd=project_base, env=env, stdout=server_out, async_=True)
    server_options = extra_env
    time.sleep(3)
    text = []
    while True:
        if server.commands[0].process and server.commands[0].poll():
            print('    Server exited with code %s' % server.commands[0].poll())
            text.extend(server_out.readlines())
            for line in text:
                print('      %s' % line.rstrip())
            print("    %s" % ("-" * 60))
            raise Exception("Server didn't start")
        line = server_out.readline()
        if line:
            text.append(line)
        if 'Database is now at level' in line:
            # Last log message before the server is running

def _nosetests():
    nosetests_cmd = os.path.join(curdir, 'bin', 'nosetests')
    zato_packages = ' '.join([item for item in glob.iglob(os.path.join(curdir, 'zato-*'))])

    run('{} {} --nocapture'.format(nosetests_cmd, zato_packages))

for key in extra_env:
        extra_env[key] = str(extra_env[key])
    if server and server_options == extra_env:
        print("  Server already started with correct options")
        return
    stop_server()
    env = os.environ.copy()
    env.update(extra_env)
    env['PORT'] = '10180'
    env['SITE_ORIGIN'] = 'localhost:10180'
    env['CONTENT_ORIGIN'] = 'localhost:10180'
    env['GA_ID'] = ''
    env['NODE_ENV'] = 'production'
    env['LOG_QUERY_LIMIT'] = '0'
    env['ENABLE_WATCHDOG'] = 'false'
    server_out = Capture()
    env_print = ['%s=%s' % (name, value) for name, value in sorted(extra_env.items())]
    if env_print:
        env_print = ' (%s)' % ' '.join(env_print)
    else:
        env_print = ''
    print('  Starting server%s' % env_print)
    server = run('./bin/run-server --no-auto', cwd=project_base, env=env, stdout=server_out, async_=True)
    server_options = extra_env
    time.sleep(3)
    text = []
    while True:
        if server.commands[0].process and server.commands[0].poll():
            print('    Server exited with code %s' % server.commands[0].poll())
            text.extend(server_out.readlines())
            for line in text:
                print('      %s' % line.rstrip())

def virsh(args):
    result = sarge.capture_stdout('virsh {}'.format(args))
    if result.returncode != 0:
        raise RuntimeError('virsh command failed with exit code {}'.format(result.returncode))
    return result.stdout.read().strip()

	@mock.patch.object(sarge.Pipeline, 'wait_events')
	@mock.patch('sarge.run')
	@mock.patch('octoprint_firmwareupdater.os.path.dirname')
	def test_flash_worker_ok(self, mock_dirname, mock_sarge_run, mock_wait_events, mock_send_status):
		# Set Up
		self.plugin = octoprint_firmwareupdater.FirmwareupdaterPlugin()
		self.plugin._settings = settings_mock()
		self.plugin._logger = logger_mock()
		self.plugin._printer = printer_mock()
		p = pipeline_mock()
		mock_sarge_run.return_value = p

		hex_file = named_temporary_file_mock()
		hex_file.close = mock.MagicMock(name='close')

		# Call test subject
		self.plugin._flash_worker(hex_file, "port")

def check_google_mx(domain):
    """
    Check Google MX DNS records

    Args:
        domain (str): DNS domain name

    Returns:
        int: 0 if OK, 1 on error

    | https://support.google.com/a/topic/2716885?hl=en&ref_topic=2426592
    """
    cmd = sarge.shell_format("dig {0} mx +short", domain)
    log.info('cmd', cmd=cmd)
    output = sarge.capture_both(cmd).stdout.text.rstrip()
    log.debug('MX', record=output)
    result = 0
    check_domain1 = "aspmx.l.google.com."
    check_domain2 = "googlemail.com."
    lines = output.split('\n')
    if not lines:
        log.error('err', msg="No MX records found for %r" % domain)
        result += 1
    for l in lines:
        l = l.lower()
        if not (l.endswith(check_domain1) or l.endswith(check_domain2)):
            result += 1
            log.error('err', msg="%r does not end with %r or %r" %
                      (l, check_domain1, check_domain2))

if (stm32flash_reset and not stm32flash_execute):
        stm32flash_args.append('-R')

    stm32flash_args.append('-w')
    stm32flash_args.append(firmware)
    stm32flash_args.append(printer_port)

    stm32flash_command = ' '.join(stm32flash_args)

    self._logger.info(u"Running '{}' in {}".format(stm32flash_command, working_dir))
    self._console_logger.info(u"")
    self._console_logger.info(stm32flash_command)

    try:
        p = sarge.run(stm32flash_command, cwd=working_dir, async=True, stdout=sarge.Capture(), stderr=sarge.Capture())
        p.wait_events()

        while p.returncode is None:
            output = p.stdout.read(timeout=0.5)
            if not output:
                p.commands[0].poll()
                continue

            for line in output.split("\n"):
                if line.endswith("\r"):
                    line = line[:-1]
                self._console_logger.info(u"> {}".format(line))

                if "Write to memory" in line:
                    self._logger.info(u"Writing memory...")
                    self._send_status("progress", subtype="writing")

while not command_done.is_set():
                                try:
                                    for line in out:
                                        echo_info(line.strip())
                                except TypeError:
                                    pass

                                try:
                                    for line in err:
                                        warn(line.strip())
                                except TypeError:
                                    pass

                                sleep(0.05)

                        p = run(cmd, stdout=out, stderr=err, env=self._env, async=True)

                        if self._verbose:
                            Thread(target=redirect).start()

                        try:
                            p.wait()
                        finally:
                            # make sure the console redirect thread is properly shutting down
                            command_done.set()

            except OSError:
                error('command: "{0}" could not be executed (not found?)'.format(cmd))