()
| 128 | |
| 129 | |
| 130 | def test_cache_profiler(): |
| 131 | with CacheProfiler() as cprof: |
| 132 | in_context_time = default_timer() |
| 133 | get(dsk2, "c") |
| 134 | results = cprof.results |
| 135 | assert all(isinstance(i, tuple) and len(i) == 5 for i in results) |
| 136 | assert cprof.start_time < in_context_time < cprof.end_time |
| 137 | |
| 138 | cprof.clear() |
| 139 | assert cprof.results == [] |
| 140 | |
| 141 | tics = [0] |
| 142 | |
| 143 | def nbytes(res): |
| 144 | tics[0] += 1 |
| 145 | return tics[0] |
| 146 | |
| 147 | with CacheProfiler(nbytes) as cprof: |
| 148 | get(dsk2, "c") |
| 149 | |
| 150 | results = cprof.results |
| 151 | assert tics[-1] == len(results) |
| 152 | assert tics[-1] == results[-1].metric |
| 153 | assert cprof._metric_name == "nbytes" |
| 154 | assert CacheProfiler(metric=nbytes, metric_name="foo")._metric_name == "foo" |
| 155 | |
| 156 | |
| 157 | @pytest.mark.parametrize( |
nothing calls this directly
no test coverage detected