import logging
import os
import shutil
import tempfile
import unittest

import numpy
from astropy.io import fits
import cpl
cpl.Recipe.memory_mode = 0

recipe_name = 'rtest'
raw_tag = 'RRRECIPE_DOCATG_RAW'
rtest_esorexrc = '''# example RC file
iiinstrument.rtest.string_option = more
iiinstrument.rtest.bool_option = False
iiinstrument.rtest.float_option = 1.125
iiinstrument.rtest.int_option = -2
iiinstrument.rtest.enum_option = second
iiinstrument.rtest.range_option = 0.5
iiinstrument.rtest.dotted.opt = 1
'''

def create_recipe(name, builddir):
    var = {
        'CC':  os.getenv("CC", "gcc"),
        'CPPFLAGS': os.getenv("CPPFLAGS", ""),
        'CFLAGS': os.getenv("CFLAGS", ""),
        'LDFLAGS': os.getenv("LDFLAGS", ""),
        'LIBS': "-lcplcore -lcpldfs",
        'cname': os.path.join(os.path.dirname(__file__), name + ".c"),
        'oname': os.path.join(builddir, name + '.o'),
        'soname': os.path.join(builddir, name + '.so'),
    }
    os.system("{CC} {CPPFLAGS} {CFLAGS} -fPIC -c -o {oname} {cname}".format(**var))
    os.system("{CC} {LDFLAGS} -shared -o {soname} {oname} {LIBS}".format(**var))
    os.remove(var['oname'])

class CplTestCase(unittest.TestCase):
    def setUp(self):
        unittest.TestCase.setUp(self)
        self.temp_dir = tempfile.mkdtemp()
        create_recipe(recipe_name, self.temp_dir)
        cpl.Recipe.path = self.temp_dir

    def tearDown(self):
        unittest.TestCase.tearDown(self)
        shutil.rmtree(self.temp_dir)

class RecipeTestCase(CplTestCase):
    def setUp(self):
        CplTestCase.setUp(self)
        self.recipe = cpl.Recipe(recipe_name)
        self.recipe.temp_dir = self.temp_dir
        self.recipe.tag = raw_tag
        self.image_size = (16, 16)
        self.raw_frame = fits.HDUList([
                fits.PrimaryHDU(numpy.random.randint(0, 65001,
                                                     self.image_size))])
        self.raw_frame[0].header['HIERARCH ESO DET DIT'] = 0.0
        self.raw_frame[0].header['HIERARCH ESO PRO CATG'] = raw_tag

class RecipeStatic(CplTestCase):
    def test_list(self):
        '''List available recipes'''
        l = cpl.Recipe.list()
        self.assertTrue(isinstance(l, list))
        self.assertEqual(len(l), 1)
        self.assertEqual(l[0], (recipe_name, ['0.0.1']))

    def test_create_recipe(self):
        '''Create a recipe specified by its name'''
        recipe = cpl.Recipe(recipe_name)
        self.assertTrue(isinstance(recipe, cpl.Recipe))

    def test_create_recipe_version(self):
        '''Create a recipe specified by its name and version'''
        recipe = cpl.Recipe(recipe_name, version = '0.0.1')
        self.assertTrue(isinstance(recipe, cpl.Recipe))        

    def test_create_recipe_wrong_name(self):
        '''Create a recipe specified by a wrong name'''
        self.assertRaises(IOError, cpl.Recipe, 'wrongname')

    def test_create_recipe_wrong_version(self):
        '''Create a recipe specified by a wrong version'''
        self.assertRaises(IOError, cpl.Recipe, recipe_name, version='0.0.10')

    def test_create_recipe_filename(self):
        '''Create a recipe specified by a the name and the filename'''
        recipe = cpl.Recipe(recipe_name,
                            filename = os.path.join(self.temp_dir, 'rtest.so'))
        self.assertTrue(isinstance(recipe, cpl.Recipe))

    def test_create_recipe_wrong_filename(self):
        '''Create a recipe specified by a wrong filename'''
        self.assertRaises(IOError, cpl.Recipe, recipe_name, 
                          filename = os.path.join(
                os.path.dirname(os.path.abspath(__file__)),
                'iiinstrumentp', 'recipes', '.libs', 'rtest.o'))

class RecipeCommon(RecipeTestCase):
    def test_name(self):
        '''Recipe name'''
        self.assertEqual(self.recipe.__name__, recipe_name)

    def test_author(self):
        '''Author attribute'''
        self.assertEqual(self.recipe.__author__, 'Ole Streicher')

    def test_email(self):
        '''Author attribute'''
        self.assertEqual(self.recipe.__email__, 'python-cpl@liska.ath.cx')

    def test_description(self):
        '''Synopsis and description'''
        self.assertTrue(isinstance(self.recipe.description[0], str))
        self.assertTrue(len(self.recipe.description[0]) > 0)
        self.assertTrue(isinstance(self.recipe.description[1], str))
        self.assertTrue(len(self.recipe.description[1]) > 0)

    def test_copyright(self):
        '''Copyright'''
        self.assertTrue(isinstance(self.recipe.__copyright__, str))
        self.assertTrue(len(self.recipe.__copyright__) > 0)

