<s> from collections import namedtuple <EOL> from dask . utils import raises <EOL> from dask . utils_test import GetFunctionTestCase <EOL> from dask import core <EOL> from dask . core import ( istask , get_dependencies , flatten , subs , <EOL> preorder_traversal , quote , _deps ) <EOL> def contains ( a , b ) : <EOL> """<STR_LIT>""" <EOL> return all ( a . get ( k ) == v for k , v in b . items ( ) ) <EOL> def inc ( x ) : <EOL> return x + <NUM_LIT:1> <EOL> def add ( x , y ) : <EOL> return x + y <EOL> def test_istask ( ) : <EOL> assert istask ( ( inc , <NUM_LIT:1> ) ) <EOL> assert not istask ( <NUM_LIT:1> ) <EOL> assert not istask ( ( <NUM_LIT:1> , <NUM_LIT:2> ) ) <EOL> f = namedtuple ( '<STR_LIT:f>' , [ '<STR_LIT:x>' , '<STR_LIT:y>' ] ) <EOL> assert not istask ( f ( sum , <NUM_LIT:2> ) ) <EOL> def test_preorder_traversal ( ) : <EOL> t = ( add , <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> assert list ( preorder_traversal ( t ) ) == [ add , <NUM_LIT:1> , <NUM_LIT:2> ] <EOL> t = ( add , ( add , <NUM_LIT:1> , <NUM_LIT:2> ) , ( add , <NUM_LIT:3> , <NUM_LIT:4> ) ) <EOL> assert list ( preorder_traversal ( t ) ) == [ add , add , <NUM_LIT:1> , <NUM_LIT:2> , add , <NUM_LIT:3> , <NUM_LIT:4> ] <EOL> t = ( add , ( sum , [ <NUM_LIT:1> , <NUM_LIT:2> ] ) , <NUM_LIT:3> ) <EOL> assert list ( preorder_traversal ( t ) ) == [ add , sum , list , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] <EOL> class TestGet ( GetFunctionTestCase ) : <EOL> get = staticmethod ( core . get ) <EOL> def test_GetFunctionTestCase_class ( ) : <EOL> class CustomTestGetFail ( GetFunctionTestCase ) : <EOL> get = staticmethod ( lambda x , y : <NUM_LIT:1> ) <EOL> custom_testget = CustomTestGetFail ( ) <EOL> raises ( AssertionError , custom_testget . test_get ) <EOL> class CustomTestGetPass ( GetFunctionTestCase ) : <EOL> get = staticmethod ( core . get ) <EOL> custom_testget = CustomTestGetPass ( ) <EOL> custom_testget . test_get ( ) <EOL> def test_memoized_get ( ) : <EOL> d = { '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : ( inc , '<STR_LIT>' ) , <EOL> '<STR_LIT>' : ( add , '<STR_LIT>' , '<STR_LIT>' ) } <EOL> try : <EOL> import toolz <EOL> except ImportError : <EOL> return <EOL> cache = dict ( ) <EOL> getm = toolz . memoize ( core . get , cache = cache , <EOL> key = lambda args , kwargs : args [ <NUM_LIT:1> : ] ) <EOL> result = getm ( d , '<STR_LIT>' , get = getm ) <EOL> assert result == <NUM_LIT:3> <EOL> assert contains ( cache , { ( '<STR_LIT>' , ) : <NUM_LIT:1> , <EOL> ( '<STR_LIT>' , ) : <NUM_LIT:2> , <EOL> ( '<STR_LIT>' , ) : <NUM_LIT:3> } ) <EOL> def test_get_laziness ( ) : <EOL> def isconcrete ( arg ) : <EOL> return isinstance ( arg , list ) <EOL> d = { '<STR_LIT:x>' : <NUM_LIT:1> , '<STR_LIT:y>' : <NUM_LIT:2> , '<STR_LIT:z>' : ( isconcrete , [ '<STR_LIT:x>' , '<STR_LIT:y>' ] ) } <EOL> assert core . get ( d , [ '<STR_LIT:x>' , '<STR_LIT:y>' ] ) == ( <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> assert core . get ( d , '<STR_LIT:z>' ) == False <EOL> def test_get_dependencies_nested ( ) : <EOL> dsk = { '<STR_LIT:x>' : <NUM_LIT:1> , '<STR_LIT:y>' : <NUM_LIT:2> , <EOL> '<STR_LIT:z>' : ( add , ( inc , [ [ '<STR_LIT:x>' ] ] ) , '<STR_LIT:y>' ) } <EOL> assert get_dependencies ( dsk , '<STR_LIT:z>' ) == set ( [ '<STR_LIT:x>' , '<STR_LIT:y>' ] ) <EOL> def test_get_dependencies_empty ( ) : <EOL> dsk = { '<STR_LIT:x>' : ( inc , ) } <EOL> assert get_dependencies ( dsk , '<STR_LIT:x>' ) == set ( ) <EOL> def test_get_dependencies_list ( ) : <EOL> dsk = { '<STR_LIT:x>' : <NUM_LIT:1> , '<STR_LIT:y>' : <NUM_LIT:2> , '<STR_LIT:z>' : [ '<STR_LIT:x>' , [ ( inc , '<STR_LIT:y>' ) ] ] } <EOL> assert get_dependencies ( dsk , '<STR_LIT:z>' ) == set ( [ '<STR_LIT:x>' , '<STR_LIT:y>' ] ) <EOL> def test_flatten ( ) : <EOL> assert list ( flatten ( ( ) ) ) == [ ] <EOL> assert list ( flatten ( '<STR_LIT:foo>' ) ) == [ '<STR_LIT:foo>' ] <EOL> def test_subs ( ) : <EOL> assert subs ( ( sum , [ <NUM_LIT:1> , '<STR_LIT:x>' ] ) , '<STR_LIT:x>' , <NUM_LIT:2> ) == ( sum , [ <NUM_LIT:1> , <NUM_LIT:2> ] ) <EOL> assert subs ( ( sum , [ <NUM_LIT:1> , [ '<STR_LIT:x>' ] ] ) , '<STR_LIT:x>' , <NUM_LIT:2> ) == ( sum , [ <NUM_LIT:1> , [ <NUM_LIT:2> ] ] ) <EOL> def test_subs_with_unfriendly_eq ( ) : <EOL> try : <EOL> import numpy as np <EOL> except : <EOL> return <EOL> else : <EOL> task = ( np . sum , np . array ( [ <NUM_LIT:1> , <NUM_LIT:2> ] ) ) <EOL> assert ( subs ( task , ( <NUM_LIT:4> , <NUM_LIT:5> ) , <NUM_LIT:1> ) == task ) is True <EOL> class MyException ( Exception ) : <EOL> pass <EOL> class F ( ) : <EOL> def __eq__ ( self , other ) : <EOL> raise MyException ( ) <EOL> task = F ( ) <EOL> assert subs ( task , <NUM_LIT:1> , <NUM_LIT:2> ) is task <EOL> def test_subs_with_surprisingly_friendly_eq ( ) : <EOL> try : <EOL> import pandas as pd <EOL> except : <EOL> return <EOL> else : <EOL> df = pd . DataFrame ( ) <EOL> assert subs ( df , '<STR_LIT:x>' , <NUM_LIT:1> ) is df <EOL> def test_quote ( ) : <EOL> literals = [ [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] , ( add , <NUM_LIT:1> , <NUM_LIT:2> ) , <EOL> [ <NUM_LIT:1> , [ <NUM_LIT:2> , <NUM_LIT:3> ] ] , ( add , <NUM_LIT:1> , ( add , <NUM_LIT:2> , <NUM_LIT:3> ) ) ] <EOL> for l in literals : <EOL> assert core . get ( { '<STR_LIT:x>' : quote ( l ) } , '<STR_LIT:x>' ) == l <EOL> def test__deps ( ) : <EOL> dsk = { '<STR_LIT:x>' : <NUM_LIT:1> , '<STR_LIT:y>' : <NUM_LIT:2> } <EOL> assert _deps ( dsk , [ '<STR_LIT:x>' , '<STR_LIT:y>' ] ) == [ '<STR_LIT:x>' , '<STR_LIT:y>' ] </s>
