def test_to_block(self):
        block = Block()
        block.from_list(range(1, 6))

        self.pointer.address = 0xabcdef
        self.pointer.to_block(block, 1)
        assert_list_equal(block[0:5].to_list(), [1, 0xef, 0xcd, 0xab, 5])

def test_write_2bpp_graphic_to_block_offset_xy():
    source = [[0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
              [0, 0, 2, 1, 2, 3, 2, 1, 2, 1],
              [0, 0, 2, 3, 1, 0, 2, 3, 2, 2],
              [0, 0, 3, 0, 3, 2, 2, 2, 0, 2],
              [0, 0, 1, 3, 3, 0, 2, 0, 2, 3],
              [0, 0, 1, 0, 1, 1, 0, 3, 3, 3],
              [0, 0, 1, 3, 3, 3, 3, 2, 1, 2],
              [0, 0, 2, 2, 3, 1, 2, 2, 1, 0],
              [0, 0, 2, 0, 3, 3, 2, 3, 1, 0],
              [0, 0, 0, 0, 0, 0, 0, 0, 0, 0]]
    target = Block()
    target.from_list([0xff] * 18)
    assert_equal(16, write_2bpp_graphic_to_block(source=source, target=target, offset=1, x=2, y=1, bit_offset=0))
    assert_list_equal(target.to_list(),
                      [0xff,
                       0b01010101,
                       0b10111010,
                       0b01100100,
                       0b11001111,
                       0b10100000,
                       0b10111101,
                       0b11100001,
                       0b01101011,
                       0b10110111,
                       0b00000111,
                       0b11111010,
                       0b01111101,
                       0b00110010,
                       0b11101100,
                       0b00110110,
                       0b10111100,
                       0xff])

    def test_pandas_series_loading(self):
        """Pandas Series objects are correctly loaded"""
        # Test valid series types
        name = ['_x', ' name']
        length = [0, 1, 2]
        index_key = [None, 'ix', 1]
        index_types = ['int', 'char', 'datetime', 'Timestamp']
        value_key = [None, 'x', 1]
        value_types = ['int', 'char', 'datetime', 'Timestamp', 'float',
                       'numpy float', 'numpy int']

        series_info = product(name, length, index_key, index_types,
                              value_key, value_types)

        for n, l, ikey, itype, vkey, vtype in series_info:
            index = sequences[itype](l)
            series = pd.Series(sequences[vtype](l), index=index, name=n,)

            vkey = vkey or series.name
            expected = [{'idx': Data.serialize(i), 'col': vkey,
                         'val': Data.serialize(v)}
                        for i, v in zip(index, series)]

            data = Data.from_pandas(series, name=n, series_key=vkey)
            nt.assert_list_equal(expected, data.values)
            nt.assert_equal(n, data.name)
            data.to_json()

        # Missing a name
        series = pd.Series(np.random.randn(10))
        data = Data.from_pandas(series)
        nt.assert_equal(data.name, 'table')

 def test_init3(self):
     # partial initialization
     expected_result = TestLine.expected_result
     line = spectro.Line(restwlen=1.282,
                         redshift=1., name='Pa_beta')
     result = [line.restwlen,line.obswlen,line.redshift,line.name]
     assert_list_equal(result, expected_result)

def test_read_2bpp_graphic_from_block_offset_xy():
    source = Block()
    source.from_list([0b01010101,
                      0b10111010,
                      0b01100100,
                      0b11001111,
                      0b10100000,
                      0b10111101,
                      0b11100001,
                      0b01101011,
                      0b10110111,
                      0b00000111,
                      0b11111010,
                      0b01111101,
                      0b00110010,
                      0b11101100,
                      0b00110110,
                      0b10111100, 5])
    target = [[0 for x in range(10)] for y in range(10)]
    assert_equal(16, read_2bpp_graphic_from_block(target=target, source=source, offset=0, x=2, y=1, bit_offset=0))
    assert_list_equal(target,
                      [[0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                       [0, 0, 2, 1, 2, 3, 2, 1, 2, 1],
                       [0, 0, 2, 3, 1, 0, 2, 3, 2, 2],
                       [0, 0, 3, 0, 3, 2, 2, 2, 0, 2],
                       [0, 0, 1, 3, 3, 0, 2, 0, 2, 3],
                       [0, 0, 1, 0, 1, 1, 0, 3, 3, 3],
                       [0, 0, 1, 3, 3, 3, 3, 2, 1, 2],
                       [0, 0, 2, 2, 3, 1, 2, 2, 1, 0],
                       [0, 0, 2, 0, 3, 3, 2, 3, 1, 0],
                       [0, 0, 0, 0, 0, 0, 0, 0, 0, 0]])

def test_comparator():
    def comparator(a, b):
        a = a.lower()
        b = b.lower()
        if a < b:
            return -1
        if a > b:
            return 1
        else:
            return 0

    comparator_name = b"CaseInsensitiveComparator"

    with tmp_db('comparator', create=False) as name:
        db = DB(name,
                create_if_missing=True,
                comparator=comparator,
                comparator_name=comparator_name)

        keys = [
            b'aaa',
            b'BBB',
            b'ccc',
        ]

        with db.write_batch() as wb:
            for key in keys:
                wb.put(key, b'')

        assert_list_equal(
            sorted(keys, key=lambda s: s.lower()),
            list(db.iterator(include_value=False)))

 def test_add_records_to_table2(self):
     expected_result = [2,[TestObsTable.obsrecord,TestObsTable.obsrecord]]
     TestObsTable.obstable.add_records_to_table([TestObsTable.obsrecord,TestObsTable.obsrecord])
     result = []
     result.append(TestObsTable.obstable.length)
     result.append(TestObsTable.obstable.records)
     assert_list_equal(result, expected_result)

 def test_empty(self):
     parsed = self.p.parse_args([])
     n.assert_false(parsed.urls)
     n.assert_false(parsed.force_colnames)
     n.assert_false(parsed.silent)
     n.assert_in('.pluplusch', parsed.cache_dir)
     n.assert_list_equal(parsed.catalog, [])

 def test_wqstd(self):
     nt.assert_true(isinstance(self.db.wqstd, pandas.DataFrame))
     nt.assert_tuple_equal(self.db.wqstd.shape, (48, 4))
     nt.assert_list_equal(
         ['parameter', 'units', 'lower_limit', 'upper_limit'],
         self.db.wqstd.columns.tolist()
     )

    def test_plain_attachment(self, gpg):
        attachments = [("blabla", None)]

        remaining_attachments = import_public_keys_from_attachments(gpg, attachments)

        assert_list_equal(attachments, remaining_attachments)
        assert not gpg.import_keys.called

def test_set_config(app1, cli, td):
    dct = {
        '1': '1',
        2: 2,
        1.1: 1.1,
        'a': 1,
        3: ['1', 2],
        4: {'a': 1, '1': '1', 2: 2}}

    # verify set_config does not delete old keys
    nt.assert_in('key1', td[app1])
    nt.assert_equal(1, td[app1]['key1'])
    td = RedisMapping()  # reset's RedisMapping's cache, from prev line
    set_config(app1, dct, cli=cli)
    nt.assert_in('key1', td[app1])
    nt.assert_equal(1, td[app1]['key1'])

    # verify set_config adds new keys
    nt.assert_equal(
        len(list(td[app1].keys())), len(list(dct.keys()) + ['key1']))
    nt.assert_true(all(
        x in (list(dct.keys()) + ['key1']) for x in td[app1].keys()))
    nt.assert_equal('1', td[app1]['1'])
    nt.assert_equal(2, td[app1][2])
    nt.assert_equal(1.1, td[app1][1.1])
    nt.assert_equal(1, td[app1]['a'])
    nt.assert_is_instance(
        td[app1][3], JSONSequence)
    nt.assert_is_instance(
        td[app1][4], JSONMapping)

    nt.assert_list_equal(list(td[app1][3]), dct[3])
    nt.assert_dict_equal(dict(td[app1][4]), dct[4])

    def test_profile_table_for_join_with_profile_attrs(self):
        profile_output = profile_table_for_join(self.table, ['attr'])

        expected_output_attrs = ['Unique values', 'Missing values', 'Comments']
        # verify whether the output dataframe has the necessary attributes.
        assert_list_equal(list(profile_output.columns.values),
                          expected_output_attrs)

        expected_unique_column = ['4 (80.0%)']
        # verify whether correct values are present in 'Unique values' column.
        assert_list_equal(list(profile_output['Unique values']),
                          expected_unique_column)

        expected_missing_column = ['1 (20.0%)']
        # verify whether correct values are present in 'Missing values' column.
        assert_list_equal(list(profile_output['Missing values']),
                          expected_missing_column)

        expected_comments = ['Joining on this attribute will ignore 1 (20.0%) rows.']
        # verify whether correct values are present in 'Comments' column.
        assert_list_equal(list(profile_output['Comments']),
                          expected_comments)

        # verify whether index name is set correctly in the output dataframe.
        assert_equal(profile_output.index.name, 'Attribute')

        expected_index_column = ['attr']
        # verify whether correct values are present in the dataframe index.
        assert_list_equal(list(profile_output.index.values),
                          expected_index_column)

def test_gzipped_files_are_iterable_as_normal():
    agz = _make_temporary_gzip(pybedtools.example_filename('a.bed'))
    agz = pybedtools.BedTool(agz)
    a = pybedtools.example_bedtool('a.bed')
    for i in agz:
        print(i)
    assert_list_equal(list(a), list(agz))

def test_simple_case():
    a = []
    b = [a]
    c = [a, b]
    for perm in permutations([a, b, c]):
        result = list(toposorted(perm, lambda x: x))
        assert_list_equal([a, b, c], result)

    def test_should_create_varnish_api_for_connected_servers(self):
        expected_construct_args = [
            call(['127.0.0.1', '6082', 1.0], 'secret-1'),
            call(['127.0.0.2', '6083', 1.0], 'secret-2'),
            call(['127.0.0.3', '6084', 1.0], 'secret-3')]
        sample_extractor = Mock(servers=servers)

        api_init_side_effect = {
            'secret-1': Exception(),
            'secret-2': None,
            'secret-3': None
        }

        with patch('vaas.cluster.cluster.ServerExtractor', Mock(return_value=sample_extractor)):
            with patch.object(
                VarnishApi, '__init__', side_effect=lambda host_port_timeout, secret: api_init_side_effect[secret]
            ) as construct_mock:
                with patch('telnetlib.Telnet.close', Mock()):
                    varnish_cluster = VarnishApiProvider()
                    api_objects = []
                    for api in varnish_cluster.get_connected_varnish_api():
                        """
                        Workaround - we cannot mock __del__ method:
                        https://docs.python.org/3/library/unittest.mock.html

                        We inject sock field to eliminate warning raised by cleaning actions in __del__ method
                        """
                        api.sock = None
                        api_objects.append(api)

                    assert_equals(2, len(api_objects))
                    assert_list_equal(expected_construct_args, construct_mock.call_args_list)

 def test_read_labels(self, mock_dat):
     
     # mock methods and properties of Datum objects
     mock_dat.return_value.ParseFromString.return_value = ""
     type(mock_dat.return_value).label = PropertyMock(side_effect=range(5))
     
     assert_list_equal(r.read_labels(self.path_lmdb), range(5))

 def test_learning_curve_from_dir(self):
     
     lc = LearningCurveFromPath(os.path.split(self.fpath)[0])
     assert_is_not_none(lc)
     train_keys, test_keys = lc.parse()
     assert_list_equal(train_keys, ['NumIters', 'Seconds', 'LearningRate', 'loss'])
     assert_list_equal(test_keys, ['NumIters', 'Seconds', 'LearningRate', 'accuracy', 'loss'])

    def test_getting_uncompleted_todos_when_todos_is_not_none(self):
        todo1 = {
            'userId': 1,
            'id': 1,
            'title': 'Make the bed',
            'completed': False
        }
        todo2 = {
            'userId': 2,
            'id': 2,
            'title': 'Walk the dog',
            'completed': True
        }

        # Configure mock to return a response with a JSON-serialized list of todos.
        self.mock_get_todos.return_value = Mock()
        self.mock_get_todos.return_value.json.return_value = [todo1, todo2]

        # Call the service, which will get a list of todos filtered on completed.
        uncompleted_todos = get_uncompleted_todos()

        # Confirm that the mock was called.
        assert_true(self.mock_get_todos.called)

        # Confirm that the expected filtered list of todos was returned.
        assert_list_equal(uncompleted_todos, [todo1])

    def test_split(self):
        
        h5_list = to.split_hdf5(self.fpath, self.dir_tmp, tot_floats=((3*4*2*3)))
        #hdf5_list = to.split_hdf5(self.fpath, self.dir_tmp, tot_floats=((10*4*2*3)))
        #hdf5_list = to.split_hdf5(self.fpath, self.dir_tmp, tot_floats=((1*4*2*3)))
        
        assert_is_instance(h5_list, list)
        assert_equals(len(h5_list), 2)
        
        name_, ext = os.path.splitext(os.path.basename(self.fpath)) 

        for p in h5_list:
            assert_in(name_, p)
            assert_true(p.endswith(ext), "Unexpected extension")
            
        offset = 0
        with h5py.File(self.fpath, 'r') as h_src:
            for p in h5_list:
                with h5py.File(p, 'r') as h:
                    assert_list_equal(['x1', 'x2'], h.keys())
                    
                    for k in h.keys():
                        min_len = min(len(h[k]), len(h_src[k]))
                        sub_actual = h[k][0:min_len]
                        sub_expected = h_src[k][offset:offset+min_len]
                        assert_true(np.all(sub_actual==sub_expected))
            
                    offset += min_len

def test_load():
    """backends.load(): works as expected.

    This is an interesting function to test, because it is just a wrapper that
    returns a TestrunResult object. So most of the testing should be happening
    in the tests for each backend.

    However, we can test this by injecting a fake backend, and ensuring that we
    get back what we expect. What we do is inject list(), which menas that we
    should get back [file_path].

    """
    backends.BACKENDS['test_backend'] = backends.register.Registry(
        extensions=['.test_extension'],
        backend=None,
        load=lambda x, y: [x],  # y is for a compression value
        meta=None,
    )

    file_path = 'foo.test_extension'
    with open(file_path, 'w') as f:
        f.write('foo')

    test = backends.load(file_path)
    nt.assert_list_equal([file_path], test)