Merge pull request #16 from ercius/update_tests

Update tests

tests/test_DM.py

@@ -1,12 +1,15 @@
+from pathlib import Path
 import pytest
-from xicam.NCEM.ingestors.DMPlugin import ingest_NCEM_DM
 
 
-# TODO: add fixture that writes temp data file with ncempy for tests
+from xicam.NCEM.ingestors.DMPlugin import ingest_NCEM_DM
+
 
 @pytest.fixture
 def DM_path():
-    return "/home/rp/data/NCEM/01_TimeSeriesImages_20images(1).dm3"
+    """DM files must be written from DM."""
+    dPath = Path.home()
+    return str(dPath / Path('data') / Path('01_TimeSeriesImages_20images.dm3'))
 
 
 def test_slicing(DM_path):

tests/test_EMD.py

@@ -9,26 +9,12 @@
 from xicam.NCEM.ingestors.EMDPlugin import ingest_NCEM_EMD, _get_slice
 from databroker.in_memory import BlueskyInMemoryCatalog
 
-# TODO: move file creation to fixture
-
 
 @pytest.fixture
-def EMD_path():
-    """
-    Write a small Berkeley EMD file to a tempfile
-
-    """
-    #return "/home/rp/data/NCEM/Acquisition_18.emd"
-    dd, _, _ = np.mgrid[0:30, 0:40, 0:50]
-    dd = dd.astype('<u2')
-
-    tmp = tempfile.NamedTemporaryFile(mode='wb')
-    tmp.close()  # need to close the file to use it later
-    fPath = str(Path(tmp.name))
-    with emd.fileEMD(fPath, readonly=False) as f0:
-        dims = emd.defaultDims(dd)
-        f0.put_emdgroup('test', dd, dims)
-    return fPath
+def temp_file():
+    tt = tempfile.NamedTemporaryFile(mode='wb')
+    tt.close()  # need to close the file to use it later
+    return Path(tt.name)
 
 
 @pytest.fixture
@@ -54,52 +40,61 @@ def EMD_multi_path():
     return fPath
 
 
-def test_slicing(EMD_path):
-    with emd.fileEMD(EMD_path) as emd_obj:
-        assert _get_slice(emd_obj, 0).shape == (40, 50)
-    docs = list(ingest_NCEM_EMD([EMD_path]))
+def test_slicing(temp_file):
+    dd = np.ones((10, 11, 12), dtype=np.uint16)
+    with emd.fileEMD(temp_file, readonly=False) as emd0:
+        dims = emd.defaultDims(dd)
+        emd0.put_emdgroup('test', dd, dims)
+    with emd.fileEMD(temp_file) as emd_obj:
+        assert _get_slice(emd_obj, 0).shape == (11, 12)
+    docs = list(ingest_NCEM_EMD([str(temp_file)]))
     event_doc = docs[2][1]
     data = event_doc['data']['raw']
-    assert data.shape == (30, 40, 50)
-    assert data[0].compute().shape == (40, 50)
-
+    assert data.shape == (10, 11, 12)
+    assert data[0].compute().shape == (11, 12)
 
-def test_ingest_emd_berkeley(EMD_path):
 
-    # Write a small Berkeley EMD file
-    #dd, _, _ = np.mgrid[0:30, 0:40, 0:50]
-    #dd = dd.astype('<u2')
-    #tmp = tempfile.TemporaryDirectory()
-    #fPath = str(Path(tmp.name) / Path('temp_emd_berkeley.emd'))
-    #with emd.fileEMD(fPath, readonly=False) as f0:
-    #    dims = emd.defaultDims(dd)
-    #    f0.put_emdgroup('test', dd, dims)
+def test_ingest_emd_berkeley(temp_file):
+    dd = np.ones((10, 11, 12), dtype=np.uint16)
+    with emd.fileEMD(temp_file, readonly=False) as emd0:
+        dims = emd.defaultDims(dd)
+        emd0.put_emdgroup('test', dd, dims)
 
     # Test slicing
-    with emd.fileEMD(EMD_path, readonly=True) as emd_obj:
+    with emd.fileEMD(temp_file) as emd_obj:
         dd0 = emd_obj.list_emds[0]['data']
-        assert dd0[0, :, :].shape == (40, 50)
+        assert dd0[0, :, :].shape == (11, 12)
 
     # Test ingest
-    docs = list(ingest_NCEM_EMD([EMD_path]))
+    docs = list(ingest_NCEM_EMD([str(temp_file)]))
     event_doc = docs[2][1]
     data = event_doc['data']['raw']
-    assert data.shape == (30, 40, 50)
-    assert data[0].compute().shape == (40, 50)
+    assert data.shape == (10, 11, 12)
+    assert data[0].compute().shape == (11, 12)
+
 
