python类String()的实例源码

def gen_operators_of_field(f_name, mongo_field, r_graphene, operators_list):
    """ Return a dict with keys as the name of the field with operator and value is the required type, for instance: 
    @param f_name: string name of the field
    @param mongo_field: object instance of mongoengine field, e.g: mongoengine.StringField()
    @param r_graphene: object instance of graphene field, e.g: graphene.String(): 
    {
        name: graphene.String()
        name__nin: graphene.List(graphene.String) ...
    }
     """

    field_with_operators = {
        f_name: field_to_id(mongo_field, r_graphene)
    }

    for op_name in operators_list:
        field_with_operators[f_name + '__' + op_name] = operators[op_name](mongo_field, r_graphene)

    if isinstance(mongo_field, fields_string_operators):
        for op in string_operators:
            field_with_operators[f_name + '__' + op] = graphene.String()

    return field_with_operators

def get_field_args(field):
    types = {
        str: graphene.String,
        bool: graphene.Boolean,
        int: graphene.Int,
        float: graphene.Float,
    }

    try:
        args = inspect.signature(field)
        result = {}
        for name, arg in args.parameters.items():
            if arg.default is not None:
                custom_type = types[type(arg.default)]
            else:
                custom_type = graphene.String

            result.update({name: custom_type()})
        return result
    except (KeyError, ValueError):
        return {}

def schema():
    class Query(graphene.ObjectType):
        test_string = graphene.String()

        def resolve_test_string(self, args, context, info):
            return 'works'

    # TODO: Implement case conversion for arg names
    class Subscription(graphene.ObjectType):
        test_subscription = graphene.String()
        test_context = graphene.String()
        test_filter = graphene.String(filterBoolean=graphene.Boolean())
        test_filter_multi = graphene.String(
            filterBoolean=graphene.Boolean(),
            a=graphene.String(),
            b=graphene.Int())
        test_channel_options = graphene.String()

        def resolve_test_subscription(self, args, context, info):
            return self

        def resolve_test_context(self, args, context, info):
            return context

        def resolve_test_filter(self, args, context, info):
            return 'good_filter' if args.get('filterBoolean') else 'bad_filter'

        def resolve_test_filter_multi(self, args, context, info):
            return 'good_filter' if args.get('filterBoolean') else 'bad_filter'

        def resolve_test_channel_options(self, args, context, info):
            return self

    return graphene.Schema(query=Query, subscription=Subscription)

def test_can_subscribe_to_more_than_one_trigger(sub_mgr):
    non_local = {'trigger_count': 0}

    query = 'subscription multiTrigger($filterBoolean: Boolean,\
            $uga: String){testFilterMulti(filterBoolean: $filterBoolean,\
            a: $uga, b: 66)}'

    def callback(err, payload):
        if err:
            sys.exit(err)
        else:
            try:
                if payload is None:
                    assert True
                else:
                    assert payload.data.get('testFilterMulti') == 'good_filter'
                    non_local['trigger_count'] += 1
            except AssertionError as e:
                sys.exit(e)
        if non_local['trigger_count'] == 2:
            sub_mgr.pubsub.greenlet.kill()

    def publish_and_unsubscribe_handler(sub_id):
        sub_mgr.publish('not_a_trigger', {'filterBoolean': False})
        sub_mgr.publish('trigger_1', {'filterBoolean': True})
        sub_mgr.publish('trigger_2', {'filterBoolean': True})
        sub_mgr.pubsub.greenlet.join()
        sub_mgr.unsubscribe(sub_id)

    p1 = sub_mgr.subscribe(query, 'multiTrigger', callback,
                           {'filterBoolean': True,
                            'uga': 'UGA'}, {}, None, None)
    p2 = p1.then(publish_and_unsubscribe_handler)
    p2.get()

def validation_schema():
    class Query(graphene.ObjectType):
        placeholder = graphene.String()

    class Subscription(graphene.ObjectType):
        test_1 = graphene.String()
        test_2 = graphene.String()

    return graphene.Schema(query=Query, subscription=Subscription)

def schema(data):
    class UserType(graphene.ObjectType):
        id = graphene.String()
        name = graphene.String()

    class Query(graphene.ObjectType):
        test_string = graphene.String()

    class Subscription(graphene.ObjectType):
        user = graphene.Field(UserType, id=graphene.String())
        user_filtered = graphene.Field(UserType, id=graphene.String())
        context = graphene.String()
        error = graphene.String()

        def resolve_user(self, args, context, info):
            id = args['id']
            name = data[args['id']]['name']
            return UserType(id=id, name=name)

        def resolve_user_filtered(self, args, context, info):
            id = args['id']
            name = data[args['id']]['name']
            return UserType(id=id, name=name)

        def resolve_context(self, args, context, info):
            return context

        def resolve_error(self, args, context, info):
            raise Exception('E1')

    return graphene.Schema(query=Query, subscription=Subscription)

