KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run Slash automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