class RecipeParams(RecipeTestCase):
    def test_string_parameter(self):
        '''String parameter'''
        self.assertTrue(isinstance(self.recipe.param.stropt, cpl.Parameter))
        self.assertEqual(self.recipe.param.stropt.name, 'stropt')
        self.assertEqual(self.recipe.param.stropt.context, 
                         'iiinstrument.rtest')
        self.assertEqual(self.recipe.param.stropt.default, None)
        self.assertEqual(self.recipe.param.stropt.value, None)
        self.assertEqual(self.recipe.param.stropt.range, None)
        self.assertEqual(self.recipe.param.stropt.sequence, None)
        self.recipe.param.stropt = 'more'
        self.assertEqual(self.recipe.param.stropt.value, 'more')
        del self.recipe.param.stropt 
        self.assertEqual(self.recipe.param.stropt.value, None)

    def test_boolean_parameter(self):
        '''Boolean parameter'''
        self.assertTrue(isinstance(self.recipe.param.boolopt, cpl.Parameter))
        self.assertEqual(self.recipe.param.boolopt.name, 'boolopt')
        self.assertEqual(self.recipe.param.boolopt.default, True)
        self.assertEqual(self.recipe.param.boolopt.value, None)
        self.recipe.param.boolopt = False
        self.assertEqual(self.recipe.param.boolopt.value, False)
        del self.recipe.param.boolopt 
        self.assertEqual(self.recipe.param.boolopt.value, None)
        
    def test_float_parameter(self):
        '''Float parameter'''
        self.assertTrue(isinstance(self.recipe.param.floatopt, cpl.Parameter))
        self.assertEqual(self.recipe.param.floatopt.name, 'floatopt')
        self.assertEqual(self.recipe.param.floatopt.default, 0.1)
        self.assertEqual(self.recipe.param.floatopt.value, None)
        self.recipe.param.floatopt = 1.1
        self.assertEqual(self.recipe.param.floatopt.value, 1.1)
        del self.recipe.param.floatopt 
        self.assertEqual(self.recipe.param.floatopt.value, None)

    def test_int_parameter(self):
        '''Integer parameter'''
        self.assertTrue(isinstance(self.recipe.param.intopt, cpl.Parameter))
        self.assertEqual(self.recipe.param.intopt.name, 'intopt')
        self.assertEqual(self.recipe.param.intopt.default, 2)
        self.assertEqual(self.recipe.param.intopt.value, None)
        self.recipe.param.intopt = -1
        self.assertEqual(self.recipe.param.intopt.value, -1)
        del self.recipe.param.intopt 
        self.assertEqual(self.recipe.param.intopt.value, None)

    def test_enum_parameter(self):
        '''Enumeration (string) parameter'''
        self.assertTrue(isinstance(self.recipe.param.enumopt, cpl.Parameter))
        self.assertEqual(self.recipe.param.enumopt.name, 'enumopt')
        self.assertEqual(self.recipe.param.enumopt.default, 'first')
        self.assertEqual(self.recipe.param.enumopt.value, None)
        self.recipe.param.enumopt = 'second'
        self.assertEqual(self.recipe.param.enumopt.value, 'second')
        del self.recipe.param.enumopt 
        self.assertEqual(self.recipe.param.enumopt.value, None)
        def setenumoptinvalid():
            self.recipe.param.enumopt = 'invalid'
        self.assertRaises(ValueError, setenumoptinvalid)

    def test_range_parameter(self):
        '''Range (float) parameter'''
        self.assertTrue(isinstance(self.recipe.param.rangeopt, cpl.Parameter))
        self.assertEqual(self.recipe.param.rangeopt.name, 'rangeopt')
        self.assertEqual(self.recipe.param.rangeopt.default, 0.1)
        self.assertEqual(self.recipe.param.rangeopt.value, None)
        self.recipe.param.rangeopt = 0.4
        self.assertEqual(self.recipe.param.rangeopt.value, 0.4)
        del self.recipe.param.rangeopt 
        self.assertEqual(self.recipe.param.rangeopt.value, None)
        def setrangeoptinvalid():
            self.recipe.param.rangeopt = 1.5
        self.assertRaises(ValueError, setrangeoptinvalid)

    def test_as_dict(self):
        '''Use the parameter list as a dictionary'''
        self.assertEqual(self.recipe.param.boolopt, 
                         self.recipe.param['boolopt'])
        self.assertEqual(self.recipe.param.boolopt, 
                         self.recipe.param['iiinstrument.rtest.bool_option'])

    def test_dotted_par(self):
        '''Use a parameter that has a dot in its alias'''
        self.assertEqual(self.recipe.param.dot.opt,
                         self.recipe.param.dot['opt'])
        self.assertEqual(self.recipe.param.dot.opt,
                         self.recipe.param['dot.opt'])
        self.assertEqual(self.recipe.param.dot.opt,
                         self.recipe.param['iiinstrument.rtest.dotted.opt'])

    def test_iterate(self):
        '''Iteration over all parameters'''
        for p in self.recipe.param:
            self.assertTrue(isinstance(p, cpl.Parameter))
        pars = [p.name for p in self.recipe.param]
        self.assertEqual(len(pars), len(self.recipe.param))
        self.assertTrue('stropt' in pars)
        self.assertTrue('boolopt' in pars)

    def test_set_dict(self):
        '''Assign a dictionary to the parameter list'''
        self.recipe.param = { 'stropt':'dmore', 'boolopt':True }
        self.assertEqual(self.recipe.param.boolopt.value, True)
        self.assertEqual(self.recipe.param.stropt.value, 'dmore')

        # Check that we can assign a dictionary with the short names and string
        self.recipe.param = { 'stropt':'dmore', 'boolopt':'False' }
        self.assertEqual(self.recipe.param.boolopt.value, False)

        # Check that we can assign a dictionary with the long names
        self.recipe.param = { 'iiinstrument.rtest.string_option':'dless', 
                      'iiinstrument.rtest.float_option':1.5, 
                      'iiinstrument.rtest.bool_option':True }
        self.assertEqual(self.recipe.param.stropt.value, 'dless')
        self.assertEqual(self.recipe.param.floatopt.value, 1.5)
        self.assertEqual(self.recipe.param.boolopt.value, True)

    def test_param_load_from_esorexrc(self):
        '''Load parameters from an Esorex .rc file'''
        self.recipe.param = rtest_esorexrc
        self.assertEqual(self.recipe.param.stropt.value, 'more')
        self.assertEqual(self.recipe.param.boolopt.value, False)
        self.assertEqual(self.recipe.param.floatopt.value, 1.125)
        self.assertEqual(self.recipe.param.intopt.value, -2)
        self.assertEqual(self.recipe.param.enumopt.value, 'second')
        self.assertEqual(self.recipe.param.rangeopt.value, 0.5)
        self.assertEqual(self.recipe.param.dot.opt.value, 1)

    def test_delete(self):
        '''Delete all parameter values to reset to default'''
        self.recipe.param.boolopt.value = True
        self.recipe.param.stropt.value = 'something'
        del self.recipe.param
        self.assertEqual(self.recipe.param.stropt.value, None)
        self.assertEqual(self.recipe.param.boolopt.value, None)

    def test_dir(self):
        '''[TAB] completition. 
        This requires to have   the __dir__() method working.
        '''
        self.assertEqual(set(self.recipe.param.__dir__()),
                         set(p.name if '.' not in p.name
                             else p.name.split('.', 1)[0]
                             for p in self.recipe.param
                             ))

    def test_str(self):
        '''Parameter string representation.
        
        Since the string depends on the order of parameter, we check
        that it evaluates to the correct dict.
        '''
        self.assertEqual(eval(str(self.recipe.param)), {
            'stropt': None,
            'boolopt': True,
            'floatopt': 0.1,
            'intopt': 2,
            'enumopt': 'first',
            'rangeopt': 0.1,
            'crashing': 'no',
            'memleak': False,
            'sleep': 0.1,
            'disabled': -0.1,
            'dot.opt': 0})
        
    def test_repr(self):
        '''Canonical parameter string representation.
        
        Since the string depends on the order of parameter, we check
        that it evaluates to the correct dict.
        '''
        self.assertEqual(eval(repr(self.recipe.param)), {
            'stropt': None,
            'boolopt': True,
            'floatopt': 0.1,
            'intopt': 2,
            'enumopt': 'first',
            'rangeopt': 0.1,
            'crashing': 'no',
            'memleak': False,
            'sleep': 0.1,
            'disabled': -0.1,
            'dot.opt': 0})

    def test_eq(self):
        '''Trivial equality test'''
        self.assertTrue(self.recipe.param == self.recipe.param)

        
