(layers)
| 252 | |
| 253 | @pytest.mark.parametrize("layers", [False, True]) |
| 254 | def test_bind(layers): |
| 255 | dsk1 = {("a-1", h1): 1, ("a-1", h2): 2} |
| 256 | dsk2 = {"b-1": (add, ("a-1", h1), ("a-1", h2))} |
| 257 | dsk3 = {"c-1": "b-1"} |
| 258 | cnt = NodeCounter() |
| 259 | dsk4 = {("d-1", h1): (cnt.f, 1), ("d-1", h2): (cnt.f, 2)} |
| 260 | dsk4b = {"e": (cnt.f, 3)} |
| 261 | |
| 262 | if layers: |
| 263 | dsk1 = HighLevelGraph({"a-1": dsk1}, {"a-1": set()}) |
| 264 | dsk2 = HighLevelGraph( |
| 265 | {"a-1": dsk1, "b-1": dsk2}, {"a-1": set(), "b-1": {"a-1"}} |
| 266 | ) |
| 267 | dsk3 = HighLevelGraph( |
| 268 | {"a-1": dsk1, "b-1": dsk2, "c-1": dsk3}, |
| 269 | {"a-1": set(), "b-1": {"a-1"}, "c-1": {"b-1"}}, |
| 270 | ) |
| 271 | dsk4 = HighLevelGraph({"d-1": dsk4, "e": dsk4b}, {"d-1": set(), "e": set()}) |
| 272 | else: |
| 273 | dsk2.update(dsk1) |
| 274 | dsk3.update(dsk2) |
| 275 | dsk4.update(dsk4b) |
| 276 | |
| 277 | # t1 = Tuple(dsk1, [("a", h1), ("a", h2)]) |
| 278 | t2 = Tuple(dsk2, ["b-1"]) |
| 279 | t3 = Tuple(dsk3, ["c-1"]) |
| 280 | t4 = Tuple(dsk4, [("d-1", h1), ("d-1", h2), "e"]) # Multiple names |
| 281 | |
| 282 | bound1 = bind(t3, t4, seed=1, assume_layers=layers) |
| 283 | cloned_a_name = clone_key("a-1", seed=1) |
| 284 | assert bound1.__dask_graph__()[cloned_a_name, h1][0] is chunks.bind |
| 285 | assert bound1.__dask_graph__()[cloned_a_name, h2][0] is chunks.bind |
| 286 | assert bound1.compute() == (3,) |
| 287 | assert cnt.n == 3 |
| 288 | |
| 289 | bound2 = bind(t3, t4, omit=t2, seed=1, assume_layers=layers) |
| 290 | cloned_c_name = clone_key("c-1", seed=1) |
| 291 | assert bound2.__dask_graph__()[cloned_c_name][0] is chunks.bind |
| 292 | assert bound2.compute() == (3,) |
| 293 | assert cnt.n == 6 |
| 294 | |
| 295 | bound3 = bind(t4, t3, seed=1, assume_layers=layers) |
| 296 | cloned_d_name = clone_key("d-1", seed=1) |
| 297 | cloned_e_name = clone_key("e", seed=1) |
| 298 | assert bound3.__dask_graph__()[cloned_d_name, h1][0] is chunks.bind |
| 299 | assert bound3.__dask_graph__()[cloned_d_name, h2][0] is chunks.bind |
| 300 | assert bound3.__dask_graph__()[cloned_e_name][0] is chunks.bind |
| 301 | assert bound3.compute() == (1, 2, 3) |
| 302 | assert cnt.n == 9 |
| 303 | |
| 304 | |
| 305 | @pytest.mark.parametrize( |
nothing calls this directly
no test coverage detected