How to use no_op method in Slash

Best Python code snippet using slash

functional_DescargaSAT.py

Source:functional_DescargaSAT.py

...11    def test_connect_disconnect(self):12        from admincfdi.pyutil import DescargaSAT13        from selenium import webdriver14        profile = webdriver.FirefoxProfile()15        def no_op(*args):16            pass17        descarga = DescargaSAT(status_callback=no_op)18        status = descarga.connect(profile, rfc=self.rfc, ciec=self.ciec)19        self.assertTrue(status)20        descarga.disconnect()21    def test_connect_fail(self):22        from admincfdi.pyutil import DescargaSAT23        from selenium import webdriver24        profile = webdriver.FirefoxProfile()25        def no_op(*args):26            pass27        descarga = DescargaSAT(status_callback=no_op)28        status = descarga.connect(profile, rfc='x', ciec='y')29        self.assertFalse(status)30        descarga.browser.close()31    def test_search_uuid(self):32        import os33        import tempfile34        from admincfdi.pyutil import DescargaSAT35        def no_op(*args):36            pass37        seccion = self.config['uuid']38        uuid = seccion['uuid']39        expected = int(seccion['expected'])40        descarga = DescargaSAT(status_callback=no_op,41                                download_callback=no_op)42        profile = descarga.get_firefox_profile('destino')43        descarga.connect(profile, rfc=self.rfc, ciec=self.ciec)44        result = descarga.search(uuid=uuid,45            dÃa='00')46        descarga.disconnect()47        self.assertEqual(expected, len(result))48    def test_uuid(self):49        import os50        import tempfile51        from admincfdi.pyutil import DescargaSAT52        def no_op(*args):53            pass54        seccion = self.config['uuid']55        uuid = seccion['uuid']56        expected = int(seccion['expected'])57        descarga = DescargaSAT(status_callback=no_op,58                               download_callback=no_op)59        with tempfile.TemporaryDirectory() as tempdir:60            destino = os.path.join(tempdir, 'cfdi-descarga')61            profile = descarga.get_firefox_profile(destino)62            descarga.connect(profile, rfc=self.rfc, ciec=self.ciec)63            docs = descarga.search(uuid=uuid, dÃa='00')64            descarga.download(docs)65            descarga.disconnect()66            self.assertEqual(expected, len(os.listdir(destino)))67    def test_rfc(self):68        import os69        import tempfile70        from admincfdi.pyutil import DescargaSAT71        def no_op(*args):72            pass73        seccion = self.config['rfc_emisor']74        rfc_emisor = seccion['rfc_emisor']75        aÃ±o = seccion['aÃ±o']76        mes = seccion['mes']77        dÃa = seccion['dÃa']78        expected = int(seccion['expected'])79        descarga = DescargaSAT(status_callback=no_op,80                               download_callback=no_op)81        with tempfile.TemporaryDirectory() as tempdir:82            destino = os.path.join(tempdir, 'cfdi-descarga')83            profile = descarga.get_firefox_profile(destino)84            descarga.connect(profile, rfc=self.rfc, ciec=self.ciec)85            docs = descarga.search(aÃ±o=aÃ±o, mes=mes, dÃa=dÃa,86                rfc_emisor=rfc_emisor)87            descarga.download(docs)88            descarga.disconnect()89            self.assertEqual(expected, len(os.listdir(destino)))90    def test_aÃ±o_mes_dÃa(self):91        import os92        import tempfile93        from admincfdi.pyutil import DescargaSAT94        def no_op(*args):95            pass96        seccion = self.config['aÃ±o_mes_dÃa']97        aÃ±o = seccion['aÃ±o']98        mes = seccion['mes']99        dÃa = seccion['dÃa']100        expected = int(seccion['expected'])101        descarga = DescargaSAT(status_callback=no_op,102                               download_callback=no_op)103        with tempfile.TemporaryDirectory() as tempdir:104            destino = os.path.join(tempdir, 'cfdi-descarga')105            profile = descarga.get_firefox_profile(destino)106            descarga.connect(profile, rfc=self.rfc, ciec=self.ciec)107            docs = descarga.search(aÃ±o=aÃ±o, mes=mes, dÃa=dÃa)108            descarga.download(docs)109            descarga.disconnect()110            self.assertEqual(expected, len(os.listdir(destino)))111    def test_hora_minuto_inicial(self):112        import os113        import tempfile114        from admincfdi.pyutil import DescargaSAT115        def no_op(*args):116            pass117        seccion = self.config['hora_minuto_inicial']118        aÃ±o = seccion['aÃ±o']119        mes = seccion['mes']120        dÃa = seccion['dÃa']121        hora_inicial =  seccion['hora_inicial']122        minuto_inicial = seccion['minuto_inicial']123        expected = int(seccion['expected'])124        descarga = DescargaSAT(status_callback=no_op,125                               download_callback=no_op)126        with tempfile.TemporaryDirectory() as tempdir:127            destino = os.path.join(tempdir, 'cfdi-descarga')128            profile = descarga.get_firefox_profile(destino)129            descarga.connect(profile, rfc=self.rfc, ciec=self.ciec)130            docs = descarga.search(aÃ±o=aÃ±o, mes=mes, dÃa=dÃa,131                                   hora_inicial=hora_inicial,132                                   minuto_inicial=minuto_inicial)133            descarga.download(docs)134            descarga.disconnect()135            self.assertEqual(expected, len(os.listdir(destino)))136    def test_hora_minuto_final(self):137        import os138        import tempfile139        from admincfdi.pyutil import DescargaSAT140        def no_op(*args):141            pass142        seccion = self.config['hora_minuto_final']143        aÃ±o = seccion['aÃ±o']144        mes = seccion['mes']145        dÃa = seccion['dÃa']146        hora_final =  seccion['hora_final']147        minuto_final = seccion['minuto_final']148        expected = int(seccion['expected'])149        descarga = DescargaSAT(status_callback=no_op,150                               download_callback=no_op)151        with tempfile.TemporaryDirectory() as tempdir:152            destino = os.path.join(tempdir, 'cfdi-descarga')153            profile = descarga.get_firefox_profile(destino)154            descarga.connect(profile, rfc=self.rfc, ciec=self.ciec)155            docs = descarga.search(aÃ±o=aÃ±o, mes=mes, dÃa=dÃa,156                                   hora_final=hora_final, minuto_final=minuto_final)157            descarga.download(docs)158            descarga.disconnect()159            self.assertEqual(expected, len(os.listdir(destino)))160    def test_segundo_inicial_final(self):161        import os162        import tempfile163        from admincfdi.pyutil import DescargaSAT164        def no_op(*args):165            pass166        seccion = self.config['segundo_inicial_final']167        aÃ±o = seccion['aÃ±o']168        mes = seccion['mes']169        dÃa = seccion['dÃa']170        hora_inicial =  seccion['hora_inicial']171        minuto_inicial = seccion['minuto_inicial']172        segundo_inicial = seccion['segundo_inicial']173        hora_final =  seccion['hora_final']174        minuto_final = seccion['minuto_final']175        segundo_final = seccion['segundo_final']176        expected = int(seccion['expected'])177        descarga = DescargaSAT(status_callback=no_op,178                               download_callback=no_op)179        with tempfile.TemporaryDirectory() as tempdir:180            destino = os.path.join(tempdir, 'cfdi-descarga')181            profile = descarga.get_firefox_profile(destino)182            descarga.connect(profile, rfc=self.rfc, ciec=self.ciec)183            docs = descarga.search(aÃ±o=aÃ±o, mes=mes, dÃa=dÃa,184                                   hora_inicial=hora_inicial, minuto_inicial=minuto_inicial,185                                   segundo_inicial=segundo_inicial,186                                   hora_final=hora_final, minuto_final=minuto_final,187                                   segundo_final=segundo_final)188            descarga.download(docs)189            descarga.disconnect()190            self.assertEqual(expected, len(os.listdir(destino)))191    def test_mes_completo(self):192        import os193        import tempfile194        from admincfdi.pyutil import DescargaSAT195        def no_op(*args):196            pass197        seccion = self.config['mes_completo_por_dÃa']198        aÃ±o = seccion['aÃ±o']199        mes = seccion['mes']200        expected = int(seccion['expected'])201        descarga = DescargaSAT(status_callback=no_op,202                               download_callback=no_op)203        with tempfile.TemporaryDirectory() as tempdir:204            destino = os.path.join(tempdir, 'cfdi-descarga')205            profile = descarga.get_firefox_profile(destino)206            descarga.connect(profile, rfc=self.rfc, ciec=self.ciec)207            docs = descarga.search(aÃ±o=aÃ±o, mes=mes, dÃa='00',208                mes_completo_por_dÃa=True)209            descarga.download(docs)210            descarga.disconnect()211            self.assertEqual(expected, len(os.listdir(destino)))212    def test_emitidas(self):213        import os214        import tempfile215        from admincfdi.pyutil import DescargaSAT216        def no_op(*args):217            pass218        seccion = self.config['emitidas']219        aÃ±o = seccion['aÃ±o']220        mes = seccion['mes']221        expected = int(seccion['expected'])222        descarga = DescargaSAT(status_callback=no_op,223                               download_callback=no_op)224        with tempfile.TemporaryDirectory() as tempdir:225            destino = os.path.join(tempdir, 'cfdi-descarga')226            profile = descarga.get_firefox_profile(destino)227            descarga.connect(profile, rfc=self.rfc, ciec=self.ciec)228            docs = descarga.search(aÃ±o=aÃ±o, mes=mes, dÃa='00',229                facturas_emitidas=True)230            descarga.disconnect()...