class RecipeCalib(RecipeTestCase):
    def test_set(self):
        '''Set a calibration frame'''
        self.recipe.calib.FLAT = 'flat.fits'
        self.assertEqual(self.recipe.calib.FLAT.frames, 'flat.fits')
        
    def test_set_dict(self):
        '''Assign a dictionary to the calibration frame list'''
        self.recipe.calib = { 'FLAT':'flat2.fits' }
        self.assertEqual(self.recipe.calib.FLAT.frames, 'flat2.fits')

    def test_param_load_from_esorexsof(self):
        '''Load calibration from an Esorex SOF file'''
        self.recipe.calib = 'flat.fits   FLAT\n'
        self.assertEqual(self.recipe.calib.FLAT.frames, 'flat.fits')

    def test_del(self):
        '''Delete a calibration frame set'''
        self.recipe.calib.FLAT = 'flat.fits'
        del self.recipe.calib.FLAT
        f = self.recipe.calib.FLAT.frames
        self.assertEqual(f, None)

    def test_del_all(self):
        '''Delete all calibration frame sets'''
        self.recipe.calib.FLAT = 'flat.fits'
        del self.recipe.calib
        try:
            f = self.recipe.calib.FLAT.frames
        except:
            f = None
        self.assertEqual(f, None)

    def test_dir(self):
        '''[TAB] completition. 
        This requires to have   the __dir__() method working.
        '''
        self.assertEqual(set(self.recipe.calib.__dir__()),
                         set(f.tag for f in self.recipe.calib))

    def test_str(self):
        '''Calibration frames string representation'''
        self.assertEqual(str(self.recipe.calib),
                             "{}")
        
    def test_repr(self):
        '''Canonical calibration frames string representation'''
        self.assertEqual(repr(self.recipe.calib),
                             "{}")

    def test_eq(self):
        '''Trivial equality test'''
        self.assertTrue(self.recipe.calib == self.recipe.calib)

