# Path: loom/test/util.py
# def for_each_dataset(fun):
#     @functools.wraps(fun)
#     def test_one(dataset):
#         paths = loom.store.get_paths(dataset, sample_count=2)
#         for key, path in loom.store.iter_paths(dataset, paths):
#             if not os.path.exists(path):
#                 raise ValueError(
#                     'missing {} at {},\n  first `python -m loom.datasets test`'
#                     .format(key, path))
#         kwargs = {'name': dataset}
#         kwargs.update(paths['query'])
#         kwargs.update(paths['consensus'])
#         kwargs.update(paths['samples'][0])
#         kwargs.update(paths['ingest'])
#         kwargs.update(paths)
#         fun(**kwargs)
# 
#     @functools.wraps(fun)
#     def test_all():
#         for dataset in loom.datasets.TEST_CONFIGS:
#             yield test_one, dataset
# 
#     return test_all
# 
# Path: loom/test/util.py
# CLEANUP_ON_ERROR = int(os.environ.get('CLEANUP_ON_ERROR', 1))
# 
# Path: loom/test/util.py
# def assert_found(*filenames):
#     for name in filenames:
#         assert_true(os.path.exists(name), 'missing file: {}'.format(name))
# 
# Path: loom/test/util.py
# def load_rows(filename):
#     rows = []
#     for string in protobuf_stream_load(filename):
#         row = Row()
#         row.ParseFromString(string)
#         rows.append(row)
#     return rows


import os
import loom.format
import loom.util
from nose.tools import assert_equal
from distributions.fileutil import tempdir
from distributions.io.stream import protobuf_stream_load
from distributions.tests.util import assert_close
from loom.test.util import for_each_dataset
from loom.test.util import CLEANUP_ON_ERROR
from loom.test.util import assert_found
from loom.test.util import load_rows
def test_import_rows(encoding, rows, rows_csv, **unused):
    with tempdir(cleanup_on_error=CLEANUP_ON_ERROR):
        rows_pbs = os.path.abspath('rows.pbs.gz')
        loom.format.import_rows(
            encoding_in=encoding,
            rows_csv_in=rows_csv,
            rows_out=rows_pbs)
        assert_found(rows_pbs)
        expected_count = sum(1 for _ in protobuf_stream_load(rows))
        actual_count = sum(1 for _ in protobuf_stream_load(rows_pbs))
        assert_equal(actual_count, expected_count)


@for_each_dataset
def test_export_rows(encoding, rows, **unused):
    with tempdir(cleanup_on_error=CLEANUP_ON_ERROR):
        rows_csv = os.path.abspath('rows_csv')
        rows_pbs = os.path.abspath('rows.pbs.gz')
        loom.format.export_rows(
            encoding_in=encoding,
            rows_in=rows,
            rows_csv_out=rows_csv,
            chunk_size=51)
        assert_found(rows_csv)
        assert_found(os.path.join(rows_csv, 'rows.0.csv.gz'))
        loom.format.import_rows(
            encoding_in=encoding,
            rows_csv_in=rows_csv,
            rows_out=rows_pbs)
        assert_found(rows_pbs)

# Path: loom/test/util.py
# def for_each_dataset(fun):
#     @functools.wraps(fun)
#     def test_one(dataset):
#         paths = loom.store.get_paths(dataset, sample_count=2)
#         for key, path in loom.store.iter_paths(dataset, paths):
#             if not os.path.exists(path):
#                 raise ValueError(
#                     'missing {} at {},\n  first `python -m loom.datasets test`'
#                     .format(key, path))
#         kwargs = {'name': dataset}
#         kwargs.update(paths['query'])
#         kwargs.update(paths['consensus'])
#         kwargs.update(paths['samples'][0])
#         kwargs.update(paths['ingest'])
#         kwargs.update(paths)
#         fun(**kwargs)
# 
#     @functools.wraps(fun)
#     def test_all():
#         for dataset in loom.datasets.TEST_CONFIGS:
#             yield test_one, dataset
# 
#     return test_all
# 
# Path: loom/test/util.py
# CLEANUP_ON_ERROR = int(os.environ.get('CLEANUP_ON_ERROR', 1))
# 
# Path: loom/test/util.py
# def assert_found(*filenames):
#     for name in filenames:
#         assert_true(os.path.exists(name), 'missing file: {}'.format(name))
# 
# Path: loom/test/util.py
# def load_rows(filename):
#     rows = []
#     for string in protobuf_stream_load(filename):
#         row = Row()
#         row.ParseFromString(string)
#         rows.append(row)
#     return rows


# Path: loom/test/test_format.py
import os
import loom.format
import loom.util
from nose.tools import assert_equal
from distributions.fileutil import tempdir
from distributions.io.stream import protobuf_stream_load
from distributions.tests.util import assert_close
from loom.test.util import for_each_dataset
from loom.test.util import CLEANUP_ON_ERROR
from loom.test.util import assert_found
from loom.test.util import load_rows

def test_import_rows(encoding, rows, rows_csv, **unused):
    with tempdir(cleanup_on_error=CLEANUP_ON_ERROR):
        rows_pbs = os.path.abspath('rows.pbs.gz')
        loom.format.import_rows(
            encoding_in=encoding,
            rows_csv_in=rows_csv,
            rows_out=rows_pbs)
        assert_found(rows_pbs)
        expected_count = sum(1 for _ in protobuf_stream_load(rows))
        actual_count = sum(1 for _ in protobuf_stream_load(rows_pbs))
        assert_equal(actual_count, expected_count)


@for_each_dataset
def test_export_rows(encoding, rows, **unused):
    with tempdir(cleanup_on_error=CLEANUP_ON_ERROR):
        rows_csv = os.path.abspath('rows_csv')
        rows_pbs = os.path.abspath('rows.pbs.gz')
        loom.format.export_rows(
            encoding_in=encoding,
            rows_in=rows,
            rows_csv_out=rows_csv,
            chunk_size=51)
        assert_found(rows_csv)
        assert_found(os.path.join(rows_csv, 'rows.0.csv.gz'))
        loom.format.import_rows(
            encoding_in=encoding,
            rows_csv_in=rows_csv,
            rows_out=rows_pbs)
        assert_found(rows_pbs)

        expected = load_rows(rows)