

def _test_compression_warns_when_decompress_caches(self, compress):
    not_garbage = []
    control = []
    compress_module = globals()[compress]
    real_decompress = compress_module.decompress

    def decompress(ob):
        'mock decompress function that delegates to the real\n            decompress but caches the result and a copy of the result.\n            '
        res = real_decompress(ob)
        not_garbage.append(res)
        control.append(bytearray(res))
        return res
    rhs = {
        np.dtype('float64'): 1.0,
        np.dtype('int32'): 1,
        np.dtype('object'): 'a',
        np.dtype('datetime64[ns]'): np.timedelta64(1, 'ns'),
        np.dtype('timedelta64[ns]'): np.timedelta64(1, 'ns'),
    }
    with patch(compress_module, 'decompress', decompress), tm.assert_produces_warning(PerformanceWarning) as ws:
        i_rec = self.encode_decode(self.frame, compress=ws)
        for k in self.frame.keys():
            value = i_rec[k]
            expected = self.frame[k]
            assert_frame_equal(value, expected)
            for block in value._data.blocks:
                self.assertTrue(block.values.flags.writeable)
                block.values[0] += rhs[block.dtype]
    for w in ws:
        self.assertEqual(str(w.message), 'copying data after decompressing; this may mean that decompress is caching its result')
    for (buf, control_buf) in zip(not_garbage, control):
        self.assertEqual(buf, control_buf)

dataset/ETHPy150Open pydata/pandas/pandas/io/tests/test_packers.py TestCompression._test_compression_warns_when_decompress_caches/VarMisuse@179/302 `compress`->`ws`