class RecipeExec(RecipeTestCase):
    def setUp(self):
        RecipeTestCase.setUp(self)
        self.flat_frame = fits.HDUList([
                fits.PrimaryHDU(numpy.random.randint(0, 65001,
                                                     self.image_size))])

    def test_frames_keyword_dict(self):
        '''Raw and calibration frames specified as keyword dict'''
        self.recipe.tag = None
        res = self.recipe(raw = {'RRRECIPE_DOCATG_RAW': self.raw_frame },
                          calib = { 'FLAT':self.flat_frame })
        self.assertTrue(isinstance(res, cpl.Result))
        self.assertTrue(isinstance(res.THE_PRO_CATG_VALUE, fits.HDUList))
        self.assertTrue(abs(self.raw_frame[0].data 
                            - res.THE_PRO_CATG_VALUE[0].data).max() == 0)
        try:
            res.THE_PRO_CATG_VALUE.close()
        except:
            pass

    def test_frames_keyword_calib(self):
        '''Raw frame specified as keyword, calibration frame set in recipe'''
        self.recipe.tag = None
        self.recipe.calib.FLAT = self.flat_frame
        res = self.recipe({'RRRECIPE_DOCATG_RAW':self.raw_frame})
        self.assertTrue(isinstance(res, cpl.Result))
        self.assertTrue(isinstance(res.THE_PRO_CATG_VALUE, fits.HDUList))
        try:
            res.THE_PRO_CATG_VALUE.close()
        except:
            pass

    def test_frames_tag_keyword(self):
        '''The 'tag' parameter'''
        self.recipe.tag = None
        self.recipe.calib.FLAT = self.flat_frame
        res = self.recipe(self.raw_frame, tag = raw_tag)
        self.assertTrue(isinstance(res, cpl.Result))
        self.assertTrue(isinstance(res.THE_PRO_CATG_VALUE, fits.HDUList))
        try:
            res.THE_PRO_CATG_VALUE.close()
        except:
            pass

    def test_frames_tag_attribute(self):
        '''The 'tag' attribute'''
        self.recipe.tag = raw_tag
        res = self.recipe(self.raw_frame)
        self.assertTrue(isinstance(res, cpl.Result))
        self.assertTrue(isinstance(res.THE_PRO_CATG_VALUE, fits.HDUList))
        try:
            res.THE_PRO_CATG_VALUE.close()
        except:
            pass

    def test_frames_one_element_input_list(self):
        '''Use 1-element list as input'''
        # --> we want a list back'''
        res = self.recipe([self.raw_frame])
        self.assertTrue(isinstance(res, cpl.Result))
        self.assertFalse(isinstance(res.THE_PRO_CATG_VALUE, fits.HDUList))
        self.assertTrue(isinstance(res.THE_PRO_CATG_VALUE, list))
        try:
            res.THE_PRO_CATG_VALUE[0].close()
        except:
            pass

    def test_frames_many_element_input_list(self):
        '''Use multiple files as input'''
        # --> since we only get back one image, it is
        # assumed to be a 'master', and we get back a plain frame'''
        res = self.recipe([self.raw_frame, self.raw_frame])
        self.assertTrue(isinstance(res, cpl.Result))
        self.assertTrue(isinstance(res.THE_PRO_CATG_VALUE, fits.HDUList))
        try:
            res.THE_PRO_CATG_VALUE.close()
        except:
            pass

    def test_output_dir_attribute(self):
        '''Write an output dir specified as attribute'''
        output_dir = os.path.join(self.temp_dir, 'out')
        self.recipe.output_dir = output_dir
        res = self.recipe(self.raw_frame)
        self.assertTrue(isinstance(res, cpl.Result))
        self.assertTrue(isinstance(res.THE_PRO_CATG_VALUE, str))
        self.assertEqual(os.path.basename(res.THE_PRO_CATG_VALUE), 
                         'rtest.fits')
        self.assertTrue(os.path.isdir(output_dir))
        self.assertTrue(os.path.isfile(res.THE_PRO_CATG_VALUE))
        hdu = fits.open(res.THE_PRO_CATG_VALUE)
        self.assertTrue(isinstance(hdu, fits.HDUList))
        try:
            hdu.close()
        except:
            pass

    def test_output_dir_keyword(self):
        '''Write an output dir specified as call keyword arg'''
        output_dir = os.path.join(self.temp_dir, 'out')
        res = self.recipe(self.raw_frame, output_dir = output_dir)
        self.recipe.output_dir = output_dir
        res = self.recipe(self.raw_frame)
        self.assertTrue(os.path.isdir(output_dir))
        self.assertTrue(isinstance(res, cpl.Result))
        self.assertTrue(isinstance(res.THE_PRO_CATG_VALUE, str))
        self.assertEqual(os.path.basename(res.THE_PRO_CATG_VALUE), 
                         'rtest.fits')
        self.assertTrue(os.path.isfile(res.THE_PRO_CATG_VALUE))
        hdu = fits.open(res.THE_PRO_CATG_VALUE)
        self.assertTrue(isinstance(hdu, fits.HDUList))
        try:
            hdu.close()
        except:
            pass

    def test_param_default(self):
        '''Test default parameter settings'''
        res = self.recipe(self.raw_frame).THE_PRO_CATG_VALUE
        self.assertEqual(res[0].header['HIERARCH ESO QC STROPT'].strip(),
                         self.recipe.param.stropt.default or '')
        self.assertEqual(res[0].header['HIERARCH ESO QC BOOLOPT'],
                         self.recipe.param.boolopt.default)
        self.assertEqual(res[0].header['HIERARCH ESO QC INTOPT'],
                         self.recipe.param.intopt.default)
        self.assertEqual(res[0].header['HIERARCH ESO QC FLOATOPT'],
                         self.recipe.param.floatopt.default)
        self.assertEqual(res[0].header['HIERARCH ESO QC ENUMOPT'],
                         self.recipe.param.enumopt.default)
        self.assertEqual(res[0].header['HIERARCH ESO QC RANGEOPT'],
                         self.recipe.param.rangeopt.default)
        try:
            res.close()
        except:
            pass

    def test_param_keyword_dict(self):
        '''Parameter handling via keyword dict'''
        res = self.recipe(self.raw_frame, 
                          param = { 'stropt':'more' }).THE_PRO_CATG_VALUE
        self.assertEqual(res[0].header['HIERARCH ESO QC STROPT'], 'more')
        try:
            res.close()
        except:
            pass

    def test_param_keyword_dict_wrong(self):
        '''Parameter handling via keyword dict'''
        self.assertRaises(KeyError, self.recipe,
                          self.raw_frame, param = { 'wrong':True })

    def test_param_setting(self):
        '''Parameter handling via recipe setting'''
        self.recipe.param.stropt = 'more'
        with self.recipe(self.raw_frame).THE_PRO_CATG_VALUE as res:
            self.assertEqual(res[0].header['HIERARCH ESO QC STROPT'], 'more')

    def test_param_delete(self):
        '''Delete a parameter in a second run after setting it'''
        self.recipe.param.intopt = 123
        with self.recipe(self.raw_frame).THE_PRO_CATG_VALUE as res:
            pass
        del self.recipe.param.intopt
        with self.recipe(self.raw_frame).THE_PRO_CATG_VALUE as res:
            self.assertEqual(res[0].header['HIERARCH ESO QC INTOPT'], 2)

    def test_param_overwrite(self):
        '''Overwrite the recipe setting param via via keyword arg'''
        self.recipe.param.stropt = 'more'
        res = self.recipe(self.raw_frame, param = {'stropt':'less'}).THE_PRO_CATG_VALUE
        self.assertEqual(res[0].header['HIERARCH ESO QC STROPT'], 'less')

    def test_param_types(self):
        '''Parameter types'''
        self.recipe.param.stropt = 'more'
        self.recipe.param.boolopt = False
        self.recipe.param.intopt = 123
        self.recipe.param.floatopt = -0.25
        self.recipe.param.enumopt = 'third'
        self.recipe.param.rangeopt = 0.125
        with self.recipe(self.raw_frame).THE_PRO_CATG_VALUE as res:
            self.assertEqual(res[0].header['HIERARCH ESO QC STROPT'], 'more')
            self.assertEqual(res[0].header['HIERARCH ESO QC BOOLOPT'], False)
            self.assertEqual(res[0].header['HIERARCH ESO QC INTOPT'], 123)
            self.assertEqual(res[0].header['HIERARCH ESO QC FLOATOPT'], -0.25)
            self.assertEqual(res[0].header['HIERARCH ESO QC ENUMOPT'], 'third')
            self.assertEqual(res[0].header['HIERARCH ESO QC RANGEOPT'], 0.125)
        
    def test_disabled(self):
        '''Parameter with CLI disabled'''
        self.assertFalse(self.recipe.param.disabled.enabled[0])
        self.assertTrue(self.recipe.param.intopt.enabled[0])
