How to use _service method in ATX

Best Python code snippet using ATX

air_conditioner.py

Source:air_conditioner.py Github

copy

Full Screen

...516 raise ValueError(f'cmd_txt not exist, {cmd_txt}')517 else:518 return cmd_dict[_cmd]519 @classmethod520 def convert_dev_specific_service(cls, pdu: list, is_fixed_len_pdu: bool) -> SAServiceBase:521 if not isinstance(pdu, list) or len(pdu) < 3:522 raise ValueError(f'service pdu invalid, {pdu}')523 if not is_fixed_len_pdu:524 raise NotImplementedError525 _pdu = list(pdu)526 logger.debug(f'parsing dev service pdu {_pdu}')527 _service = SAServiceBase.from_fixed_len_pdu(pdu=_pdu)528 if _service.service_id == ACServiceIDEnum.POWER_RW:529 _service = ACPowerService.from_fixed_len_pdu(pdu=_pdu)530 elif _service.service_id == ACServiceIDEnum.OP_MODE_RW:531 _service = ACOpModeService.from_fixed_len_pdu(pdu=_pdu)532 elif _service.service_id == ACServiceIDEnum.FAN_LEVEL_RW:533 _service = ACFanLevelService.from_fixed_len_pdu(pdu=_pdu)534 elif _service.service_id == ACServiceIDEnum.TEMPERATURE_CFG_RW:...

Full Screen

Full Screen

consts.py

Source:consts.py Github

copy

Full Screen

