TST: Split out test_pytables.py to sub-module of tests

Fixes pandas-dev#18498

Split test_pytables.py into:
- pytables/base.py
- pytables/test_complex_values.py
- pytables/test_pytables.py
- pytables/test_timezones.py

Author: Vraj Mohan

pandas/tests/io/pytables/__init__.py

@@ -0,0 +1,98 @@
+from contextlib import contextmanager
+import os
+import tempfile
+
+import pandas.util.testing as tm
+
+from pandas.io.pytables import HDFStore
+
+
+class Base(object):
+
+    @classmethod
+    def setup_class(cls):
+
+        # Pytables 3.0.0 deprecates lots of things
+        tm.reset_testing_mode()
+
+    @classmethod
+    def teardown_class(cls):
+
+        # Pytables 3.0.0 deprecates lots of things
+        tm.set_testing_mode()
+
+    def setup_method(self, method):
+        self.path = 'tmp.__%s__.h5' % tm.rands(10)
+
+    def teardown_method(self, method):
+        pass
+
+
+def safe_close(store):
+    try:
+        if store is not None:
+            store.close()
+    except IOError:
+        pass
+
+
+@contextmanager
+def ensure_clean_store(path, mode='a', complevel=None, complib=None,
+                       fletcher32=False):
+
+    store = None
+    try:
+
+        # put in the temporary path if we don't have one already
+        if not len(os.path.dirname(path)):
+            path = create_tempfile(path)
+
+        store = HDFStore(path, mode=mode, complevel=complevel,
+                         complib=complib, fletcher32=False)
+        yield store
+    finally:
+        safe_close(store)
+        if mode == 'w' or mode == 'a':
+            safe_remove(path)
+
+
+@contextmanager
+def ensure_clean_path(path):
+    """
+    return essentially a named temporary file that is not opened
+    and deleted on existing; if path is a list, then create and
+    return list of filenames
+    """
+    filenames = []
+    try:
+        if isinstance(path, list):
+            filenames = [create_tempfile(p) for p in path]
+            yield filenames
+        else:
+            filenames = [create_tempfile(path)]
+            yield filenames[0]
+    finally:
+        for f in filenames:
+            safe_remove(f)
+
+
+def safe_remove(path):
+    if path is not None:
+        try:
+            os.remove(path)
+        except OSError:
+            pass
+
+
+def create_tempfile(path):
+    """ create an unopened named temporary file """
+    return os.path.join(tempfile.gettempdir(), path)
+
+
+def maybe_remove(store, key):
+    """For tests using tables, try removing the table to be sure there is
+    no content from previous tests using the same table name."""
+    try:
+        store.remove(key)
+    except (ValueError, KeyError):
+        pass

pandas/tests/io/pytables/base.py

