def test_lazy_load_index():
    f = StringIO()
    dump({'wakka': 42}, f)
    f.seek(0)
    lj = LazyJSON(f)
    assert_equal({'wakka': 10, '__total__': 0}, lj.offsets)
    assert_equal({'wakka': 2, '__total__': 14}, lj.sizes)

    def test_default_diverging_vlims(self):

        p = mat._HeatMapper(self.df_norm, **self.default_kws)
        vlim = max(abs(self.x_norm.min()), abs(self.x_norm.max()))
        nt.assert_equal(p.vmin, -vlim)
        nt.assert_equal(p.vmax, vlim)
        nt.assert_true(p.divergent)

 def test_tickabels_off(self):
     kws = self.default_kws.copy()
     kws['xticklabels'] = False
     kws['yticklabels'] = False
     p = mat._HeatMapper(self.df_norm, **kws)
     nt.assert_equal(p.xticklabels, [])
     nt.assert_equal(p.yticklabels, [])

    def test_fetch_library_name_personal(self):
        self.node_settings.library_id = 'personal'

        assert_equal(
            self.node_settings.fetch_library_name,
            'My library'
        )

    def test_selected_library_name_empty(self):
        self.node_settings.library_id = None

        assert_equal(
            self.node_settings.fetch_library_name,
            ''
        )

def test_fail_fetch_haxby_simple():
    # Test a dataset fetching failure to validate sandboxing
    local_url = "file://" + os.path.join(datadir, "pymvpa-exampledata.tar.bz2")
    datasetdir = os.path.join(tmpdir, 'haxby2001_simple', 'pymvpa-exampledata')
    os.makedirs(datasetdir)
    # Create a dummy file. If sandboxing is successful, it won't be overwritten
    dummy = open(os.path.join(datasetdir, 'attributes.txt'), 'w')
    dummy.write('stuff')
    dummy.close()

    path = 'pymvpa-exampledata'

    opts = {'uncompress': True}
    files = [
        (os.path.join(path, 'attributes.txt'), local_url, opts),
        # The following file does not exists. It will cause an abortion of
        # the fetching procedure
        (os.path.join(path, 'bald.nii.gz'), local_url, opts)
    ]

    assert_raises(IOError, utils._fetch_files,
                  os.path.join(tmpdir, 'haxby2001_simple'), files,
                  verbose=0)
    dummy = open(os.path.join(datasetdir, 'attributes.txt'), 'r')
    stuff = dummy.read(5)
    dummy.close()
    assert_equal(stuff, 'stuff')

def test_subproc_hello_mom_second():
    fst = "echo 'hello'"
    sec = "echo 'mom'"
    s = '{0}; {1}'.format(fst, sec)
    exp = '{0}; $[{1}]'.format(fst, sec)
    obs = subproc_toks(s, lexer=LEXER, mincol=len(fst), returnline=True)
    assert_equal(exp, obs)

def test_subproc_hello_mom_first():
    fst = "echo 'hello'"
    sec = "echo 'mom'"
    s = '{0}; {1}'.format(fst, sec)
    exp = '$[{0}]; {1}'.format(fst, sec)
    obs = subproc_toks(s, lexer=LEXER, maxcol=len(fst)+1, returnline=True)
    assert_equal(exp, obs)

def test_unicode_decode_error():
    # decode_error default to strict, so this should fail
    # First, encode (as bytes) a unicode string.
    text = "J'ai mang\xe9 du kangourou  ce midi, c'\xe9tait pas tr\xeas bon."
    text_bytes = text.encode('utf-8')

    # Then let the Analyzer try to decode it as ascii. It should fail,
    # because we have given it an incorrect encoding.
    wa = CountVectorizer(ngram_range=(1, 2), encoding='ascii').build_analyzer()
    assert_raises(UnicodeDecodeError, wa, text_bytes)

    ca = CountVectorizer(analyzer='char', ngram_range=(3, 6),
                         encoding='ascii').build_analyzer()
    assert_raises(UnicodeDecodeError, ca, text_bytes)

    # Check the old interface
    with warnings.catch_warnings(record=True) as w:
        warnings.simplefilter("always")

        ca = CountVectorizer(analyzer='char', ngram_range=(3, 6),
                             charset='ascii').build_analyzer()
        assert_raises(UnicodeDecodeError, ca, text_bytes)

        assert_equal(len(w), 1)
        assert_true(issubclass(w[0].category, DeprecationWarning))
        assert_true("charset" in str(w[0].message).lower())