+def test_multi_device(temp_file):
+    dd = np.ones((10, 11, 12), dtype=np.uint16)
+    with emd.fileEMD(temp_file, readonly=False) as emd0:
+        dims = emd.defaultDims(dd)
+        emd0.put_emdgroup('test', dd, dims)
+        # Change shape and write again to simulate a second data set
+        dd2 = dd.reshape(5, 22, 12)
+        dims2 = emd.defaultDims(dd2)
+        emd0.put_emdgroup('test2', dd2, dims2)
+    del dd, dd2, dims, dims2
 
-def test_multi_device(EMD_multi_path):
-    docs = list(ingest_NCEM_EMD([EMD_multi_path]))
-    #print(docs)
+    # Ingest and get the first data set.
+    docs = list(ingest_NCEM_EMD([str(temp_file)]))
     event_doc = docs[2][1]
     data = event_doc['data']['raw']
-    assert data.shape == (30, 40, 50)
-    assert data[0].compute().shape == (40, 50)
+    assert data.shape == (10, 11, 12)
+    assert data[0].compute().shape == (11, 12)
 
     event_doc = docs[4][1]
     data = event_doc['data']['raw']
-    assert data.shape == (60, 80, 100)
-    assert data[0].compute().shape == (80, 100)
+    assert data.shape == (5, 22, 12)
+    assert data[0].compute().shape == (22, 12)
 
     catalog = BlueskyInMemoryCatalog()
     start = docs[0][1]

tests/test_MRC.py

@@ -8,44 +8,24 @@
 from xicam.NCEM.ingestors.MRCPlugin import ingest_NCEM_MRC
 
 
-# TODO: move file creation to fixture
-
 @pytest.fixture
-def mrc_path():
-    """Write a small MRC file to a temporary directory and file location.
-
-    THIS DOES NOT WORK. NOT SURE WHY. -PAE
-
-    Returns
-    -------
-        : string
-            The path to the temporary file.
-    """
-    dd, _, _ = np.mgrid[0:30, 0:40, 0:50]
-    dd = dd.astype('<u2')
-
-    tmp = tempfile.TemporaryDirectory()
-    fPath = Path(tmp.name) / Path('temp_mrc.mrc')
-
-    mrc.mrcWriter(str(fPath), dd, (0.1, 0.2, 0.3))
-
-    return str(fPath)
+def temp_file():
+    tt = tempfile.NamedTemporaryFile(mode='wb')
+    tt.close() # need to close the file to use it later
+    return Path(tt.name)
 
 
-def test_ingest():
+def test_ingest(temp_file):
+    # Write out a temporary mrc file
+    mrc.mrcWriter(temp_file, np.ones((10, 11, 12), dtype=np.float32), (1, 2, 3))
 
-    # Write a small mrc file
-    dd, _, _ = np.mgrid[0:30, 0:40, 0:50]
-    dd = dd.astype('<u2')
-    tmp = tempfile.TemporaryDirectory()
-    fPath = str(Path(tmp.name) / Path('temp_mrc.mrc'))
-    mrc.mrcWriter(fPath, dd, (0.1, 0.2, 0.3))
+    assert temp_file.exists() is True
 
     # Test
-    with mrc.fileMRC(fPath) as mrc_obj:
-        assert mrc_obj.getSlice(0).shape == (40, 50)
-    docs = list(ingest_NCEM_MRC([fPath]))
+    with mrc.fileMRC(temp_file) as mrc_obj:
+        assert mrc_obj.getSlice(0).shape == (11, 12)
+    docs = list(ingest_NCEM_MRC([temp_file]))
     event_doc = docs[2][1]
     data = event_doc['data']['raw']
-    assert data.shape == (30, 40, 50)
-    assert data[0].compute().shape == (40, 50)
+    assert data.shape == (10, 11, 12)
+    assert data[0].compute().shape == (11, 12)

tests/test_SER.py

@@ -1,17 +1,17 @@
+from pathlib import Path
+
 import pytest
-from ncempy.io import ser
 from xicam.NCEM.ingestors.SERPlugin import ingest_NCEM_SER, _get_slice
 
 
-# TODO: add fixture that writes temp data file with ncempy for tests
-
 @pytest.fixture
 def SER_path():
-    return "/home/rp/data/NCEM/10_series_1.ser"
+    """SER files cant be written. We can only use data from a microscope"""
+    dPath = Path.home()
+    return str(dPath / Path('data') / Path('10_series_1.ser'))
 
 
 def test_slicing(SER_path):
-    # with ser.fileSER(SER_path) as ser_obj:
     assert _get_slice(SER_path, 0).shape == (512, 512)
     docs = list(ingest_NCEM_SER([SER_path]))
     event_doc = docs[2][1]