test_no_op.py

Source:test_no_op.py

1import math2import os3import shutil4from pathlib import Path5from django.test import TestCase, tag6import pandas as pd7import scipy.stats8from data_refinery_common.models import (9    Organism,10    OriginalFile,11    OriginalFileSampleAssociation,12    ProcessorJob,13    ProcessorJobOriginalFileAssociation,14    Sample,15)16from data_refinery_workers.processors import no_op17from data_refinery_workers.processors.testing_utils import assertMostlyAgrees18def prepare_job(job_info: dict) -> ProcessorJob:19    job = ProcessorJob()20    job.pipeline_applied = "NO_OP"21    job.save()22    og_file = OriginalFile()23    og_file.source_filename = job_info["source_filename"]24    og_file.filename = job_info["filename"]25    og_file.absolute_file_path = job_info["absolute_file_path"]26    og_file.is_downloaded = True27    og_file.save()28    sample = Sample()29    sample.accession_code = job_info["accession_code"]30    sample.title = job_info["accession_code"]31    sample.platform_accession_code = job_info["platform_accession_code"]32    manufacturer = job_info.get("manufacturer", None)33    if manufacturer is not None:34        sample.manufacturer = manufacturer35    # The illumina samples need the human organism36    if manufacturer == "ILLUMINA":37        homo_sapiens = Organism(name="HOMO_SAPIENS", taxonomy_id=9606, is_scientific_name=True)38        homo_sapiens.save()39        sample.organism = homo_sapiens40    sample.save()41    assoc = OriginalFileSampleAssociation()42    assoc.original_file = og_file43    assoc.sample = sample44    assoc.save()45    assoc1 = ProcessorJobOriginalFileAssociation()46    assoc1.original_file = og_file47    assoc1.processor_job = job48    assoc1.save()49    return job50def assertRunsSuccessfully(test_case: TestCase, job: ProcessorJob) -> dict:51    final_context = no_op.no_op_processor(job.pk)52    test_case.assertTrue(final_context["success"])53    test_case.assertTrue(os.path.exists(final_context["output_file_path"]))54    test_case.assertEqual(len(final_context["samples"]), 1)55    test_case.assertEqual(len(final_context["computed_files"]), 1)56    for sample in final_context["samples"]:57        for cf in final_context["computed_files"]:58            test_case.assertTrue(cf in sample.computed_files.all())59    # Return final_context so we can perform additional checks manually60    return final_context61class NOOPTestCase(TestCase):62    def setUp(self):63        """Cleanup work dirs from previous test runs that may have failed."""64        job_dirs = list(Path("/home/user/data_store").glob("**/processor_job_*"))65        for job_dir in job_dirs:66            shutil.rmtree(str(job_dir), ignore_errors=True)67    @tag("no_op")68    def test_convert_simple_pcl_with_header(self):69        """PCL with header70        > ID_REF, VALUE71        """72        job = prepare_job(73            {74                "accession_code": "GSM1234847",75                "source_filename": "https://www.ebi.ac.uk/arrayexpress/experiments/E-GEOD-51013/",76                "filename": "GSM1234847_sample_table.txt",77                "absolute_file_path": "/home/user/data_store/raw/TEST/NO_OP/GSM1234847_sample_table.txt",78                "platform_accession_code": "A-AFFY-38",79            }80        )81        final_context = assertRunsSuccessfully(self, job)82        expected_data = pd.read_csv(83            "/home/user/data_store/TEST/NO_OP/EXPECTED/gene_converted_GSM1234847-tbl-1.txt",84            sep="\t",85            index_col=0,86        )["VALUE"]87        actual_data = pd.read_csv(final_context["output_file_path"], sep="\t", index_col=0)["VALUE"]88        assertMostlyAgrees(self, expected_data, actual_data)89    @tag("no_op")90    def test_convert_simple_pcl_with_no_header(self):91        """PCL with no header, ex:92        > AFFX-BioB-3_at  0.7421875693        """94        job = prepare_job(95            {96                "accession_code": "GSM1234847",97                "source_filename": "https://www.ebi.ac.uk/arrayexpress/experiments/E-GEOD-51013/",98                "filename": "GSM1234847_sample_table.txt",99                "absolute_file_path": "/home/user/data_store/raw/TEST/NO_OP/GSM1234847_sample_table_headerless.txt",100                "platform_accession_code": "A-AFFY-38",101            }102        )103        final_context = assertRunsSuccessfully(self, job)104        expected_data = pd.read_csv(105            "/home/user/data_store/TEST/NO_OP/EXPECTED/gene_converted_GSM1234847-tbl-1.txt",106            sep="\t",107            index_col=0,108        )["VALUE"]109        actual_data = pd.read_csv(final_context["output_file_path"], sep="\t", index_col=0)["VALUE"]110        assertMostlyAgrees(self, expected_data, actual_data)111    @tag("no_op")112    def test_convert_processed_illumina(self):113        """Illumina file with header, ex:114        > Reporter Identifier VALUE   Detection Pval115        > ILMN_1343291    14.943602   0116        > ILMN_1343295    13.528082   0117        """118        job = prepare_job(119            {120                "accession_code": "GSM557500",121                "source_filename": "https://www.ebi.ac.uk/arrayexpress/experiments/E-GEOD-22433/",122                "filename": "GSM557500-tbl-1.txt",123                "absolute_file_path": "/home/user/data_store/raw/TEST/NO_OP/GSM557500-tbl-1.txt",124                "platform_accession_code": "A-MEXP-1171",125                "manufacturer": "ILLUMINA",126            }127        )128        final_context = assertRunsSuccessfully(self, job)129        self.assertTrue(no_op.check_output_quality(final_context["output_file_path"]))130        # To:131        # ENSG00000156508 14.943602132        # ENSG00000111640 13.528082133        expected_data = pd.read_csv(134            "/home/user/data_store/TEST/NO_OP/EXPECTED/gene_converted_GSM557500-tbl-1.txt",135            sep="\t",136            names=["", "VALUE"],137            index_col=0,138        )["VALUE"]139        actual_data = pd.read_csv(140            final_context["output_file_path"], sep="\t", names=["", "VALUE"], index_col=0141        )["VALUE"]142        assertMostlyAgrees(self, expected_data, actual_data)143    @tag("no_op")144    def test_convert_illumina_no_header(self):145        """Illumina file without header, ex:146        > ILMN_1885639    10.0000   0.7931147        > ILMN_2209417    10.0000   0.2029148        > ILMN_1765401    152.0873  0.0000149        """150        job = prepare_job(151            {152                "accession_code": "GSM1089291",153                "source_filename": "https://github.com/AlexsLemonade/refinebio/files/2255178/GSM1089291-tbl-1.txt",154                "filename": "GSM1089291-tbl-1.txt",155                "absolute_file_path": "/home/user/data_store/raw/TEST/NO_OP/GSM1089291-tbl-1.txt",156                "platform_accession_code": "A-MEXP-1171",157                "manufacturer": "ILLUMINA",158            }159        )160        final_context = assertRunsSuccessfully(self, job)161        self.assertTrue(no_op.check_output_quality(final_context["output_file_path"]))162        # To:163        # ENSG00000105675 10164        # ENSG00000085721 152.0873165        # ENSG00000278494 152.0873166        expected_data = pd.read_csv(167            "/home/user/data_store/TEST/NO_OP/EXPECTED/gene_converted_GSM1089291-tbl-1.txt",168            sep="\t",169            names=["", "VALUE"],170            index_col=0,171        )["VALUE"]172        actual_data = pd.read_csv(173            final_context["output_file_path"], sep="\t", names=["", "VALUE"], index_col=0174        )["VALUE"]175        assertMostlyAgrees(self, expected_data, actual_data)176    @tag("no_op")177    def test_convert_illumina_bad_cols(self):178        """179        In future, this test may be deprecated. For now it just alerts that it needs attention.180        """181        job = ProcessorJob()182        job.pipeline_applied = "NO_OP"183        job.save()184        # ex:185        # ILMN_1885639    10.0000 0.7931  11.0000 0.123186        # ILMN_2209417    10.0000 0.2029  11.1234 0.543187        # LMN_1765401    152.0873    0.0000  99.999  0.19188        og_file = OriginalFile()189        og_file.source_filename = (190            "https://github.com/AlexsLemonade/refinebio/files/2255178/GSM1089291-tbl-1-modified.txt"191        )192        og_file.filename = "GSM1089291-tbl-1-modified.txt"193        og_file.absolute_file_path = (194            "/home/user/data_store/raw/TEST/NO_OP/GSM1089291-tbl-1-modified.txt"195        )196        og_file.is_downloaded = True197        og_file.save()198        homo_sapiens = Organism(name="HOMO_SAPIENS", taxonomy_id=9606, is_scientific_name=True)199        homo_sapiens.save()200        sample = Sample()201        sample.accession_code = "GSM557500"202        sample.title = "GSM557500"203        sample.platform_accession_code = "A-MEXP-1171"204        sample.manufacturer = "ILLUMINA"205        sample.organism = homo_sapiens206        sample.save()207        assoc = OriginalFileSampleAssociation()208        assoc.original_file = og_file209        assoc.sample = sample210        assoc.save()211        assoc1 = ProcessorJobOriginalFileAssociation()212        assoc1.original_file = og_file213        assoc1.processor_job = job214        assoc1.save()215        final_context = no_op.no_op_processor(job.pk)216        self.assertFalse(final_context["success"])...