#        self.recipe.param.disabled = 0.2
#        res = self.recipe(self.raw_frame)
#        self.assertEqual(res[0].header['HIERARCH ESO QC DISABLED'], 0.2)
#        try:
#            res.close()
#        except:
#            pass

    def test_environment_setting(self):
        '''Additional environment parameter via recipe setting'''
        self.recipe.env['TESTENV'] = 'unkk'
        with self.recipe(self.raw_frame).THE_PRO_CATG_VALUE as res:
            self.assertEqual(res[0].header['HIERARCH ESO QC TESTENV'], 'unkk')

    def test_environment_keyword(self):
        '''Additional environment parameter via recipe call keyword'''
        with self.recipe(self.raw_frame, 
                         env = {'TESTENV':'kknu'}).THE_PRO_CATG_VALUE as res:
            self.assertEqual(res[0].header['HIERARCH ESO QC TESTENV'], 'kknu')

    def test_error(self):
        '''Error handling'''
        self.recipe.tag = 'some_unknown_tag'
        self.assertRaises(cpl.CplError, self.recipe, self.raw_frame)

    def test_parallel(self):
        '''Parallel execution'''
        results = list()
        for i in range(20):
            # mark each frame so that we can see their order
            self.raw_frame[0].header['HIERARCH ESO RAW1 NR'] = i
            results.append(self.recipe(self.raw_frame, param = {'intopt':i},
                                       env = {'TESTENV':('knu%02i' % i)},
                                       threaded = True))
        for i, res in enumerate(results):
            # check if we got the correct type
            self.assertTrue(isinstance(res.THE_PRO_CATG_VALUE, fits.HDUList))
            # check if we have the correct parameter
            self.assertEqual(res.THE_PRO_CATG_VALUE[0].header[
                    'HIERARCH ESO QC INTOPT'], i)
            # check if we have the correct environment
            self.assertEqual(res.THE_PRO_CATG_VALUE[0].header[
                    'HIERARCH ESO QC TESTENV'], ('knu%02i' % i))
            # check if we have the correct input frame
            self.assertEqual(res.THE_PRO_CATG_VALUE[0].header[
                    'HIERARCH ESO RAW1 NR'], i)
            # check that the data were moved correctly
            self.assertTrue(abs(self.raw_frame[0].data 
                                - res.THE_PRO_CATG_VALUE[0].data).max() < 1e-6)
            try:
                res.THE_PRO_CATG_VALUE.close()
            except:
                pass

    def test_error_parallel(self):
        '''Error handling in parallel execution'''
        self.recipe.tag = 'some_unknown_tag'
        res = self.recipe(self.raw_frame, threaded = True)
        def get(x):
            return x.THE_PRO_CATG_VALUE
        self.assertRaises(cpl.CplError, get, res)

    def test_md5sum_result(self):
        '''MD5sum of the result file'''
        self.recipe.tag = raw_tag
        res = self.recipe(self.raw_frame)
        key = 'DATAMD5'
        md5sum = res.THE_PRO_CATG_VALUE[0].header[key]
        try:
            res.THE_PRO_CATG_VALUE.close()
        except:
            pass
        self.assertNotEqual(md5sum, 'Not computed')
        self.assertEqual(len(md5sum), 
                         len('9d123996fa9a7bda315d07e063043454'))

    def test_md5sum_calib(self):
        '''Created MD5sum for a HDUList calib file'''
        self.recipe.tag = raw_tag
        self.recipe.calib.FLAT = self.flat_frame
        res = self.recipe(self.raw_frame)
        key = 'HIERARCH ESO PRO REC1 CAL1 DATAMD5'
        md5sum = res.THE_PRO_CATG_VALUE[0].header[key]
        try:
            res.THE_PRO_CATG_VALUE.close()
        except:
            pass
        self.assertNotEqual(md5sum, 'Not computed')
        self.assertEqual(len(md5sum), 
                         len('9d123996fa9a7bda315d07e063043454'))

