Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _is_ref method in localstack

Best Python code snippet using localstack_python

json_schema.py

Source:json_schema.py

...81    if not re.match(r'^#/.*', ref):82        raise JSONSchemaError('Invalid format for `$ref`: "{}"'.format(ref))83    return _get_ref(schema,84                    re.split('/', re.sub(r'^#/', '', ref)))85def _is_ref(schema):86    """87    Given a JSON Schema compatible dict, returns True when the schema implements `$ref`88    NOTE: `$ref` OVERRIDES all other keys present in a schema89    :param schema:90    :return: Boolean91    """92    return '$ref' in schema93def _is_allof(schema):94    """95    Given a JSON Schema compatible dict, returns True when the schema implements `allOf`.96    :param schema:97    :return: Boolean98    """99    return not _is_ref(schema) and 'allOf' in schema100def is_anyof(schema):101    """102    Given a JSON Schema compatible dict, returns True when the schema implements `anyOf`.103    :param schema:104    :return: Boolean105    """106    return not _is_ref(schema) and not _is_allof(schema) and 'anyOf' in schema107def is_object(schema):108    """109    Given a JSON Schema compatible dict, returns True when schema's type allows being an Object.110    :param schema: dict, JSON Schema111    :return: Boolean112    """113    return not _is_ref(schema) and not is_anyof(schema) and not _is_allof(schema) \114           and (OBJECT in get_type(schema)115                or 'properties' in schema116                or not schema)117def is_iterable(schema):118    """119    Given a JSON Schema compatible dict, returns True when schema's type allows being iterable (ie, 'array')120    :param schema: dict, JSON Schema121    :return: Boolean122    """123    return not _is_ref(schema) \124           and ARRAY in get_type(schema) \125           and 'items' in schema126def is_nullable(schema):127    """128    Given a JSON Schema compatible dict, returns True when schema's type allows being 'null'129    :param schema: dict, JSON Schema130    :return: Boolean131    """132    return NULL in get_type(schema)133def is_literal(schema):134    """135    Given a JSON Schema compatible dict, returns True when schema's type allows being a literal136    (ie, 'integer', 'number', etc.)137    :param schema: dict, JSON Schema138    :return: Boolean139    """140    return not {STRING, INTEGER, NUMBER, BOOLEAN}.isdisjoint(set(get_type(schema)))141def is_datetime(schema):142    """143    Given a JSON Schema compatible dict, returns True when schema's type allows being a date-time144    :param schema: dict, JSON Schema145    :return: Boolean146    """147    return STRING in get_type(schema) and schema.get('format') == DATE_TIME_FORMAT148def make_nullable(schema):149    """150    Given a JSON Schema dict, returns the dict but makes the `type` `null`able.151    `is_nullable` will return true on the output.152    :return: dict, JSON Schema153    """154    t = get_type(schema)155    if NULL in t:156        return schema157    ret_schema = deepcopy(schema)158    ret_schema['type'] = t + [NULL]159    return ret_schema160class Cachable(dict):161    '''162    The simplified json_schemas we produce are idempotent. ie, if you simplify a simplified163    json_schema, it will return the same thing. We wrap the `dict` object with a few164    helpers which extend it so that we avoid recursion in some instances.165    '''166    def __init__(self, raw_dict, simplified=True):167        self._c = None168        super(Cachable, self).__init__(self, **raw_dict)169    def __hash__(self):170        return self._comparator().__hash__()171    def deepcopy(self):172        s = deepcopy(self)173        s._c = self._c174        return s175    def _comparator(self):176        if not self._c:177            self._c = json.dumps(self, sort_keys=True)178        return self._c179    def __lt__(self, other):180        return self._comparator() < other._comparator()181def _allof_sort_key(schema):182    '''183    We prefer scalars over combinations.184    With scalars we prefer date-times over strings.185    With combinations, we prefer objects.186    With all, we prefer nullables.187    '''188    if is_nullable(schema):189        sort_value = 0190    else:191        sort_value = 1192    if is_datetime(schema):193        sort_value += 0194    elif is_literal(schema):195        sort_value += 10196    elif is_object(schema):197        sort_value += 100198    elif is_iterable(schema):199        sort_value += 200200    else:201        # Unknown schema...maybe a $ref?202        sort_value += 1000203    return sort_value204def _simplify__allof__merge__objects(schemas):205    ret_schema = schemas[0]206    # Merge objects together preferring later allOfs over earlier207    next_schemas = schemas[1:]208    while next_schemas and is_object(next_schemas[0]):209        ret_schema['properties'] = {210            **ret_schema.get('properties', {}),211            **next_schemas[0].get('properties', {})}212        next_schemas = next_schemas[1:]213    return ret_schema214def _simplify__allof__merge__iterables(root_schema, schemas):215    ret_schema = schemas[0]216    # Recurse on all of the item schemas to create a single item schema217    item_schemas = []218    next_schemas = schemas219    while next_schemas and is_iterable(next_schemas[0]):220        item_schemas.append(next_schemas[0]['items'])221        next_schemas = next_schemas[1:]222    ret_schema['items'] = _helper_simplify(root_schema, {'allOf': item_schemas})223    return ret_schema224def _simplify__allof(root_schema, child_schema):225    simplified_schemas = [226        _helper_simplify(root_schema, schema)227        for schema in child_schema['allOf']]228    schemas = sorted(simplified_schemas, key=_allof_sort_key)229    ret_schema = schemas[0]230    if is_object(ret_schema):231        return _simplify__allof__merge__objects(schemas)232    if is_iterable(ret_schema):233        return _simplify__allof__merge__iterables(root_schema, schemas)234    return ret_schema235def _simplify__implicit_anyof(root_schema, schema):236    '''237    Typically literals are simple and have at most two types, one of which being NULL.238    However, they _can_ have many types wrapped up inside them as an implicit `anyOf`.239    Since we support `anyOf`, it is simpler to unwrap and "flatten" this implicit240    combination type.241    '''242    schemas = []243    types = set(get_type(schema))244    if types == {NULL}:245        return Cachable({'type': [NULL]})246    types.discard(NULL)247    if is_datetime(schema):248        schemas.append(Cachable({249            'type': [STRING],250            'format': DATE_TIME_FORMAT251        }))252        types.remove(STRING)253    if is_object(schema):254        properties = {}255        for field, field_json_schema in schema.get('properties', {}).items():256            properties[field] = _helper_simplify(root_schema, field_json_schema)257        schemas.append({258            'type': [OBJECT],259            'properties': properties260        })261        types.discard(OBJECT)262    if is_iterable(schema):263        schemas.append({264            'type': [ARRAY],265            'items': _helper_simplify(root_schema, schema.get('items', {}))266        })267        types.remove(ARRAY)268    schemas += [{'type': [t]} for t in types]269    if is_nullable(schema):270        schemas = [make_nullable(s) for s in schemas]271    return _helper_simplify(root_schema, {'anyOf': [Cachable(s) for s in schemas]})272def _simplify__anyof(root_schema, schema):273    '''274    `anyOf` clauses are merged/simplified according to the following rules (these _are_ recursive):275    - all literals are dedupped276    - all objects are merged into the same object schema, with sub-schemas being grouped as simplified `anyOf` schemas277    - all iterables' `items` schemas are merged as simplified `anyOf` schemas278    - all `anyOf`s are flattened to the topmost279    - if there is only a single element in an `anyOf`, that is denested280    - if any `anyOf`s are nullable, all are nullable281    '''282    schemas = [283            _helper_simplify(root_schema, schema)284            for schema in schema['anyOf']]285    literals = set()286    any_nullable = False287    any_merged_objects = False288    merged_object_properties = {}289    any_merged_iters = False290    merged_item_schemas = []291    while schemas:292        sub_schema = schemas.pop()293        any_nullable = any_nullable or is_nullable(sub_schema)294        if is_literal(sub_schema):295            literals.add(sub_schema)296        elif is_anyof(sub_schema):297            # Flatten potentially deeply nested `anyOf`s298            schemas += sub_schema['anyOf']299        elif is_object(sub_schema):300            any_merged_objects = True301            for k, s in sub_schema.get('properties', {}).items():302                if k in merged_object_properties:303                    merged_object_properties[k].append(s)304                else:305                    merged_object_properties[k] = [s]306        elif is_iterable(sub_schema):307            any_merged_iters = True308            merged_item_schemas.append(sub_schema['items'])309    merged_schemas = set()310    for l in literals:311        s = l312        if any_nullable:313            s = make_nullable(l)314        merged_schemas.add(Cachable(s))315    if any_merged_objects:316        for k, v in merged_object_properties.items():317            merged_object_properties[k] = _helper_simplify(root_schema, {'anyOf': v})318        s = {319            'type': [OBJECT],320            'properties': merged_object_properties321        }322        if any_nullable:323            s = make_nullable(s)324        merged_schemas.add(Cachable(s))325    if any_merged_iters:326        merged_item_schemas = _helper_simplify(root_schema, {'anyOf': merged_item_schemas})327        s = {328            'type': [ARRAY],329            'items': merged_item_schemas330        }331        if any_nullable:332            s = make_nullable(s)333        merged_schemas.add(Cachable(s))334    if len(merged_schemas) == 1:335        return merged_schemas.pop()336    return Cachable({'anyOf': sorted(merged_schemas)})337def _helper_simplify(root_schema, child_schema):338    # We check this value to make simplify a noop for schemas which have _already_ been simplified339    if isinstance(child_schema, Cachable):340        return child_schema341    ## Refs override all other type definitions342    if _is_ref(child_schema):343        try:344            ret_schema = _helper_simplify(root_schema, get_ref(root_schema, child_schema['$ref']))345        except RecursionError:346            raise JSONSchemaError('`$ref` path "{}" is recursive'.format(get_ref(root_schema, child_schema['$ref'])))347    elif _is_allof(child_schema):348        ret_schema = _simplify__allof(root_schema, child_schema)349    elif is_anyof(child_schema):350        ret_schema = _simplify__anyof(root_schema, child_schema)351    else:352        ret_schema = _simplify__implicit_anyof(root_schema, child_schema)353    if 'default' in child_schema:354        ret_schema['default'] = child_schema.get('default')355    return Cachable(ret_schema)356def simplify(schema):...

