def test_check_totals_calls_run_once(self):
        prob = Problem()
        root = prob.root = Group()
        root.add('p1', IndepVarComp('x', 1.0), promotes=['*'])
        root.add('p2', IndepVarComp('y', 1.0), promotes=['*'])
        root.add('comp', Paraboloid(), promotes=['*'])

        prob.driver.add_desvar('x')
        prob.driver.add_desvar('y')
        prob.driver.add_objective('f_xy')

        prob.setup(check=False)

        prob['x'] = 5.0
        prob['y'] = 2.0

        iostream = StringIO()

        data = prob.check_total_derivatives(out_stream=iostream)

        self.assertAlmostEqual(first=prob["f_xy"],
                               second= (prob['x']-3.0)**2 \
                                       + prob['x']*prob['y'] \
                                       + (prob['y']+4.0)**2 - 3.0,
                               places=5,
                               msg="check partial derivatives did not call"
                                   "run_once on the driver as expected.")

        self.assertEqual(first=iostream.getvalue()[:39],
                         second="Executing model to populate unknowns...",
                         msg="check partial derivatives failed to run driver once")

 def test_no_children(self):
     my_task = Task("t2", [""], file_dep=['d2.txt'])
     output = StringIO()
     cmd = CmdFactory(Graphx, outstream=output, task_list=[my_task])
     cmd._execute(graph_type='json', no_children=True)
     got = output.getvalue()
     self.assertNotIn("d2.txt", got)