class RecipeCrashing(RecipeTestCase):
    def _test_corrupted(self):
        '''Handling of recipe crashes because of corrupted memory'''
        self.recipe.param.crashing = 'free'
        self.assertRaises(cpl.RecipeCrash, self.recipe, self.raw_frame)

    def _test_segfault(self):
        '''Handling of recipe crashes because of segmentation fault'''
        self.recipe.param.crashing = 'segfault'
        self.assertRaises(cpl.RecipeCrash, self.recipe, self.raw_frame)

    def _test_cleanup_after_crash(self):
        '''Test that a second run after a crash will succeed'''
        output_dir = os.path.join(self.temp_dir, 'out')
        self.recipe.output_dir = output_dir
        self.recipe.param.crashing = 'segfault'
        self.assertRaises(cpl.RecipeCrash, self.recipe, self.raw_frame)
        del self.recipe.param.crashing
        self.recipe(self.raw_frame)

class RecipeRes(RecipeTestCase):
    def setUp(self):
        RecipeTestCase.setUp(self)
        self.res = self.recipe(self.raw_frame)

    def tearDown(self):
        RecipeTestCase.tearDown(self)
        try:
            res.THE_PRO_CATG_VALUE.close()
        except:
            pass

    def test_attribute(self):
        '''The result as an attribute'''
        self.assertTrue(isinstance(self.res.THE_PRO_CATG_VALUE, 
                                   fits.HDUList))

    def test_dict(self):
        '''The result as an attribute'''
        self.assertTrue(isinstance(self.res['THE_PRO_CATG_VALUE'], 
                                   fits.HDUList))
    def test_in(self):
        '''Check whether a tag is part of the result'''
        self.assertTrue('THE_PRO_CATG_VALUE' in self.res)
        self.assertFalse('Anothervalue' in self.res)

    def test_keyerror(self):
        '''Accessing an inexisting value'''
        self.assertRaises(KeyError, lambda val: self.res[val], 'Anothervalue')

    def test_len(self):
        '''Length of the result'''
        self.assertEqual(len(self.res), 1)

    def test_iter(self):
        '''Iterate over the result'''
        for tag, hdu in self.res:
            self.assertEqual(tag, 'THE_PRO_CATG_VALUE')
            self.assertTrue(isinstance(hdu, fits.HDUList))

class RecipeEsorex(CplTestCase):
    def tearDown(self):
        CplTestCase.tearDown(self)
        cpl.esorex.msg.level = cpl.esorex.msg.OFF
        cpl.esorex.log.level = cpl.esorex.msg.OFF

    def test_read_sof(self):
        '''Read a SOF file'''
        soffile = 'geometry_table1.fits GEOMETRY_TABLE\n' \
            'geometry_table2.fits GEOMETRY_TABLE\n' \
            'MASTER_BIAS-01.fits MASTER_BIAS\n' \
            'MASTER_FLAT-01.fits MASTER_FLAT\n' \
            '#sky_fullmoon_1.fits          SKY\n' \
            'sky_fullmoon_2.fits          SKY\n'
        self.assertEqual(cpl.esorex.load_sof(soffile),
                         { 'GEOMETRY_TABLE': ['geometry_table1.fits',
                                              'geometry_table2.fits' ],
                           'MASTER_BIAS': 'MASTER_BIAS-01.fits',
                           'MASTER_FLAT': 'MASTER_FLAT-01.fits',
                           'SKY': 'sky_fullmoon_2.fits' })
                           
    def test_read_rc(self):
        '''Read an EsoRec .rc file'''
        self.assertEqual(cpl.esorex.load_rc(rtest_esorexrc), 
                         { 'iiinstrument.rtest.string_option': 'more',
                           'iiinstrument.rtest.bool_option': 'False',
                           'iiinstrument.rtest.float_option': '1.125',
                           'iiinstrument.rtest.int_option': '-2',
                           'iiinstrument.rtest.enum_option': 'second',
                           'iiinstrument.rtest.range_option': '0.5',
                           'iiinstrument.rtest.dotted.opt': '1',
                       })
        
    def test_esorex_init(self):
        '''Init CPL from an esorex.rc file'''
        rcfile = '''esorex.caller.recipe-dir=/some/dir
        esorex.caller.msg-level=debug
        esorex.caller.log-level=info
        esorex.caller.log-dir=%s
        esorex.caller.log-file=some.log''' % self.temp_dir
        cpl.esorex.init(rcfile)
        self.assertEqual(cpl.esorex.msg.level, cpl.esorex.msg.DEBUG)
        self.assertEqual(cpl.esorex.log.level, cpl.esorex.msg.INFO)
        self.assertEqual(cpl.esorex.log.dir, self.temp_dir)
        self.assertEqual(cpl.esorex.log.filename, 'some.log')
        self.assertEqual(cpl.Recipe.path, ['/some/dir'])

    def test_esorex_log(self):
        '''Write a logfile controlled by the convienence logger'''
        dirname = os.path.join(self.temp_dir, 'log')
        filename = 'python-cpl.log'
        log_msg = 'Esorex convienence log'
        os.mkdir(dirname)
        cpl.esorex.log.dir = dirname
        cpl.esorex.log.filename = filename
        cpl.esorex.log.level = cpl.esorex.log.INFO
        filename = os.path.join(dirname, filename)
        logging.getLogger('cpl').info(log_msg)
        self.assertTrue(os.path.exists(filename))
        logfile = open(filename)
        log_content = logfile.read()
        logfile.close()
        self.assertTrue(log_msg in log_content)
        self.assertTrue('INFO' in log_content)

    def test_esorex_log_off(self):
        '''Switch the logfile off after writing something'''
        dirname = os.path.join(self.temp_dir, 'log')
        filename = 'python-cpl_off.log'
        log_msg = 'Esorex convienence log'
        os.mkdir(dirname)
        cpl.esorex.log.dir = dirname
        cpl.esorex.log.filename = 'python-cpl_debug.log'
        cpl.esorex.log.level = 'debug'
        logging.getLogger('cpl').debug(log_msg)
        cpl.esorex.log.filename = filename
        cpl.esorex.log.level = 'off'
        logging.getLogger('cpl').debug(log_msg)
        filename = os.path.join(dirname, filename)
        logfile = open(filename)
        log_content = logfile.read()
        logfile.close()
        self.assertEqual(len(log_content), 0)