upload.py

Source:upload.py

1import sys2import argparse3import subprocess4import tempfile5import os6import json7from .common.utils import *8from .common.globals import *9def parse_args():10    parser = argparse.ArgumentParser(description='upload a package and ALL its dependencies to a remote server')11    parser.add_argument('reference',12                        metavar='reference',13                        type=str,14                        help='conanfile dir or reference')15    parser.add_argument('--profile', '-pr',16                        type=str,17                        metavar='profile',18                        help="Apply the specified profile to the install command")19    parser.add_argument('--remote', '-r',20                        type=str,21                        metavar='remote',22                        help="conan remote for upload",23                        required=True)24    parser.add_argument('--options', '-o',25                        type=str,26                        metavar='PROFILE',27                        nargs="+",28                        help="cDefine options values, e.g., -o Pkg:with_qt=True")29    parser.add_argument('--env', '-e',30                        type=str,31                        metavar='e',32                        nargs="+",33                        help="co for upload")34    parser.add_argument('--settings', '-s',35                        type=str,36                        metavar='SETTINGS',37                        nargs="+",38                        help="Settings to build the package, overwriting the defaults. e.g., -s compiler=gcc")39    parser.add_argument('--check',40                        action='store_true',41                        help="perform an integrity check, using the manifests,before upload ")42    parser.add_argument('--all',43                        action='store_true',44                        help="Upload both package recipe and packages")45    parser.add_argument('--skip-upload',46                        action='store_true',47                        help="Upload both package recipe and packages")48    parser.add_argument('--confirm', '-c',49                        action='store_true',50                        help="Upload all matching recipes without confirmation")51    parser.add_argument('--dry-run',52                        action='store_true',53                        help="Do not upload, only show what would be done")54    parser.add_argument('--parallel',55                        action='store_true',56                        help="Upload files in parallel using multiple "57                             "threads the default number of launched threads is 8")58    parser.add_argument('--retry',59                        type=int,60                        metavar="RETRY",61                        help="In case of fail retries to upload again the specified times.")62    parser.add_argument('--retry-wait',63                        type=int,64                        metavar="SEC",65                        help="Waits specified seconds before retry again")66    parser.add_argument('-no', '--no-overwrite',67                        type=str,68                        metavar="{all, recipe}",69                        help="Uploads package only if recipe is the same as the remote one")70    return parser.parse_args()71def get_packages(json_data):72    output = []73    error = False74    for p in json_data:75        _ref = None76        _id = None77        _binary = None78        _recipe = None79        _is_ref = False80        try:81            _is_ref = p['is_ref']82        except KeyError:83            pass84        try:85            _ref = p['reference']86        except KeyError:87            pass88        try:89            _id = p['id']90        except KeyError:91            pass92        try:93            _binary = p['binary']94        except KeyError:95            pass96        try:97            _recipe = p['recipe']98        except KeyError:99            pass100        if not _is_ref:101            continue102        if _ref == None or _ref == "" or _id == None or _id == "":103            print("ERROR: parsing error of json content")104            print(_ref, _id, _binary, _recipe)105            error=True106        107        if ( _binary != "Cache" or _recipe != "Cache" ):108            print("ERROR: can not upload package %s:%s : package does not exist" % (_ref, _id))109            error=True110        output.append((_ref,_id,_binary, _recipe ))111        112    if error:113        print("exiting due to previous errors.")114        exit(1)115    116    return output117def upload_packages(packages, args):118    upload_cmd = [conan_exe, 'upload']119    if args.all:120        upload_cmd += ['--all']121    if args.skip_upload:122        upload_cmd += ['--skip-upload']123    if args.confirm:124        upload_cmd += ['--confirm']125    if args.parallel:126        upload_cmd += ['--parallel']127    128    if args.retry:129        upload_cmd += ['--retry', args.retry]130    if args.retry_wait:131        upload_cmd += ['--retry-wait', args.retry_wait]132    if args.no_overwrite:133        upload_cmd += ['--no-overwrite', args.no_overwrite]134    135    if args.remote:136        upload_cmd += ['--remote', args.remote]137    138    for i in packages:139        cmd = upload_cmd + [ i[0] + ":" + i[1] ]140        print("uploading package %s:%s binary %s, recipe: %s" % i)141        if not args.dry_run:142            subprocess.check_call(cmd)143if __name__ == "__main__":144    check_python3()145    args = parse_args()146    json_file = tempfile.mkstemp()[1]147    info_command = [conan_exe, 'info', args.reference, '-j', json_file]148    if args.profile:149        info_command += ['-pr', args.profile]150    if args.options:151        for i in args.options:152            info_command += ['-o', i]153    if args.settings:154        for i in args.settings:155            info_command += ['-s', i]156    if args.env:157        for i in args.env:158            info_command += ['-e', i]159    subprocess.check_call(info_command)160    with open(json_file,'rb') as f:161        jsonData = json.load(f)162    upload_packages(get_packages(jsonData), args)163    os.remove(json_file)...

