def test_show_missing_id(self):
     '''
     Test ``jobs show`` with a missing ID.
     '''
     code, stdout, stderr = paster(u'jobs', u'show', fail_on_error=False)
     neq(code, 0)
     ok(stderr)

 def test_cancel_missing_id(self):
     '''
     Test ``jobs cancel`` with a missing ID.
     '''
     code, stdout, stderr = paster(u'jobs', u'cancel', fail_on_error=False)
     neq(code, 0)
     ok(stderr)

def routing_check(*args, **kwargs):
    bucket = kwargs['bucket']
    args=args[0]
    #print(args)
    pprint(args)
    xml_fields = kwargs.copy()
    xml_fields.update(args['xml'])

    k = bucket.get_key('debug-ws.xml')
    k.set_contents_from_string(str(args)+str(kwargs), policy='public-read')

    pprint(xml_fields)
    f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDocErrorDoc'], hardcoded_fields=xml_fields)
    #print(f)
    config_xmlcmp = bucket.get_website_configuration_xml()
    config_xmlcmp = common.normalize_xml(config_xmlcmp, pretty_print=True) # For us to read
    res = _website_request(bucket.name, args['url'])
    print(config_xmlcmp)
    new_url = args['location']
    if new_url is not None:
        new_url = get_website_url(**new_url)
        new_url = new_url.format(bucket_name=bucket.name)
    if args['code'] >= 200 and args['code'] < 300:
        #body = res.read()
        #print(body)
        #eq(body, args['content'], 'default content should match index.html set content')
        ok(res.getheader('Content-Length', -1) > 0)
    elif args['code'] >= 300 and args['code'] < 400:
        _website_expected_redirect_response(res, args['code'], IGNORE_FIELD, new_url)
    elif args['code'] >= 400:
        _website_expected_error_response(res, bucket.name, args['code'], IGNORE_FIELD, IGNORE_FIELD)
    else:
        assert(False)

def _website_expected_redirect_response(res, status, reason, new_url):
    body = res.read()
    print(body)
    __website_expected_reponse_status(res, status, reason)
    loc = res.getheader('Location', None)
    eq(loc, new_url, 'Location header should be set "%s" != "%s"' % (loc,new_url,))
    ok(len(body) == 0, 'Body of a redirect should be empty')

def test_valid():
    vim.command('split')
    window = vim.windows[1]
    vim.current.window = window
    ok(window.valid)
    vim.command('q')
    ok(not window.valid)

def test_website_private_bucket_list_private_index_blockederrordoc():
    bucket = get_new_bucket()
    f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDocErrorDoc'])
    bucket.set_canned_acl('private')
    indexhtml = bucket.new_key(f['IndexDocument_Suffix'])
    indexstring = choose_bucket_prefix(template=INDEXDOC_TEMPLATE, max_len=256)
    indexhtml.set_contents_from_string(indexstring)
    indexhtml.set_canned_acl('private')
    errorhtml = bucket.new_key(f['ErrorDocument_Key'])
    errorstring = choose_bucket_prefix(template=ERRORDOC_TEMPLATE, max_len=256)
    errorhtml.set_contents_from_string(errorstring)
    errorhtml.set_canned_acl('private')
    #time.sleep(1)
    while bucket.get_key(f['ErrorDocument_Key']) is None:
        time.sleep(SLEEP_INTERVAL)

    res = _website_request(bucket.name, '')
    body = res.read()
    print(body)
    _website_expected_error_response(res, bucket.name, 403, 'Forbidden', 'AccessDenied', content=_website_expected_default_html(Code='AccessDenied'), body=body)
    ok(errorstring not in body, 'error content should match error.html set content')

    indexhtml.delete()
    errorhtml.delete()
    bucket.delete()

 def test_is_running(self):
     """
     Test ``Service.is_running``.
     """
     service = TimedService(1)
     ok(not service.is_running())
     start(service)
     ok(service.is_running())

def test_name():
    vim.command("new")
    eq(vim.current.buffer.name, "")
    new_name = vim.eval("tempname()")
    vim.current.buffer.name = new_name
    eq(vim.current.buffer.name, new_name)
    vim.command("w!")
    ok(os.path.isfile(new_name))
    os.unlink(new_name)

def test_command():
    fname = tempfile.mkstemp()[1]
    vim.command("new")
    vim.command("edit %s" % fname)
    vim.command("normal itesting\npython\napi")
    vim.command("w")
    ok(os.path.isfile(fname))
    eq(file(fname).read(), "testing\npython\napi\n")
    os.unlink(fname)

 def dependency_order(self):
     with Workspace() as bit:
         with bit.test as test: pass
         with bit.test2 as test: pass
         bit.test << bit.test2
         ok('test2' not in bit.dependencies)
         ok('test2' not in bit.order)
         ok('test2' in bit.test.dependencies)
         ok('test2' in bit.test.order)
         ok(isinstance(bit.test.test2, Target))

def __website_expected_reponse_status(res, status, reason):
    if not isinstance(status, collections.Container):
        status = set([status])
    if not isinstance(reason, collections.Container):
        reason = set([reason])

    if status is not IGNORE_FIELD:
        ok(res.status in status, 'HTTP code was %s should be %s' % (res.status, status))
    if reason is not IGNORE_FIELD:
        ok(res.reason in reason, 'HTTP reason was was %s should be %s' % (res.reason, reason))

 def spawn(self):
     with Target('test', None) as target:
         target.cache = 'build'
         with target.task as task:
             eq(task.name, 'task')
         with target.task('words') as task:
             eq(task.name, 'words')
         ok('words' in target.dependencies)
         ok('task' in target.dependencies)
         eq(target.task, target.dependencies['words'])
         eq(target.task, target.dependencies['task'])

