<commit_before># -*- coding: utf-8 -*-

"""Test context."""

#------------------------------------------------------------------------------
# Imports
#------------------------------------------------------------------------------

import os

import numpy as np
from numpy.testing import assert_array_equal as ae
from pytest import fixture, yield_fixture, mark
from ipyparallel.tests.clienttest import ClusterTestCase, add_engines

from ..context import Context, _iter_chunks_dask


#------------------------------------------------------------------------------
# Fixtures
#------------------------------------------------------------------------------

@yield_fixture
def context(tempdir):
    ctx = Context('{}/cache/'.format(tempdir))
    yield ctx


#------------------------------------------------------------------------------
# Test context
#------------------------------------------------------------------------------

def test_iter_chunks_dask():
    from dask.array import from_array

    x = np.arange(10)
    da = from_array(x, chunks=(3,))
    assert len(list(_iter_chunks_dask(da))) == 4


def test_context_map(context):
    def f(x):
        return x * x

    args = range(10)
    assert context.map(f, args) == [x * x for x in range(10)]


def test_context_dask(context):
    from dask.array import from_array

    def square(x):
        return x * x

    x = np.arange(10)
    da = from_array(x, chunks=(3,))
    res = context.map_dask_array(square, da)
    ae(res.compute(), x ** 2)


#------------------------------------------------------------------------------
# ipyparallel tests
#------------------------------------------------------------------------------

@yield_fixture(scope='module')
def ipy_client():

    def iptest_stdstreams_fileno():
        return os.open(os.devnull, os.O_WRONLY)

    # OMG-THIS-IS-UGLY-HACK: monkey-patch this global object to avoid
    # using the nose iptest extension (we're using pytest).
    # See https://github.com/ipython/ipython/blob/master/IPython/testing/iptest.py#L317-L319  # noqa
    from ipyparallel import Client
    import ipyparallel.tests
    ipyparallel.tests.nose.iptest_stdstreams_fileno = iptest_stdstreams_fileno

    # Start two engines engine (one is launched by setup()).
    ipyparallel.tests.setup()
    ipyparallel.tests.add_engines(1)
    yield Client(profile='iptest')
    ipyparallel.tests.teardown()


def test_client(ipy_client):
    print(ipy_client.ids)<commit_msg>Done ipy_client fixtures<commit_after># -*- coding: utf-8 -*-

"""Test context."""

#------------------------------------------------------------------------------
# Imports
#------------------------------------------------------------------------------

import os

import numpy as np
from numpy.testing import assert_array_equal as ae
from pytest import yield_fixture

from ..context import Context, _iter_chunks_dask


#------------------------------------------------------------------------------
# Fixtures
#------------------------------------------------------------------------------

@yield_fixture
def context(tempdir):
    ctx = Context('{}/cache/'.format(tempdir))
    yield ctx


#------------------------------------------------------------------------------
# Test context
#------------------------------------------------------------------------------

def test_iter_chunks_dask():
    from dask.array import from_array

    x = np.arange(10)
    da = from_array(x, chunks=(3,))
    assert len(list(_iter_chunks_dask(da))) == 4


def test_context_map(context):
    def f(x):
        return x * x

    args = range(10)
    assert context.map(f, args) == [x * x for x in range(10)]


def test_context_dask(context):
    from dask.array import from_array

    def square(x):
        return x * x

    x = np.arange(10)
    da = from_array(x, chunks=(3,))
    res = context.map_dask_array(square, da)
    ae(res.compute(), x ** 2)


#------------------------------------------------------------------------------
# ipyparallel tests
#------------------------------------------------------------------------------

@yield_fixture(scope='module')
def ipy_client():

    def iptest_stdstreams_fileno():
        return os.open(os.devnull, os.O_WRONLY)

    # OMG-THIS-IS-UGLY-HACK: monkey-patch this global object to avoid
    # using the nose iptest extension (we're using pytest).
    # See https://github.com/ipython/ipython/blob/master/IPython/testing/iptest.py#L317-L319  # noqa
    from ipyparallel import Client
    import ipyparallel.tests
    ipyparallel.tests.nose.iptest_stdstreams_fileno = iptest_stdstreams_fileno

    # Start two engines engine (one is launched by setup()).
    ipyparallel.tests.setup()
    ipyparallel.tests.add_engines(1)
    yield Client(profile='iptest')
    ipyparallel.tests.teardown()


def test_client_1(ipy_client):
    assert ipy_client.ids == [0, 1]


def test_client_2(ipy_client):
    assert ipy_client[:].map_sync(lambda x: x * x, [1, 2, 3]) == [1, 4, 9]
{'fraction_non_alphanumeric': 0.3412474849094567, 'fraction_numerical': 0.01006036217303823, 'mean_word_length': 1.401449275362319, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1532778', 'n_tokens_mistral': 1415, 'n_tokens_neox': 1349, 'n_words': 312}