def test_wilsonLT_Defaults_FeatureInput1():
    '''Confirm default FeatureInput values.'''
    G = la.input_.Geometry
    load_params = {
        'R': 12e-3,                                    # specimen radius
        'a': 7.5e-3,                                   # support ring radius
        'p': 5,                                        # points/layer
        'P_a': 1,                                      # applied load
        'r': 2e-4,                                     # radial distance from center loading
    }

#     mat_props = {'HA' : [5.2e10, 0.25],
#                  'PSu' : [2.7e9, 0.33],
#                  }

    mat_props = {'Modulus': {'HA': 5.2e10, 'PSu': 2.7e9},
                 'Poissons': {'HA': 0.25, 'PSu': 0.33}}

    '''Find way to compare materials DataFrames and Geo_objects .'''
    actual = dft.FeatureInput
    expected = {
        'Geometry': G('400-[200]-800'),
        'Parameters': load_params,
        'Properties': mat_props,
        'Materials': ['HA', 'PSu'],
        'Model': 'Wilson_LT',
        'Globals': None
    }
    ##del actual['Geometry']
    ##del actual['Materials']
    nt.assert_equal(actual, expected)

def test_recursive_solve():
    rec_sudoku = " ,  ,  ,  , 5, 3,  ,  ,  ;\n" + \
                 "1,  ,  , 6,  ,  ,  ,  , 8;\n" + \
                 " , 5,  ,  ,  , 1,  , 4,  ;\n" + \
                 "4,  ,  ,  , 9,  , 5, 3,  ;\n" + \
                 " ,  , 9, 7,  , 6, 8,  ,  ;\n" + \
                 " , 2, 7,  , 3,  ,  ,  , 6;\n" + \
                 " , 4,  , 1,  ,  ,  , 8,  ;\n" + \
                 "2,  ,  ,  ,  , 7,  ,  , 1;\n" + \
                 " ,  ,  , 3, 2,  ,  ,  ,  ;\n"

    solution = "6, 8, 4, 2, 5, 3, 1, 7, 9;\n" + \
               "1, 9, 3, 6, 7, 4, 2, 5, 8;\n" + \
               "7, 5, 2, 9, 8, 1, 6, 4, 3;\n" + \
               "4, 1, 6, 8, 9, 2, 5, 3, 7;\n" + \
               "5, 3, 9, 7, 1, 6, 8, 2, 4;\n" + \
               "8, 2, 7, 4, 3, 5, 9, 1, 6;\n" + \
               "3, 4, 5, 1, 6, 9, 7, 8, 2;\n" + \
               "2, 6, 8, 5, 4, 7, 3, 9, 1;\n" + \
               "9, 7, 1, 3, 2, 8, 4, 6, 5;\n"

    sudoku = Sudoku()
    sudoku.read_string(rec_sudoku)
    sudoku.recursive_solve()
    assert_equal(str(sudoku), solution)

    def test_equality_encoding_realm_emptyValues(self):
        expected_value = ({
            'oauth_nonce': ['4572616e48616d6d65724c61686176'],
            'oauth_timestamp': ['137131200'],
            'oauth_consumer_key': ['0685bd9184jfhq22'],
            'oauth_something': [' Some Example'],
            'oauth_signature_method': ['HMAC-SHA1'],
            'oauth_version': ['1.0'],
            'oauth_token': ['ad180jjd733klru7'],
            'oauth_empty': [''],
            'oauth_signature': ['wOJIO9A2W5mFwDgiDvZbTSMK/PY='],
            }, 'Examp%20le'
        )
        assert_equal(expected_value, parse_authorization_header_value('''\
            OAuth\
\
            realm="Examp%20le",\
            oauth_consumer_key="0685bd9184jfhq22",\
            oauth_token="ad180jjd733klru7",\
            oauth_signature_method="HMAC-SHA1",\
            oauth_signature="wOJIO9A2W5mFwDgiDvZbTSMK%2FPY%3D",\
            oauth_timestamp="137131200",\
            oauth_nonce="4572616e48616d6d65724c61686176",\
            oauth_version="1.0",\
            oauth_something="%20Some+Example",\
            oauth_empty=""\
        '''), "parsing failed.")

    def _req(cls, method, *args, **kwargs):
        use_token = kwargs.pop('use_token', True)
        token = kwargs.pop('token', None)
        if use_token and token is None:
            admin = kwargs.pop('admin', False)
            if admin:
                if cls._admin_token is None:
                    cls._admin_token = get_auth_token(ADMIN_USERNAME,
                        ADMIN_PASSWORD)
                token = cls._admin_token
            else:
                if cls._token is None:
                    cls._token = get_auth_token(USERNAME, PASSWORD)
                token = cls._token

        if use_token:
            headers = kwargs.get('headers', {})
            headers.setdefault('Authorization', 'Token ' + token)
            kwargs['headers'] = headers

        expected = kwargs.pop('expected', 200)
        resp = requests.request(method, *args, **kwargs)
        if expected is not None:
            if hasattr(expected, '__iter__'):
                assert_in(resp.status_code, expected,
                    "Expected http status in %s, received %s" % (expected,
                        resp.status_code))
            else:
                assert_equal(resp.status_code, expected,
                    "Expected http status %s, received %s" % (expected,
                        resp.status_code))
        return resp

 def test_valid_signature(self):
     for example in self._examples:
         client_shared_secret = example["private_key"]
         client_certificate = example["certificate"]
         public_key = example["public_key"]
         url = example["url"]
         method = example["method"]
         oauth_params = example["oauth_params"]
         expected_signature = example["oauth_signature"]
         # Using the RSA private key.
         assert_equal(expected_signature,
                      generate_rsa_sha1_signature(client_shared_secret,
                                              method=method,
                                              url=url,
                                              oauth_params=oauth_params
                                              )
         )
         # Using the X.509 certificate.
         assert_true(verify_rsa_sha1_signature(
             client_certificate, expected_signature,
             method, url, oauth_params))
         # Using the RSA public key.
         assert_true(verify_rsa_sha1_signature(
             public_key, expected_signature,
             method, url, oauth_params))

 def test_valid_base_string(self):
     base_string = generate_signature_base_string("POST",
                                                   "http://example.com/request?b5=%3D%253D&a3=a&c%40=&a2=r%20b&c2&a3=2+q"
                                                   ,
                                                   self.oauth_params)
     assert_equal(base_string,
                  "POST&http%3A%2F%2Fexample.com%2Frequest&a2%3Dr%2520b%26a3%3D2%2520q%26a3%3Da%26b5%3D%253D%25253D%26c%2540%3D%26c2%3D%26oauth_consumer_key%3D9djdj82h48djs9d2%26oauth_nonce%3D7d8f3e4a%26oauth_signature_method%3DHMAC-SHA1%26oauth_timestamp%3D137131201%26oauth_token%3Dkkk9d7dh3k39sjv7")

 def test_strict_type_guessing_with_large_file(self):
     fh = horror_fobj('211.csv')
     rows = CSVTableSet(fh).tables[0]
     offset, headers = headers_guess(rows.sample)
     rows.register_processor(offset_processor(offset + 1))
     types = [StringType, IntegerType, DecimalType, DateUtilType]
     guessed_types = type_guess(rows.sample, types, True)
     assert_equal(len(guessed_types), 96)
     assert_equal(guessed_types, [
         IntegerType(), StringType(),
         StringType(), StringType(), StringType(), StringType(),
         IntegerType(), StringType(), StringType(), StringType(),
         StringType(), StringType(), StringType(), StringType(),
         StringType(), StringType(), StringType(), StringType(),
         StringType(), StringType(), StringType(), StringType(),
         StringType(), StringType(), StringType(), StringType(),
         StringType(), IntegerType(), StringType(), DecimalType(),
         DecimalType(), StringType(), StringType(), StringType(),
         StringType(), StringType(), StringType(), StringType(),
         StringType(), StringType(), StringType(), StringType(),
         StringType(), StringType(), StringType(), StringType(),
         StringType(), StringType(), StringType(), StringType(),
         StringType(), StringType(), StringType(), StringType(),
         IntegerType(), StringType(), StringType(), StringType(),
         StringType(), StringType(), StringType(), StringType(),
         StringType(), StringType(), StringType(), StringType(),
         StringType(), StringType(), StringType(), StringType(),
         IntegerType(), StringType(), StringType(), StringType(),
         StringType(), StringType(), StringType(), StringType(),
         StringType(), StringType(), StringType(), StringType(),
         StringType(), StringType(), StringType(), StringType(),
         StringType(), StringType(), StringType(), DateUtilType(),
         DateUtilType(), DateUtilType(), DateUtilType(), StringType(),
         StringType(), StringType()])