1# Copyright 2013: Mirantis Inc.2# All Rights Reserved.3#4# Licensed under the Apache License, Version 2.0 (the "License"); you may5# not use this file except in compliance with the License. You may obtain6# a copy of the License at7#8# http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the13# License for the specific language governing permissions and limitations14# under the License.15"""16There is a lot of situations when we would like to work with Enum or Const.17E.g. work around Tasks. We would like to use Enum in DB to store status of task18and also in migration that creates DB and in business logic to set some status19so as to avoid copy paste or direct usage of enums values we create singletons20for each enum. (e.g. TaskStatus)21"""22from rally.common import utils23JSON_SCHEMA = "http://json-schema.org/draft-04/schema"24JSON_SCHEMA7 = "http://json-schema.org/draft-07/schema"25class _TaskStatus(utils.ImmutableMixin, utils.EnumMixin):26 """Consts that represents task possible states."""27 INIT = "init"28 VALIDATING = "validating"29 VALIDATED = "validated"30 VALIDATION_FAILED = "validation_failed"31 RUNNING = "running"32 FINISHED = "finished"33 CRASHED = "crashed"34 ABORTING = "aborting"35 SLA_FAILED = "sla_failed"36 SOFT_ABORTING = "soft_aborting"37 ABORTED = "aborted"38 PAUSED = "paused"39class _SubtaskStatus(utils.ImmutableMixin, utils.EnumMixin):40 """Consts that represents task possible states."""41 INIT = "init"42 VALIDATING = "validating"43 VALIDATED = "validated"44 VALIDATION_FAILED = "validation_failed"45 RUNNING = "running"46 FINISHED = "finished"47 CRASHED = "crashed"48 ABORTING = "aborting"49 SLA_FAILED = "sla_failed"50 SOFT_ABORTING = "soft_aborting"51 ABORTED = "aborted"52 PAUSED = "paused"53class _DeployStatus(utils.ImmutableMixin, utils.EnumMixin):54 DEPLOY_INIT = "deploy->init"55 DEPLOY_STARTED = "deploy->started"56 DEPLOY_SUBDEPLOY = "deploy->subdeploy"57 DEPLOY_FINISHED = "deploy->finished"58 DEPLOY_FAILED = "deploy->failed"59 DEPLOY_INCONSISTENT = "deploy->inconsistent"60 CLEANUP_STARTED = "cleanup->started"61 CLEANUP_FINISHED = "cleanup->finished"62 CLEANUP_FAILED = "cleanup->failed"63class _EndpointPermission(utils.ImmutableMixin, utils.EnumMixin):64 ADMIN = "admin"65 USER = "user"66class _EndpointType(utils.ImmutableMixin, utils.EnumMixin):67 INTERNAL = "internal"68 ADMIN = "admin"69 PUBLIC = "public"70class _Service(utils.ImmutableMixin, utils.EnumMixin):71 """OpenStack services names, by rally convention."""72 NOVA = "nova"73 NOVA_NET = "nova-network"74 CINDER = "cinder"75 MANILA = "manila"76 EC2 = "ec2"77 GLANCE = "glance"78 CLOUD = "cloud"79 HEAT = "heat"80 KEYSTONE = "keystone"81 NEUTRON = "neutron"82 DESIGNATE = "designate"83 CEILOMETER = "ceilometer"84 MONASCA = "monasca"85 S3 = "s3"86 SENLIN = "senlin"87 TROVE = "trove"88 SAHARA = "sahara"89 SWIFT = "swift"90 MISTRAL = "mistral"91 MURANO = "murano"92 IRONIC = "ironic"93 GNOCCHI = "gnocchi"94 MAGNUM = "magnum"95 WATCHER = "watcher"96class _ServiceType(utils.ImmutableMixin, utils.EnumMixin):97 """OpenStack services types, mapped to service names."""98 VOLUME = "volume"99 SHARE = "share"100 EC2 = "ec2"101 IMAGE = "image"102 CLOUD = "cloudformation"103 ORCHESTRATION = "orchestration"104 IDENTITY = "identity"105 CLUSTERING = "clustering"106 COMPUTE = "compute"107 NETWORK = "network"108 DNS = "dns"109 METERING = "metering"110 MONITORING = "monitoring"111 S3 = "s3"112 DATABASE = "database"113 DATA_PROCESSING = "data-processing"114 DATA_PROCESSING_MOS = "data_processing"115 OBJECT_STORE = "object-store"116 WORKFLOW_EXECUTION = "workflowv2"117 APPLICATION_CATALOG = "application-catalog"118 BARE_METAL = "baremetal"119 METRIC = "metric"120 CONTAINER_INFRA = "container-infra"121 INFRA_OPTIM = "infra-optim"122 def __init__(self):123 self.__names = {124 self.CLUSTERING: _Service.SENLIN,125 self.COMPUTE: _Service.NOVA,126 self.VOLUME: _Service.CINDER,127 self.SHARE: _Service.MANILA,128 self.EC2: _Service.EC2,129 self.IMAGE: _Service.GLANCE,130 self.CLOUD: _Service.CLOUD,131 self.ORCHESTRATION: _Service.HEAT,132 self.IDENTITY: _Service.KEYSTONE,133 self.NETWORK: _Service.NEUTRON,134 self.DNS: _Service.DESIGNATE,135 self.METERING: _Service.CEILOMETER,136 self.MONITORING: _Service.MONASCA,137 self.S3: _Service.S3,138 self.DATABASE: _Service.TROVE,139 self.DATA_PROCESSING: _Service.SAHARA,140 self.DATA_PROCESSING_MOS: _Service.SAHARA,141 self.OBJECT_STORE: _Service.SWIFT,142 self.WORKFLOW_EXECUTION: _Service.MISTRAL,143 self.APPLICATION_CATALOG: _Service.MURANO,144 self.BARE_METAL: _Service.IRONIC,145 self.METRIC: _Service.GNOCCHI,146 self.CONTAINER_INFRA: _Service.MAGNUM,147 self.INFRA_OPTIM: _Service.WATCHER,148 }149 def __getitem__(self, service_type):150 """Mapping protocol to service names.151 :param name: str, service name152 :returns: str, service type153 """154 return self.__names[service_type]155class _HookStatus(utils.ImmutableMixin, utils.EnumMixin):156 """Hook result statuses."""157 SUCCESS = "success"158 FAILED = "failed"159 VALIDATION_FAILED = "validation_failed"160class _TagType(utils.ImmutableMixin, utils.EnumMixin):161 TASK = "task"162 SUBTASK = "subtask"163 VERIFICATION = "verification"164class _VerifierStatus(utils.ImmutableMixin, utils.EnumMixin):165 """Verifier statuses."""166 INIT = "init"167 INSTALLING = "installing"168 INSTALLED = "installed"169 UPDATING = "updating"170 EXTENDING = "extending"171 FAILED = "failed"172# NOTE(andreykurilin): In case of updating these statuses, please do not forget173# to update doc reference too174class _VerificationStatus(utils.ImmutableMixin, utils.EnumMixin):175 """Verification statuses."""176 INIT = "init"177 RUNNING = "running"178 FINISHED = "finished"179 FAILED = "failed"180 CRASHED = "crashed"181class _TimeFormat(utils.ImmutableMixin, utils.EnumMixin):182 """International time formats"""183 ISO8601 = "%Y-%m-%dT%H:%M:%S%z"184TaskStatus = _TaskStatus()185SubtaskStatus = _SubtaskStatus()186DeployStatus = _DeployStatus()187EndpointPermission = _EndpointPermission()188ServiceType = _ServiceType()189Service = _Service()190EndpointType = _EndpointType()191HookStatus = _HookStatus()192TagType = _TagType()193VerifierStatus = _VerifierStatus()194VerificationStatus = _VerificationStatus()...

Full Screen

Full Screen

dispatcher.py

Source:dispatcher.py Github

