linpershey/process_mining

1from pm4py.objects.log.importer.xes import importer as xes_importer2from pm4py.objects.log.exporter.xes import exporter as xes_exporter3from tests.constants import INPUT_DATA_DIR, OUTPUT_DATA_DIR, PROBLEMATIC_XES_DIR, COMPRESSED_INPUT_DATA4import logging5import unittest6import importlib.util7import os8 9 10class XesImportExportTest(unittest.TestCase):11    def test_importExportXEStoXES(self):12        # to avoid static method warnings in tests,13        # that by construction of the unittest package have to be expressed in such way14        self.dummy_variable = "dummy_value"15        log = xes_importer.apply(os.path.join(INPUT_DATA_DIR, "running-example.xes"))16        xes_exporter.apply(log, os.path.join(OUTPUT_DATA_DIR, "running-example-exported.xes"))17        log_imported_after_export = xes_importer.apply(18            os.path.join(OUTPUT_DATA_DIR, "running-example-exported.xes"))19        self.assertEqual(len(log), len(log_imported_after_export))20        os.remove(os.path.join(OUTPUT_DATA_DIR, "running-example-exported.xes"))21 22    def test_importExportProblematicLogs(self):23        # to avoid static method warnings in tests,24        # that by construction of the unittest package have to be expressed in such way25        self.dummy_variable = "dummy_value"26        logs = os.listdir(PROBLEMATIC_XES_DIR)27        for log_name in logs:28            log_full_path = os.path.join(PROBLEMATIC_XES_DIR, log_name)29            try:30                output_log_path = os.path.join(OUTPUT_DATA_DIR, log_name)31                log = xes_importer.apply(log_full_path)32                xes_exporter.apply(log, output_log_path)33                log_imported_after_export = xes_importer.apply(output_log_path)34                self.assertEqual(len(log), len(log_imported_after_export))35                os.remove(output_log_path)36            except SyntaxError as e:37                logging.info("SyntaxError on log " + str(log_name) + ": " + str(e))38 39    def test_importExportXESfromGZIP_imp1(self):40        # to avoid static method warnings in tests,41        # that by construction of the unittest package have to be expressed in such way42        self.dummy_variable = "dummy_value"43        log = xes_importer.apply(os.path.join(COMPRESSED_INPUT_DATA, "01_running-example.xes.gz"))44        xes_exporter.apply(log, os.path.join(OUTPUT_DATA_DIR, "01-running-example.xes"),45                           parameters={xes_exporter.Variants.ETREE.value.Parameters.COMPRESS: True})46        os.remove(os.path.join(OUTPUT_DATA_DIR, "01-running-example.xes.gz"))47 48    def test_importXESfromGZIP_imp2(self):49        # to avoid static method warnings in tests,50        # that by construction of the unittest package have to be expressed in such way51        self.dummy_variable = "dummy_value"52        log = xes_importer.apply(os.path.join(COMPRESSED_INPUT_DATA, "01_running-example.xes.gz"))53        del log54 55    def test_rustxes_xes_import(self):56        if importlib.util.find_spec("rustxes"):57            log = xes_importer.apply(os.path.join(INPUT_DATA_DIR, "receipt.xes"), variant=xes_importer.Variants.RUSTXES)58            self.assertEqual(len(log), 1434)59 60    def test_rustxes_xesgz_import(self):61        if importlib.util.find_spec("rustxes"):62            log = xes_importer.apply(os.path.join(INPUT_DATA_DIR, "bpic2012.xes.gz"), variant=xes_importer.Variants.RUSTXES)63            self.assertEqual(len(log), 13087)64 65 66if __name__ == "__main__":67    unittest.main()68