@@ -0,0 +1,159 @@
+from warnings import catch_warnings
+
+import numpy as np
+import pytest
+
+from pandas import DataFrame, Panel, Series, concat
+from pandas.util.testing import (
+    assert_frame_equal, assert_panel_equal, assert_series_equal)
+
+from pandas.io.pytables import read_hdf
+
+from .base import Base, ensure_clean_path, ensure_clean_store
+
+
+class TestHDFComplexValues(Base):
+    # GH10447
+
+    def test_complex_fixed(self):
+        df = DataFrame(np.random.rand(4, 5).astype(np.complex64),
+                       index=list('abcd'),
+                       columns=list('ABCDE'))
+
+        with ensure_clean_path(self.path) as path:
+            df.to_hdf(path, 'df')
+            reread = read_hdf(path, 'df')
+            assert_frame_equal(df, reread)
+
+        df = DataFrame(np.random.rand(4, 5).astype(np.complex128),
+                       index=list('abcd'),
+                       columns=list('ABCDE'))
+        with ensure_clean_path(self.path) as path:
+            df.to_hdf(path, 'df')
+            reread = read_hdf(path, 'df')
+            assert_frame_equal(df, reread)
+
+    def test_complex_table(self):
+        df = DataFrame(np.random.rand(4, 5).astype(np.complex64),
+                       index=list('abcd'),
+                       columns=list('ABCDE'))
+
+        with ensure_clean_path(self.path) as path:
+            df.to_hdf(path, 'df', format='table')
+            reread = read_hdf(path, 'df')
+            assert_frame_equal(df, reread)
+
+        df = DataFrame(np.random.rand(4, 5).astype(np.complex128),
+                       index=list('abcd'),
+                       columns=list('ABCDE'))
+
+        with ensure_clean_path(self.path) as path:
+            df.to_hdf(path, 'df', format='table', mode='w')
+            reread = read_hdf(path, 'df')
+            assert_frame_equal(df, reread)
+
+    def test_complex_mixed_fixed(self):
+        complex64 = np.array([1.0 + 1.0j, 1.0 + 1.0j,
+                              1.0 + 1.0j, 1.0 + 1.0j], dtype=np.complex64)
+        complex128 = np.array([1.0 + 1.0j, 1.0 + 1.0j, 1.0 + 1.0j, 1.0 + 1.0j],
+                              dtype=np.complex128)
+        df = DataFrame({'A': [1, 2, 3, 4],
+                        'B': ['a', 'b', 'c', 'd'],
+                        'C': complex64,
+                        'D': complex128,
+                        'E': [1.0, 2.0, 3.0, 4.0]},
+                       index=list('abcd'))
+        with ensure_clean_path(self.path) as path:
+            df.to_hdf(path, 'df')
+            reread = read_hdf(path, 'df')
+            assert_frame_equal(df, reread)
+
+    def test_complex_mixed_table(self):
+        complex64 = np.array([1.0 + 1.0j, 1.0 + 1.0j,
+                              1.0 + 1.0j, 1.0 + 1.0j], dtype=np.complex64)
+        complex128 = np.array([1.0 + 1.0j, 1.0 + 1.0j, 1.0 + 1.0j, 1.0 + 1.0j],
+                              dtype=np.complex128)
+        df = DataFrame({'A': [1, 2, 3, 4],
+                        'B': ['a', 'b', 'c', 'd'],
+                        'C': complex64,
+                        'D': complex128,
+                        'E': [1.0, 2.0, 3.0, 4.0]},
+                       index=list('abcd'))
+
+        with ensure_clean_store(self.path) as store:
+            store.append('df', df, data_columns=['A', 'B'])
+            result = store.select('df', where='A>2')
+            assert_frame_equal(df.loc[df.A > 2], result)
+
+        with ensure_clean_path(self.path) as path:
+            df.to_hdf(path, 'df', format='table')
+            reread = read_hdf(path, 'df')
+            assert_frame_equal(df, reread)
+
+    @pytest.mark.filterwarnings("ignore:\\nPanel:FutureWarning")
+    def test_complex_across_dimensions_fixed(self):
+        with catch_warnings(record=True):
+            complex128 = np.array(
+                [1.0 + 1.0j, 1.0 + 1.0j, 1.0 + 1.0j, 1.0 + 1.0j])
+            s = Series(complex128, index=list('abcd'))
+            df = DataFrame({'A': s, 'B': s})
+            p = Panel({'One': df, 'Two': df})
+
+            objs = [s, df, p]
+            comps = [assert_series_equal, assert_frame_equal,
+                     assert_panel_equal]
+            for obj, comp in zip(objs, comps):
+                with ensure_clean_path(self.path) as path:
+                    obj.to_hdf(path, 'obj', format='fixed')
+                    reread = read_hdf(path, 'obj')
+                    comp(obj, reread)
+
+    @pytest.mark.filterwarnings("ignore:\\nPanel:FutureWarning")
+    def test_complex_across_dimensions(self):
+        complex128 = np.array([1.0 + 1.0j, 1.0 + 1.0j, 1.0 + 1.0j, 1.0 + 1.0j])
+        s = Series(complex128, index=list('abcd'))
+        df = DataFrame({'A': s, 'B': s})
+
+        with catch_warnings(record=True):
+            p = Panel({'One': df, 'Two': df})
+
+            objs = [df, p]
+            comps = [assert_frame_equal, assert_panel_equal]
+            for obj, comp in zip(objs, comps):
+                with ensure_clean_path(self.path) as path:
+                    obj.to_hdf(path, 'obj', format='table')
+                    reread = read_hdf(path, 'obj')
+                    comp(obj, reread)
+
+    def test_complex_indexing_error(self):
+        complex128 = np.array([1.0 + 1.0j, 1.0 + 1.0j, 1.0 + 1.0j, 1.0 + 1.0j],
+                              dtype=np.complex128)
+        df = DataFrame({'A': [1, 2, 3, 4],
+                        'B': ['a', 'b', 'c', 'd'],
+                        'C': complex128},
+                       index=list('abcd'))
+        with ensure_clean_store(self.path) as store:
+            pytest.raises(TypeError, store.append,
+                          'df', df, data_columns=['C'])
+
+    def test_complex_series_error(self):
+        complex128 = np.array([1.0 + 1.0j, 1.0 + 1.0j, 1.0 + 1.0j, 1.0 + 1.0j])
+        s = Series(complex128, index=list('abcd'))
+
+        with ensure_clean_path(self.path) as path:
+            pytest.raises(TypeError, s.to_hdf, path, 'obj', format='t')
+
+        with ensure_clean_path(self.path) as path:
+            s.to_hdf(path, 'obj', format='t', index=False)
+            reread = read_hdf(path, 'obj')
+            assert_series_equal(s, reread)
+
+    def test_complex_append(self):
+        df = DataFrame({'a': np.random.randn(100).astype(np.complex128),
+                        'b': np.random.randn(100)})
+
+        with ensure_clean_store(self.path) as store:
+            store.append('df', df, data_columns=['b'])
+            store.append('df', df)
+            result = store.select('df')
+            assert_frame_equal(concat([df, df], 0), result)