Repository: glotzerlab/signac
Pull request #129 (2019-02-19 02:18:48)

Implement the H5StoreManager class.
## Description

Introduces the `DictManager` and the `H5StoreManager` classes that are exposed as `job.stores` and `project.stores`.

## Motivation and Context

It would be highly advantageous to enable users to easily manage multiple HDF5-files instead of just one. There are plenty of reasons why that is a good idea:

1. While the HDF5-format is generally mutable, it is fundamentally designed to be used as an immutable data container. In fact, even the latest version is internally restricted to append-only operations and a file will continue to grow with repeated addition and edits. The only way to reduce the file size is to create a new file (e.g. with `h5repack`).
2. A single large file is difficult to synchronize. Even if we only change a few MBs or even KBs of a multiple TBs-large file, there is a large chance that synchronization tools like `rsync` or GLOBUS will have to transfer the complete file. Furthermore, all data stored within one large file share one modification time stamp, making fine-grained synchronization even more difficult. Finally, synchronization of parts of a file are impossible.
3. A single large file, especially when it is repeatedly mutated poses a data corruption risk. The HDF5 library uses file locks to reduce the risk of data corruption, but there is no journaling as of yet and the corruption of parts of the file, e.g., through a crashed application (think job killed due to walltime stop etc.) might corrupt the whole file leading to complete data loss.
4. The HDF5-format does not play well with concurrent/parallel access. While the single-writer-multiple-reader (SWMR) mode is implemented for some libraries, it comes with severe limitations and requires a special access style that cannot be easily made transparent to the user. Existence checks etc would be much easier to execute concurrently if they are kept to the file system level without needing to open the file.

## Types of Changes
<!-- Please select all items that apply either now or after creating the pull request: -->
- [ ] Documentation update
- [ ] Bug fix
- [x] New feature
- [ ] Breaking change<sup>1</sup>

This pull request is based on
- [ ] *master*, because it is a bug fix or an update to the documentation.
- [x] *develop*, because it introduces a new feature.

<sup>1</sup>The change breaks (or has the potential to break) existing functionality.

## Checklist:
<!-- Please select all items that apply either now or after creating the pull request. -->
<!-- If you are unsure about any of these items, do not hesitate to ask! -->
- [x] I am familiar with the [**Contributing Guidelines**](https://github.com/glotzerlab/signac/blob/master/CONTRIBUTING.md).
- [x] I agree with the terms of the [**Contributor Agreement**](https://github.com/glotzerlab/signac/blob/master/ContributorAgreement.md).
- [x] My name is on the [list of contributors](https://github.com/glotzerlab/signac/blob/master/contributors.txt).
- [x] My code follows the [code style guideline](https://github.com/glotzerlab/signac/blob/master/CONTRIBUTING.md#code-style) of this project.
- [x] The changes introduced by this pull request are covered by existing or newly introduced tests.

If necessary:
- [x] I have updated the API documentation as part of the package doc-strings.
- [ ] I have created a separate pull request to update the [framework documentation](https://docs.signac.io/) on [signac-docs](https://github.com/glotzerlab/signac-docs) and linked it here.
- [x] I have updated the [changelog](https://github.com/glotzerlab/signac/blob/master/changelog.txt).


csadorf: @glotzerlab/signac-maintainers This is ready for review.
csadorf: > LGTM, I don't work with the H5Store enough to approve the PR but looking over it I don't see anything obviously wrong. We should make sure to update [glotzerlab/signac-docs#6](https://github.com/glotzerlab/signac-docs/pull/6) in light of the changes made here.

Thank you, that's a good point and I will go through the docs before merging.
vyasr: I will review this tomorrow, trying to finish up another task before I start going back through signac PRs.
csadorf: @vyasr I've addressed all of your comments.

Patch:
diff --git a/.circleci/config.yml b/.circleci/config.yml
index e25a1334..485cc47e 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -52,6 +52,7 @@ jobs:
               sudo apt update -qq && sudo apt install -qq libhdf5-dev
             fi
             pip install -U pip==18
+            pip install -U codecov
             pip install -U -e . -r requirements-dev.txt
 
       - save_cache:
@@ -63,8 +64,9 @@ jobs:
           name: run tests
           command: |
             . venv/bin/activate
-            coverage run --source=signac/ -m unittest discover tests/ -v
-            coverage report -i --include="signac*"
+            coverage run -m unittest discover tests/ -v
+            coverage report -i
+            codecov
 
       - store_artifacts:
           path: test-reports
diff --git a/.coveragerc b/.coveragerc
index 3b0be4af..baec5b67 100644
--- a/.coveragerc
+++ b/.coveragerc
@@ -1,2 +1,7 @@
 [run]
-omit = */configobj/*.py
+source=signac
+omit=
+    signac/common/configobj/*.py
+    signac/common/tempdir.py
+    signac/common/tqdm/*.py
+    signac/common/six.py
diff --git a/changelog.txt b/changelog.txt
index 3d706c26..a57856b3 100644
--- a/changelog.txt
+++ b/changelog.txt
@@ -24,7 +24,7 @@ Highlights
 Added
 +++++
 
- - Adds an ``H5Store`` class, useful for storing array-like data with an HDF5 backend. Accessible via ``with job.data:`` or ``with project.data:``.
+ - Adds an ``H5Store`` and a ``H5StoreManager`` class, useful for storing array-like data with an HDF5-backend. Those classes are exposed as part of the ``job.data``, ``job.stores``, ``project.data``, and ``project.stores``, properties.
  - Adds the ``signac.get_job()`` and the ``signac.Project.get_job()`` functions which allow users to get a job handle by switching into or providing the job's workspace directory.
  - Add automatic cast of numpy arrays to lists when storing them within a `JSONDict`, e.g., a `job.statepoint` or `job.document`.
  - Enable `Collection` class to manage collections stored in gzip files.
@@ -32,6 +32,11 @@ Added
  - The `Job.init()` method returns the job to allow one-line job creation and initialization.
  - Add HTML representation of instances of `Project` and `JobsCursor` targeted at Jupyter Notebooks (requires pandas).
 
+Changed
++++++++
+
+ - `Project.find_jobs()` and `Project.find_statepoints()` no longer have the unused argument `index`.
+
 Fixed
 +++++
 
diff --git a/signac/contrib/job.py b/signac/contrib/job.py
index 8c71c3ec..3ddb44e1 100644
--- a/signac/contrib/job.py
+++ b/signac/contrib/job.py
@@ -11,7 +11,7 @@ from ..common import six
 from ..core.json import json, CustomJSONEncoder
 from ..core.attrdict import SyncedAttrDict
 from ..core.jsondict import JSONDict
-from ..core.h5store import H5Store
+from ..core.h5store import H5StoreManager
 from .hashing import calc_id
 from .utility import _mkdir_p
 from .errors import DestinationExistsError, JobsCorruptedError
@@ -53,8 +53,7 @@ class Job(object):
     FN_DOCUMENT = 'signac_job_document.json'
     "The job's document filename."
 
-    FN_DATA = 'signac_data.h5'
-    "The job's datastore filename."
+    KEY_DATA = 'signac_data'
 
     def __init__(self, project, statepoint, _id=None):
         self._project = project
@@ -77,9 +76,8 @@ class Job(object):
         self._fn_doc = os.path.join(self._wd, self.FN_DOCUMENT)
         self._document = None
 
-        # Prepare job datastore
-        self._fn_data = os.path.join(self._wd, self.FN_DATA)
-        self._data = None
+        # Prepare job h5-stores
+        self._stores = H5StoreManager(self._wd)
 
         # Prepare current working directory for context management
         self._cwd = list()
@@ -163,7 +161,6 @@ class Job(object):
         self._wd = dst._wd
         self._fn_doc = dst._fn_doc
         self._document = None
-        self._fn_data = dst._fn_data
         self._data = None
         self._cwd = list()
         logger.info("Moved '{}' -> '{}'.".format(self, dst))
@@ -263,21 +260,47 @@ class Job(object):
     def doc(self, new_doc):
         self.document = new_doc
 
+    @property
+    def stores(self):
+        """Access HDF5-stores associated wit this job.
+
+        Use this property to access an HDF5-file within the job's workspace
+        directory using the H5Store dict-like interface.
+
+        This is an example for accessing an HDF5-file called 'my_data.h5' within
+        the job's workspace:
+
+            job.stores['my_data']['array'] = np.random((32, 4))
+
+        This is equivalent to:
+
+            H5Store(job.fn('my_data.h5'))['array'] = np.random((32, 4))
+
+        Both the `job.stores` and the `H5Store` itself support attribute
+        access. The above example could therefore also be expressed as
+
+            job.stores.my_data.array = np.random((32, 4))
+
+        :return: The HDF5-Store manager for this job.
+        :rtype: :class:`~..core.h5store.H5StoreManager
+        """
+        return self.init()._stores
+
     @property
     def data(self):
         """The data associated with this job.
 
+        Equivalent to:
+
+            return job.stores['signac_data']
+
         :return: An HDF5-backed datastore.
-        :rtype: :class:`~signac.core.h5store.H5Store`"""
-        if self._data is None:
-            self.init()
-            self._data = H5Store(filename=self._fn_data)
-        return self._data
+        :rtype: :class:`~..core.h5store.H5Store`"""
+        return self.stores[self.KEY_DATA]
 
     @data.setter
     def data(self, new_data):
-        self._data.clear()
-        self._data.update(new_data)
+        self.stores[self.KEY_DATA] = new_data
 
     def _init(self, force=False):
         fn_manifest = os.path.join(self._wd, self.FN_MANIFEST)
diff --git a/signac/contrib/project.py b/signac/contrib/project.py
index 50460ad2..4e06d4c7 100644
--- a/signac/contrib/project.py
+++ b/signac/contrib/project.py
@@ -18,7 +18,7 @@ from multiprocessing.pool import ThreadPool
 from .. import syncutil
 from ..core.json import json
 from ..core.jsondict import JSONDict
-from ..core.h5store import H5Store
+from ..core.h5store import H5StoreManager
 from .collection import Collection
 from ..common import six
 from ..common.config import load_config
@@ -124,8 +124,8 @@ class Project(object):
     FN_DOCUMENT = 'signac_project_document.json'
     "The project's document filename."
 
-    FN_DATA = 'signac_data.h5'
-    "The project's datastore filename."
+    KEY_DATA = 'signac_data'
+    "The project's datastore key."
 
     FN_STATEPOINTS = 'signac_statepoints.json'
     "The default filename to read from and write statepoints to."
@@ -147,10 +147,6 @@ class Project(object):
         self._fn_doc = os.path.join(self._rd, self.FN_DOCUMENT)
         self._document = None
 
-        # Prepare project datastore
-        self._fn_data = os.path.join(self._rd, self.FN_DATA)
-        self._data = None
-
         # Internal caches
         self._index_cache = dict()
         self._sp_cache = dict()
@@ -305,21 +301,48 @@ class Project(object):
     def doc(self, new_doc):
         self.document = new_doc
 
+    @property
+    def stores(self):
+        """Access HDF5-stores associated wit this project.
+
+        Use this property to access an HDF5-file within the project's root
+        directory using the H5Store dict-like interface.
+
+        This is an example for accessing an HDF5-file called 'my_data.h5' within
+        the project's root directory:
+
+            project.stores['my_data']['array'] = np.random((32, 4))
+
+        This is equivalent to:
+
+            H5Store(project.fn('my_data.h5'))['array'] = np.random((32, 4))
+
+        Both the `project.stores` and the `H5Store` itself support attribute
+        access. The above example could therefore also be expressed as
+
+            project.stores.my_data.array = np.random((32, 4))
+
+        :return: The HDF5-Store manager for this project.
+        :rtype: :class:`~..core.h5store.H5StoreManager
+        """
+        return H5StoreManager(self._rd)
+
     @property
     def data(self):
         """The data associated with this project.
 
+        Equivalent to:
+
+            return project.store['signac_data']
+
         :return: An HDF5-backed datastore.
-        :rtype: :class:`~signac.core.h5store.H5Store`
+        :rtype: :class:`~..core.h5store.H5Store`
         """
-        if self._data is None:
-            self._data = H5Store(filename=self._fn_data)
-        return self._data
+        return self.stores[self.KEY_DATA]
 
     @data.setter
     def data(self, new_data):
-        self._data.clear()
-        self._data.update(new_data)
+        self.stores[self.KEY_DATA] = new_data
 
     def open_job(self, statepoint=None, id=None):
         """Get a job handle associated with a statepoint.
@@ -522,7 +545,7 @@ class Project(object):
             search_index = JobSearchIndex(index)
         return search_index.find_job_ids(filter=filter, doc_filter=doc_filter)
 
-    def find_jobs(self, filter=None, doc_filter=None, index=None):
+    def find_jobs(self, filter=None, doc_filter=None):
         """Find all jobs in the project's workspace.
 
         The optional filter arguments must be a Mapping of key-value pairs and
@@ -530,10 +553,6 @@ class Project(object):
         statepoints, whereas the `doc_filter` argument compares against job
         document keys.
 
-        .. note::
-            Providing a pre-calculated index may vastly increase the
-            performance of this function.
-
         :param filter: A mapping of key-value pairs that all
             indexed job statepoints are compared against.
         :type filter: Mapping
@@ -630,7 +649,7 @@ class Project(object):
         """
         return self.find_jobs().to_dataframe(*args, **kwargs)
 
-    def find_statepoints(self, filter=None, doc_filter=None, index=None, skip_errors=False):
+    def find_statepoints(self, filter=None, doc_filter=None, skip_errors=False):
         """Find all statepoints in the project's workspace.
 
         This function is deprecated.
@@ -648,11 +667,7 @@ class Project(object):
             "The Project.find_statepoints() method is deprecated.",
             DeprecationWarning)
 
-        if index is None:
-            index = self.index(include_job_document=False)
-        if skip_errors:
-            index = _skip_errors(index, logger.critical)
-        jobs = self.find_jobs(filter, doc_filter, index)
+        jobs = self.find_jobs(filter, doc_filter)
         if skip_errors:
             jobs = _skip_errors(jobs, logger.critical)
         for job in jobs:
diff --git a/signac/core/dict_manager.py b/signac/core/dict_manager.py
new file mode 100644
index 00000000..eb6293f8
--- /dev/null
+++ b/signac/core/dict_manager.py
@@ -0,0 +1,115 @@
+# Copyright (c) 2019 The Regents of the University of Michigan
+# All rights reserved.
+# This software is licensed under the BSD 3-Clause License.
+"Basic wrapper to access multiple different data stores."
+import os
+import re
+import errno
+import uuid
+
+from ..common import six
+
+
+class DictManager(object):
+
+    cls = None
+    suffix = None
+
+    __slots__ = ['_prefix', '_dict_registry']
+
+    def __init__(self, prefix):
+        assert self.cls is not None, "Subclasses of DictManager must define the cls variable."
+        assert self.suffix is not None, "Subclasses of DictManager must define the suffix variable."
+        self._prefix = os.path.abspath(prefix)
+        self._dict_registry = dict()
+
+    @property
+    def prefix(self):
+        return self._prefix
+
+    def __eq__(self, other):
+        return os.path.realpath(self.prefix) == os.path.realpath(other.prefix) and \
+            self.suffix == other.suffix
+
+    def __repr__(self):
+        return "{}(prefix='{}')".format(type(self).__name__, os.path.relpath(self.prefix))
+
+    __str__ = __repr__
+
+    def __getitem__(self, key):
+        if key not in self._dict_registry:
+            self._dict_registry[key] = self.cls(os.path.join(self.prefix, key) + self.suffix)
+        return self._dict_registry[key]
+
+    def __setitem__(self, key, value):
+        tmp_key = str(uuid.uuid4())
+        try:
+            self[tmp_key].update(value)
+            if six.PY2:
+                os.rename(self[tmp_key].filename, self[key].filename)
+            else:
+                os.replace(self[tmp_key].filename, self[key].filename)
+        except (IOError, OSError) as error:
+            if error.errno == errno.ENOENT and not len(value):
+                raise ValueError("Cannot asssign empty value!")
+            else:
+                raise error
+        except Exception as error:
+            try:
+                del self[tmp_key]
+            except KeyError:
+                pass
+            raise error
+        else:
+            del self._dict_registry[key]
+
+    def __delitem__(self, key):
+        try:
+            os.unlink(self[key].filename)
+        except (IOError, OSError) as error:
+            if error.errno == errno.ENOENT:
+                raise KeyError(key)
+            else:
+                raise error
+
+    def __getattr__(self, name):
+        try:
+            return super(DictManager, self).__getattribute__(name)
+        except AttributeError:
+            if name.startswith('__') or name in self.__slots__:
+                raise
+            try:
+                return self.__getitem__(name)
+            except KeyError:
+                raise AttributeError(name)
+
+    def __setattr__(self, name, value):
+        if name.startswith('__') or name in self.__slots__:
+            super(DictManager, self).__setattr__(name, value)
+        else:
+            self.__setitem__(name, value)
+
+    def __delattr__(self, name):
+        if name.startswith('__') or name in self.__slots__:
+            super(DictManager, self).__delattr__(name)
+        else:
+            self.__delitem__(name)
+
+    def __iter__(self):
+        for fn in os.listdir(self.prefix):
+            m = re.match('(.*){}'.format(self.suffix), fn)
+            if m:
+                yield m.groups()[0]
+
+    def keys(self):
+        return iter(self)
+
+    def __len__(self):
+        return len(list(self.keys()))
+
+    def __getstate__(self):
+        return dict(_prefix=self._prefix, _dict_registry=self._dict_registry)
+
+    def __setstate__(self, d):
+        self._prefix = d['_prefix']
+        self._dict_registry = d['_dict_registry']
diff --git a/signac/core/h5store.py b/signac/core/h5store.py
index 7f8fcbb0..10f022af 100644
--- a/signac/core/h5store.py
+++ b/signac/core/h5store.py
@@ -10,6 +10,7 @@ import array
 from threading import RLock
 
 from ..common import six
+from .dict_manager import DictManager
 
 if six.PY2:
     from collections import Mapping
@@ -284,6 +285,10 @@ class H5Store(MutableMapping):
         self._file = None
         self._kwargs = kwargs
 
+    @property
+    def filename(self):
+        return self._filename
+
     def __repr__(self):
         return "<{}(filename={})>".format(type(self).__name__, os.path.relpath(self._filename))
 
@@ -359,7 +364,10 @@ class H5Store(MutableMapping):
 
     def flush(self):
         """Flush the underlying HDF5-file."""
-        self._file.flush()
+        if self._file is None:
+            raise H5StoreClosedError(self._filename)
+        else:
+            self._file.flush()
 
     def __getitem__(self, key):
         key = key if key.startswith('/') else '/' + key
@@ -392,6 +400,12 @@ class H5Store(MutableMapping):
         else:
             self.__setitem__(key, value)
 
+    def __delattr__(self, key):
+        if key.startswith('__') or key in self.__slots__:
+            super(H5Store, self).__delattr__(key)
+        else:
+            self.__delitem__(key)
+
     def __iter__(self):
         with _ensure_open(self):
             # The generator below should be refactored to use 'yield from'
@@ -428,3 +442,8 @@ class H5Store(MutableMapping):
         """
         with _ensure_open(self):
             self._file.clear()
+
+
+class H5StoreManager(DictManager):
+    cls = H5Store
+    suffix = '.h5'


Test patch:
diff --git a/tests/test_h5store_manager.py b/tests/test_h5store_manager.py
new file mode 100644
index 00000000..dd40ae2a
--- /dev/null
+++ b/tests/test_h5store_manager.py
@@ -0,0 +1,98 @@
+# Copyright (c) 2019 The Regents of the University of Michigan
+# All rights reserved.
+# This software is licensed under the BSD 3-Clause License.
+import os
+import unittest
+import pickle
+
+from signac.core.h5store import H5StoreManager
+from signac.common import six
+if six.PY2:
+    from tempdir import TemporaryDirectory
+else:
+    from tempfile import TemporaryDirectory
+
+
+class TestH5StoreManager(unittest.TestCase):
+
+    def setUp(self):
+        self._tmp_dir = TemporaryDirectory(prefix='h5store_')
+        self.addCleanup(self._tmp_dir.cleanup)
+        self.store = H5StoreManager(prefix=self._tmp_dir.name)
+        with open(os.path.join(self._tmp_dir.name, 'other_file.txt'), 'w') as file:
+            file.write(r'blank\n')
+
+    def test_repr(self):
+        self.assertEqual(eval(repr(self.store)), self.store)
+
+    def test_str(self):
+        self.assertEqual(eval(str(self.store)), self.store)
+
+    def test_set(self):
+        self.assertEqual(len(self.store), 0)
+        self.assertNotIn('test', self.store)
+        for value in ('', [], {}):
+            with self.assertRaises(ValueError):
+                self.store['test'] = value
+        for value in (True, 0, 0.0, 1, 1.0, None):
+            with self.assertRaises(TypeError):
+                self.store['test'] = value
+        for value in ('abc'):
+            with self.assertRaises(ValueError):
+                self.store['test'] = value
+
+        # Assigning a dictionary is the intended use case
+        self.store['test'] = dict(foo=True)
+        self.assertEqual(len(self.store), 1)
+        self.assertIn('test', self.store)
+
+    def test_set_iterable(self):
+        self.assertEqual(len(self.store), 0)
+        self.assertNotIn('test', self.store)
+        self.store['test'] = list(dict(foo=True).items())
+        self.assertEqual(len(self.store), 1)
+        self.assertIn('test', self.store)
+
+    def test_set_get(self):
+        self.assertEqual(len(self.store), 0)
+        self.assertNotIn('test', self.store)
+        self.store['test']['foo'] = 'bar'
+        self.assertIn('test', self.store)
+        self.assertEqual(len(self.store), 1)
+        self.assertIn('foo', self.store['test'])
+
+    def test_del(self):
+        self.assertEqual(len(self.store), 0)
+        self.assertNotIn('test', self.store)
+        self.store['test']['foo'] = 'bar'
+        self.assertIn('test', self.store)
+        self.assertEqual(len(self.store), 1)
+        self.assertIn('foo', self.store['test'])
+        with self.assertRaises(KeyError):
+            del self.store['invalid']
+        del self.store['test']
+        self.assertEqual(len(self.store), 0)
+        self.assertNotIn('test', self.store)
+
+    def test_iteration(self):
+        keys = ['foo', 'bar', 'baz']
+        for key in keys:
+            self.store[key] = dict(test=True)
+        self.assertEqual(list(sorted(keys)), list(sorted(self.store)))
+        self.assertEqual(list(sorted(keys)), list(sorted(self.store.keys())))
+
+    def test_contains(self):
+        keys = ['foo', 'bar', 'baz']
+        for key in keys:
+            self.assertNotIn(key, self.store)
+        for key in keys:
+            self.store[key] = dict(test=True)
+        for key in keys:
+            self.assertIn(key, self.store)
+
+    def test_pickle(self):
+        self.assertEqual(pickle.loads(pickle.dumps(self.store)), self.store)
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/test_job.py b/tests/test_job.py
index dc5f80b6..87dad692 100644
--- a/tests/test_job.py
+++ b/tests/test_job.py
@@ -1383,6 +1383,382 @@ class JobClosedDataTest(JobOpenDataTest):
     def open_data(job):
         yield
 
+    def test_implicit_initialization(self):
+        job = self.open_job(test_token)
+        self.assertNotIn('test', job.stores)
+        self.assertNotIn('foo', job.stores.test)
+        self.assertEqual(list(job.stores.keys()), [])
+        self.assertEqual(list(job.stores), [])
+        self.assertNotIn('test', job.stores)
+        job.stores.test.foo = True
+        self.assertIn('test', job.stores)
+        self.assertIn('foo', job.stores.test)
+        self.assertEqual(list(job.stores.keys()), ['test'])
+        self.assertEqual(list(job.stores), ['test'])
+
+
+class JobOpenCustomDataTest(BaseJobTest):
+
+    @staticmethod
+    @contextmanager
+    def open_data(job):
+        with job.stores.test:
+            yield
+
+    def test_get_set(self):
+        key = 'get_set'
+        d = testdata()
+        job = self.open_job(test_token)
+        with self.open_data(job):
+            self.assertFalse(bool(job.stores.test))
+            self.assertEqual(len(job.stores.test), 0)
+            self.assertNotIn(key, job.stores.test)
+            job.stores.test[key] = d
+            self.assertTrue(bool(job.stores.test))
+            self.assertEqual(len(job.stores.test), 1)
+            self.assertIn(key, job.stores.test)
+            self.assertEqual(job.stores.test[key], d)
+            self.assertEqual(job.stores.test.get(key), d)
+            self.assertEqual(job.stores.test.get('bs', d), d)
+
+    def test_del(self):
+        key = 'del0'
+        key1 = 'del1'
+        d = testdata()
+        d1 = testdata()
+        job = self.open_job(test_token)
+        with self.open_data(job):
+            self.assertEqual(len(job.stores.test), 0)
+            self.assertNotIn(key, job.stores.test)
+            job.stores.test[key] = d
+            self.assertEqual(len(job.stores.test), 1)
+            self.assertIn(key, job.stores.test)
+            job.stores.test[key1] = d1
+            self.assertEqual(len(job.stores.test), 2)
+            self.assertIn(key, job.stores.test)
+            self.assertIn(key1, job.stores.test)
+            self.assertEqual(job.stores.test[key], d)
+            self.assertEqual(job.stores.test[key1], d1)
+            del job.stores.test[key]
+            self.assertEqual(len(job.stores.test), 1)
+            self.assertIn(key1, job.stores.test)
+            self.assertNotIn(key, job.stores.test)
+
+    def test_get_set_data(self):
+        key = 'get_set'
+        d = testdata()
+        job = self.open_job(test_token)
+        with self.open_data(job):
+            self.assertFalse(bool(job.stores.test))
+            self.assertEqual(len(job.stores.test), 0)
+            self.assertNotIn(key, job.stores.test)
+            job.stores.test[key] = d
+            self.assertTrue(bool(job.stores.test))
+            self.assertEqual(len(job.stores.test), 1)
+            self.assertIn(key, job.stores.test)
+            self.assertEqual(job.stores.test[key], d)
+            self.assertEqual(job.stores.test.get(key), d)
+            self.assertEqual(job.stores.test.get('bs', d), d)
+
+    def test_set_set_data(self):
+        key0, key1 = 'set_set0', 'set_set1'
+        d0, d1 = testdata(), testdata()
+        job = self.open_job(test_token)
+        with self.open_data(job):
+            self.assertFalse(bool(job.stores.test))
+            self.assertEqual(len(job.stores.test), 0)
+            self.assertNotIn(key0, job.stores.test)
+            job.stores.test[key0] = d0
+            self.assertTrue(bool(job.stores.test))
+            self.assertEqual(len(job.stores.test), 1)
+            self.assertIn(key0, job.stores.test)
+            self.assertEqual(job.stores.test[key0], d0)
+        job = self.open_job(test_token)
+        with self.open_data(job):
+            self.assertTrue(bool(job.stores.test))
+            self.assertEqual(len(job.stores.test), 1)
+            self.assertIn(key0, job.stores.test)
+            self.assertEqual(job.stores.test[key0], d0)
+        job = self.open_job(test_token)
+        with self.open_data(job):
+            job.stores.test[key1] = d1
+            self.assertTrue(bool(job.stores.test))
+            self.assertEqual(len(job.stores.test), 2)
+            self.assertIn(key0, job.stores.test)
+            self.assertIn(key1, job.stores.test)
+            self.assertEqual(job.stores.test[key0], d0)
+            self.assertEqual(job.stores.test[key1], d1)
+
+    def test_get_set_nested(self):
+        d0 = testdata()
+        d1 = testdata()
+        d2 = testdata()
+        assert d0 != d1 != d2
+        job = self.open_job(test_token)
+        with self.open_data(job):
+            self.assertEqual(len(job.stores.test), 0)
+            self.assertNotIn('key0', job.stores.test)
+            job.stores.test['key0'] = d0
+            self.assertEqual(len(job.stores.test), 1)
+            self.assertIn('key0', job.stores.test)
+            self.assertEqual(job.stores.test['key0'], d0)
+            with self.assertRaises(AttributeError):
+                job.stores.test.key0.key1
+            job.stores.test.key0 = {'key1': d0}
+            self.assertEqual(len(job.stores.test), 1)
+            self.assertIn('key0', job.stores.test)
+            self.assertEqual(dict(job.stores.test), {'key0': {'key1': d0}})
+            self.assertEqual(job.stores.test['key0'], {'key1': d0})
+            self.assertEqual(job.stores.test['key0']['key1'], d0)
+            self.assertEqual(job.stores.test.key0, {'key1': d0})
+            self.assertEqual(job.stores.test.key0.key1, d0)
+            job.stores.test.key0.key1 = d1
+            self.assertEqual(job.stores.test, {'key0': {'key1': d1}})
+            self.assertEqual(job.stores.test['key0'], {'key1': d1})
+            self.assertEqual(job.stores.test['key0']['key1'], d1)
+            self.assertEqual(job.stores.test.key0, {'key1': d1})
+            self.assertEqual(job.stores.test.key0.key1, d1)
+            job.stores.test['key0']['key1'] = d2
+            self.assertEqual(job.stores.test, {'key0': {'key1': d2}})
+            self.assertEqual(job.stores.test['key0'], {'key1': d2})
+            self.assertEqual(job.stores.test['key0']['key1'], d2)
+            self.assertEqual(job.stores.test.key0, {'key1': d2})
+            self.assertEqual(job.stores.test.key0.key1, d2)
+
+    def test_get_set_nested_data(self):
+        d0 = testdata()
+        d1 = testdata()
+        d2 = testdata()
+        assert d0 != d1 != d2
+        job = self.open_job(test_token)
+        with self.open_data(job):
+            self.assertEqual(len(job.stores.test), 0)
+            self.assertNotIn('key0', job.stores.test)
+            job.stores.test['key0'] = d0
+            self.assertEqual(len(job.stores.test), 1)
+            self.assertIn('key0', job.stores.test)
+            self.assertEqual(job.stores.test['key0'], d0)
+            with self.assertRaises(AttributeError):
+                job.stores.test.key0.key1
+            job.stores.test.key0 = {'key1': d0}
+            self.assertEqual(len(job.stores.test), 1)
+            self.assertIn('key0', job.stores.test)
+            self.assertEqual(dict(job.stores.test), {'key0': {'key1': d0}})
+            self.assertEqual(job.stores.test['key0'], {'key1': d0})
+            self.assertEqual(job.stores.test['key0']['key1'], d0)
+            self.assertEqual(job.stores.test.key0, {'key1': d0})
+            self.assertEqual(job.stores.test.key0.key1, d0)
+            job.stores.test.key0.key1 = d1
+            self.assertEqual(job.stores.test, {'key0': {'key1': d1}})
+            self.assertEqual(job.stores.test['key0'], {'key1': d1})
+            self.assertEqual(job.stores.test['key0']['key1'], d1)
+            self.assertEqual(job.stores.test.key0, {'key1': d1})
+            self.assertEqual(job.stores.test.key0.key1, d1)
+            job.stores.test['key0']['key1'] = d2
+            self.assertEqual(job.stores.test, {'key0': {'key1': d2}})
+            self.assertEqual(job.stores.test['key0'], {'key1': d2})
+            self.assertEqual(job.stores.test['key0']['key1'], d2)
+            self.assertEqual(job.stores.test.key0, {'key1': d2})
+            self.assertEqual(job.stores.test.key0.key1, d2)
+
+    def test_assign(self):
+        key = 'assign'
+        d0 = testdata()
+        d1 = testdata()
+        job = self.open_job(test_token)
+        with self.open_data(job):
+            self.assertEqual(len(job.stores.test), 0)
+            job.stores.test[key] = d0
+            self.assertEqual(len(job.stores.test), 1)
+            self.assertEqual(dict(job.stores.test), {key: d0})
+            with self.assertRaises(ValueError):
+                job.stores.test = d1
+            job.stores.test = {key: d1}
+            self.assertEqual(len(job.stores.test), 1)
+            self.assertEqual(dict(job.stores.test), {key: d1})
+
+    def test_assign_data(self):
+        key = 'assign'
+        d0 = testdata()
+        d1 = testdata()
+        job = self.open_job(test_token)
+        with self.open_data(job):
+            self.assertEqual(len(job.stores.test), 0)
+            job.stores.test[key] = d0
+            self.assertEqual(len(job.stores.test), 1)
+            self.assertEqual(dict(job.stores.test), {key: d0})
+            with self.assertRaises(ValueError):
+                job.stores.test = d1
+            job.stores.test = {key: d1}
+            self.assertEqual(len(job.stores.test), 1)
+            self.assertEqual(dict(job.stores.test), {key: d1})
+
+    def test_copy_data(self):
+        key = 'get_set'
+        d = testdata()
+        job = self.open_job(test_token)
+        with self.open_data(job):
+            job.stores.test[key] = d
+            self.assertTrue(bool(job.stores.test))
+            self.assertEqual(len(job.stores.test), 1)
+            self.assertIn(key, job.stores.test)
+            self.assertEqual(job.stores.test[key], d)
+            self.assertEqual(job.stores.test.get(key), d)
+            self.assertEqual(job.stores.test.get('bs', d), d)
+            copy = dict(job.stores.test)
+            self.assertTrue(bool(copy))
+            self.assertEqual(len(copy), 1)
+            self.assertIn(key, copy)
+            self.assertEqual(copy[key], d)
+            self.assertEqual(copy.get(key), d)
+            self.assertEqual(copy.get('bs', d), d)
+
+    def test_update(self):
+        key = 'get_set'
+        d = testdata()
+        job = self.open_job(test_token)
+        with self.open_data(job):
+            job.stores.test.update({key: d})
+            self.assertIn(key, job.stores.test)
+
+    def test_clear_data(self):
+        key = 'clear'
+        d = testdata()
+        job = self.open_job(test_token)
+        with self.open_data(job):
+            job.stores.test[key] = d
+            self.assertIn(key, job.stores.test)
+            self.assertEqual(len(job.stores.test), 1)
+            job.stores.test.clear()
+            self.assertNotIn(key, job.stores.test)
+            self.assertEqual(len(job.stores.test), 0)
+
+    def test_reopen(self):
+        key = 'reopen'
+        d = testdata()
+        job = self.open_job(test_token)
+        with self.open_data(job):
+            job.stores.test[key] = d
+            self.assertIn(key, job.stores.test)
+            self.assertEqual(len(job.stores.test), 1)
+        job2 = self.open_job(test_token)
+        with self.open_data(job2):
+            self.assertIn(key, job2.stores.test)
+            self.assertEqual(len(job2.stores.test), 1)
+
+    def test_concurrency(self):
+        key = 'concurrent'
+        d = testdata()
+        job = self.open_job(test_token)
+        job2 = self.open_job(test_token)
+        with self.open_data(job):
+            with self.open_data(job2):
+                self.assertNotIn(key, job.stores.test)
+                self.assertNotIn(key, job2.stores.test)
+                job.stores.test[key] = d
+                self.assertIn(key, job.stores.test)
+                self.assertIn(key, job2.stores.test)
+
+    def test_remove(self):
+        key = 'remove'
+        job = self.open_job(test_token)
+        job.remove()
+        d = testdata()
+        with self.open_data(job):
+            job.stores.test[key] = d
+            self.assertIn(key, job.stores.test)
+            self.assertEqual(len(job.stores.test), 1)
+        fn_test = os.path.join(job.workspace(), 'test')
+        with open(fn_test, 'w') as file:
+            file.write('test')
+        self.assertTrue(os.path.isfile(fn_test))
+        job.remove()
+        with self.open_data(job):
+            self.assertNotIn(key, job.stores.test)
+        self.assertFalse(os.path.isfile(fn_test))
+
+    def test_clear_job(self):
+        key = 'clear'
+        job = self.open_job(test_token)
+        self.assertNotIn(job, self.project)
+        job.clear()
+        self.assertNotIn(job, self.project)
+        job.clear()
+        self.assertNotIn(job, self.project)
+        job.init()
+        self.assertIn(job, self.project)
+        job.clear()
+        self.assertIn(job, self.project)
+        job.clear()
+        job.clear()
+        self.assertIn(job, self.project)
+        d = testdata()
+        with self.open_data(job):
+            job.stores.test[key] = d
+            self.assertIn(job, self.project)
+            self.assertIn(key, job.stores.test)
+            self.assertEqual(len(job.stores.test), 1)
+        job.clear()
+        with self.open_data(job):
+            self.assertEqual(len(job.stores.test), 0)
+        with open(job.fn('test'), 'w') as file:
+            file.write('test')
+        self.assertTrue(job.isfile('test'))
+        self.assertIn(job, self.project)
+        job.clear()
+        self.assertFalse(job.isfile('test'))
+        with self.open_data(job):
+            self.assertEqual(len(job.stores.test), 0)
+
+    def test_reset(self):
+        key = 'reset'
+        job = self.open_job(test_token)
+        self.assertNotIn(job, self.project)
+        job.reset()
+        self.assertIn(job, self.project)
+        with self.open_data(job):
+            self.assertEqual(len(job.stores.test), 0)
+            job.stores.test[key] = testdata()
+            self.assertEqual(len(job.stores.test), 1)
+        job.reset()
+        self.assertIn(job, self.project)
+        with self.open_data(job):
+            self.assertEqual(len(job.stores.test), 0)
+
+    def test_data(self):
+        key = 'test_data'
+        job = self.open_job(test_token)
+
+        def check_content(key, d):
+            self.assertEqual(job.stores.test[key], d)
+            self.assertEqual(getattr(job.stores.test, key), d)
+            self.assertEqual(dict(job.stores.test)[key], d)
+            self.assertEqual(job.stores.test[key], d)
+            self.assertEqual(getattr(job.stores.test, key), d)
+            self.assertEqual(dict(job.stores.test)[key], d)
+
+        with self.open_data(job):
+            d = testdata()
+            job.stores.test[key] = d
+            check_content(key, d)
+            d2 = testdata()
+            job.stores.test[key] = d2
+            check_content(key, d2)
+            d3 = testdata()
+            job.stores.test[key] = d3
+            check_content(key, d3)
+            d4 = testdata()
+            setattr(job.stores.test, key, d4)
+            check_content(key, d4)
+
+
+class JobClosedCustomDataTest(JobOpenCustomDataTest):
+
+    @staticmethod
+    @contextmanager
+    def open_data(job):
+        yield
+
 
 if __name__ == '__main__':
     unittest.main()
diff --git a/tests/test_project.py b/tests/test_project.py
index 2b609aa7..64f81349 100644
--- a/tests/test_project.py
+++ b/tests/test_project.py
@@ -1394,7 +1394,7 @@ class ProjectRepresentationTest(BaseProjectTest):
                 repr(self.project)
             with self.subTest(type='html'):
                 for use_pandas in (True, False):
-                    type(self.project)._use_use_pandas_for_html_repr = use_pandas
+                    type(self.project)._use_pandas_for_html_repr = use_pandas
                     with self.subTest(use_pandas=use_pandas):
                         if use_pandas and not PANDAS:
                             raise unittest.SkipTest('requires use_pandas')
@@ -1409,7 +1409,7 @@ class ProjectRepresentationTest(BaseProjectTest):
                         repr(self.project.find_jobs(filter_))
                     with self.subTest(type='html'):
                         for use_pandas in (True, False):
-                            type(self.project)._use_use_pandas_for_html_repr = use_pandas
+                            type(self.project)._use_pandas_for_html_repr = use_pandas
                             with self.subTest(use_pandas=use_pandas):
                                 if use_pandas and not PANDAS:
                                     raise unittest.SkipTest('requires use_pandas')