copy

Full Screen

1#!/usr/bin/false2# The Dispatcher class functions simple in a way to help manage3# Hook and function calls by the hosting Service object.4#5# This class is a thread that waits until an order is given and will6# then process the messages in a FIFO order.7#8# System Management Daemon9#10# Copyright (C) 2016 - 2022 iDigitalFlame11#12# This program is free software: you can redistribute it and/or modify13# it under the terms of the GNU General Public License as published by14# the Free Software Foundation, either version 3 of the License, or15# any later version.16#17# This program is distributed in the hope that it will be useful,18# but WITHOUT ANY WARRANTY; without even the implied warranty of19# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the20# GNU General Public License for more details.21#22# You should have received a copy of the GNU General Public License23# along with this program. If not, see <https://www.gnu.org/licenses/>.24#25from threading import Thread, Event26from lib.loader import load_modules27from lib.structs.hook import HookDaemon28from lib.structs.message import Message29from lib.constants import (30 HOOK_OK,31 HOOK_LOG,32 LOG_LEVEL,33 HOOK_DAEMON,34 HOOK_RELOAD,35 HOOK_STARTUP,36 HOOK_SHUTDOWN,37)38class Dispatcher(Thread):39 def __init__(self, service, directory):40 Thread.__init__(self, name="SMDDispatcherThread", daemon=False)41 self._hooks = None42 self._daemons = None43 self._dir = directory44 self._running = False45 self._service = service46 self._messages = list()47 self._waiting = Event()48 def run(self):49 self._running = True50 self._waiting.clear()51 self._service.debug("Starting Dispatcher processing thread..")52 self._hooks = load_modules(self._service, self._dir)53 if HOOK_STARTUP in self._hooks:54 self._service.debug("Running Startup Hooks..")55 r = self._hooks[HOOK_STARTUP].run(self._service, Message(HOOK_STARTUP))56 if len(r) > 0:57 self._service.send(None, r)58 del r59 try:60 self._service.write()61 except OSError as err:62 self._service.error(63 f'Error saving configuration "{self._service.get_file()}"!', err=err64 )65 if HOOK_DAEMON in self._hooks:66 self._daemons = HookDaemon(self._service, self._hooks[HOOK_DAEMON])67 self._daemons.start()68 while self._running:69 if len(self._messages) == 0:70 self._waiting.clear()71 self._waiting.wait()72 while len(self._messages) > 0:73 m = self._messages.pop()74 if not m.is_valid():75 self._service.warning("Received an invalid Message!")76 del m77 continue78 if m.message.header() == HOOK_LOG:79 self80 self._service.set_level(m.message.get("level", LOG_LEVEL))81 if self._service.is_server() and not m.message.is_multicast():82 self._service.info("Forwarding log message to clients..")83 self._service.send(None, m.message.set_multicast())84 del m85 continue86 if m.message.header() == HOOK_RELOAD:87 self._service.debug(88 f'Reloading the configuration from "{self._service.get_file()}"..'89 )90 try:91 self._service.read()92 except OSError as err:93 self._service.error(94 f'Error reading configuration "{self._service.get_file()}"!',95 err=err,96 )97 if m.message.header() not in self._hooks:98 if m.message.header() != HOOK_OK:99 self._service.warning(100 f"Received an un-hooked Message 0x{m.message.header():02X}!"101 )102 del m103 continue104 try:105 self._service.debug(106 f"Running Hooks for 0x{m.message.header():02X}."107 )108 r = self._hooks[m.message.header()].run(self._service, m.message)109 if len(r) > 0:110 self._service.send(m.eid, r)111 del r112 except Exception as err:113 self._service.error("Error processing Message request!", err=err)114 finally:115 del m116 if not self._running:117 return118 self.stop()119 def stop(self):120 self._service.debug("Stopping Dispatcher thread..")121 if HOOK_SHUTDOWN in self._hooks:122 self._service.debug("Running shutdown hooks..")123 self._hooks[HOOK_SHUTDOWN].run(self._service, Message(header=HOOK_SHUTDOWN))124 try:125 self._service.write()126 except OSError as err:127 self._service.error(128 f'Could not save configuration "{self._service.get_file()}"!', err=err129 )130 self._running = False131 self._waiting.set()132 if self._daemons is None:133 return134 self._daemons.stop()135 def add(self, eid, message):136 if not self._running:137 return138 self._messages.append(DispatchMessage(eid, message))139 self._waiting.set()140class DispatchMessage(object):141 def __init__(self, eid, message):142 self.eid = eid143 self.message = message144 def is_valid(self):145 return isinstance(self.message, Message) and isinstance(146 self.message.header(), int...

Full Screen

Full Screen

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:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run ATX 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!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful