Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use _run_transformers method in lisa

Best Python code snippet using lisa_python

__init__.py

Source:__init__.py

...124    try:125        transformer.transform_bundle(bundle_path, bundle_manifest_path)126    except Exception as ex:127        _log_error(f"{bundle_uuid}.{bundle_version}", ex, traceback.format_exc(), extractor)128def _run_transformers(extractor: DSSExtractor, transformers: list):129    """130    Executes a provided list of transformers.131    :param extractor: DSSExtractor132    :param transformers: List of MetadataToPsvTransformers to execute133    """134    logger.info(f"ETL: All bundles downloaded. Performing final PSV transformations.")135    for transformer in transformers:136        try:137            transformer.transform(os.path.join(extractor.sd, 'bundles'))138        except Exception as ex:139            _log_error(transformer.__class__.__name__, ex, traceback.format_exc(), extractor)140    logger.info(f"ETL: All transformations complete.")141def finalizer_reload(extractor: DSSExtractor):142    """143    Final callback during ETL for loading an empty database. Invokes Loader.144    """145    _run_transformers(extractor, [146        FeatureTransformer(extractor.sd),147        AnalysisTransformer(extractor.sd),148        SpecimenLibraryTransformer(extractor.sd),149        ProjectPublicationContributorTransformer(extractor.sd)150    ])151    upload_and_load(extractor.sd, is_update=False)152def finalizer_update(extractor: DSSExtractor):153    """154    Final callback during ETL to perform database updates. Invokes Loader with update flag.155    :return:156    """157    _run_transformers(extractor, [158        FeatureTransformer(extractor.sd),159        AnalysisTransformer(extractor.sd),160        SpecimenLibraryTransformer(extractor.sd),161        ProjectPublicationContributorTransformer(extractor.sd)162    ])163    upload_and_load(extractor.sd, is_update=True)164def finalizer_notification(extractor: DSSExtractor):165    """166    Final callback during ETL to handle DSS notifications. Invokes Loader with update flag.167    :return:168    """169    _run_transformers(extractor, [170        AnalysisTransformer(extractor.sd),171        SpecimenLibraryTransformer(extractor.sd),172        ProjectPublicationContributorTransformer(extractor.sd)173    ])174    upload_and_load(extractor.sd, is_update=True)175def _log_error(entity: str, exception: Exception, trace: str, extractor: DSSExtractor):176    """177    Logs an ETL error and exception stack trace to a file.178    Error messages and exceptions are written to 'errors.txt'179    A list of failed entities are written to 'failed_transforms.txt'180    :param entity: A MetadataToPsvTransformer, or a bundle FQID for CellExpressionTransformer errors181    :param exception: Thrown exception string182    :param trace: Exception stack trace183    :param extractor: DSSExtractor...

test_transformer.py

Source:test_transformer.py

...57    def test_transformer_keep_all_values(self) -> None:58        # no value is overridden, all values are kept59        transformers = self._generate_transformers_runbook(2)60        runbook_builder = self._generate_runbook_builder(transformers)61        result = transformer._run_transformers(runbook_builder)62        self._validate_variables(63            {64                "v0": "original",65                "va": "original",66                "t0_v0": "0_0 processed",67                "t1_v0": "1_0 processed",68                "t1_v1": "1_1 processed",69            },70            result,71        )72    def test_transformer_overridden_values(self) -> None:73        # value is overridden74        transformers = self._generate_transformers_runbook(2)75        transformers[0].rename = {"t0_v0": "v0"}76        transformers[1].rename = {"t1_v0": "v0", "t1_v1": "v1"}77        runbook_builder = self._generate_runbook_builder(transformers)78        result = transformer._run_transformers(runbook_builder)79        self._validate_variables(80            {81                "va": "original",82                "v0": "1_0 processed",83                "v1": "1_1 processed",84            },85            result,86        )87    def test_transformer_rename_not_exist(self) -> None:88        # not exist name raise exception89        transformers = self._generate_transformers_runbook(1)90        transformers[0].rename = {"v0": "v0_1"}91        runbook_builder = self._generate_runbook_builder(transformers)92        with self.assertRaises(LisaException) as cm:93            transformer._run_transformers(runbook_builder)94        self.assertEqual("unmatched rename variable: {'v0': 'v0_1'}", str(cm.exception))95    def test_transformer_customized_prefix(self) -> None:96        # modified prefix97        transformers = self._generate_transformers_runbook(1)98        transformers[0].prefix = "v0_1"99        runbook_builder = self._generate_runbook_builder(transformers)100        result = transformer._run_transformers(runbook_builder)101        self._validate_variables(102            {103                "v0": "original",104                "va": "original",105                "v0_1_v0": "0_0 processed",106            },107            result,108        )109    def test_transformer_no_name(self) -> None:110        # no name, the type will be used. in name111        transformers_data: List[Any] = [{"type": MOCK, "items": {"v0": "v0_1"}}]112        transformers = schema.load_by_type_many(schema.Transformer, transformers_data)113        runbook_builder = self._generate_runbook_builder(transformers)114        result = transformer._run_transformers(runbook_builder)115        self._validate_variables(116            {117                "v0": "original",118                "va": "original",119                "mock_v0": "v0_1 processed",120            },121            result,122        )123    def test_transformer_skip_disabled(self) -> None:124        # the second transformer should be skipped, so the value is original.125        transformers = self._generate_transformers_runbook(2)126        transformers[0].rename = {"t0_v0": "v0"}127        transformers[0].enabled = False128        runbook_builder = self._generate_runbook_builder(transformers)129        result = transformer._run_transformers(runbook_builder)130        self._validate_variables(131            {132                "v0": "original",133                "va": "original",134                "t1_v0": "1_0 processed",135                "t1_v1": "1_1 processed",136            },137            result,138        )139    def _validate_variables(140        self, expected: Dict[str, str], actual: Dict[str, VariableEntry]141    ) -> None:142        actual_pairs = {name: value.data for name, value in actual.items()}143        self.assertDictEqual(expected, actual_pairs)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.