Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use is_sub_dict method in localstack
Best Python code snippet using localstack_python
loader.py
Source:loader.py 
...23from server import server_constructor24from terminal import MDTerminal25from updater import Updater26from utils import state_cache, Messenger, pretty_time, SignalHandlerDummy27def is_sub_dict(key, data: dict):28 return isinstance(data.get(key), dict) and data[key]29def prepare_detectors(detector, is_extract=False) -> set or dict:30 def is_uni(cls_, info_) -> bool:31 return isinstance(cls_, info_) or issubclass(cls_, info_)32 types = (detectors.Detector, str) if is_extract else (detectors.Detector,)33 if is_uni(detector, types):34 detector = [detector]35 if not isinstance(detector, (list, tuple, set)):36 detector = []37 if is_extract:38 detector = set(x if isinstance(x, str) else x.NAME for x in detector if is_uni(x, types))39 else:40 detector = {x.NAME: x for x in detector if is_uni(x, types)}41 return detector42class Loader(Owner):43 def __init__(self, init_cfg: dict, init_state: dict, path: dict, sig: SignalHandlerDummy):44 self._sig = sig45 self.reload = False46 self._restore_filename = None47 self._lock = threading.Lock()48 self._stts_lock = threading.Lock()49 self._join_lock = threading.Lock()50 self._pub = PubSub()51 self._sig.set_wakeup_callback(lambda: self.sub_call('default', 'terminal_stop', True))52 self._logger = logger.Logger(self)53 proxies.add_logger(self._logger.add('Proxy'))54 self._cfg = ConfigHandler(cfg=init_cfg, state=init_state, path=path, log=self._logger.add('CFG'), owner=self)55 self._logger.init(cfg=self._cfg, owner=self)56 self._log = self._logger.add('SYSTEM')57 self._listen = Listener(cfg=self._cfg, log=self._logger.add('REC'), owner=self)58 self._notifier = MajordomoNotifier(cfg=self._cfg, log=self._logger.add('Notifier'), owner=self)59 self._tts = stts.TextToSpeech(cfg=self._cfg, log=self._logger.add('TTS'))60 self._play = Player(cfg=self._cfg, log=self._logger.add('Player'), owner=self)61 self._music = music_constructor(cfg=self._cfg, logger=self._logger, owner=self)62 self._stt = stts.SpeechToText(cfg=self._cfg, log=self._logger.add('STT'), owner=self)63 self._mm = ModuleManager(cfg=self._cfg, log=self._logger.add('MM'), owner=self)64 self._updater = Updater(cfg=self._cfg, log=self._logger.add('Updater'), owner=self)65 self._backup = Backup(cfg=self._cfg, log=self._logger.add('Backup'), owner=self)66 self._terminal = MDTerminal(cfg=self._cfg, log=self._logger.add('Terminal'), owner=self)67 self._server = server_constructor(cfg=self._cfg, logger=self._logger, owner=self)68 self._plugins = Plugins(cfg=self._cfg, log=self._logger.add('Plugins'), owner=self)69 self._duplex_pool = DuplexPool(cfg=self._cfg, log=self._logger.add('DP'), owner=self)70 self._discovery = DiscoveryServer(cfg=self._cfg, log=self._logger.add('Discovery'))71 def start_all_systems(self):72 self._music.start()73 self._play.start()74 self._play.say_info(F('ÐÑивеÑÑÑвÑÑ. ÐолоÑовой ÑеÑминал наÑÑÑаиваеÑÑÑ, ÑÑи... два... один...'), 0, wait=0.5)75 self._stt.start()76 self._cfg.start()77 self._notifier.start()78 self._mm.start()79 self._updater.start()80 self._terminal.start()81 self._server.start()82 self._discovery.start()83 self._backup.start()84 self._plugins.start()85 self.messenger(lambda: self.log(available_version_msg(self._cfg.version_info), logger.INFO), None)86 self.sub_call('default', 'version', self._cfg.version_str)87 self.volume_callback(self.get_volume())88 def stop_all_systems(self):89 self._cfg.config_save(final=True)90 self.join_thread(self._plugins)91 self._mm.stop()92 self.join_thread(self._discovery)93 self.join_thread(self._server)94 self.join_thread(self._terminal)95 self.join_thread(self._backup)96 self.join_thread(self._updater)97 self.join_thread(self._notifier)98 self.join_thread(self._duplex_pool)99 self._play.quiet()100 self._play.kill_popen()101 self._play.say_info(F('ÐолоÑовой ÑеÑминал завеÑÑÐ°ÐµÑ ÑÐ²Ð¾Ñ ÑабоÑÑ.'))102 self._stt.stop()103 self._play.stop()104 self.join_thread(self._music)105 if self._restore_filename:106 self._backup.restore(self._restore_filename)107 self._restore_filename = ''108 self.join_thread(self._logger.remote_log)109 self._pub.stopping = True110 self._logger.join()111 self._pub.join()112 self._pub.report()113 def log(self, msg: str, lvl=logger.DEBUG):114 self._log(msg, lvl)115 def join_thread(self, obj):116 def obj_log(msg_: str, lvl=logger.DEBUG):117 if log_present:118 obj.log(msg_, lvl)119 def diagnostic_msg() -> str:120 _call = getattr(obj, 'diagnostic_msg', None)121 return ' {}'.format(_call()) if callable(_call) else ''122 with self._join_lock:123 close_signal = getattr(obj, 'close_signal', None)124 if close_signal:125 close_signal()126 if not obj.work:127 return128 log_present = callable(getattr(obj, 'log', None))129 obj.work = False130 obj_log('stopping...')131 stop_time = time.time()132 obj.join()133 stop_time = time.time() - stop_time134 if not obj.is_alive():135 obj_log('stop.', logger.INFO)136 else:137 obj_log('stopping error.', logger.ERROR)138 name_ = '.'.join(getattr(obj.log, 'name', [''])) if log_present else None139 name_ = name_ or str(obj)140 msg = 'Thread \'{}\' stuck and not stopping in {}!{}'.format(141 name_, pretty_time(stop_time), diagnostic_msg())142 self.log(msg, logger.ERROR)143 def subscribe(self, event, callback, channel='default') -> bool:144 return self._pub.subscribe(event, callback, channel)145 def unsubscribe(self, event, callback, channel='default') -> bool:146 return self._pub.unsubscribe(event, callback, channel)147 def registration(self, event: str, channel='default'):148 return self._pub.registration(event, channel)149 def has_subscribers(self, event: str, channel='default') -> bool:150 return self._pub.has_subscribers(event, channel)151 def events_list(self, channel='default') -> list:152 return self._pub.events_list(channel)153 def send_notify(self, event: str, *args, **kwargs):154 return self._pub.sub_call('default', event, *args, **kwargs)155 def sub_call(self, channel: str, event: str, *args, **kwargs):156 return self._pub.sub_call(channel, event, *args, **kwargs)157 @staticmethod158 def messenger(call, callback, *args, **kwargs) -> bool:159 return Messenger(call, callback, *args, **kwargs)()160 def insert_module(self, module) -> bool:161 return self._mm.insert_module(module)162 def extract_module(self, callback) -> bool:163 return self._mm.extract_module(callback)164 def insert_detectors(self, detector):165 detector = prepare_detectors(detector)166 if not detector:167 return168 def callback():169 with self._lock:170 detectors.DETECTORS.update(detector)171 self.__reconfigure_terminal(detector)172 # noinspection PyTypeChecker173 self.terminal_call('callme', callback, save_time=False)174 def extract_detectors(self, detector):175 detector = prepare_detectors(detector, True)176 if not detector:177 return178 def callback():179 with self._lock:180 [detectors.DETECTORS.pop(x, None) for x in detector]181 self.__reconfigure_terminal(detector)182 # noinspection PyTypeChecker183 self.terminal_call('callme', callback, save_time=False)184 def add_stt_provider(self, name: str, entrypoint) -> bool:185 with self._stts_lock:186 if name not in STT.PROVIDERS:187 STT.PROVIDERS[name] = entrypoint188 return True189 return False190 def remove_stt_provider(self, name: str):191 with self._stts_lock:192 try:193 return STT.PROVIDERS.pop(name)194 except KeyError:195 return None196 def add_tts_provider(self, name: str, entrypoint) -> bool:197 with self._stts_lock:198 if name not in TTS.PROVIDERS:199 TTS.PROVIDERS[name] = entrypoint200 return True201 return False202 def remove_tts_provider(self, name: str):203 with self._stts_lock:204 try:205 return TTS.PROVIDERS.pop(name)206 except KeyError:207 return None208 def tts_providers(self) -> list:209 with self._stts_lock:210 return list(TTS.PROVIDERS.keys())211 def stt_providers(self) -> list:212 with self._stts_lock:213 return list(STT.PROVIDERS.keys())214 def is_tts_provider(self, name: str) -> bool:215 return name in TTS.PROVIDERS216 def is_stt_provider(self, name: str) -> bool:217 return name in STT.PROVIDERS218 def plugins_status(self, state: str) -> dict:219 return self._plugins.status(state)220 def get_plugin(self, name: str) -> object:221 try:222 return self._plugins.modules[name]223 except KeyError:224 raise RuntimeError('Plugin \'{}\' not found'.format(name))225 except Exception as e:226 raise RuntimeError('Error accessing to plugin \'{}\': {}'.format(name, e))227 def list_notifications(self) -> list:228 return self._notifier.list_notifications()229 def add_notifications(self, events: list, is_self=False) -> list:230 return self._notifier.add_notifications(events, is_self)231 def remove_notifications(self, events: list) -> list:232 return self._notifier.remove_notifications(events)233 def say(self, msg: str, lvl: int = 2, alarm=None, wait=0, is_file: bool = False, blocking: int = 0):234 self._play.say(msg, lvl, alarm, wait, is_file, blocking)235 def play(self, file, lvl: int = 2, wait=0, blocking: int = 0):236 self._play.play(file, lvl, wait, blocking)237 def say_info(self, msg: str, lvl: int = 2, alarm=None, wait=0, is_file: bool = False):238 self._play.say_info(msg, lvl, alarm, wait, is_file)239 def set_lvl(self, lvl: int) -> bool:240 return self._play.set_lvl(lvl)241 def clear_lvl(self):242 self._play.clear_lvl()243 def quiet(self):244 self._play.quiet()245 def full_quiet(self):246 self._play.full_quiet()247 def really_busy(self) -> bool:248 return self._play.really_busy()249 @state_cache(interval=0.008)250 def noising(self) -> bool:251 return self._play.noising()252 def kill_popen(self):253 self._play.kill_popen()254 def listen(self, hello: str = '', deaf: bool = True, voice: bool = False) -> tuple:255 return self._stt.listen(hello, deaf, voice)256 def voice_record(self, hello: str or None, save_to: str, convert_rate=None, convert_width=None, limit=8):257 return self._stt.voice_record(hello, save_to, convert_rate, convert_width, limit)258 def voice_recognition(self, audio, quiet: bool = False, fusion=None) -> str:259 return self._stt.voice_recognition(audio, quiet, fusion)260 @property261 def max_mic_index(self) -> int:262 return self._stt.max_mic_index263 @max_mic_index.setter264 def max_mic_index(self, val: int):265 self._stt.max_mic_index = val266 @property267 def mic_index(self) -> int:268 return self._stt.get_mic_index()269 def phrase_from_files(self, files: list) -> tuple:270 return self._stt.phrase_from_files(files)271 def multiple_recognition(self, file_or_adata, providers: list) -> list:272 return self._stt.multiple_recognition(file_or_adata, providers)273 @property274 def sys_say_chance(self) -> bool:275 return self._stt.sys_say.chance276 def music_state(self) -> str:277 return self._music.state()278 def music_play(self, uri):279 self._music.play(uri)280 def music_pause(self, paused=None):281 self._music.pause(paused)282 @property283 def music_plays(self) -> bool:284 return self._music.plays285 @property286 def music_volume(self):287 return self._music.volume288 @music_volume.setter289 def music_volume(self, vol):290 self._music.volume = vol291 @property292 def music_real_volume(self):293 return self._music.real_volume294 @music_real_volume.setter295 def music_real_volume(self, vol):296 self._music.real_volume = vol297 @property298 def music_track_name(self) -> str or None:299 return self._music.get_track_name()300 def tts(self, msg, realtime: bool = True):301 return self._tts.tts(msg, realtime)302 def ask_again_callback(self):303 self._pub.call('ask_again')304 def voice_activated_callback(self):305 self._pub.call('voice_activated')306 def speech_recognized_callback(self, status: bool):307 if status and self._cfg.gts('alarm_recognized'):308 self.play(self._cfg.path['bimp'])309 self._pub.call('speech_recognized_{}success'.format('' if status else 'un'))310 def record_callback(self, start_stop: bool):311 self._pub.call('start_record' if start_stop else 'stop_record')312 def say_callback(self, start_stop: bool):313 self._pub.call('start_talking' if start_stop else 'stop_talking')314 def speech_recognized(self, start_stop: bool):315 self._pub.call('start_recognized' if start_stop else 'stop_recognized')316 def music_status_callback(self, status: str):317 self._pub.call('music_status', status if status is not None else 'error')318 def music_volume_callback(self, volume: int):319 self._pub.call('music_volume', volume if volume is not None else -1)320 def volume_callback(self, volume: int):321 self._pub.call('volume', volume)322 @property323 def srv_ip(self) -> str:324 return self._cfg['smarthome']['ip']325 def update(self):326 self._updater.update()327 def manual_rollback(self):328 self._updater.manual_rollback()329 def backup_manual(self):330 self._backup.manual_backup()331 def backup_restore(self, filename: str):332 if not self._restore_filename and filename:333 self._restore_filename = filename334 self.die_in(3, reload=True)335 def backup_list(self) -> list:336 return self._backup.backup_list()337 def modules_tester(self, phrase: str, call_me=None, rms=None, model=None):338 return self._mm.tester(phrase, call_me, rms, model)339 def die_in(self, wait, reload=False):340 self.reload = reload341 self._sig.die_in(wait)342 @property343 def get_volume_status(self) -> dict:344 music_volume = self._music.real_volume345 return {'volume': self.get_volume(), 'music_volume': music_volume if music_volume is not None else -1}346 def terminal_call(self, cmd: str, data='', lvl: int = 0, save_time: bool = True):347 self._terminal.call(cmd, data, lvl, save_time)348 def terminal_listen(self) -> bool:349 return self._terminal.listening350 def recognition_forever(self, interrupt_check: callable, callback: callable):351 return self._listen.recognition_forever(interrupt_check, callback)352 def get_vad_detector(self, source_or_mic, vad_mode=None, vad_lvl=None, energy_lvl=None, energy_dynamic=None):353 return self._listen.get_vad_detector(source_or_mic, vad_mode, vad_lvl, energy_lvl, energy_dynamic)354 def detected_fake(self, text: str, rms=None, model=None, cb=None):355 self._listen.detected_fake(text, rms, model, cb)356 def listener_listen(self, r=None, mic=None, vad=None):357 return self._listen.listen(r, mic, vad)358 def background_listen(self):359 return self._listen.background_listen()360 def get_volume(self) -> int:361 control = self._cfg.gt('volume', 'line_out', '')362 card = self._cfg.gt('volume', 'card', 0)363 if not control or control == volume_.UNDEFINED:364 return -2365 return volume_.get_volume(control, card)366 def set_volume(self, vol) -> int:367 control = self._cfg.gt('volume', 'line_out', '')368 card = self._cfg.gt('volume', 'card', 0)369 if not control or control == volume_.UNDEFINED:370 return -2371 try:372 return volume_.set_volume(vol, control, card)373 except RuntimeError as e:374 self.log('set_volume({}): {}'.format(vol, e), logger.WARN)375 return -1376 def settings_from_inside(self, cfg: dict) -> bool:377 with self._lock:378 return self._cfg.update_from_dict(cfg)379 def settings_from_srv(self, cfg: str or dict) -> dict:380 # Reload modules if their settings could be changes381 with self._lock:382 diff = self._cfg.update_from_external(cfg)383 reload_terminal = False384 detector_reconfigure = False385 vad_reconfigure = False386 if diff is None:387 self._cfg.print_cfg_no_change()388 return {}389 lang, lang_check = None, None390 if is_sub_dict('settings', diff) and ('lang' in diff['settings'] or 'lang_check' in diff['settings']):391 # re-init lang392 lang = diff['settings'].pop('lang', None)393 lang_check = diff['settings'].pop('lang_check', None)394 self._cfg.lang_init()395 if lang:396 # reload phrases397 self._stt.reload()398 # reload modules399 self._mm.reload()400 if is_sub_dict('models', diff):401 # reload models. Reload terminal - later402 self._cfg.models_load()403 reload_terminal = True404 if is_sub_dict('log', diff):405 # reload logger406 self._logger.reload()407 if is_sub_dict('cache', diff):408 # re-check tts cache409 self._cfg.tts_cache_check()410 if is_sub_dict('proxy', diff):411 # re-init proxy412 self._cfg.proxies_init()413 if is_sub_dict('music', diff):414 # reconfigure music server415 self.music_reload()416 if is_sub_dict('update', diff):417 # update 'update' interval418 self._updater.reload()419 if is_sub_dict('backup', diff):420 # update backup interval421 self._backup.reload()422 if is_sub_dict('smarthome', diff):423 if 'allow_addresses' in diff['smarthome']:424 # re-init allow ip addresses425 self._cfg.allow_addresses_init()426 if 'disable_server' in diff['smarthome']:427 # handle [smarthome] disable_server428 self.messenger(self.server_reload, None)429 # resubscribe430 self._notifier.reload(diff)431 self._duplex_pool.reload()432 if is_sub_dict('noise_suppression', diff):433 # reconfigure APM. Reload terminal - later434 self._cfg.apm_configure()435 reload_terminal = True436 if is_sub_dict('listener', diff):437 reload_terminal = True438 detector_reconfigure = 'detector' in diff['listener']439 vad_reconfigure = bool([key for key in ('vad_mode', 'vad_chrome') if key in diff['listener']])440 if is_sub_dict('settings', diff) or reload_terminal:441 # reload terminal442 # noinspection PyTypeChecker443 self.terminal_call('reload', (detector_reconfigure, vad_reconfigure), save_time=False)444 # restore lang's445 if lang is not None:446 diff['settings']['lang'] = lang447 if lang_check is not None:448 diff['settings']['lang_check'] = lang_check449 # check and reload plugins450 self._plugins.reload(diff)451 self._cfg.print_cfg_change()452 self._cfg.config_save()453 return diff454 def music_reload(self):...
test_flask_ligand.py
Source:test_flask_ligand.py
...30 dictionary.31 """32 small_dict = {"one": 1, "two": 2}33 big_dict = {"one": 1, "two": 2, "three": 3}34 assert FlaskLigandTestHelpers.is_sub_dict(small_dict, big_dict)35 def test_is_sub_dict_order_mismatch(self, pytester):36 """37 Verify that the 'is_sub_dict' helper will correctly identify a dictionary is in fact a subset of another38 dictionary regardless of the order of items.39 """40 small_dict = {"two": 2, "one": 1}41 big_dict = {"one": 1, "three": 3, "two": 2}42 assert FlaskLigandTestHelpers.is_sub_dict(small_dict, big_dict)43 def test_is_sub_dict_same_dict(self, pytester):44 """45 Verify that the 'is_sub_dict' helper will correctly identify a dictionary is considered a sub-dict of itself.46 """47 same_dict = {"uno": 1, "dos": 2}48 assert FlaskLigandTestHelpers.is_sub_dict(same_dict, same_dict)49 def test_is_sub_dict_identical_items(self, pytester):50 """51 Verify that the 'is_sub_dict' helper will correctly identify a dictionary with the same exact items of another52 dictionary is considered a sub-dict.53 """54 first_dict = {"uno": 1, "dos": 2}55 second_dict = {"uno": 1, "dos": 2}56 assert FlaskLigandTestHelpers.is_sub_dict(first_dict, second_dict)57class TestNegativeFlaskLigandTestHelpers(object):58 """Negative test cases the 'FlaskLigandTestHelpers' helpers class."""59 def test_is_sub_dict_different_dicts(self, pytester):60 """61 Verify that the 'is_sub_dict' helper will reject comparisons of dictionary that have the same size,62 but different values.63 """64 first_dict = {"uno": 1, "dos": 2}65 second_dict = {"totally": "different", "dictionaries": "all together"}66 assert not FlaskLigandTestHelpers.is_sub_dict(first_dict, second_dict)67 def test_is_sub_dict_key_mismatch(self, pytester):68 """69 Verify that the 'is_sub_dict' helper will reject comparisons of dictionaries that have the same size,70 but different keys.71 """72 small_dict = {"uno": 1, "dos": 2}73 big_dict = {"one": 1, "two": 2}74 assert not FlaskLigandTestHelpers.is_sub_dict(small_dict, big_dict)75 def test_is_sub_dict_value_mismatch(self, pytester):76 """77 Verify that the 'is_sub_dict' helper will reject comparisons of dictionaries that have the same size,78 but different values.79 """80 small_dict = {"one": "uno", "two": "dos"}81 big_dict = {"one": 1, "two": 2}...
learn_dict.py
Source:learn_dict.py
1#!/usr/bin/env python2# -*- coding: utf-8 -*-3def is_sub_dict(less, more):4 return less.items() <= more.items()5def dict_has_subset(more, less):6 for k, v in less.items():7 if k not in more:8 return False9 if more[k] != v:10 return False11 return True12def test_is_sub_dict():13 less = {'a': 1}14 more = {'a': 1, 'b': 2}15 assert is_sub_dict(less, more)16 assert dict_has_subset(more, less)17 less.update(a='a')18 assert is_sub_dict(less, more) is False19 assert dict_has_subset(more, less) is False20if __name__ == '__main__':...
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 localstack 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?
