from __future__ import absolute_import, division, print_function

import pytest
pytest.importorskip('dask')

from odo.backends.dask import append, Array, merge
from dask.array.core import insert_to_ooc
from dask import core
from odo import convert, into
from odo.utils import tmpfile
import numpy as np
import bcolz


def eq(a, b):
    c = a == b
    if isinstance(c, np.ndarray):
        c = c.all()
    return c


def test_convert():
    x = np.arange(600).reshape((20, 30))
    d = convert(Array, x, blockshape=(4, 5))

    assert isinstance(d, Array)


def test_convert_to_numpy_array():
    x = np.arange(600).reshape((20, 30))
    d = convert(Array, x, blockshape=(4, 5))
    x2 = convert(np.ndarray, d)

    assert eq(x, x2)


def test_append_to_array():
    x = np.arange(600).reshape((20, 30))
    a = into(Array, x, blockshape=(4, 5))
    b = bcolz.zeros(shape=(0, 30), dtype=x.dtype)

    append(b, a)
    assert eq(b[:], x)

    with tmpfile('hdf5') as fn:
        h = into(fn+'::/data', a)
        assert eq(h[:], x)
        h.file.close()


def test_into_inplace():
    x = np.arange(600).reshape((20, 30))
    a = into(Array, x, blockshape=(4, 5))
    b = bcolz.zeros(shape=(20, 30), dtype=x.dtype)

    append(b, a, inplace=True)
    assert eq(b[:], x)


def test_insert_to_ooc():
    x = np.arange(600).reshape((20, 30))
    y = np.empty(shape=x.shape, dtype=x.dtype)
    a = convert(Array, x, blockshape=(4, 5))

    dsk = insert_to_ooc(y, a)
    core.get(merge(dsk, a.dask), list(dsk.keys()))

    assert eq(y, x)


def test__array__():
    x = np.arange(600).reshape((20, 30))
    d = convert(Array, x, blockshape=(4, 5))

    assert eq(x, np.array(d))

@@ -143,15 +143,8 @@
 end,
- Array,
  mer
@@ -187,16 +187,53 @@
 t_to_ooc
+, Array%0Afrom dask.bag.core import Bag
 %0Afrom da
@@ -262,16 +262,24 @@
 o import
+ chunks,
  convert
@@ -284,16 +284,26 @@
 rt, into
+, TextFile
 %0Afrom od
@@ -324,16 +324,27 @@
  tmpfile
+, filetexts
 %0Aimport 
@@ -359,21 +359,71 @@
  np%0A
-import bcolz%0A
+%0A####################%0A# dask.array tests #%0A####################
 %0A%0Ade
@@ -856,32 +856,73 @@
 end_to_array():%0A
+    bcolz = pytest.importorskip('bcolz')%0A
     x = np.arang
@@ -1219,24 +1219,65 @@
 _inplace():%0A
+    bcolz = pytest.importorskip('bcolz')%0A
     x = np.a
@@ -1857,8 +1857,737 @@
 ray(d))%0A
+%0A##################%0A# dask.bag tests #%0A##################%0A%0A%0Adef inc(x):%0A    return x + 1%0A%0Adsk = %7B('x', 0): (range, 5),%0A       ('x', 1): (range, 5),%0A       ('x', 2): (range, 5)%7D%0A%0AL = list(range(5)) * 3%0A%0Ab = Bag(dsk, 'x', 3)%0A%0Adef test_convert_bag_to_list():%0A    assert convert(list, b) == L%0A%0Adef test_convert_logfiles_to_bag():%0A    with filetexts(%7B'a1.log': 'Hello%5CnWorld', 'a2.log': 'Hola%5CnMundo'%7D) as fns:%0A        logs = chunks(TextFile)(list(map(TextFile, fns)))%0A        b = convert(Bag, logs)%0A        assert isinstance(b, Bag)%0A        assert 'a1.log' in str(b.dask.values())%0A        assert convert(list, b) == convert(list, logs)%0A%0A%0Adef test_sequence():%0A    b = into(Bag, %5B1, 2, 3%5D)%0A    assert set(b.map(inc)) == set(%5B2, 3, 4%5D)%0A

Add dask.bag.into tests