def _entries_to_ldif(entries):
    """Format LDAP entries as LDIF"""
    io = StringIO()
    writer = LDIFWriter(io)
    for entry in entries:
        writer.unparse(str(entry.dn), dict(entry.raw))
    return io.getvalue()

    def test_simple(self):
        # This test just checks the output of the program against
        # some expected output
        from six import StringIO

        out = StringIO()
        zdd.do_zdd(Arguments(), out)
        output = json.loads(out.getvalue())
        output['labels']['HAPROXY_DEPLOYMENT_STARTED_AT'] = ""

        expected = json.loads('''{
  "acceptedResourceRoles": [
    "*",
    "slave_public"
  ],
  "container": {
    "docker": {
      "forcePullImage": true,
      "image": "brndnmtthws/nginx-echo-sleep",
      "network": "BRIDGE",
      "portMappings": [
        {
          "containerPort": 8080,
          "hostPort": 0,
          "servicePort": 10001
        }
      ]
    },
    "type": "DOCKER"
  },
  "cpus": 0.1,
  "healthChecks": [
    {
      "gracePeriodSeconds": 15,
      "intervalSeconds": 3,
      "maxConsecutiveFailures": 10,
      "path": "/",
      "portIndex": 0,
      "protocol": "HTTP",
      "timeoutSeconds": 15
    }
  ],
  "id": "/nginx-blue",
  "instances": 1,
  "labels": {
    "HAPROXY_0_PORT": "10000",
    "HAPROXY_APP_ID": "nginx",
    "HAPROXY_DEPLOYMENT_ALT_PORT": "10001",
    "HAPROXY_DEPLOYMENT_COLOUR": "blue",
    "HAPROXY_DEPLOYMENT_GROUP": "nginx",
    "HAPROXY_DEPLOYMENT_NEW_INSTANCES": "0",
    "HAPROXY_DEPLOYMENT_STARTED_AT": "2016-02-01T15:51:38.184623",
    "HAPROXY_DEPLOYMENT_TARGET_INSTANCES": "3",
    "HAPROXY_GROUP": "external"
  },
  "mem": 65
}
''')
        expected['labels']['HAPROXY_DEPLOYMENT_STARTED_AT'] = ""
        self.assertEqual(output, expected)

    def test_adds_supergroup(self):
        out = StringIO()
        fxa_id = uuid.uuid4().hex
        call_command(
            'createsuperuser',
            interactive=False,
            username='myusername',
            email='[email protected]',
            add_to_supercreate_group=True,
            fxa_id=fxa_id,
            stdout=out)

        user = UserProfile.objects.get(username='myusername')
        assert user.email == '[email protected]'
        assert user.read_dev_agreement
        assert user.groups.filter(rules='Accounts:SuperCreate').exists()

        response = json.loads(out.getvalue())

        assert response == {
            'username': 'myusername',
            'email': '[email protected]',
            'api-key': ANY,
            'api-secret': ANY,
            'fxa-id': fxa_id,
        }

 def check_roundtrip(self, code1, filename="internal"):
     ast1 = compile(code1, filename, "exec", ast.PyCF_ONLY_AST)
     unparse_buffer = StringIO()
     unparse.Unparser(ast1, unparse_buffer)
     code2 = unparse_buffer.getvalue()
     ast2 = compile(code2, filename, "exec", ast.PyCF_ONLY_AST)
     self.assertASTEqual(ast1, ast2)

    def render(self, data, media_type=None, renderer_context={}, writer_opts=None):
        """
        Renders serialized *data* into CSV. For a dictionary:
        """
        if data is None:
            return ''

        if not isinstance(data, list):
            data = [data]

        if writer_opts is not None:
            log.warning('The writer_opts argument is deprecated. Pass the '
                        'writer_opts attribute into renderer_context instead.')

        writer_opts = renderer_context.get('writer_opts', writer_opts or self.writer_opts or {})

        renderer_context = renderer_context or {}
        header = renderer_context.get('header', self.header)
        labels = renderer_context.get('labels', self.labels)

        table = self.tablize(data, header=header, labels=labels)
        csv_buffer = StringIO()
        csv_writer = csv.writer(csv_buffer, **writer_opts)
        for row in table:
            # Assume that strings should be encoded as UTF-8
            csv_writer.writerow([
                elem.encode('utf-8') if isinstance(elem, text_type) and PY2 else elem
                for elem in row
            ])

        return csv_buffer.getvalue()

    def get(self, *args, **kwargs):
        # type: (*Any, **Any) -> None
        fh = StringIO()
        w_csv = csv.writer(fh, lineterminator="\n")

        # header
        w_csv.writerow(
            [
                "username",
                "created_at",
                "type",
                "size",
                "fingerprint",
                "fingerprint_sha256",
                "comment",
            ]
        )

        with closing(Session()) as session:
            user_key_list = session.query(PublicKey, User).filter(User.id == PublicKey.user_id)
            for key, user in user_key_list:
                w_csv.writerow(
                    [
                        user.name,
                        key.created_on.isoformat(),
                        key.key_type,
                        key.key_size,
                        key.fingerprint,
                        key.fingerprint_sha256,
                        key.comment,
                    ]
                )

        self.set_header("Content-Type", "text/csv")
        self.write(fh.getvalue())

    def render(self, input_data, media_type=None, renderer_context=None):
        """
        Renders serialized *data* into CSV. For a dictionary:
        """
        if input_data is None:
            return ''

        data = input_data
        if not isinstance(data, list):
            data = input_data.get('results', [input_data])

        table = self.tablize(data)



        csv_buffer = StringIO()
        csv_writer = csv.writer(csv_buffer, dialect='excel-tab')
        for row in table:
            # Assume that strings should be encoded as UTF-8
            csv_writer.writerow([
                elem.encode('utf-8') if isinstance(elem, text_type) and PY2 else elem
                for elem in row
            ])

        return csv_buffer.getvalue()

