test_bulkIO.py

import unittest
import pathlib
import io
import re


import icartt

try:
    from _utils import compareFiles  # we're executing from the directory of this script
except ImportError:
    from ._utils import compareFiles  # we're executing from another directory

# working directory, example files
wd = pathlib.Path(__file__).parent / "example_data"

# file : (ffi, nlscom, nlncom, nHeaderLines, exception) <- want
fileinfo = {
    # should warn; has multiple keywords per line in normalComments
    "AROTAL-RAY_DC8_20040715_R1.ict": (2110, 1, 19, 68, None),
    "AR_DC8_20050203_R0.ict": (2110, 0, 18, 54, None),  # warns
    "BB-FLUX_CU-SOF_20180808_R2.ict": (1001, 0, 18, 38, None),  # ok
    "DC8-20160517.ict": (1001, 0, 18, 36, None),  # ok
    "discoveraq-CO2_p3b_20140721_R0.ict": (1001, 1, 18, 37, None),  # ok
    # ok
    "DISCOVERAQ-NOXYO3_P3B_20140720_R0.ict": (1001, 0, 27, 47, None),
    "Dongdaemun_NIER_20160520_RA.ict": (1001, 0, 18, 36, None),  # warns
    "HOX_DC8_20040712_R0.ict": (1001, 0, 18, 36, None),  # ok
    # warns
    "korusaq-flexpart-dc8_trajectory_20160529_R2.ict": (2110, 27, 20, 101, None),
    # ok
    "korusaq-mrg01-HANSEO-KING-AIR_merge_20160507_RA.ict": (1001, 0, 18, 45, None),
    # error: 2310 not implemented
    "LIDARO3_WP3_20040830_R0.ict": (2310, 0, 18, 46, NotImplementedError),
    "NOx_RHBrown_20040830_R0.ict": (1001, 0, 18, 41, None),  # ok
    # error: invalid number of variables / columns
    "output.ict": (1001, 8, 17, 41, Exception),
    "PAVE-AR_DC8_20050203_R0.ict": (2110, 1, 18, 55, None),  #  warns
    # ok
    "SEAC4RS-PTRMS-acetaldehyde_DC8_20130806_R1.ict": (1001, 0, 26, 44, None),
    "bt_Munich_2020061000_72.ict.txt": (1001, 29, 18, 91, None),  # warns
    # warns (variable names)
    "korusaq-mrg10-dc8_merge_20160510_R4.ict": (1001, 0, 29, 397, None),
}


# TODO? dataset -> close file pointer after read


class BulkIOTestCase(unittest.TestCase):
    def setUp(self):
        self.files_ok = list((wd / "expect_ok").glob("*.ict"))
        self.files_warn = list((wd / "expect_warn").glob("*.ict")) + list(
            (wd / "example_data" / "expect_warn").glob("*.txt")
        )
        self.files_fail = list((wd / "expect_fail").glob("*.ict"))

    def tearDown(self):
        pass

    def testOpen(self):
        for fn in self.files_ok:
            with self.subTest(msg=f"Opening test file {str(fn)}"):
                ict = icartt.Dataset(fn, loadData=False)
                self.assertEqual(type(ict), icartt.Dataset)

                # assert that we have correct number of header lines
                self.assertEqual(
                    (len(ict.specialComments), ict.normalComments.nlines, ict.nHeader),
                    fileinfo[fn.name][1:4],
                )

    def testReadData(self):
        for fn in self.files_ok:
            with self.subTest(msg=f"Reading data from test file {str(fn)}"):
                ict = icartt.Dataset(fn, loadData=True)
                self.assertEqual(type(ict), icartt.Dataset)
                m = re.search("R([a-zA-Z0-9]).ict", fn.name)
                if m:
                    self.assertEqual(m.groups()[0], ict.revision)

    def testWriteHeader(self):
        for fn in self.files_ok:
            with self.subTest(msg=f"Writing header for test file {str(fn)}"):
                ict = icartt.Dataset(fn, loadData=False)
                strIn = open(fn, "r", encoding="utf-8")
                strOut = io.StringIO()
                ict.writeHeader(f=strOut)
                self.assertTrue(compareFiles(fn, strIn, strOut, nlines=ict.nHeader))

    def testWrite(self):
        for fn in self.files_ok:
            with self.subTest(msg=f"Writing data for test file {str(fn)}"):
                ict = icartt.Dataset(fn, loadData=True)
                strIn = open(fn, "r", encoding="utf-8")
                strOut = io.StringIO()
                ict.write(f=strOut)
                self.assertTrue(compareFiles(fn, strIn, strOut))

    # @pytest.mark.filterwarnings('ignore::UserWarning')
    def testInvalid(self):
        for fn in self.files_fail:
            with self.subTest(msg=f"Opening invalid file {str(fn)}"):
                try:
                    _ = icartt.Dataset(fn, loadData=True)
                except fileinfo[fn.name][-1]:
                    pass
                else:
                    self.fail("expected to fail")

    def testWarnings(self):
        for fn in self.files_warn:
            with self.assertWarns(Warning):
                _ = icartt.Dataset(fn, loadData=True)


if __name__ == "__main__":  # pragma: no cover
    # unittest.main()

    import warnings

    with warnings.catch_warnings():
        warnings.simplefilter("ignore", category=UserWarning)
        unittest.main()