def test_D8_D4_route():
    """
    Tests the functionality of D4 routing.
    """
    frD8.route_flow()
    frD4.route_flow()
    lfD8.map_depressions()
    lfD4.map_depressions()
    assert_equal(lfD8.number_of_lakes, 1)
    assert_equal(lfD4.number_of_lakes, 3)

    flow_recD8 = np.array([ 0,  1,  2,  3,  4,  5,  6,  7, 16, 10, 16, 10, 18,
                           13, 14, 14, 15, 16, 10, 18, 20, 21, 16, 16, 16, 18,
                           33, 27, 28, 28, 24, 24, 24, 32, 34, 35, 35, 38, 32,
                           32, 32, 41, 42, 43, 44, 45, 46, 47, 48])
    flow_recD4 = np.array([ 0,  1,  2,  3,  4,  5,  6,  7,  7, 10, 17, 10, 11,
                           13, 14, 14, 15, 16, 17, 18, 20, 21, 21, 16, 17, 18,
                           33, 27, 28, 28, 29, 24, 31, 32, 34, 35, 35, 36, 37,
                           32, 33, 41, 42, 43, 44, 45, 46, 47, 48])
    assert_array_equal(mg1.at_node['flow__receiver_node'], flow_recD8)
    assert_array_equal(mg2.at_node['flow__receiver_node'], flow_recD4)
    assert_array_almost_equal(mg1.at_node['drainage_area'].reshape((7,7))[:,
                                  0].sum(),
                              mg2.at_node['drainage_area'].reshape((7,7))[:,
                                  0].sum())

 def test_clear_cache (self):
     self.cache.clear_cache()
     assert_equal(self.cache.cache,
         {"IF"   : {},
          "CONF" : [],
          "EXEC" : []}
         )