def test_command():
    fname = tempfile.mkstemp()[1]
    vim.command('new')
    vim.command('edit %s' % fname)
    # skip the "press return" state, which does not handle deferred calls
    vim.input('\r')
    vim.command('normal itesting\npython\napi')
    vim.command('w')
    ok(os.path.isfile(fname))
    eq(open(fname).read(), 'testing\npython\napi\n')
    os.unlink(fname)

def test_get_mirrors_with_all():
    cfg = get_config()
    mirrors = list(mirror.get_mirrors(cfg, "baz"))
    eq(0, len(mirrors))

    cfg.add_section("mirror github")
    cfg.set("mirror github", "repos", "@all")
    cfg.set("mirror github", "uri", "[email protected]:res0nat0r/%s.git")
    mirrors = list(mirror.get_mirrors(cfg, "baz"))
    ok("[email protected]:res0nat0r/baz.git" in mirrors)
    eq(1, len(mirrors))

def test_position():
    height = vim.windows[0].height
    width = vim.windows[0].width
    vim.command('split')
    vim.command('vsplit')
    eq((vim.windows[0].row, vim.windows[0].col), (0, 0))
    vsplit_pos = width / 2
    split_pos = height / 2
    eq(vim.windows[1].row, 0)
    ok(vsplit_pos - 1 <= vim.windows[1].col <= vsplit_pos + 1)
    ok(split_pos - 1 <= vim.windows[2].row <= split_pos + 1)
    eq(vim.windows[2].col, 0)

 def test_list_default_queue(self):
     '''
     Test output of ``jobs list`` for default queue.
     '''
     job = self.enqueue()
     stdout = paster(u'jobs', u'list')[1]
     fields = stdout.split()
     eq(len(fields), 3)
     dt = datetime.datetime.strptime(fields[0], u'%Y-%m-%dT%H:%M:%S')
     now = datetime.datetime.utcnow()
     ok(abs((now - dt).total_seconds()) < 10)
     eq(fields[1], job.id)
     eq(fields[2], jobs.DEFAULT_QUEUE_NAME)

def validate_response(r, **options):
    " Validate webtest response "

    # Validate: status, status_int, status_code
    ok(r.status.startswith("%d " % r.status_int))
    ok(r.status_int >= 100)
    eq(r.status_int, r.status_code)
    eq(r.status_int, options.get("status_int", 200))

    # Validate: headers
    " Ref: https://en.wikipedia.org/wiki/ASCII#ASCII_printable_characters "
    message = "Must be printable ASCII characters: %s"
    for key, value in r.headers.iteritems():
        ok(re.match("^[\x20-\x7e]*$", value), message % repr(value))
    # endfold

    eq(r.content_type, options.get("content_type", "text/plain"))

    # Validate: status: 200
    if r.status_int == 200:
        ok("location" not in r.headers)

    # Validate: status: 301, 302
    if r.status_int in [301, 302]:
        eq(urllib.unquote(r.headers["location"]), options["location"])
        eq(r.normal_body, "")

 def test_comments(self):
     now = datetime.datetime.now()
     now_ts = int(time.mktime(now.timetuple()))
     before_ts = int(time.mktime((now - datetime.timedelta(minutes=15)).timetuple()))
     message = 'test message ' + str(now_ts)
     comment_id = dog.Comment.create(handle=TEST_USER, message=message)['comment']['id']
     time.sleep(self.wait_time)
     event = dog.Event.get(comment_id)
     eq(event['event']['text'], message)
     dog.Comment.update(comment_id, handle=TEST_USER, message=message + ' updated')
     time.sleep(self.wait_time)
     event = dog.Event.get(comment_id)
     eq(event['event']['text'], message + ' updated')
     reply_id = dog.Comment.create(handle=TEST_USER, message=message + ' reply',
                                   related_event_id=comment_id)['comment']['id']
     time.sleep(3)
     stream = dog.Event.query(start=before_ts, end=now_ts + 100)['events']
     ok(stream is not None, msg="No events found in stream")
     ok(isinstance(stream, list), msg="Event stream is not a list")
     ok(len(stream) > 0, msg="No events found in stream")
     comment_ids = [x['id'] for x in stream[0]['comments']]
     ok(reply_id in comment_ids,
        msg="Should find {0} in {1}".format(reply_id, comment_ids))
     # Delete the reply
     dog.Comment.delete(reply_id)
     # Then the post itself
     dog.Comment.delete(comment_id)
     time.sleep(self.wait_time)
     try:
         dog.Event.get(comment_id)
     except:
         pass
     else:
         assert False

def test_getAliasRepos():
    cfg = RawConfigParser()
    cfg.add_section('gitosis')
    cfg.set('gitosis', 'foo', 'bar baz')
    repos = list(util.getAliasRepositories(cfg, 'foo'))
    ok('bar' in repos)
    ok('baz' in repos)
    eq(2, len(repos))
    
    cfg.set('gitosis', 'foobar', '@foo foo')
    repos = list(util.getAliasRepositories(cfg, 'foobar'))
    ok('bar' in repos)
    ok('baz' in repos)
    ok('foo' in repos)
    eq(3, len(repos))

def test_list_runtime_paths():
    # Is this the default runtime path list?
    homedir = os.environ['HOME'] + '/.nvim'
    dflt_rtp = [
        homedir,
        '/usr/local/share/nvim/vimfiles',
        '/usr/local/share/nvim',
        '/usr/local/share/nvim/vimfiles/after'
    ]
    # If the runtime is installed the default path
    # is nvim/runtime
    dflt_rtp2 = list(dflt_rtp)
    dflt_rtp2[2] += '/runtime'

    rtp = vim.list_runtime_paths()
    ok(rtp == dflt_rtp or rtp == dflt_rtp2)