text: import atexit
  import tempfile
  import unittest
  from numbers import Integral
  
  import numpy as np
  import pytest
  from numcodecs import Adler32, Zlib
  from numpy.testing import assert_array_equal
  
  import zarr
  from zarr.convenience import (
      consolidate_metadata,
      copy,
      copy_store,
      load,
      open,
      open_consolidated,
      save,
      save_group,
      save_array,
      copy_all,
  )
  from zarr.core import Array
  from zarr.errors import CopyError
  from zarr.hierarchy import Group, group
  from zarr.storage import (ConsolidatedMetadataStore, MemoryStore,
                            atexit_rmtree, getsize)
  
  
  def test_open_array(path_type):
  
      store = tempfile.mkdtemp()
      atexit.register(atexit_rmtree, store)
      store = path_type(store)
  
      # open array, create if doesn't exist
      z = open(store, mode='a', shape=100)
      assert isinstance(z, Array)
      assert z.shape == (100,)
  
      # open array, overwrite
      z = open(store, mode='w', shape=200)
      assert isinstance(z, Array)
      assert z.shape == (200,)
  
      # open array, read-only
      z = open(store, mode='r')
      assert isinstance(z, Array)
      assert z.shape == (200,)
      assert z.read_only
  
      # path not found
      with pytest.raises(ValueError):
          open('doesnotexist', mode='r')
  
  
  def test_open_group(path_type):
  
      store = tempfile.mkdtemp()
      atexit.register(atexit_rmtree, store)
      store = path_type(store)
  
      # open group, create if doesn't exist
      g = open(store, mode='a')
      g.create_group('foo')
      assert isinstance(g, Group)
      assert 'foo' in g
  
      # open group, overwrite
      g = open(store, mode='w')
      assert isinstance(g, Group)
      assert 'foo' not in g
  
      # open group, read-only
      g = open(store, mode='r')
      assert isinstance(g, Group)
      assert g.read_only
  
  
  def test_save_errors():
      with pytest.raises(ValueError):
          # no arrays provided
          save_group('data/group.zarr')
      with pytest.raises(ValueError):
          # no arrays provided
          save('data/group.zarr')
  
  
  def test_lazy_loader():
      foo = np.arange(100)
      bar = np.arange(100, 0, -1)
      store = 'data/group.zarr'
      save(store, foo=foo, bar=bar)
      loader = load(store)
      assert 'foo' in loader
      assert 'bar' in loader
      assert 'baz' not in loader
      assert len(loader) == 2
      assert sorted(loader) == ['bar', 'foo']
      assert_array_equal(foo, loader['foo'])
      assert_array_equal(bar, loader['bar'])
  
  
  def test_consolidate_metadata():
  
      # setup initial data
      store = MemoryStore()
      z = group(store)
      z.create_group('g1')
      g2 = z.create_group('g2')
      g2.attrs['hello'] = 'world'
      arr = g2.create_dataset('arr', shape=(20, 20), chunks=(5, 5), dtype='f8')
      assert 16 == arr.nchunks
      assert 0 == arr.nchunks_initialized
      arr.attrs['data'] = 1
      arr[:] = 1.0
      assert 16 == arr.nchunks_initialized
  
      # perform consolidation
      out = consolidate_metadata(store)
      assert isinstance(out, Group)
      assert '.zmetadata' in store
      for key in ['.zgroup',
                  'g1/.zgroup',
                  'g2/.zgroup',
                  'g2/.zattrs',
                  'g2/arr/.zarray',
                  'g2/arr/.zattrs']:
          del store[key]
  
      # open consolidated
      z2 = open_consolidated(store)
      assert ['g1', 'g2'] == list(z2)
      assert 'world' == z2.g2.attrs['hello']
      assert 1 == z2.g2.arr.attrs['data']
      assert (z2.g2.arr[:] == 1.0).all()
      assert 16 == z2.g2.arr.nchunks
      assert 16 == z2.g2.arr.nchunks_initialized
  
      # tests del/write on the store
      cmd = ConsolidatedMetadataStore(store)
      with pytest.raises(PermissionError):
          del cmd['.zgroup']
      with pytest.raises(PermissionError):
          cmd['.zgroup'] = None
  
      # test getsize on the store
      assert isinstance(getsize(cmd), Integral)
  
      # test new metadata are not writeable
      with pytest.raises(PermissionError):
          z2.create_group('g3')
      with pytest.raises(PermissionError):
          z2.create_dataset('spam', shape=42, chunks=7, dtype='i4')
      with pytest.raises(PermissionError):
          del z2['g2']
  
      # test consolidated metadata are not writeable
      with pytest.raises(PermissionError):
          z2.g2.attrs['hello'] = 'universe'
      with pytest.raises(PermissionError):
          z2.g2.arr.attrs['foo'] = 'bar'
  
      # test the data are writeable
      z2.g2.arr[:] = 2
      assert (z2.g2.arr[:] == 2).all()
  
      # test invalid modes
      with pytest.raises(ValueError):
          open_consolidated(store, mode='a')
      with pytest.raises(ValueError):
          open_consolidated(store, mode='w')
  
      # make sure keyword arguments are passed through without error
      open_consolidated(store, cache_attrs=True, synchronizer=None)
  
  
  def test_consolidated_with_chunk_store():
      # setup initial data
      store = MemoryStore()
      chunk_store = MemoryStore()
      z = group(store, chunk_store=chunk_store)
      z.create_group('g1')
      g2 = z.create_group('g2')
      g2.attrs['hello'] = 'world'
      arr = g2.create_dataset('arr', shape=(20, 20), chunks=(5, 5), dtype='f8')
      assert 16 == arr.nchunks
      assert 0 == arr.nchunks_initialized
      arr.attrs['data'] = 1
      arr[:] = 1.0
      assert 16 == arr.nchunks_initialized
  
      # perform consolidation
      out = consolidate_metadata(store)
      assert isinstance(out, Group)
      assert '.zmetadata' in store
      for key in ['.zgroup',
                  'g1/.zgroup',
                  'g2/.zgroup',
                  'g2/.zattrs',
                  'g2/arr/.zarray',
                  'g2/arr/.zattrs']:
          del store[key]
      # open consolidated
      z2 = open_consolidated(store, chunk_store=chunk_store)
      assert ['g1', 'g2'] == list(z2)
      assert 'world' == z2.g2.attrs['hello']
      assert 1 == z2.g2.arr.attrs['data']
      assert (z2.g2.arr[:] == 1.0).all()
      assert 16 == z2.g2.arr.nchunks
      assert 16 == z2.g2.arr.nchunks_initialized
  
      # test the data are writeable
      z2.g2.arr[:] = 2
      assert (z2.g2.arr[:] == 2).all()
  
      # test invalid modes
      with pytest.raises(ValueError):
          open_consolidated(store, mode='a', chunk_store=chunk_store)
      with pytest.raises(ValueError):
          open_consolidated(store, mode='w', chunk_store=chunk_store)
  
      # make sure keyword arguments are passed through without error
      open_consolidated(store, cache_attrs=True, synchronizer=None,
                        chunk_store=chunk_store)
  
  
  @pytest.mark.parametrize("options", (
      {"dimension_separator": "/"},
      {"dimension_separator": "."},
      {"dimension_separator": None},
  ))
  def test_save_array_separator(tmpdir, options):
      data = np.arange(6).reshape((3, 2))
      url = tmpdir.join("test.zarr")
      save_array(url, data, **options)
  
  
  class TestCopyStore(unittest.TestCase):
  
      def setUp(self):
          source = dict()
          source['foo'] = b'xxx'
          source['bar/baz'] = b'yyy'
          source['bar/qux'] = b'zzz'
          self.source = source
  
      def test_no_paths(self):
          source = self.source
          dest = dict()
          copy_store(source, dest)
          assert len(source) == len(dest)
          for key in source:
              assert source[key] == dest[key]
  
      def test_source_path(self):
          source = self.source
          # paths should be normalized
          for source_path in 'bar', 'bar/', '/bar', '/bar/':
              dest = dict()
              copy_store(source, dest, source_path=source_path)
              assert 2 == len(dest)
              for key in source:
                  if key.startswith('bar/'):
                      dest_key = key.split('bar/')[1]
                      assert source[key] == dest[dest_key]
                  else:
                      assert key not in dest
  
      def test_dest_path(self):
          source = self.source
          # paths should be normalized
          for dest_path in 'new', 'new/', '/new', '/new/':
              dest = dict()
              copy_store(source, dest, dest_path=dest_path)
              assert len(source) == len(dest)
              for key in source:
                  dest_key = 'new/' + key
                  assert source[key] == dest[dest_key]
  
      def test_source_dest_path(self):
          source = self.source
          # paths should be normalized
          for source_path in 'bar', 'bar/', '/bar', '/bar/':
              for dest_path in 'new', 'new/', '/new', '/new/':
                  dest = dict()
                  copy_store(source, dest, source_path=source_path,
                             dest_path=dest_path)
                  assert 2 == len(dest)
                  for key in source:
                      if key.startswith('bar/'):
                          dest_key = 'new/' + key.split('bar/')[1]
                          assert source[key] == dest[dest_key]
                      else:
                          assert key not in dest
                          assert ('new/' + key) not in dest
  
      def test_excludes_includes(self):
          source = self.source
  
          # single excludes
          dest = dict()
          excludes = 'f.*'
          copy_store(source, dest, excludes=excludes)
          assert len(dest) == 2
          assert 'foo' not in dest
  
          # multiple excludes
          dest = dict()
          excludes = 'b.z', '.*x'
          copy_store(source, dest, excludes=excludes)
          assert len(dest) == 1
          assert 'foo' in dest
          assert 'bar/baz' not in dest
          assert 'bar/qux' not in dest
  
          # excludes and includes
          dest = dict()
          excludes = 'b.*'
          includes = '.*x'
          copy_store(source, dest, excludes=excludes, includes=includes)
          assert len(dest) == 2
          assert 'foo' in dest
          assert 'bar/baz' not in dest
          assert 'bar/qux' in dest
  
      def test_dry_run(self):
          source = self.source
          dest = dict()
          copy_store(source, dest, dry_run=True)
          assert 0 == len(dest)
  
      def test_if_exists(self):
          source = self.source
          dest = dict()
          dest['bar/baz'] = b'mmm'
  
          # default ('raise')
          with pytest.raises(CopyError):
              copy_store(source, dest)
  
          # explicit 'raise'
          with pytest.raises(CopyError):
              copy_store(source, dest, if_exists='raise')
  
          # skip
          copy_store(source, dest, if_exists='skip')
          assert 3 == len(dest)
          assert dest['foo'] == b'xxx'
          assert dest['bar/baz'] == b'mmm'
          assert dest['bar/qux'] == b'zzz'
  
          # replace
          copy_store(source, dest, if_exists='replace')
          assert 3 == len(dest)
          assert dest['foo'] == b'xxx'
          assert dest['bar/baz'] == b'yyy'
          assert dest['bar/qux'] == b'zzz'
  
          # invalid option
          with pytest.raises(ValueError):
              copy_store(source, dest, if_exists='foobar')
  
  
  def check_copied_array(original, copied, without_attrs=False,
                         expect_props=None):
  
      # setup
      source_h5py = original.__module__.startswith('h5py.')
      dest_h5py = copied.__module__.startswith('h5py.')
      zarr_to_zarr = not (source_h5py or dest_h5py)
      h5py_to_h5py = source_h5py and dest_h5py
      zarr_to_h5py = not source_h5py and dest_h5py
      h5py_to_zarr = source_h5py and not dest_h5py
      if expect_props is None:
          expect_props = dict()
      else:
          expect_props = expect_props.copy()
  
      # common properties in zarr and h5py
      for p in 'dtype', 'shape', 'chunks':
          expect_props.setdefault(p, getattr(original, p))
  
      # zarr-specific properties
      if zarr_to_zarr:
          for p in 'compressor', 'filters', 'order', 'fill_value':
              expect_props.setdefault(p, getattr(original, p))
  
      # h5py-specific properties
      if h5py_to_h5py:
          for p in ('maxshape', 'compression', 'compression_opts', 'shuffle',
                    'scaleoffset', 'fletcher32', 'fillvalue'):
              expect_props.setdefault(p, getattr(original, p))
  
      # common properties with some name differences
      if h5py_to_zarr:
          expect_props.setdefault('fill_value', original.fillvalue)
      if zarr_to_h5py:
          expect_props.setdefault('fillvalue', original.fill_value)
  
      # compare properties
      for k, v in expect_props.items():
          assert v == getattr(copied, k)
  
      # compare data
      assert_array_equal(original[:], copied[:])
  
      # compare attrs
      if without_attrs:
          for k in original.attrs.keys():
              assert k not in copied.attrs
      else:
          assert sorted(original.attrs.items()) == sorted(copied.attrs.items())
  
  
  def check_copied_group(original, copied, without_attrs=False, expect_props=None,
                         shallow=False):
  
      # setup
      if expect_props is None:
          expect_props = dict()
      else:
          expect_props = expect_props.copy()
  
      # compare children
      for k, v in original.items():
          if hasattr(v, 'shape'):
              assert k in copied
              check_copied_array(v, copied[k], without_attrs=without_attrs,
                                 expect_props=expect_props)
          elif shallow:
              assert k not in copied
          else:
              assert k in copied
              check_copied_group(v, copied[k], without_attrs=without_attrs,
                                 shallow=shallow, expect_props=expect_props)
  
      # compare attrs
      if without_attrs:
          for k in original.attrs.keys():
              assert k not in copied.attrs
      else:
          assert sorted(original.attrs.items()) == sorted(copied.attrs.items())
  
  
  def test_copy_all():
      """
      https://github.com/zarr-developers/zarr-python/issues/269
  
      copy_all used to not copy attributes as `.keys()` does not return hidden `.zattrs`.
  
      """
      original_group = zarr.group(store=MemoryStore(), overwrite=True)
      original_group.attrs["info"] = "group attrs"
      original_subgroup = original_group.create_group("subgroup")
      original_subgroup.attrs["info"] = "sub attrs"
  
      destination_group = zarr.group(store=MemoryStore(), overwrite=True)
  
      # copy from memory to directory store
      copy_all(
          original_group,
          destination_group,
          dry_run=False,
      )
  
      assert destination_group.attrs["info"] == "group attrs"
      assert destination_group.subgroup.attrs["info"] == "sub attrs"
  
  
  class TestCopy:
      @pytest.fixture(params=[False, True], ids=['zarr', 'hdf5'])
      def source(self, request, tmpdir):
          def prep_source(source):
              foo = source.create_group('foo')
              foo.attrs['experiment'] = 'weird science'
              baz = foo.create_dataset('bar/baz', data=np.arange(100), chunks=(50,))
              baz.attrs['units'] = 'metres'
              if request.param:
                  extra_kws = dict(compression='gzip', compression_opts=3, fillvalue=84,
                                   shuffle=True, fletcher32=True)
              else:
                  extra_kws = dict(compressor=Zlib(3), order='F', fill_value=42, filters=[Adler32()])
              source.create_dataset('spam', data=np.arange(100, 200).reshape(20, 5),
                                    chunks=(10, 2), dtype='i2', **extra_kws)
              return source
  
          if request.param:
              h5py = pytest.importorskip('h5py')
              fn = tmpdir.join('source.h5')
              with h5py.File(str(fn), mode='w') as h5f:
                  yield prep_source(h5f)
          else:
              yield prep_source(group())
  
      @pytest.fixture(params=[False, True], ids=['zarr', 'hdf5'])
      def dest(self, request, tmpdir):
          if request.param:
              h5py = pytest.importorskip('h5py')
              fn = tmpdir.join('dest.h5')
              with h5py.File(str(fn), mode='w') as h5f:
                  yield h5f
          else:
              yield group()
  
      def test_copy_array(self, source, dest):
          # copy array with default options
          copy(source['foo/bar/baz'], dest)
          check_copied_array(source['foo/bar/baz'], dest['baz'])
          copy(source['spam'], dest)
          check_copied_array(source['spam'], dest['spam'])
  
      def test_copy_bad_dest(self, source, dest):
          # try to copy to an array, dest must be a group
          dest = dest.create_dataset('eggs', shape=(100,))
          with pytest.raises(ValueError):
              copy(source['foo/bar/baz'], dest)
  
      def test_copy_array_name(self, source, dest):
          # copy array with name
          copy(source['foo/bar/baz'], dest, name='qux')
          assert 'baz' not in dest
          check_copied_array(source['foo/bar/baz'], dest['qux'])
  
      def test_copy_array_create_options(self, source, dest):
          dest_h5py = dest.__module__.startswith('h5py.')
  
          # copy array, provide creation options
          compressor = Zlib(9)
          create_kws = dict(chunks=(10,))
          if dest_h5py:
              create_kws.update(compression='gzip', compression_opts=9,
                                shuffle=True, fletcher32=True, fillvalue=42)
          else:
              create_kws.update(compressor=compressor, fill_value=42, order='F',
                                filters=[Adler32()])
          copy(source['foo/bar/baz'], dest, without_attrs=True, **create_kws)
          check_copied_array(source['foo/bar/baz'], dest['baz'],
                             without_attrs=True, expect_props=create_kws)
  
      def test_copy_array_exists_array(self, source, dest):
          # copy array, dest array in the way
          dest.create_dataset('baz', shape=(10,))
  
          # raise
          with pytest.raises(CopyError):
              # should raise by default
              copy(source['foo/bar/baz'], dest)
          assert (10,) == dest['baz'].shape
          with pytest.raises(CopyError):
              copy(source['foo/bar/baz'], dest, if_exists='raise')
          assert (10,) == dest['baz'].shape
  
          # skip
          copy(source['foo/bar/baz'], dest, if_exists='skip')
          assert (10,) == dest['baz'].shape
  
          # replace
          copy(source['foo/bar/baz'], dest, if_exists='replace')
          check_copied_array(source['foo/bar/baz'], dest['baz'])
  
          # invalid option
          with pytest.raises(ValueError):
              copy(source['foo/bar/baz'], dest, if_exists='foobar')
  
      def test_copy_array_exists_group(self, source, dest):
          # copy array, dest group in the way
          dest.create_group('baz')
  
          # raise
          with pytest.raises(CopyError):
              copy(source['foo/bar/baz'], dest)
          assert not hasattr(dest['baz'], 'shape')
          with pytest.raises(CopyError):
              copy(source['foo/bar/baz'], dest, if_exists='raise')
          assert not hasattr(dest['baz'], 'shape')
  
          # skip
          copy(source['foo/bar/baz'], dest, if_exists='skip')
          assert not hasattr(dest['baz'], 'shape')
  
          # replace
          copy(source['foo/bar/baz'], dest, if_exists='replace')
          check_copied_array(source['foo/bar/baz'], dest['baz'])
  
      def test_copy_array_skip_initialized(self, source, dest):
          dest_h5py = dest.__module__.startswith('h5py.')
  
          dest.create_dataset('baz', shape=(100,), chunks=(10,), dtype='i8')
          assert not np.all(source['foo/bar/baz'][:] == dest['baz'][:])
  
          if dest_h5py:
              with pytest.raises(ValueError):
                  # not available with copy to h5py
                  copy(source['foo/bar/baz'], dest, if_exists='skip_initialized')
  
          else:
              # copy array, dest array exists but not yet initialized
              copy(source['foo/bar/baz'], dest, if_exists='skip_initialized')
              check_copied_array(source['foo/bar/baz'], dest['baz'])
  
              # copy array, dest array exists and initialized, will be skipped
              dest['baz'][:] = np.arange(100, 200)
              copy(source['foo/bar/baz'], dest, if_exists='skip_initialized')
              assert_array_equal(np.arange(100, 200), dest['baz'][:])
              assert not np.all(source['foo/bar/baz'][:] == dest['baz'][:])
  
      def test_copy_group(self, source, dest):
          # copy group, default options
          copy(source['foo'], dest)
          check_copied_group(source['foo'], dest['foo'])
  
      def test_copy_group_no_name(self, source, dest):
          with pytest.raises(TypeError):
              # need a name if copy root
              copy(source, dest)
  
          copy(source, dest, name='root')
          check_copied_group(source, dest['root'])
  
      def test_copy_group_options(self, source, dest):
          # copy group, non-default options
          copy(source['foo'], dest, name='qux', without_attrs=True)
          assert 'foo' not in dest
          check_copied_group(source['foo'], dest['qux'], without_attrs=True)
  
      def test_copy_group_shallow(self, source, dest):
          # copy group, shallow
          copy(source, dest, name='eggs', shallow=True)
          check_copied_group(source, dest['eggs'], shallow=True)
  
      def test_copy_group_exists_group(self, source, dest):
          # copy group, dest groups exist
          dest.create_group('foo/bar')
          copy(source['foo'], dest)
          check_copied_group(source['foo'], dest['foo'])
  
      def test_copy_group_exists_array(self, source, dest):
          # copy group, dest array in the way
          dest.create_dataset('foo/bar', shape=(10,))
  
          # raise
          with pytest.raises(CopyError):
              copy(source['foo'], dest)
          assert dest['foo/bar'].shape == (10,)
          with pytest.raises(CopyError):
              copy(source['foo'], dest, if_exists='raise')
          assert dest['foo/bar'].shape == (10,)
  
          # skip
          copy(source['foo'], dest, if_exists='skip')
          assert dest['foo/bar'].shape == (10,)
  
          # replace
          copy(source['foo'], dest, if_exists='replace')
          check_copied_group(source['foo'], dest['foo'])
  
      def test_copy_group_dry_run(self, source, dest):
          # dry run, empty destination
          n_copied, n_skipped, n_bytes_copied = \
              copy(source['foo'], dest, dry_run=True, return_stats=True)
          assert 0 == len(dest)
          assert 3 == n_copied
          assert 0 == n_skipped
          assert 0 == n_bytes_copied
  
          # dry run, array exists in destination
          baz = np.arange(100, 200)
          dest.create_dataset('foo/bar/baz', data=baz)
          assert not np.all(source['foo/bar/baz'][:] == dest['foo/bar/baz'][:])
          assert 1 == len(dest)
  
          # raise
          with pytest.raises(CopyError):
              copy(source['foo'], dest, dry_run=True)
          assert 1 == len(dest)
  
          # skip
          n_copied, n_skipped, n_bytes_copied = \
              copy(source['foo'], dest, dry_run=True, if_exists='skip',
                   return_stats=True)
          assert 1 == len(dest)
          assert 2 == n_copied
          assert 1 == n_skipped
          assert 0 == n_bytes_copied
          assert_array_equal(baz, dest['foo/bar/baz'])
  
          # replace
          n_copied, n_skipped, n_bytes_copied = \
              copy(source['foo'], dest, dry_run=True, if_exists='replace',
                   return_stats=True)
          assert 1 == len(dest)
          assert 3 == n_copied
          assert 0 == n_skipped
          assert 0 == n_bytes_copied
          assert_array_equal(baz, dest['foo/bar/baz'])
  
      def test_logging(self, source, dest, tmpdir):
          # callable log
          copy(source['foo'], dest, dry_run=True, log=print)
  
          # file name
          fn = str(tmpdir.join('log_name'))
          copy(source['foo'], dest, dry_run=True, log=fn)
  
          # file
          with tmpdir.join('log_file').open(mode='w') as f:
              copy(source['foo'], dest, dry_run=True, log=f)
  
          # bad option
          with pytest.raises(TypeError):
              copy(source['foo'], dest, dry_run=True, log=True)
  