OP_ref_to_image_plane.py

Source:OP_ref_to_image_plane.py

...28        return True29        # return context.object and context.object.type == 'EMPTY'\30        #     and context.object.empty_display_type == 'IMAGE' and context.object.data31    @staticmethod32    def _is_ref(o):33        return o and o.type == 'EMPTY' and o.empty_display_type == 'IMAGE' and o.data34    def invoke(self, context, event):35        self.shader = fn.get_prefs().shader36        self.use_driver = fn.get_prefs().use_driver37        return context.window_manager.invoke_props_dialog(self) # , width=45038    ## optional draw (just to add open addon pref button)39    def draw(self, context):40        layout = self.layout41        layout.use_property_split = True42        col = layout.column(align=False)43        col.prop(self, 'shader')44        col.prop(self, 'name_from')45        col.prop(self, 'del_ref')46        row = col.row(align=True)47        row.label(text='') # 'Change Default Settings:'48        row.operator("rtp.open_addon_prefs", text="", icon='PREFERENCES')49    def execute(self, context):50        pool = [o for o in context.selected_objects]51        if context.object and context.object not in pool:52            pool.append(context.object)53        converted = 054        for o in pool:55            if not self._is_ref(o):56                continue57            fn.convert_empty_image_to_mesh(58                context, 59                o, 60                name_from=self.name_from, 61                delete_ref=self.del_ref, 62                shader=self.shader,)63            64            converted += 165        if not converted:66            self.report({'ERROR'}, 'Nothing converted')67            return {"CANCELLED"}68        self.report({'INFO'}, f'{converted} converted to mesh plane')69        return {"FINISHED"}...

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.