def bottlestats(server_name, path=''):
    """Renders a GET request, by showing this nodes stats and children."""
    path = path.lstrip('/')
    parts = path.split('/')
    if not parts[0]:
        parts = parts[1:]
    stat_dict = util.lookup(scales.getStats(), parts)

    if stat_dict is None:
        abort(404, "Not Found")
        return

    output = StringIO()
    output_format = request.query.get('format', 'html')
    query = request.query.get('query', None)
    if output_format == 'json':
        response.content_type = "application/json"
        formats.jsonFormat(output, stat_dict, query)
    elif output_format == 'prettyjson':
        formats.jsonFormat(output, stat_dict, query, pretty=True)
        response.content_type = "application/json"
    else:
        formats.htmlHeader(output, '/' + path, server_name, query)
        formats.htmlFormat(output, tuple(parts), stat_dict, query)
        response.content_type = "text/html"

    return output.getvalue()

    def cmd(self, command, checks=None, allowed_exceptions=None, debug=False): #pylint: disable=no-self-use
        allowed_exceptions = allowed_exceptions or []
        if not isinstance(allowed_exceptions, list):
            allowed_exceptions = [allowed_exceptions]

        if self._debug or debug:
            print('\n\tRUNNING: {}'.format(command))
        command_list = shlex.split(command)
        output = StringIO()
        try:
            cli_main(command_list, file=output)
        except Exception as ex: # pylint: disable=broad-except
            ex_msg = str(ex)
            if not next((x for x in allowed_exceptions if x in ex_msg), None):
                raise ex
        self._track_executed_commands(command_list)
        result = output.getvalue().strip()
        output.close()

        if self._debug or debug:
            print('\tRESULT: {}\n'.format(result))

        if checks:
            checks = [checks] if not isinstance(checks, list) else checks
            for check in checks:
                check.compare(result)

        result = result or '{}'
        try:
            return json.loads(result)
        except Exception: # pylint: disable=broad-except
            return result

class _Buffer(object):

    def __init__(self, stream):
        self._stream = stream
        self._buffer = StringIO()

    def fileno(self):
        return self._stream.fileno()

    def __getattr__(self, attr):
        # this happens on unpickling
        if attr == '_buffer':
            raise AttributeError("No _buffer yet")
        return getattr(self._buffer, attr)

    def __le__(self, obj):
        return self._buffer.getvalue() == obj

    def __eq__(self, obj):
        return self._buffer.getvalue() == obj

    def __str__(self):
        return self._buffer.getvalue()

    def __repr__(self):
        return repr(self._buffer.getvalue())

def run_layer(options, layer_name, layer, tests, setup_layers,
              failures, errors):

    output = options.output
    gathered = []
    gather_layers(layer, gathered)
    needed = dict([(l, 1) for l in gathered])
    if options.resume_number != 0:
        output.info("Running %s tests:" % layer_name)
    tear_down_unneeded(options, needed, setup_layers)

    if options.resume_layer is not None:
        output.info_suboptimal("  Running in a subprocess.")

    try:
        setup_layer(options, layer, setup_layers)
    except zope.testrunner.interfaces.EndRun:
        raise
    except Exception:
        f = StringIO()
        traceback.print_exc(file=f)
        output.error(f.getvalue())
        errors.append((SetUpLayerFailure(layer), sys.exc_info()))
        return 0
    else:
        return run_tests(options, tests, layer_name, failures, errors)

def runquery_csv():
	global out

	q = frappe.form_dict.get('query')

	rep_name = frappe.form_dict.get('report_name')
	if not frappe.form_dict.get('simple_query'):

		# Report Name
		if not rep_name:
			rep_name = get_sql_tables(q)[0]

	if not rep_name: rep_name = 'DataExport'

	rows = [[rep_name], out['colnames']] + out['values']

	from six import StringIO
	import csv

	f = StringIO()
	writer = csv.writer(f)
	for r in rows:
		# encode only unicode type strings and not int, floats etc.
		writer.writerow(map(lambda v: isinstance(v, text_type) and v.encode('utf-8') or v, r))

	f.seek(0)
	out['result'] = text_type(f.read(), 'utf-8')
	out['type'] = 'csv'
	out['doctype'] = rep_name