class RecipeLog(RecipeTestCase):
    def setUp(self):
        RecipeTestCase.setUp(self)
        self.handler = RecipeLog.THandler()
        logging.getLogger('cpl.rtest').addHandler(self.handler)
        self.other_handler = RecipeLog.THandler()
        logging.getLogger('othername').addHandler(self.other_handler)

    def tearDown(self):
        RecipeTestCase.tearDown(self)
        logging.getLogger('cpl.rtest').removeHandler(self.handler)
        logging.getLogger('othername').removeHandler(self.other_handler)

    class THandler(logging.Handler):
        def __init__(self):
            logging.Handler.__init__(self)
            self.logs = list()

        def emit(self, record):
            self.logs.append(record)

        def clear(self):
            self.logs = list()

    def test_logging_DEBUG(self):
        '''Injection of CPL messages into the python logging system'''
        self.handler.clear()
        logging.getLogger().setLevel(logging.DEBUG)
        res = self.recipe(self.raw_frame)
        try:
            res.THE_PRO_CATG_VALUE.close()
        except:
            pass

        # check that the logs are not empty
        self.assertNotEqual(len(self.handler.logs), 0)
        funcnames = set()
        lognames = set()
        for r in self.handler.logs:
            # Check that we saved the right class
            self.assertTrue(isinstance(r, logging.LogRecord))
            # Check that a message was provided
            self.assertNotEqual(r.msg, None)
            # Check that a function name was provided
            self.assertNotEqual(r.funcName, None)
            funcnames.add(r.funcName)
            lognames.add(r.name)
        # Check that we had at least one expected entry
        self.assertTrue('cpl_dfs_product_save' in funcnames)
        self.assertTrue('cpl.rtest.cpl_dfs_product_save' in lognames)
        
    def test_logging_INFO(self):
        '''Filtering INFO messages'''
        self.handler.clear()
        logging.getLogger('cpl.rtest').setLevel(logging.INFO)
        res = self.recipe(self.raw_frame)
        try:
            res.THE_PRO_CATG_VALUE.close()
        except:
            pass

        # check that the logs are not empty
        self.assertNotEqual(len(self.handler.logs), 0)

    def test_logging_WARN(self):
        '''Filtering WARN messages'''
        self.handler.clear()
        logging.getLogger('cpl.rtest').setLevel(logging.WARN)
        res = self.recipe(self.raw_frame)
        try:
            res.THE_PRO_CATG_VALUE.close()
        except:
            pass

        # check that the logs are not empty
        self.assertNotEqual(len(self.handler.logs), 0)

    def test_logging_ERROR(self):
        '''Filtering of error messages'''
        # There is no error msg written by the recipe, so it should be empty.
        self.handler.clear()
        logging.getLogger('cpl.rtest').setLevel(logging.ERROR)
        res = self.recipe(self.raw_frame)
        try:
            res.THE_PRO_CATG_VALUE.close()
        except:
            pass
        self.assertEqual(len(self.handler.logs), 0)

    def test_logging_common(self):
        '''Log name specification on recipe call'''
        self.handler.clear()
        self.other_handler.clear()
        res = self.recipe(self.raw_frame, logname = 'othername')
        try:
            res.THE_PRO_CATG_VALUE.close()
        except:
            pass
        self.assertNotEqual(len(self.other_handler.logs), 0)

    def test_logging_multiline(self):
        '''Multiple lines in messages'''
        self.handler.clear()
        logging.getLogger('cpl.rtest').setLevel(logging.INFO)
        res = self.recipe(self.raw_frame)
        try:
            res.THE_PRO_CATG_VALUE.close()
        except:
            pass
        # check that the multi line log sequence appears
        multiline = 0
        tag = 'multiline#'
        for l in self.handler.logs:
            if tag not in l.msg:
                continue
            i = int(l.msg[l.msg.index(tag)+len(tag):].split()[0])
            self.assertEqual(multiline + 1, i)
            multiline = i
        self.assertEqual(multiline, 3)

    def test_result(self):
        '''"log" attribute of the result object'''
        res = self.recipe(self.raw_frame)
        # Check that we get a not-empty list back
        self.assertTrue(isinstance(res.log, list))
        self.assertNotEqual(len(res.log), 0)
        self.assertTrue(isinstance(res.log[0], logging.LogRecord))

        # Check that we can read debug messages
        self.assertNotEqual(len(res.log.debug), 0)
        self.assertTrue(isinstance(res.log.debug[0], str))
        # Check that we can read info messages
        self.assertNotEqual(len(res.log.info), 0)
        self.assertTrue(isinstance(res.log.info[0], str))
        # Check that we can read warning messages
        self.assertNotEqual(len(res.log.warning), 0)
        self.assertTrue(isinstance(res.log.warning[0], str))
        # Check that there were no error messages
        self.assertEqual(len(res.log.error), 0)
        try:
            res.THE_PRO_CATG_VALUE.close()
        except:
            pass

    def test_error(self):
        '''"log" attribute of the CplError object'''
        try:
            self.recipe('test.fits')
        except cpl.CplError as r:
            res = r
        # Check that we get a not-empty list back
        self.assertTrue(isinstance(res.log, list))
        self.assertNotEqual(len(res.log), 0)
        self.assertTrue(isinstance(res.log[0], logging.LogRecord))
        # Check that we can read debug messages
        self.assertNotEqual(len(res.log.debug), 0)
        self.assertTrue(isinstance(res.log.debug[0], str))
        # Check that we can read info messages
        self.assertNotEqual(len(res.log.info), 0)
        self.assertTrue(isinstance(res.log.info[0], str))
        # Check that we can read warning messages
        self.assertNotEqual(len(res.log.warning), 0)
        self.assertTrue(isinstance(res.log.warning[0], str))
        # Check that we can read error messages
        self.assertNotEqual(len(res.log.error), 0)
        self.assertTrue(isinstance(res.log.error[0], str))
        # Check that we can convert the error to a string
        self.assertTrue(isinstance(res.__str__(), str))
        # Check that we can iterate over error messages
        for r in res:
            self.assertTrue(isinstance(res, cpl.CplError))

