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

How to use singleton method in tempest

Best Python code snippet using tempest_python

IonMeshDiscoveryManager.py

Source:IonMeshDiscoveryManager.py

1#!/usr/bin/env python2# Copyright (C) 2017 Thermo Fisher Scientific. All Rights Reserved.3import dbus, gobject, avahi4from dbus.mainloop.glib import DBusGMainLoop5import threading6import time7import copy8import socket9from iondb.utils.hostip import gethostfqdn10class IonMeshDiscoveryManager(threading.Thread):11    """This class will monitor avahi via dbus in order to detect the current mesh network setup"""12    # singleton instance13    __singleton = None14    # list of computers in the mesh15    __meshComputers = list()16    # registration type17    __REG_TYPE = "_ionMesh._tcp"18    # thread lock19    __threadLock = threading.Lock()20    # local host name21    __localhost = gethostfqdn()22    def __new__(cls, *args, **kwargs):23        if not cls.__singleton:24            cls.__singleton = super(IonMeshDiscoveryManager, cls).__new__(25                cls, *args, **kwargs26            )27            gobject.threads_init()28            # setup dbus stuff29            cls.__singleton.__bus = dbus.SystemBus(mainloop=DBusGMainLoop())30            cls.__singleton.__server = dbus.Interface(31                cls.__singleton.__bus.get_object(32                    avahi.DBUS_NAME, avahi.DBUS_PATH_SERVER33                ),34                avahi.DBUS_INTERFACE_SERVER,35            )36            # Look for self.regtype services and hook into callbacks37            cls.__singleton.__browser = dbus.Interface(38                cls.__singleton.__bus.get_object(39                    avahi.DBUS_NAME,40                    cls.__singleton.__server.ServiceBrowserNew(41                        avahi.IF_UNSPEC,42                        avahi.PROTO_UNSPEC,43                        cls.__singleton.__REG_TYPE,44                        "",45                        dbus.UInt32(0),46                    ),47                ),48                avahi.DBUS_INTERFACE_SERVICE_BROWSER,49            )50            cls.__singleton.__browser.connect_to_signal(51                "ItemNew", cls.__singleton.__serviceFound52            )53            cls.__singleton.__browser.connect_to_signal(54                "ItemRemove", cls.__singleton.__serviceRemoved55            )56            cls.__singleton.__loop = gobject.MainLoop()57            # initialize threading super object58            threading.Thread.__init__(cls.__singleton)59            cls.__singleton.start()60            time.sleep(1)61        return cls.__singleton62    def stop(self):63        """64        Call this to stop the thread as one would expect65        """66        self.__loop.quit()67    def run(self):68        """69        Method called with the thread object's "start" method is called70        """71        self.__loop.run()72    def getMeshComputers(self):73        """74        This will create a copy of the list of the computers in the mesh by hostname75        :return: List of hostnames converted to fully qualified domain names76        """77        self.__threadLock.acquire()78        try:79            ret = []80            for name in self.__meshComputers:81                fqdn = socket.getfqdn(name)82                if fqdn not in ["localhost", "tsvm"]:83                    ret.append(fqdn)84            return ret85        finally:86            self.__threadLock.release()87    def __serviceFound(self, interface, protocol, name, stype, domain, flags):88        """Callback for when a service needs to be added to the mesh list"""89        # skip local services90        # http://sources.debian.net/src/avahi/0.6.32-1/avahi-python/avahi/__init__.py/?hl=50#L5091        if flags & avahi.LOOKUP_RESULT_LOCAL:92            return93        # add the computer name to the list of mesh computers94        self.__threadLock.acquire()95        try:96            if name not in self.__meshComputers:97                self.__meshComputers.append(str(name))98        finally:99            self.__threadLock.release()100    def __serviceRemoved(self, interface, protocol, name, stype, domain, flags):101        """102        Callback for when a service needs to be removed from the mesh list103        :param interface:104        :param protocol:105        :param name:106        :param stype:107        :param domain:108        :param flags:109        """110        self.__threadLock.acquire()111        try:112            if name in self.__meshComputers:113                self.__meshComputers.remove(str(name))114        finally:115            self.__threadLock.release()116def getLocalComputer():117    """Gets the localhost name"""118    return gethostfqdn()119if __name__ == "__main__":120    mesh1 = IonMeshDiscoveryManager()121    try:122        while True:123            time.sleep(1)124            print("*************************")125            print(mesh1.getMeshComputers())126    except KeyboardInterrupt:127        pass128    finally:...

singleton.py

Source:singleton.py

1#!/usr/bin/env python2#3# Copyright 2007 Google Inc.4#5# Licensed under the Apache License, Version 2.0 (the "License");6# you may not use this file except in compliance with the License.7# You may obtain a copy of the License at8#9#     http://www.apache.org/licenses/LICENSE-2.010#11# Unless required by applicable law or agreed to in writing, software12# distributed under the License is distributed on an "AS IS" BASIS,13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.14# See the License for the specific language governing permissions and15# limitations under the License.16#17# Copyright 2011 Google Inc. All Rights Reserved.18"""Thread-safe implementation of a singleton decorator.19Sometimes, only a single instance of a class should ever be created. This file20provides a wrapper that turns a class into a Singleton. The constructor may21only be called once; a static method Singleton() provides access to the22constructed instance. If Singleton() is called before the constructor, or if23the constructor is called multiple times, Errors are raised. This wrapper is24thread-safe; calls to the constructor and Singleton() method are protected25by per-class locks.26Singletons are often associated with bad coding practices; see27http://wiki/Main/SingletonsConsideredDangerous and decide if you should28really be using this functionality. Consider alternatives, like the29"Borg pattern" where object state (instead of object identity) is shared.30To make your singletons more testable, use the following idiom:31class A(...):32  "All the complicated code goes in here, can be tested normally..."33  ..34  ..35@singleton.Singleton36class B(A):37  "Singleton instance of A"38Example usage:39from google.pyglib import singleton40@singleton.Singleton41class Foo(object):42  "Example singleton"43a = Foo()44b = Foo.Singleton()45c = Foo.Singleton()46assert a == b47assert b == c48"""49import threading50_CLASS_LOCKS = {}  # Holds the per-class locks.51_CLASS_LOCKS_LOCK = threading.Lock()  # Lock for obtaining a per-class lock.52_INSTANCES = {}  # Mapping from class to instantiated object.53class Error(Exception):54  """Base error class."""55class AlreadyHasSingletonMethodError(Error):56  """Raised if the class already defines a Singleton() method."""57  def __init__(self, cls):58    Error.__init__(self)59    self.cls = cls60  def __str__(self):61    return 'Class already has a Singleton() method: %s' % self.cls62class NotConstructedError(Error):63  """Raised if the constructor has not been called yet."""64  def __init__(self, cls):65    Error.__init__(self)66    self.cls = cls67  def __str__(self):68    return 'Constructor has not yet been called for class %s' % self.cls69class ConstructorCalledAgainError(Error):70  """Raised if the constructor is called twice for a singleton."""71  def __init__(self, cls, args, kws):72    Error.__init__(self)73    self.cls = cls74    self.args = args75    self.kws = kws76  def __str__(self):77    return ('Constructor called (again) on class %s with args %s and kws %s'78            % (self.cls, self.args, self.kws))79def _GetClassLock(cls):80  """Returns the lock associated with the class."""81  with _CLASS_LOCKS_LOCK:82    if cls not in _CLASS_LOCKS:83      _CLASS_LOCKS[cls] = threading.Lock()84    return _CLASS_LOCKS[cls]85def Singleton(cls):86  """Turn a class into a singleton.87  One call to the constructor is allowed. After that, all future calls88  must be to the Singleton() static method.89  This code is multithread-safe. Shared locks are held for brief periods90  of time; when a class is instantiated, it uses a class specific lock.91  Args:92    cls: The class to decorate.93  Returns:94    The singleton class. Note that this class is an extension95    of the class it is decorating.96  Raises:97    AlreadyHasSingletonMethodError: If the class has a Singleton method98      defined.99  """100  if hasattr(cls, 'Singleton'):101    raise AlreadyHasSingletonMethodError(cls)102  class _Singleton(cls):103    def __init__(self, *args, **kws):104      class_lock = _GetClassLock(cls)105      with class_lock:106        if cls not in _INSTANCES:107          cls.__init__(self, *args, **kws)108          _INSTANCES[cls] = self109        else:110          raise ConstructorCalledAgainError(cls, args, kws)111    @staticmethod112    def Singleton():113      class_lock = _GetClassLock(cls)114      with class_lock:115        if cls not in _INSTANCES:116          raise NotConstructedError(cls)117        else:118          return _INSTANCES[cls]...

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.