Commit 4d3bdd36 authored by payno's avatar payno
Browse files

[core][test] add `TestSaveFlowAuto` to make sure the process is correctly saved.

parent 7cd757c0
......@@ -30,6 +30,7 @@ __date__ = "06/11/2019"
import unittest
import os
from PyMca5.PyMcaDataDir import PYMCA_DATA_DIR
from xas.io import XASWriter
from silx.io.url import DataUrl
from xas.core.process.exafs import pymca_exafs, PyMca_exafs
from xas.core.process.ft import pymca_ft, PyMca_ft
......@@ -42,6 +43,7 @@ from xas.core.types import XASObject
import tempfile
import shutil
import numpy
import h5py
class TestStreamSingleSpectrum(unittest.TestCase):
......@@ -119,7 +121,6 @@ class TestWorkflowAndH5LinkedFile(unittest.TestCase):
def test_h5_link_xas_object(self):
"""Test that the processing can be stored continuously on a .h5 file"""
self.assertTrue(self.xas_obj.linked_h5_file is not None)
out = PyMca_normalization()(xas_obj=self.xas_obj)
out = PyMca_exafs()(xas_obj=out)
......@@ -161,9 +162,44 @@ class TestWorkflowAndH5LinkedFile(unittest.TestCase):
self.assertEqual(len(processes), 0)
class TestSaveFlowAuto(unittest.TestCase):
"""Test that the processing can be stored continuously on a temporary
.h5 file then dump without lost during call to 'dump_xas' of the XASObject
"""
def setUp(self):
self.output_dir = tempfile.mkdtemp()
spectra = spectra_utils.create_dataset(shape=(10, 10, 10))
energy = numpy.linspace(start=3.26, stop=3.96, num=10)
self.xas_obj = XASObject(spectra=spectra, energy=energy)
self.h5_file = os.path.join(self.output_dir, 'output_file.h5')
def tearDown(self):
shutil.rmtree(self.output_dir)
def test(self):
self.assertTrue(self.xas_obj.linked_h5_file is not None)
out = PyMca_normalization()(xas_obj=self.xas_obj)
out = PyMca_exafs()(xas_obj=out)
out = PyMca_k_weight()(xas_obj=out)
out = PyMca_ft()(xas_obj=out)
fina_xas_obj = PyMca_normalization()(xas_obj=out)
writer = XASWriter()
writer.output_file = self.h5_file
writer(fina_xas_obj)
with h5py.File(self.h5_file) as hdf:
self.assertTrue('scan1' in hdf.keys())
self.assertTrue('data' in hdf['scan1'].keys())
self.assertTrue('absorbed_beam' in hdf['scan1'].keys())
self.assertTrue('monochromator' in hdf['scan1'].keys())
self.assertTrue('xas_process_1' in hdf['scan1'].keys())
def suite():
test_suite = unittest.TestSuite()
for ui in (TestStreamSingleSpectrum, TestWorkflowAndH5LinkedFile):
for ui in (TestStreamSingleSpectrum, TestWorkflowAndH5LinkedFile,
TestSaveFlowAuto):
test_suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(ui))
return test_suite
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment