(T, m, dask_cluster)
| 444 | @pytest.mark.filterwarnings("ignore:\\s+Port 8787 is already in use:UserWarning") |
| 445 | @pytest.mark.parametrize("T, m", test_data) |
| 446 | def test_stimped(T, m, dask_cluster): |
| 447 | if T.ndim > 1: |
| 448 | T = T.copy() |
| 449 | T = T[0] |
| 450 | n = 3 |
| 451 | seed = np.random.randint(100000) |
| 452 | with Client(dask_cluster) as dask_client: |
| 453 | np.random.seed(seed) |
| 454 | ref = aamp_stimped(dask_client, T, m) |
| 455 | for i in range(n): |
| 456 | ref.update() |
| 457 | |
| 458 | np.random.seed(seed) |
| 459 | cmp = stimped(dask_client, T, m, normalize=False) |
| 460 | for i in range(n): |
| 461 | cmp.update() |
| 462 | |
| 463 | # Compare raw pan |
| 464 | ref_PAN = ref._PAN |
| 465 | cmp_PAN = cmp._PAN |
| 466 | |
| 467 | naive.replace_inf(ref_PAN) |
| 468 | naive.replace_inf(cmp_PAN) |
| 469 | |
| 470 | npt.assert_almost_equal(ref_PAN, cmp_PAN) |
| 471 | |
| 472 | # Compare transformed pan |
| 473 | npt.assert_almost_equal(ref.PAN_, cmp.PAN_) |
| 474 | |
| 475 | |
| 476 | @pytest.mark.filterwarnings("ignore", category=NumbaPerformanceWarning) |
nothing calls this directly
no test coverage detected