| 2088 | |
| 2089 | |
| 2090 | def test_store_delayed_target(): |
| 2091 | from dask.delayed import delayed |
| 2092 | |
| 2093 | d = da.ones((4, 4), chunks=(2, 2)) |
| 2094 | a, b = d + 1, d + 2 |
| 2095 | |
| 2096 | # empty buffers to be used as targets |
| 2097 | targs = {} |
| 2098 | |
| 2099 | def make_target(key): |
| 2100 | a = np.empty((4, 4)) |
| 2101 | targs[key] = a |
| 2102 | return a |
| 2103 | |
| 2104 | # delayed calls to these targets |
| 2105 | atd = delayed(make_target)("at") |
| 2106 | btd = delayed(make_target)("bt") |
| 2107 | |
| 2108 | # test not keeping result |
| 2109 | st = store([a, b], [atd, btd]) |
| 2110 | |
| 2111 | at = targs["at"] |
| 2112 | bt = targs["bt"] |
| 2113 | |
| 2114 | assert st is None |
| 2115 | assert_eq(at, a) |
| 2116 | assert_eq(bt, b) |
| 2117 | |
| 2118 | # test keeping result |
| 2119 | for st_compute in [False, True]: |
| 2120 | targs.clear() |
| 2121 | st = store([a, b], [atd, btd], return_stored=True, compute=st_compute) |
| 2122 | if st_compute: |
| 2123 | for arr in st: |
| 2124 | assert_has_persisted_data(arr) |
| 2125 | |
| 2126 | st = dask.compute(*st) |
| 2127 | |
| 2128 | at = targs["at"] |
| 2129 | bt = targs["bt"] |
| 2130 | |
| 2131 | assert st is not None |
| 2132 | assert isinstance(st, tuple) |
| 2133 | assert all([isinstance(v, np.ndarray) for v in st]) |
| 2134 | assert_eq(at, a) |
| 2135 | assert_eq(bt, b) |
| 2136 | assert_eq(st[0], a) |
| 2137 | assert_eq(st[1], b) |
| 2138 | |
| 2139 | pytest.raises(ValueError, lambda at=at, bt=bt: store([a], [at, bt])) |
| 2140 | pytest.raises(ValueError, lambda at=at: store(at, at)) |
| 2141 | pytest.raises(ValueError, lambda at=at, bt=bt: store([at, bt], [at, bt])) |
| 2142 | |
| 2143 | |
| 2144 | def test_store(): |