literals.pyi

Source:literals.pyi

1"""2Type annotations for appflow service literal definitions.3[Open documentation](https://vemel.github.io/boto3_stubs_docs/mypy_boto3_appflow/literals.html)4Usage::5    ```python6    from mypy_boto3_appflow.literals import AggregationTypeType7    data: AggregationTypeType = "None"8    ```9"""10import sys11if sys.version_info >= (3, 8):12    from typing import Literal13else:14    from typing_extensions import Literal15__all__ = (16    "AggregationTypeType",17    "AmplitudeConnectorOperatorType",18    "ConnectionModeType",19    "ConnectorTypeType",20    "DataPullModeType",21    "DatadogConnectorOperatorType",22    "DynatraceConnectorOperatorType",23    "ExecutionStatusType",24    "FileTypeType",25    "FlowStatusType",26    "GoogleAnalyticsConnectorOperatorType",27    "InforNexusConnectorOperatorType",28    "MarketoConnectorOperatorType",29    "OperatorPropertiesKeysType",30    "OperatorType",31    "PrefixFormatType",32    "PrefixTypeType",33    "PrivateConnectionProvisioningFailureCauseType",34    "PrivateConnectionProvisioningStatusType",35    "S3ConnectorOperatorType",36    "S3InputFileTypeType",37    "SAPODataConnectorOperatorType",38    "SalesforceConnectorOperatorType",39    "ScheduleFrequencyTypeType",40    "ServiceNowConnectorOperatorType",41    "SingularConnectorOperatorType",42    "SlackConnectorOperatorType",43    "TaskTypeType",44    "TrendmicroConnectorOperatorType",45    "TriggerTypeType",46    "VeevaConnectorOperatorType",47    "WriteOperationTypeType",48    "ZendeskConnectorOperatorType",49)50AggregationTypeType = Literal["None", "SingleFile"]51AmplitudeConnectorOperatorType = Literal["BETWEEN"]52ConnectionModeType = Literal["Private", "Public"]53ConnectorTypeType = Literal[54    "Amplitude",55    "CustomerProfiles",56    "Datadog",57    "Dynatrace",58    "EventBridge",59    "Googleanalytics",60    "Honeycode",61    "Infornexus",62    "LookoutMetrics",63    "Marketo",64    "Redshift",65    "S3",66    "SAPOData",67    "Salesforce",68    "Servicenow",69    "Singular",70    "Slack",71    "Snowflake",72    "Trendmicro",73    "Upsolver",74    "Veeva",75    "Zendesk",76]77DataPullModeType = Literal["Complete", "Incremental"]78DatadogConnectorOperatorType = Literal[79    "ADDITION",80    "BETWEEN",81    "DIVISION",82    "EQUAL_TO",83    "MASK_ALL",84    "MASK_FIRST_N",85    "MASK_LAST_N",86    "MULTIPLICATION",87    "NO_OP",88    "PROJECTION",89    "SUBTRACTION",90    "VALIDATE_NON_NEGATIVE",91    "VALIDATE_NON_NULL",92    "VALIDATE_NON_ZERO",93    "VALIDATE_NUMERIC",94]95DynatraceConnectorOperatorType = Literal[96    "ADDITION",97    "BETWEEN",98    "DIVISION",99    "EQUAL_TO",100    "MASK_ALL",101    "MASK_FIRST_N",102    "MASK_LAST_N",103    "MULTIPLICATION",104    "NO_OP",105    "PROJECTION",106    "SUBTRACTION",107    "VALIDATE_NON_NEGATIVE",108    "VALIDATE_NON_NULL",109    "VALIDATE_NON_ZERO",110    "VALIDATE_NUMERIC",111]112ExecutionStatusType = Literal["Error", "InProgress", "Successful"]113FileTypeType = Literal["CSV", "JSON", "PARQUET"]114FlowStatusType = Literal["Active", "Deleted", "Deprecated", "Draft", "Errored", "Suspended"]115GoogleAnalyticsConnectorOperatorType = Literal["BETWEEN", "PROJECTION"]116InforNexusConnectorOperatorType = Literal[117    "ADDITION",118    "BETWEEN",119    "DIVISION",120    "EQUAL_TO",121    "MASK_ALL",122    "MASK_FIRST_N",123    "MASK_LAST_N",124    "MULTIPLICATION",125    "NO_OP",126    "PROJECTION",127    "SUBTRACTION",128    "VALIDATE_NON_NEGATIVE",129    "VALIDATE_NON_NULL",130    "VALIDATE_NON_ZERO",131    "VALIDATE_NUMERIC",132]133MarketoConnectorOperatorType = Literal[134    "ADDITION",135    "BETWEEN",136    "DIVISION",137    "GREATER_THAN",138    "LESS_THAN",139    "MASK_ALL",140    "MASK_FIRST_N",141    "MASK_LAST_N",142    "MULTIPLICATION",143    "NO_OP",144    "PROJECTION",145    "SUBTRACTION",146    "VALIDATE_NON_NEGATIVE",147    "VALIDATE_NON_NULL",148    "VALIDATE_NON_ZERO",149    "VALIDATE_NUMERIC",150]151OperatorPropertiesKeysType = Literal[152    "CONCAT_FORMAT",153    "DATA_TYPE",154    "DESTINATION_DATA_TYPE",155    "EXCLUDE_SOURCE_FIELDS_LIST",156    "LOWER_BOUND",157    "MASK_LENGTH",158    "MASK_VALUE",159    "MATH_OPERATION_FIELDS_ORDER",160    "SOURCE_DATA_TYPE",161    "SUBFIELD_CATEGORY_MAP",162    "TRUNCATE_LENGTH",163    "UPPER_BOUND",164    "VALIDATION_ACTION",165    "VALUE",166    "VALUES",167]168OperatorType = Literal[169    "ADDITION",170    "BETWEEN",171    "CONTAINS",172    "DIVISION",173    "EQUAL_TO",174    "GREATER_THAN",175    "GREATER_THAN_OR_EQUAL_TO",176    "LESS_THAN",177    "LESS_THAN_OR_EQUAL_TO",178    "MASK_ALL",179    "MASK_FIRST_N",180    "MASK_LAST_N",181    "MULTIPLICATION",182    "NOT_EQUAL_TO",183    "NO_OP",184    "PROJECTION",185    "SUBTRACTION",186    "VALIDATE_NON_NEGATIVE",187    "VALIDATE_NON_NULL",188    "VALIDATE_NON_ZERO",189    "VALIDATE_NUMERIC",190]191PrefixFormatType = Literal["DAY", "HOUR", "MINUTE", "MONTH", "YEAR"]192PrefixTypeType = Literal["FILENAME", "PATH", "PATH_AND_FILENAME"]193PrivateConnectionProvisioningFailureCauseType = Literal[194    "ACCESS_DENIED", "CONNECTOR_AUTHENTICATION", "CONNECTOR_SERVER", "INTERNAL_SERVER", "VALIDATION"195]196PrivateConnectionProvisioningStatusType = Literal["CREATED", "FAILED", "PENDING"]197S3ConnectorOperatorType = Literal[198    "ADDITION",199    "BETWEEN",200    "DIVISION",201    "EQUAL_TO",202    "GREATER_THAN",203    "GREATER_THAN_OR_EQUAL_TO",204    "LESS_THAN",205    "LESS_THAN_OR_EQUAL_TO",206    "MASK_ALL",207    "MASK_FIRST_N",208    "MASK_LAST_N",209    "MULTIPLICATION",210    "NOT_EQUAL_TO",211    "NO_OP",212    "PROJECTION",213    "SUBTRACTION",214    "VALIDATE_NON_NEGATIVE",215    "VALIDATE_NON_NULL",216    "VALIDATE_NON_ZERO",217    "VALIDATE_NUMERIC",218]219S3InputFileTypeType = Literal["CSV", "JSON"]220SAPODataConnectorOperatorType = Literal[221    "ADDITION",222    "BETWEEN",223    "CONTAINS",224    "DIVISION",225    "EQUAL_TO",226    "GREATER_THAN",227    "GREATER_THAN_OR_EQUAL_TO",228    "LESS_THAN",229    "LESS_THAN_OR_EQUAL_TO",230    "MASK_ALL",231    "MASK_FIRST_N",232    "MASK_LAST_N",233    "MULTIPLICATION",234    "NOT_EQUAL_TO",235    "NO_OP",236    "PROJECTION",237    "SUBTRACTION",238    "VALIDATE_NON_NEGATIVE",239    "VALIDATE_NON_NULL",240    "VALIDATE_NON_ZERO",241    "VALIDATE_NUMERIC",242]243SalesforceConnectorOperatorType = Literal[244    "ADDITION",245    "BETWEEN",246    "CONTAINS",247    "DIVISION",248    "EQUAL_TO",249    "GREATER_THAN",250    "GREATER_THAN_OR_EQUAL_TO",251    "LESS_THAN",252    "LESS_THAN_OR_EQUAL_TO",253    "MASK_ALL",254    "MASK_FIRST_N",255    "MASK_LAST_N",256    "MULTIPLICATION",257    "NOT_EQUAL_TO",258    "NO_OP",259    "PROJECTION",260    "SUBTRACTION",261    "VALIDATE_NON_NEGATIVE",262    "VALIDATE_NON_NULL",263    "VALIDATE_NON_ZERO",264    "VALIDATE_NUMERIC",265]266ScheduleFrequencyTypeType = Literal["BYMINUTE", "DAILY", "HOURLY", "MONTHLY", "ONCE", "WEEKLY"]267ServiceNowConnectorOperatorType = Literal[268    "ADDITION",269    "BETWEEN",270    "CONTAINS",271    "DIVISION",272    "EQUAL_TO",273    "GREATER_THAN",274    "GREATER_THAN_OR_EQUAL_TO",275    "LESS_THAN",276    "LESS_THAN_OR_EQUAL_TO",277    "MASK_ALL",278    "MASK_FIRST_N",279    "MASK_LAST_N",280    "MULTIPLICATION",281    "NOT_EQUAL_TO",282    "NO_OP",283    "PROJECTION",284    "SUBTRACTION",285    "VALIDATE_NON_NEGATIVE",286    "VALIDATE_NON_NULL",287    "VALIDATE_NON_ZERO",288    "VALIDATE_NUMERIC",289]290SingularConnectorOperatorType = Literal[291    "ADDITION",292    "DIVISION",293    "EQUAL_TO",294    "MASK_ALL",295    "MASK_FIRST_N",296    "MASK_LAST_N",297    "MULTIPLICATION",298    "NO_OP",299    "PROJECTION",300    "SUBTRACTION",301    "VALIDATE_NON_NEGATIVE",302    "VALIDATE_NON_NULL",303    "VALIDATE_NON_ZERO",304    "VALIDATE_NUMERIC",305]306SlackConnectorOperatorType = Literal[307    "ADDITION",308    "BETWEEN",309    "DIVISION",310    "EQUAL_TO",311    "GREATER_THAN",312    "GREATER_THAN_OR_EQUAL_TO",313    "LESS_THAN",314    "LESS_THAN_OR_EQUAL_TO",315    "MASK_ALL",316    "MASK_FIRST_N",317    "MASK_LAST_N",318    "MULTIPLICATION",319    "NO_OP",320    "PROJECTION",321    "SUBTRACTION",322    "VALIDATE_NON_NEGATIVE",323    "VALIDATE_NON_NULL",324    "VALIDATE_NON_ZERO",325    "VALIDATE_NUMERIC",326]327TaskTypeType = Literal[328    "Arithmetic", "Filter", "Map", "Map_all", "Mask", "Merge", "Truncate", "Validate"329]330TrendmicroConnectorOperatorType = Literal[331    "ADDITION",332    "DIVISION",333    "EQUAL_TO",334    "MASK_ALL",335    "MASK_FIRST_N",336    "MASK_LAST_N",337    "MULTIPLICATION",338    "NO_OP",339    "PROJECTION",340    "SUBTRACTION",341    "VALIDATE_NON_NEGATIVE",342    "VALIDATE_NON_NULL",343    "VALIDATE_NON_ZERO",344    "VALIDATE_NUMERIC",345]346TriggerTypeType = Literal["Event", "OnDemand", "Scheduled"]347VeevaConnectorOperatorType = Literal[348    "ADDITION",349    "BETWEEN",350    "CONTAINS",351    "DIVISION",352    "EQUAL_TO",353    "GREATER_THAN",354    "GREATER_THAN_OR_EQUAL_TO",355    "LESS_THAN",356    "LESS_THAN_OR_EQUAL_TO",357    "MASK_ALL",358    "MASK_FIRST_N",359    "MASK_LAST_N",360    "MULTIPLICATION",361    "NOT_EQUAL_TO",362    "NO_OP",363    "PROJECTION",364    "SUBTRACTION",365    "VALIDATE_NON_NEGATIVE",366    "VALIDATE_NON_NULL",367    "VALIDATE_NON_ZERO",368    "VALIDATE_NUMERIC",369]370WriteOperationTypeType = Literal["INSERT", "UPDATE", "UPSERT"]371ZendeskConnectorOperatorType = Literal[372    "ADDITION",373    "DIVISION",374    "GREATER_THAN",375    "MASK_ALL",376    "MASK_FIRST_N",377    "MASK_LAST_N",378    "MULTIPLICATION",379    "NO_OP",380    "PROJECTION",381    "SUBTRACTION",382    "VALIDATE_NON_NEGATIVE",383    "VALIDATE_NON_NULL",384    "VALIDATE_NON_ZERO",385    "VALIDATE_NUMERIC",...

modbus_server.py

Source:modbus_server.py

1from platform_driver.interfaces.modbus_tk.server import Server2from platform_driver.interfaces.modbus_tk import helpers3from platform_driver.interfaces.modbus_tk.client import Client, Field4from platform_driver.interfaces.modbus_tk.maps import Map, Catalog5import serial6from struct import pack, unpack7import logging8logger = logging.getLogger(__name__)9class ModbusTkClient (Client):10    """11        Testing for tcp transport12    """13    byte_order = helpers.BIG_ENDIAN14    addressing = helpers.ADDRESS_OFFSET15    unsigned_short = Field("unsigned_short", 0, helpers.USHORT, 'PPM', 0, helpers.no_op, helpers.REGISTER_READ_WRITE, helpers.OP_MODE_READ_WRITE)16    unsigned_int = Field("unsigned_int", 1, helpers.UINT, 'PPM', 0, helpers.no_op, helpers.REGISTER_READ_WRITE, helpers.OP_MODE_READ_WRITE)17    unsigned_long = Field("unsigned_long", 3, helpers.UINT64, 'PPM', 0, helpers.no_op, helpers.REGISTER_READ_WRITE, helpers.OP_MODE_READ_WRITE)18    sample_short = Field("sample_short", 7, helpers.SHORT, 'PPM', 0, helpers.no_op, helpers.REGISTER_READ_WRITE, helpers.OP_MODE_READ_WRITE)19    sample_int = Field("sample_int", 8, helpers.INT, 'PPM', 0, helpers.no_op, helpers.REGISTER_READ_WRITE, helpers.OP_MODE_READ_WRITE)20    sample_float = Field("sample_float", 10, helpers.FLOAT, 'PPM', 2, helpers.no_op, helpers.REGISTER_READ_WRITE, helpers.OP_MODE_READ_WRITE)21    sample_long = Field("sample_long", 12, helpers.INT64, 'PPM', 0, helpers.no_op, helpers.REGISTER_READ_WRITE, helpers.OP_MODE_READ_WRITE)22    sample_bool = Field('sample_bool', 16, helpers.BOOL, 'PPM', 0, helpers.no_op, helpers.COIL_READ_WRITE, helpers.OP_MODE_READ_WRITE)23    sample_str = Field("sample_str", 17, helpers.string(4), "bytes", 4, helpers.no_op, helpers.REGISTER_READ_WRITE, helpers.OP_MODE_READ_WRITE)24class WattsOn (Client):25    """26        Testing Stub for WattsOn modbus device.27    """28    byte_order = helpers.BIG_ENDIAN29    addressing = helpers.ADDRESS_OFFSET30    active_power_total = Field.default_holding_register('Active Power Total', 0x200, helpers.FLOAT, "kW", helpers.no_op)31    active_power_A = Field.default_holding_register('Active Power A', 0x232, helpers.FLOAT, "kW", helpers.no_op)32    apparent_power_A = Field.default_holding_register('Apparent Power A', 0x23E, helpers.FLOAT, "kW", helpers.no_op)33    net_total_energy = Field.default_holding_register('Net Total Energy', 0x1100, helpers.FLOAT, "kWh", helpers.no_op)34    voltage_A = Field.default_holding_register('Voltage A', 0x220, helpers.FLOAT, "V", helpers.no_op)35    current_A = Field.default_holding_register('Current A', 0x22C, helpers.FLOAT, "A", helpers.no_op)36class PPSPi32Client (Client):37    """38        Define some regiesters to PPSPi32Client39    """40    def __init__(self, *args, **kwargs):41        super(PPSPi32Client, self).__init__(*args, **kwargs)42    byte_order = helpers.BIG_ENDIAN43    addressing = helpers.ADDRESS_OFFSET44    BigUShort = Field("BigUShort", 0, helpers.USHORT, 'PPM', 2, helpers.no_op, helpers.REGISTER_READ_WRITE, helpers.OP_MODE_READ_WRITE)45    BigUInt = Field("BigUInt", 1, helpers.UINT, 'PPM', 2, helpers.no_op, helpers.REGISTER_READ_WRITE, helpers.OP_MODE_READ_WRITE)46    BigULong = Field("BigULong", 3, helpers.UINT64, 'PPM', 2, helpers.no_op, helpers.REGISTER_READ_WRITE, helpers.OP_MODE_READ_WRITE)47    BigShort = Field("BigShort", 7, helpers.SHORT, 'PPM', 2, helpers.no_op, helpers.REGISTER_READ_WRITE, helpers.OP_MODE_READ_WRITE)48    BigInt = Field("BigInt", 8, helpers.INT, 'PPM', 2, helpers.no_op, helpers.REGISTER_READ_WRITE, helpers.OP_MODE_READ_WRITE)49    BigFloat = Field("BigFloat", 10, helpers.FLOAT, 'PPM', 2, helpers.no_op, helpers.REGISTER_READ_WRITE, helpers.OP_MODE_READ_WRITE)50    BigLong = Field("BigLong", 12, helpers.INT64, 'PPM', 2, helpers.no_op, helpers.REGISTER_READ_WRITE, helpers.OP_MODE_READ_WRITE)51    LittleUShort = Field(52        "LittleUShort", 100, helpers.USHORT, 'PPM', 2, helpers.no_op, helpers.REGISTER_READ_WRITE, helpers.OP_MODE_READ_WRITE)53    LittleUInt = Field(54        "LittleUInt", 101, helpers.UINT, 'PPM', 2, helpers.no_op, helpers.REGISTER_READ_WRITE, helpers.OP_MODE_READ_WRITE)55    LittleULong = Field(56        "LittleULong", 103, helpers.UINT64, 'PPM', 2, helpers.no_op, helpers.REGISTER_READ_WRITE, helpers.OP_MODE_READ_WRITE)57    LittleShort = Field(58        "LittleShort", 107, helpers.SHORT, 'PPM', 2, helpers.no_op, helpers.REGISTER_READ_WRITE, helpers.OP_MODE_READ_WRITE)59    LittleInt = Field(60        "LittleInt", 108, helpers.INT, 'PPM', 2, helpers.no_op, helpers.REGISTER_READ_WRITE, helpers.OP_MODE_READ_WRITE)61    LittleFloat = Field(62        "LittleFloat", 110, helpers.FLOAT, 'PPM', 2, helpers.no_op, helpers.REGISTER_READ_WRITE, helpers.OP_MODE_READ_WRITE)63    LittleLong = Field(64        "LittleLong", 112, helpers.INT64, 'PPM', 2, helpers.no_op, helpers.REGISTER_READ_WRITE, helpers.OP_MODE_READ_WRITE)65def watts_on_server():66    # For rtu transport: connect to the Elkor Watts On meter by usb to the RS-485 interface67    # Can define ModbusClient2 by Map or defined the class as ModbusClient1 or ModbusClient268    # modbus_map = Map(69    #     map_dir='/Users/anhnguyen/repos/kisensum-volttron/volttron/services/core/PlatformDriverAgent/platform_driver/interfaces/modbus_tk/maps',70    #     addressing='offset', name='watts_on', file='watts_on.csv', endian='big')71    # ModbusClient2 = modbus_map.get_class()72    ModbusClient2 = Catalog()['watts_on'].get_class()73    client = ModbusClient2(slave_address=2, verbose=True)74    client.set_transport_rtu('/dev/tty.usbserial-AL00IEEY',75                             115200,76                             serial.EIGHTBITS,77                             serial.PARITY_NONE,78                             serial.STOPBITS_ONE,79                             False)80    # Get reading values for defined registers81    print(dict((field.name, value) for field, value, timestamp in client.dump_all()))82    setattr(client, "serial_baud_rate", 115)83    client.write_all()84    print(getattr(client, "serial_baud_rate"))85if __name__ == '__main__':86    # For tcp transport87    ModbusClient = Catalog()['modbus_tk_test'].get_class()88    server_process = Server(address='127.0.0.1', port=5020)89    server_process.define_slave(1, ModbusClient, unsigned=True)90    server_process.start()91    # For rtu transport...

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.