def test_Logisticdegenerate():
    X = W((40,10))
    X[:,0] = X[:,1] + X[:,2]
    Y = np.greater(W((40,)), 0)
    cmodel = GLM(design=X, family=family.Binomial())
    results = cmodel.fit(Y)
    assert_equal(results.df_resid, 31)

def test_zero_byte_string():
    # Tests hack to allow chars of non-zero length, but 0 bytes
    # make reader-like thing
    str_io = cStringIO()
    r = _make_readerlike(str_io, boc.native_code)
    c_reader = m5u.VarReader5(r)
    tag_dt = np.dtype([('mdtype', 'u4'), ('byte_count', 'u4')])
    tag = np.zeros((1,), dtype=tag_dt)
    tag['mdtype'] = mio5p.miINT8
    tag['byte_count'] = 1
    hdr = m5u.VarHeader5()
    # Try when string is 1 length
    hdr.set_dims([1,])
    _write_stream(str_io, tag.tostring() + asbytes('        '))
    str_io.seek(0)
    val = c_reader.read_char(hdr)
    assert_equal(val, ' ')
    # Now when string has 0 bytes 1 length
    tag['byte_count'] = 0
    _write_stream(str_io, tag.tostring())
    str_io.seek(0)
    val = c_reader.read_char(hdr)
    assert_equal(val, ' ')
    # Now when string has 0 bytes 4 length
    str_io.seek(0)
    hdr.set_dims([4,])
    val = c_reader.read_char(hdr)
    assert_array_equal(val, [' '] * 4)