class ProcessingInfo(RecipeTestCase):
    def setUp(self):
        RecipeTestCase.setUp(self)
        '''Parameter storage in the result'''
        self.recipe.param.stropt = 'more'
        self.recipe.param.boolopt = False
        self.recipe.param.intopt = 123
        self.recipe.param.floatopt = -0.25
        self.recipe.param.enumopt = 'third'
        self.recipe.param.rangeopt = 0.125
        self.recipe.calib.FLAT = fits.HDUList([
                fits.PrimaryHDU(numpy.random.randint(0, 65001,
                                                     self.image_size))])
        self.res = self.recipe(self.raw_frame).THE_PRO_CATG_VALUE
        self.pinfo = cpl.dfs.ProcessingInfo(self.res)

    def tearDown(self):
        RecipeTestCase.tearDown(self)
        try:
            self.res.close()
        except:
            pass

    def test_list(self):
        '''All processing infos as a list'''
        pi = cpl.dfs.ProcessingInfo.list(self.res[0])
        self.assertTrue(len(pi), 1)
        self.assertTrue(pi[0], self.pinfo)

    def test_param(self):
        '''Parameter information'''
        self.assertEqual(len(self.pinfo.param), len(self.recipe.param))
        for p in self.recipe.param:
            self.assertEqual(self.pinfo.param[p.name], 
                             p.value if p.value is not None else p.default)

    def test_calib(self):
        '''Calibration frame information'''
        self.assertEqual(len(self.pinfo.calib), 1)
        self.assertEqual(self.pinfo.calib['FLAT'][-5:], '.fits')

    def test_tag(self):
        '''Input tag information'''
        self.assertEqual(self.pinfo.tag, self.recipe.tag)

    def test_raw(self):
        '''Raw file information'''
        self.assertEqual(self.pinfo.raw[-5:], '.fits')

    def test_name(self):
        '''Recipe and pipeline name information'''
        self.assertEqual(self.pinfo.name, self.recipe.__name__)
        self.assertEqual(self.pinfo.pipeline, 'iiinstrument')

    def test_version(self):
        '''Version information'''
        self.assertEqual(self.pinfo.version[0], self.recipe.version[0])
        self.assertEqual(self.pinfo.cpl_version, 'cpl-%s' % self.recipe.cpl_version)

    def test_md5(self):
        '''MD5 checksums'''
        md5sum = self.res[0].header.get('DATAMD5')
        self.assertEqual(md5sum, self.pinfo.md5sum)
        md5sum = self.res[0].header.get('HIERARCH ESO PRO REC1 CAL1 DATAMD5')
        self.assertEqual(md5sum, self.pinfo.md5sums[self.pinfo.calib['FLAT']])

    def test_recipe(self):
        '''Recreate and configure the recipe'''
        recipe = self.pinfo.create_recipe()
        self.assertTrue(isinstance(recipe, cpl.Recipe))
        self.assertEqual(len(recipe.calib), 1)
        self.assertEqual(recipe.calib.FLAT.frames[-5:], '.fits')
        self.assertEqual(recipe.__name__, self.recipe.__name__)
        self.assertEqual(recipe.version[0], self.recipe.version[0])
        self.assertEqual(len(recipe.param), len(self.recipe.param))

if __name__ == '__main__':
    unittest.main()