def get_correct_indentation_diff(code, filename):
    """
    Generate a diff to make code correctly indented.

    :param code: a string containing a file's worth of Python code
    :param filename: the filename being considered (used in diff generation only)
    :returns: a unified diff to make code correctly indented, or
              None if code is already correctedly indented
    """
    code_buffer = StringIO(code)
    output_buffer = StringIO()
    reindenter = reindent.Reindenter(code_buffer)
    reindenter.run()
    reindenter.write(output_buffer)
    reindent_output = output_buffer.getvalue()
    output_buffer.close()
    if code != reindent_output:
        diff_generator = difflib.unified_diff(code.splitlines(True), reindent_output.splitlines(True),
                                              fromfile=filename, tofile=filename + " (reindented)")
        # work around http://bugs.python.org/issue2142
        diff_tuple = map(clean_diff_line_for_python_bug_2142, diff_generator)
        diff = "".join(diff_tuple)
        return diff
    else:
        return None

 def test_cmds_with_params(self, commands):
     output = StringIO()
     cmd = CmdFactory(TabCompletion, task_loader=DodoTaskLoader(),
                      outstream=output, cmds=commands)
     cmd.execute({'shell':'zsh', 'hardcode_tasks': False}, [])
     got = output.getvalue()
     assert "tabcompletion: generate script" in got

def test_packed_workflow_execution(wf_path, job_path, namespaced, tmpdir):
    load_tool.loaders = {}

    document_loader, workflowobj, uri = fetch_document(
        get_data(wf_path), resolver=tool_resolver)

    document_loader, _, processobj, metadata, uri = validate_document(
        document_loader, workflowobj, uri, [], {})

    packed = json.loads(print_pack(document_loader, processobj, uri, metadata))

    assert not namespaced or "$namespaces" in packed

    wf_packed_handle, wf_packed_path = tempfile.mkstemp()
    with open(wf_packed_path, 'w') as temp_file:
        json.dump(packed, temp_file)

    normal_output = StringIO()
    packed_output = StringIO()

    normal_params = ['--outdir', str(tmpdir), get_data(wf_path), get_data(job_path)]
    packed_params = ['--outdir', str(tmpdir), '--debug', get_data(wf_packed_path), get_data(job_path)]

    assert main(normal_params, stdout=normal_output) == 0
    assert main(packed_params, stdout=packed_output) == 0

    assert json.loads(packed_output.getvalue()) == json.loads(normal_output.getvalue())

    os.close(wf_packed_handle)
    os.remove(wf_packed_path)

 def test_hardcoded_tasks(self, commands):
     output = StringIO()
     cmd = CmdFactory(TabCompletion, task_loader=FakeLoader(),
                      outstream=output, cmds=commands)
     cmd.execute({'shell':'zsh', 'hardcode_tasks': True}, [])
     got = output.getvalue()
     assert 't1' in got

    def __call__(self, environ, start_response):
        script_name = environ.get('SCRIPT_NAME', '')
        path_info = environ.get('PATH_INFO', '')
        sent = []
        written_response = StringIO()

        def replacement_start_response(status, headers, exc_info=None):
            if not self.should_filter(status, headers):
                return start_response(status, headers, exc_info)
            else:
                sent[:] = [status, headers, exc_info]
                return written_response.write

        app_iter = self.app(environ, replacement_start_response)
        if not sent:
            return app_iter
        status, headers, exc_info = sent
        try:
            for chunk in app_iter:
                written_response.write(chunk)
        finally:
            if hasattr(app_iter, 'close'):
                app_iter.close()
        body = written_response.getvalue()
        status, headers, body = self.filter(
            script_name, path_info, environ, status, headers, body)
        start_response(status, headers, exc_info)
        return [body]

 def test_isatty_false(self):
     w1 = StringIO()
     w1.isatty = lambda: True
     w2 = StringIO()
     w2.isatty = lambda: True
     writer = action.Writer(w1, w2)
     assert writer.isatty()