test_version_metadata.py 文件源码

def test_version_metadata_with_streaming(self, api, opener):

        np.random.seed(123)
        times = pd.date_range('2000-01-01', '2001-12-31', name='time')
        annual_cycle = np.sin(2 * np.pi * (times.dayofyear / 365.25 - 0.28))
        base = 10 + 15 * np.array(annual_cycle).reshape(-1, 1)

        tmin_values = base + 3 * np.random.randn(annual_cycle.size, 3)
        tmax_values = base + 3 * np.random.randn(annual_cycle.size, 3)

        ds = xr.Dataset({'tmin': (('time', 'location'), tmin_values),
                         'tmax': (('time', 'location'), tmax_values)},
                        {'time': times, 'location': ['IA', 'IN', 'IL']})

        var = api.create('streaming_test')
        with var.get_local_path(
                bumpversion='patch',
                dependencies={'arch1': '0.1.0', 'arch2': '0.2.0'}) as f:

            ds.to_netcdf(f)
            ds.close()

        assert var.get_history()[-1]['dependencies']['arch2'] == '0.2.0'

        tmin_values = base + 10 * np.random.randn(annual_cycle.size, 3)
        ds.update({'tmin': (('time', 'location'), tmin_values)})

        with var.get_local_path(
                bumpversion='patch',
                dependencies={'arch1': '0.1.0', 'arch2': '1.2.0'}) as f:

            with xr.open_dataset(f) as ds:

                mem = ds.load()
                ds.close()

            mem.to_netcdf(f)

        assert var.get_history()[-1]['dependencies']['arch2'] == '1.2.0'
        assert var.get_history()[-1][
            'checksum'] != var.get_history()[-2]['checksum']