def test_should_call_unsubscribe_when_client_closes_cxn(server_with_mocks):
    node_script = '''
        module.paths.push('{0}')
        WebSocket = require('ws')
        const SubscriptionClient =
        require('subscriptions-transport-ws').SubscriptionClient
        const client = new SubscriptionClient('ws://localhost:{1}/socket')
        client.subscribe({{
            query: `subscription useInfo($id: String) {{
            user(id: $id) {{
              id
              name
            }}
          }}`,
            operationName: 'useInfo',
            variables: {{
              id: 3,
            }},
          }}, function (error, result) {{
            // nothing
          }}
        )
        setTimeout(() => {{
            client.client.close()
        }}, 500)
    '''.format(
        os.path.join(os.path.dirname(__file__), 'node_modules'), TEST_PORT)
    try:
        subprocess.check_output(
            ['node', '-e', node_script], stderr=subprocess.STDOUT, timeout=1)
    except:
        while True:
            mock = server_with_mocks.get_nowait()
            if mock.name == 'on_unsubscribe':
                mock.assert_called_once()
                break

def test_should_trigger_on_unsubscribe_when_client_unsubscribes(
        server_with_mocks):
    node_script = '''
        module.paths.push('{0}')
        WebSocket = require('ws')
        const SubscriptionClient =
        require('subscriptions-transport-ws').SubscriptionClient
        const client = new SubscriptionClient('ws://localhost:{1}/socket')
        const subId = client.subscribe({{
            query: `subscription useInfo($id: String) {{
            user(id: $id) {{
              id
              name
            }}
          }}`,
            operationName: 'useInfo',
            variables: {{
              id: 3,
            }},
          }}, function (error, result) {{
            // nothing
          }})
        client.unsubscribe(subId)
    '''.format(
        os.path.join(os.path.dirname(__file__), 'node_modules'), TEST_PORT)
    try:
        subprocess.check_output(
            ['node', '-e', node_script], stderr=subprocess.STDOUT, timeout=.2)
    except:
        while True:
            mock = server_with_mocks.get_nowait()
            if mock.name == 'on_unsubscribe':
                mock.assert_called_once()
                break

def convert_column_to_string(type, attribute, registry=None):
    if attribute.is_hash_key:
        return ID(description=attribute.attr_name, required=not attribute.null)

    return String(description=getattr(attribute, 'attr_name'),
                  required=not (getattr(attribute, 'null', True)))

def convert_date_to_string(type, attribute, registry=None):
    return String(description=getattr(attribute, 'attr_name'),
                  required=not (getattr(attribute, 'null', True)))

def convert_scalar_list_to_list(type, attribute, registry=None):
    return List(String, description=attribute.attr_name)

def convert_list_to_list(type, attribute, registry=None):
    return List(String, description=attribute.attr_name)

def test_should_datetime_convert_string():
    assert_attribute_conversion(UTCDateTimeAttribute(), graphene.String)

def test_should_binary_convert_string():
    assert_attribute_conversion(BinaryAttribute(), graphene.String)

def convert_form_field_to_string(field):
    return String(description=field.help_text, required=field.required)

def test_should_char_convert_string():
    assert_conversion(models.CharField, graphene.String)

def test_should_text_convert_string():
    assert_conversion(models.TextField, graphene.String)

def test_should_slug_convert_string():
    assert_conversion(models.SlugField, graphene.String)

def test_should_url_convert_string():
    assert_conversion(models.URLField, graphene.String)

def test_should_ipaddress_convert_string():
    assert_conversion(models.GenericIPAddressField, graphene.String)

def test_should_file_convert_string():
    assert_conversion(models.FileField, graphene.String)

def test_should_image_convert_string():
    assert_conversion(models.ImageField, graphene.String)

def test_should_time_convert_string():
    assert_conversion(forms.TimeField, graphene.String)

def test_should_date_time_convert_string():
    assert_conversion(forms.DateTimeField, graphene.String)

def test_should_char_convert_string():
    assert_conversion(forms.CharField, graphene.String)

def test_should_email_convert_string():
    assert_conversion(forms.EmailField, graphene.String)

def test_should_slug_convert_string():
    assert_conversion(forms.SlugField, graphene.String)

def test_should_choice_convert_string():
    assert_conversion(forms.ChoiceField, graphene.String)

def test_should_base_field_convert_string():
    assert_conversion(forms.Field, graphene.String)

def test_should_regex_convert_string():
    assert_conversion(forms.RegexField, graphene.String, '[0-9]+')