KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _service method in ATX

Best Python code snippet using ATX

air_conditioner.py

Source:air_conditioner.py

...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:...

consts.py

Source:consts.py

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()...

dispatcher.py

Source:dispatcher.py

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...

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.