Best Python code snippet using playwright-python
team_common.py
Source:team_common.py  
...183        Check if the union tag is ``alert_only``.184        :rtype: bool185        """186        return self._tag == 'alert_only'187    def is_stop_sync(self):188        """189        Check if the union tag is ``stop_sync``.190        :rtype: bool191        """192        return self._tag == 'stop_sync'193    def is_other(self):194        """195        Check if the union tag is ``other``.196        :rtype: bool197        """198        return self._tag == 'other'199    def _process_custom_annotations(self, annotation_type, field_path, processor):200        super(MemberSpaceLimitType, self)._process_custom_annotations(annotation_type, field_path, processor)201MemberSpaceLimitType_validator = bv.Union(MemberSpaceLimitType)...web.py
Source:web.py  
...77                eel.sleep(0.1)78            if STOP_SYNC:79                break80@eel.expose81def stop_sync():82    global STOP_SYNC83    STOP_SYNC = True84def save_hue_connection_to_file():85    save_file = open(SAVE_FILE_PATH, 'w')86    json.dump(HUE_CONNECTION, save_file)87def load_hue_connection_from_file():88    try:89        save_file = open(SAVE_FILE_PATH, 'r')90        data = json.load(save_file)91        HUE_CONNECTION['ip'] = data['ip']92        HUE_CONNECTION['lights'] = data['lights']93        HUE_CONNECTION['brightness'] = data['brightness']94        HUE_CONNECTION['sim'] = data['sim'] or 'AC'95        HUE_CONNECTION['colors'] = data['colors']96        HUE_CONNECTION['auto_sync'] = data['auto_sync']97    except (FileNotFoundError, KeyError) as error:98        print(error)99        HUE_CONNECTION['ip'] = ''100        HUE_CONNECTION['lights'] = []101        HUE_CONNECTION['brightness'] = 255102        HUE_CONNECTION['sim'] = 'AC'103        HUE_CONNECTION['colors'] = {104            'No_Flag': '',105            'Blue_Flag': GUI_COLOR_BLUE_FLAG,106            'Yellow_Flag': GUI_COLOR_YELLOW_FLAG,107            'Black_Flag': '',108            'White_Flag': GUI_COLOR_WHITE_FLAG,109            'Checkered_Flag': '',110            'Penalty_Flag': GUI_COLOR_PENALTY_FLAG,111            'Green_Flag': GUI_COLOR_GREEN_FLAG,112            'Orange_Flag': GUI_COLOR_ORANGE_FLAG113        }114        HUE_CONNECTION['auto_sync'] = False115def bridge_connection_works() -> bool:116    if HUE_CONNECTION['ip'] == '':117        return False118    else:119        try:120            Bridge(HUE_CONNECTION['ip'])121            return True122        except:123            return False124def get_lights_from_bridge(bridge: Bridge) -> []:125    light_options = []126    for light in bridge.get_light_objects():127        light_options.append(light.name)128    return light_options129def raise_color(color_hex: str):130    if color_hex == '' or color_hex == '#000000':131        for light in HUE_CONNECTION['lights']:132            Bridge(HUE_CONNECTION['ip']).set_light(light, {'transitiontime': 0, 'on': False})133    else:134        converter = Converter()135        color_xy = converter.hex_to_xy(color_hex.replace('#', ''))136        for light in HUE_CONNECTION['lights']:137            Bridge(HUE_CONNECTION['ip']).set_light(light, {'transitiontime': 0, 'on': True, 'bri': int(HUE_CONNECTION['brightness']),138                                                           'xy': color_xy})139def raise_ac_flag(flag: ac.ACFlagType):140    if flag == ac.ACFlagType.AC_NO_FLAG:141        raise_color(HUE_CONNECTION['colors']['No_Flag'])142    if flag == ac.ACFlagType.AC_BLUE_FLAG:143        raise_color(HUE_CONNECTION['colors']['Blue_Flag'])144    if flag == ac.ACFlagType.AC_YELLOW_FLAG:145        raise_color(HUE_CONNECTION['colors']['Yellow_Flag'])146    if flag == ac.ACFlagType.AC_BLACK_FLAG:147        raise_color(HUE_CONNECTION['colors']['Black_Flag'])148    if flag == ac.ACFlagType.AC_WHITE_FLAG:149        raise_color(HUE_CONNECTION['colors']['White_Flag'])150    if flag == ac.ACFlagType.AC_CHECKERED_FLAG:151        raise_color(HUE_CONNECTION['colors']['Checkered_Flag'])152    if flag == ac.ACFlagType.AC_PENALTY_FLAG:153        raise_color(HUE_CONNECTION['colors']['Penalty_Flag'])154def raise_acc_flag(flag: acc.ACCFlagType):155    if flag == acc.ACCFlagType.ACC_NO_FLAG:156        raise_color(HUE_CONNECTION['colors']['No_Flag'])157    if flag == acc.ACCFlagType.ACC_BLUE_FLAG:158        raise_color(HUE_CONNECTION['colors']['Blue_Flag'])159    if flag == acc.ACCFlagType.ACC_YELLOW_FLAG:160        raise_color(HUE_CONNECTION['colors']['Yellow_Flag'])161    if flag == acc.ACCFlagType.ACC_BLACK_FLAG:162        raise_color(HUE_CONNECTION['colors']['Black_Flag'])163    if flag == acc.ACCFlagType.ACC_WHITE_FLAG:164        raise_color(HUE_CONNECTION['colors']['White_Flag'])165    if flag == acc.ACCFlagType.ACC_CHECKERED_FLAG:166        raise_color(HUE_CONNECTION['colors']['Checkered_Flag'])167    if flag == acc.ACCFlagType.ACC_PENALTY_FLAG:168        raise_color(HUE_CONNECTION['colors']['Penalty_Flag'])169    if flag == acc.ACCFlagType.ACC_GREEN_FLAG:170        raise_color(HUE_CONNECTION['colors']['Green_Flag'])171    if flag == acc.ACCFlagType.ACC_ORANGE_FLAG:172        raise_color(HUE_CONNECTION['colors']['Orange_Flag'])173def raise_iracing_flag(flag: iracing.IRacingGUIFlagType):174    if flag == iracing.IRacingGUIFlagType.IRACING_NO_FLAG:175        raise_color(HUE_CONNECTION['colors']['No_Flag'])176    if flag == iracing.IRacingGUIFlagType.IRACING_BLUE_FLAG:177        raise_color(HUE_CONNECTION['colors']['Blue_Flag'])178    if flag == iracing.IRacingGUIFlagType.IRACING_YELLOW_FLAG:179        raise_color(HUE_CONNECTION['colors']['Yellow_Flag'])180    if flag == iracing.IRacingGUIFlagType.IRACING_BLACK_FLAG:181        raise_color(HUE_CONNECTION['colors']['Black_Flag'])182    if flag == iracing.IRacingGUIFlagType.IRACING_WHITE_FLAG:183        raise_color(HUE_CONNECTION['colors']['White_Flag'])184    if flag == iracing.IRacingGUIFlagType.IRACING_CHEQUERED_FLAG:185        raise_color(HUE_CONNECTION['colors']['Checkered_Flag'])186    if flag == iracing.IRacingGUIFlagType.IRACING_RED_FLAG:187        raise_color(HUE_CONNECTION['colors']['Penalty_Flag'])188    if flag == iracing.IRacingGUIFlagType.IRACING_GREEN_FLAG:189        raise_color(HUE_CONNECTION['colors']['Green_Flag'])190    if flag == iracing.IRacingGUIFlagType.IRACING_MEATBALL_FLAG:191        raise_color(HUE_CONNECTION['colors']['Orange_Flag'])192def sync_ac_color():193    flag = ac.get_flag()194    raise_ac_flag(flag)195def sync_acc_color():196    flag = acc.get_flag()197    raise_acc_flag(flag)198def sync_iracing_color():199    flag = iracing.get_flag()200    raise_iracing_flag(flag)201def close_callback(route, websockets):202    if not websockets:203        stop_sync()204        exit()205def resource_path(rel_path):206    """ Get absolute path to resource, works for dev and for PyInstaller """207    try:208        base_path = sys._MEIPASS209    except Exception:210        base_path = os.path.abspath(".")211    return os.path.join(base_path, rel_path)212if __name__ == '__main__':213    eel.init('web')214    eel.browsers.set_path('electron', resource_path('node_modules\electron\dist\electron.exe'))...base_transfomer.py
Source:base_transfomer.py  
1# -*- coding: utf-8 -*-2import logging3from ...models_access import (4    OdooProductAccess, ProductSyncAccess, AmazonProductAccess5)6from ...model_names.shared_names import (7    MODEL_NAME_FIELD, RECORD_ID_FIELD,8)9from ...model_names.product_sync import (10    SYNC_TYPE_FIELD,11    SYNC_DELETE, SYNC_CREATE, SYNC_DEACTIVATE,12)13from ..amazon_names import AMAZON_ID_FIELD, AMAZON_SKU_FIELD14_logger = logging.getLogger(__name__)15class BaseTransformer(object):16    """17    This is the base transform18    """19    def __init__(self, env):20        self._odoo_product = OdooProductAccess(env)21        self._product_sync = ProductSyncAccess(env)22        self._amazon_product = AmazonProductAccess(env)23        self._product = None24    @staticmethod25    def _raise_exception(field_name):26        template = "Invalid {} value in Sync transformation"27        raise ValueError(template.format(field_name))28    @staticmethod29    def _check_string(sync_value, field_name, field_value):30        # add field to sync value, raise an exception if the value is invalid31        if field_value:32            field_value = field_value.strip()33            if field_value:34                sync_value[field_name] = field_value35                return36        # otherwise raise an exception for required field37        BaseTransformer._raise_exception(field_name)38    @staticmethod39    def _add_string(sync_value, field_name, field_value):40        # add valid field value to sync value41        if field_value:42            field_value = field_value.strip()43            if field_value:44                sync_value[field_name] = field_value45    @staticmethod46    def _remove_syncs(sync_ops, removing_ops):47        for sync_op in removing_ops:48            sync_ops = sync_ops - sync_op49        return sync_ops50    def _merge_others(self, sync_op, sync_ops):51        """52        This is stub that to be implement in a child class if53        it needs to do other work54        """55        pass56    # the default implementation, update transform should combine values57    def _check_redundant(self, sync_ops):58        _logger.debug("check and remove redundant syncs.")59        processed = set()60        redundant = []61        for sync_op in sync_ops:62            sync_key = (sync_op[MODEL_NAME_FIELD], sync_op[RECORD_ID_FIELD])63            if sync_key in processed:64                self._product_sync.set_sync_redundant(sync_op)65                redundant.append(sync_op)66            else:67                processed.add(sync_key)68                # a hook method that might be implemented in a subclass69                self._merge_others(sync_op, sync_ops)70        _logger.debug("Found {} redundant syncs.".format(len(redundant)))71        return BaseTransformer._remove_syncs(sync_ops, redundant)72    def _convert_sync(self, sync_op):73        """74        To be called and extended in subclass to convert more fields75        """76        sync_value = {AMAZON_ID_FIELD: sync_op.id}77        sku = OdooProductAccess.get_sku(self._product)78        BaseTransformer._check_string(sync_value, AMAZON_SKU_FIELD, sku)79        return sync_value80    def _check_stop(self, sync_op):81        stop_sync = False82        self._product = self._odoo_product.get_existed_product(sync_op)83        # for all but delete, we want to make sure the product exists84        # no need to check Amazon Product table because both85        # waiting syncs are checked before switch to new86        if sync_op[SYNC_TYPE_FIELD] != SYNC_DELETE:87            if self._product:88                if self._odoo_product.is_sync_active_product(89                        self._product):90                    # may be unnecessary but does not hurt91                    if sync_op[SYNC_TYPE_FIELD] == SYNC_DEACTIVATE:92                        stop_sync = True93                else:94                    if sync_op[SYNC_TYPE_FIELD] != SYNC_DEACTIVATE:95                        stop_sync = True96            else:97                stop_sync = True98        return stop_sync99    def _transform_sync(self, sync_op, invalid_ops, sync_values):100        if self._check_stop(sync_op):101            log_template = "Product not found or sync disabled " \102                           "for sync id {0}. Skip it."103            _logger.debug(log_template.format(sync_op.id))104            ProductSyncAccess.set_sync_no_product(sync_op)105            invalid_ops.append(sync_op)106        else:107            sync_value = self._convert_sync(sync_op)108            if sync_value:109                sync_values.append(sync_value)110            else:111                log_template = "Sync id {0} has empty value. Skip it."112                _logger.debug(log_template.format(sync_op.id))113                ProductSyncAccess.update_sync_new_empty_value(sync_op)114                invalid_ops.append(sync_op)115    def transform(self, sync_ops):116        # we change sync_ops record set because making a copy117        # creates a new record set that is saved in table.118        sync_ops = self._check_redundant(sync_ops)119        sync_values = []120        invalid_ops = []121        for sync_op in sync_ops:122            try:123                self._transform_sync(sync_op, invalid_ops, sync_values)124                # some pending write syncs or newly-switched new125                # write syncs are made redundant by delete and create126                if sync_op[SYNC_TYPE_FIELD] in [SYNC_CREATE, SYNC_DELETE]:127                    self._product_sync.find_set_redundant(sync_op)128            except Exception as ex:129                log_template = "Sync transform error for sync id {0}  " \130                               "Exception: {1}."131                _logger.debug(log_template.format(sync_op.id, ex.message))132                ProductSyncAccess.update_sync_new_exception(sync_op, ex)133                invalid_ops.append(sync_op)134        sync_ops = BaseTransformer._remove_syncs(sync_ops, invalid_ops)135        assert(len(sync_ops) == len(sync_values))...run.py
Source:run.py  
...44        except exceptions.AspenError as aspen_err:45            if aspen_err.fatal:46                logger.error('Stopping sync thread due to previous error: %s', aspen_err)47                SYNC_THREAD_STATE = SyncThreadState.STOPPED48                stop_sync()49            else:50                logger.warning('Caught a non-fatal AspenError: %s', aspen_err)51    SYNC_THREAD_STATE = SyncThreadState.STOPPED52@FLASK_APP.route('/api/v1/cache/<mgt_res_grp>/<cluster>/<app>', methods=['DELETE'])53def clear_app_from_cache(mgt_res_grp, cluster, app):54    client = redis.get_client()55    redis.clear_cache_for_cluster_and_app(client, mgt_res_grp, cluster, app)56    return jsonify(message='Cache cleared')57@FLASK_APP.route('/api/v1/cache/<mgt_res_grp>/<cluster>', methods=['DELETE'])58def clear_cluster_from_cache(mgt_res_grp, cluster):59    client = redis.get_client()60    redis.clear_cache_for_cluster(client, mgt_res_grp, cluster)61    return jsonify(message='Cache cleared')62@FLASK_APP.route('/api/v1/sync/start', methods=['GET'])63def start_sync():64    logger = logging.getLogger(__name__)65    logger.info('Starting sync thread')66    sync_thread = thread.get_sync_thread()67    if not sync_thread:68        thread.create_and_start_sync_thread(sync_routine)69        return jsonify(message='Sync thread created and started')70    elif sync_thread.stopped():71        sync_thread.stopped = False72        return jsonify(message='Sync thread started'), 20073    else:74        return jsonify(message='Sync thread already running'), 40475def stop_sync():76    logger = logging.getLogger(__name__)77    logger.info('Stopping sync thread')78    sync_thread = thread.get_sync_thread()79    if sync_thread and not sync_thread.stopped():80        sync_thread.stop()81        return True82    else:83        return False84@FLASK_APP.route('/api/v1/sync/stop', methods=['GET'])85def stop_sync_and_return():86    got_stopped = stop_sync()87    if got_stopped:88        return jsonify(message='Sync thread stopped'), 20089    else:90        return jsonify(message='Sync thread already stopped'), 40491@FLASK_APP.route('/api/v1/status/<mgt_res_grp>', methods=['GET'])92def get_status(mgt_res_grp):93    logger = logging.getLogger(__name__)94    logger.info('Returning status')95    redis_client = redis.get_client()96    cache_size = redis.get_management_key_count(redis_client, mgt_res_grp)97    sync_thread = thread.get_sync_thread()98    if not sync_thread:99        running = False100    else:...LambdaTest’s Playwright tutorial will give you a broader idea about the Playwright automation framework, its unique features, and use cases with examples to exceed your understanding of Playwright testing. This tutorial will give A to Z guidance, from installing the Playwright framework to some best practices and advanced concepts.
Get 100 minutes of automation test minutes FREE!!
