Python hypothesis.strategies 模块，text() 实例源码

def test_transaction_specification_for_avtalegiro_payment_request(
        tn, ln, cn, text):
    original = netsgiro.records.TransactionSpecification(
        service_code=netsgiro.ServiceCode.AVTALEGIRO,
        transaction_type=(
            netsgiro.TransactionType.AVTALEGIRO_WITH_BANK_NOTIFICATION),
        transaction_number=tn,
        line_number=ln,
        column_number=cn,
        text=text,
    )

    ocr = original.to_ocr()
    record = netsgiro.records.TransactionSpecification.from_string(ocr)

    assert record.transaction_number == tn
    assert record.line_number == ln
    assert record.column_number == cn
    assert len(record.text) == 40
    assert record.text == original.text

def test_can_return_text_even_with_binary_content_type_configured(
        create_event):
    demo = app.Chalice('demo-app')

    @demo.route('/index')
    def index_view():
        return app.Response(
            status_code=200,
            body='Plain text',
            headers={'Content-Type': 'text/plain'})

    event = create_event('/index', 'GET', {})
    event['headers']['Accept'] = 'application/octet-stream'
    response = demo(event, context=None)
    assert response['statusCode'] == 200
    assert response['body'] == 'Plain text'
    assert response['headers']['Content-Type'] == 'text/plain'

def spark_application(app_id):
    """Mock of the Spark jobs REST resource."""
    if 'last' in request.args:
        return jsonify(redis.get(request.base_url))

    d = st.fixed_dictionaries({
        'jobId': st.integers(0),
        'name': st.text(),
        'submissionTime': st.text(),
        'completionTime': st.text(),
        'stageIds': st.lists(st.integers(0), average_size=3),
        'status': st.sampled_from(['SUCCEEDED', 'RUNNING', 'FAILED']),
        'numTasks': st.integers(0),
        'numActiveTasks': st.integers(0),
        'numCompletedTasks': st.integers(0),
        'numSkippedTasks': st.integers(0),
        'numFailedTasks': st.integers(0),
        'numActiveStages': st.integers(0),
        'numCompletedStages': st.integers(0),
        'numSkippedStages': st.integers(0),
        'numFailedStages': st.integers(0),
    })
    result = json.dumps(st.lists(d, average_size=3).example())
    redis.set(request.base_url, result)
    return jsonify(result)

def services(
        draw,
        ids=uuids(),
        names=text(),
        descriptions=text(),
        registration_schemas=dictionaries(text(), text()),
        result_schemas=dictionaries(text(), text()),
        are_available=booleans(),
        service_job_lists=job_lists(),
        timeouts=timedeltas()
) -> ServiceInterface:
    return Service(
        draw(ids), draw(names), draw(descriptions),
        draw(registration_schemas), draw(result_schemas),
        draw(are_available), draw(service_job_lists),
        draw(timeouts)
    )

def dns_labels():
    """
    Strategy for generating limited charset DNS labels.
    """
    # This is too limited, but whatever
    return (
        s.text(
            u'abcdefghijklmnopqrstuvwxyz0123456789-',
            min_size=1, max_size=25)
        .filter(
            lambda s: not any([
                s.startswith(u'-'),
                s.endswith(u'-'),
                s.isdigit(),
                s[2:4] == u'--',
            ])))

def test_fsnative(text):
    assert isinstance(fsnative(text), fsnative)

def test_text2fsn(text):
    assert isinstance(text2fsn(text), fsnative)

def test_text_fsn_roudntrip(text):
    if u"\x00" in text:
        return
    assert isinstance(fsn2text(text2fsn(text)), text_type)

def digits(min_size=10, max_size=None):
    max_size = max_size or min_size
    return st.text(string.digits, min_size=min_size, max_size=max_size)

def test_transaction_amount_item_3_for_ocr_giro_transactions(tn, text):
    original = netsgiro.records.TransactionAmountItem3(
        service_code=netsgiro.ServiceCode.OCR_GIRO,
        transaction_type=(
            netsgiro.TransactionType.PURCHASE_WITH_TEXT),
        transaction_number=tn,
        text=text,
    )

    ocr = original.to_ocr()
    record = netsgiro.records.TransactionAmountItem3.from_string(ocr)

    assert record.transaction_number == tn
    assert record.text == original.text

def random_prefix(draw):
    #limited to unicode letters, see
    #https://en.wikipedia.org/wiki/Unicode_character_property#General_Category
    categories = ['Ll', 'Lt', 'Lm', 'Lo']
    characters = st.lists(st.characters(whitelist_categories=categories), min_size = 1)
    prefix = st.text(alphabet = draw(characters), min_size = 1)
    return draw(prefix)

def test_booleanvar_true(text):
    assert BooleanVar()._to_python(text) is True

def test_booleanvar_false(text):
    assert BooleanVar()._to_python(text) is False

def test_booleanvar_other(text):
    assume(text not in (TRUE_VALUES + FALSE_VALUES))
    with pytest.raises(ValueError):
        BooleanVar()._to_python(text)

def test_stringvar_return_the_same_string(text):
    assert StringVar()._to_python(text) == text

def enums_of_primitives(draw):
    """Generate enum classes with primitive values."""
    if is_py2:
        names = draw(st.sets(st.text(alphabet=string.ascii_letters,
                                     min_size=1),
                             min_size=1))
    else:
        names = draw(st.sets(st.text(min_size=1), min_size=1))
    n = len(names)
    vals = draw(st.one_of(st.sets(st.one_of(
            st.integers(),
            st.floats(allow_nan=False),
            st.text(min_size=1)),
        min_size=n, max_size=n)))
    return Enum('HypEnum', list(zip(names, vals)))

def str_attrs(draw, defaults=None):
    """
    Generate a tuple of an attribute and a strategy that yields strs for that
    attribute.
    """
    default = NOTHING
    if defaults is True or (defaults is None and draw(st.booleans())):
        default = draw(st.text())
    return ((attr.ib(default=default), st.text()))

def dict_attrs(draw, defaults=None):
    """
    Generate a tuple of an attribute and a strategy that yields dictionaries
    for that attribute. The dictionaries map strings to integers.
    """
    default = NOTHING
    val_strat = st.dictionaries(keys=st.text(), values=st.integers())
    if defaults is True or (defaults is None and draw(st.booleans())):
        default_val = draw(val_strat)
        default = attr.Factory(lambda: default_val)
    return ((attr.ib(default=default), val_strat))

def test_text_and_title(title, text):
    assert(str(Abbreviation(title, text)) == "<abbr title=\"" + title + "\">" + text + "</abbr>")

def test_initialism():
    assert(str(Abbreviation("Title", "text", True)) == "<abbr title=\"Title\" class=\"initialism\">text</abbr>")

def test_custom_class_ident_style_and_attrs():
    assert(str(Abbreviation("Title", "text", True, cl='abclass', ident='123', style="font-size:0.9em;", attrs={"data-test": 'abc'}))
           == "<abbr title=\"Title\" id=\"123\" class=\"initialism abclass\" style=\"font-size:0.9em;\" data-test=\"abc\">text</abbr>")

def test_hypothesis_text(s, each_text_maker):
    # have to get a fresh text maker EACH time function is invoked, due to Hypothesis quirk explained here:
    # http://hypothesis.works/articles/hypothesis-pytest-fixtures/
    text_maker = each_text_maker.clone()

    _input_tokenized = text_maker.input_text(s)

    sentences = text_maker.make_sentences(300)

    word_set_comparison = helpers.WordSetComparison(generated_tokens=sentences, input_tokenized=_input_tokenized)
    assert word_set_comparison.output_is_valid_strict()

def _device_list(minimum):
    """
    Get a device generating strategy.

    :param int minimum: the minimum number of devices, must be at least 0
    """
    return strategies.lists(
       strategies.text(
          alphabet=string.ascii_letters + "/",
          min_size=1
       ),
       min_size=minimum
    )

def _device_list(minimum):
    """
    Get a device generating strategy.

    :param int minimum: the minimum number of devices, must be at least 0
    """
    return strategies.lists(
       strategies.text(
          alphabet=string.ascii_letters + "/",
          min_size=1
       ),
       min_size=minimum
    )

def jsonify(text):
    """Substitute for flask.jsonify which accepts an already encoded
    JSON string and makes a response with 200 status and application/json
    Content-Type header.
    """
    return make_response(text, 200, {'Content-Type': 'application/json'})

def test_example(self) -> None:
        """
        Test a successful example of creating an Authorization header.
        """
        result = authorization_header(
            access_key=b'my_access_key',
            secret_key=b'my_secret_key',
            method=GET,
            content=b'{"something": "other"}',
            content_type='text/example',
            date='Sun, 22 Apr 2012 08:49:37 GMT.',
            request_path='/example_path',
        )

        assert result == b'VWS my_access_key:CetfV6Yl/3mSz/Xl0c+O1YjXKYg='

def jobs(
        draw,
        services=service_generator(),
        parameters=dictionaries(text(), text()),
) -> Job:
    return Job.new(
        draw(services),
        draw(parameters)
    )

def services(
        draw,
        names=text(),
        descriptions=text(),
        registration=dictionaries(text(), text()),
        results=dictionaries(text(), text())
) -> Service:
    return Service.new(
        name=draw(names),
        description=draw(descriptions),
        registration_schema=draw(registration),
        result_schema=draw(results)
    )

def services(
        draw,
        service_id=uuids(), name=text(), description=text(),
        is_available=booleans()
) -> IService:
    return Service(
        draw(service_id), draw(name), draw(description), draw(is_available)
    )

def _valid_post_requests(
        draw,
        names: str=text(),
        descriptions: str=text(),
        job_registration_schemas: dict=dictionaries(text(), text()),
        job_result_schemas: dict=dictionaries(text(), text())
) -> dict:
    return {
        'name': draw(names),
        'description': draw(descriptions),
        'job_registration_schema': draw(job_registration_schemas),
        'job_result_schema': draw(job_result_schemas)
    }

def api_exceptions(
        draw,
        status_codes=integers(min_value=400, max_value=599),
        titles=text(),
        details=text()
) -> APIExceptionInterface:
    return _APIException(
        draw(status_codes), draw(titles), draw(details)
    )

def jobs(
        draw,
        ids=uuids(),
        statuses=sampled_from(JobInterface.JobStatus),
        parameters=dictionaries(text(), text()),
        results=dictionaries(text(), text()),
        dates_submitted=datetimes(),
        registration_schemas=dictionaries(text(), text()),
        result_schemas=dictionaries(text(), text())
) -> JobInterface:
    """

    :param draw: A function that can take a strategy and draw a datum from it
    :param ids: A hypothesis strategy (statisticians should read "random
        variable"), that represents the set of all valid job IDs
    :param statuses: A hypothesis strategy that samples from the set of all
        allowed job statuses
    :param parameters: A hypothesis strategy that samples from all job
        parameters
    :param results: A hypothesis strategy that represents the possible results
    :param dates_submitted: A hypothesis strategy that represents the
        possible dates that can be submitted
    :param registration_schemas: The possible job registration schemas
    :param result_schemas: The possible job result schemas
    :return: A randomly-generated implementation of :class:`JobInterface`
    """
    return Job(
        draw(ids), draw(statuses), draw(parameters), draw(results),
        draw(dates_submitted),
        draw(registration_schemas),
        draw(result_schemas)
    )

def list_of_strings(draw):
        return [draw(text(min_size=1, average_size=70))
                for i in range(draw(integers(min_value=0, max_value=100)))]

def _device_list(minimum):
    """
    Get a device generating strategy.

    :param int minimum: the minimum number of devices, must be at least 0
    """
    return strategies.lists(
       strategies.text(
          alphabet=string.ascii_letters + "/",
          min_size=1
       ),
       min_size=minimum
    )

def list_of_strings(draw):
        return [draw(text(min_size=1, average_size=70))
                for i in range(draw(integers(min_value=0, max_value=100)))]

def valid_identifier(**kwargs):
    """Return a strategy which generates a valid Python Identifier"""
    if 'min_size' not in kwargs:
        kwargs['min_size'] = 4
    return hs.text(alphabet="abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789", **kwargs)\
        .filter(lambda x: x[0].isalpha() and x.isidentifier() and not (iskeyword(x)))

def _parse_text(source: Union[str, NodeNG]) -> Tuple[astroid.Module, TypeInferer]:
    """Parse source code text and output an AST with type inference performed."""
    # TODO: apparently no literal syntax for empty set in Python3, also cannot do set()
    # TODO: Deal with special case later.
    # if isinstance(source, astroid.Set) and len(list(source.elts)) == 0:
    #     source = f'{set({})}'
    if not isinstance(source, str):  # It's an astroid node
        source = source.as_string()
    module = astroid.parse(source)
    type_inferer = TypeInferer()
    type_inferer.environment_transformer().visit(module)
    type_inferer.type_inference_transformer().visit(module)
    return module, type_inferer

def urls():
    """
    Strategy for generating ``twisted.python.url.URL``\s.
    """
    return s.builds(
        URL,
        scheme=s.just(u'https'),
        host=dns_names(),
        path=s.lists(s.text(
            max_size=64,
            alphabet=s.characters(blacklist_characters=u'/?#',
                                  blacklist_categories=('Cs',))
        ), min_size=1, max_size=10))

def test_start_responding(self, token):
        """
        Calling ``start_responding`` makes an appropriate resource available.
        """
        challenge = challenges.HTTP01(token=token)
        response = challenge.response(RSA_KEY_512)

        responder = HTTP01Responder()

        challenge_resource = Resource()
        challenge_resource.putChild(b'acme-challenge', responder.resource)
        root = Resource()
        root.putChild(b'.well-known', challenge_resource)
        client = StubTreq(root)

        encoded_token = challenge.encode('token')
        challenge_url = URL(host=u'example.com', path=[
            u'.well-known', u'acme-challenge', encoded_token]).asText()

        self.assertThat(client.get(challenge_url),
                        succeeded(MatchesStructure(code=Equals(404))))

        responder.start_responding(u'example.com', challenge, response)
        self.assertThat(client.get(challenge_url), succeeded(MatchesAll(
            MatchesStructure(
                code=Equals(200),
                headers=AfterPreprocessing(
                    methodcaller('getRawHeaders', b'content-type'),
                    Equals([b'text/plain']))),
            AfterPreprocessing(methodcaller('content'), succeeded(
                Equals(response.key_authorization.encode('utf-8'))))
        )))

        # Starting twice before stopping doesn't break things
        responder.start_responding(u'example.com', challenge, response)
        self.assertThat(client.get(challenge_url),
                        succeeded(MatchesStructure(code=Equals(200))))

        responder.stop_responding(u'example.com', challenge, response)
        self.assertThat(client.get(challenge_url),
                        succeeded(MatchesStructure(code=Equals(404))))

def test_headers_to_scrapy():
    assert headers_to_scrapy(None) == Headers()
    assert headers_to_scrapy({}) == Headers()
    assert headers_to_scrapy([]) == Headers()

    html_headers = Headers({'Content-Type': 'text/html'})

    assert headers_to_scrapy({'Content-Type': 'text/html'}) == html_headers
    assert headers_to_scrapy([('Content-Type', 'text/html')]) == html_headers
    assert headers_to_scrapy([{'name': 'Content-Type', 'value': 'text/html'}]) == html_headers

def test_headers_to_scrapy():
    assert headers_to_scrapy(None) == Headers()
    assert headers_to_scrapy({}) == Headers()
    assert headers_to_scrapy([]) == Headers()

    html_headers = Headers({'Content-Type': 'text/html'})

    assert headers_to_scrapy({'Content-Type': 'text/html'}) == html_headers
    assert headers_to_scrapy([('Content-Type', 'text/html')]) == html_headers
    assert headers_to_scrapy([{'name': 'Content-Type', 'value': 'text/html'}]) == html_headers

def test_headers_to_scrapy():
    assert headers_to_scrapy(None) == Headers()
    assert headers_to_scrapy({}) == Headers()
    assert headers_to_scrapy([]) == Headers()

    html_headers = Headers({'Content-Type': 'text/html'})

    assert headers_to_scrapy({'Content-Type': 'text/html'}) == html_headers
    assert headers_to_scrapy([('Content-Type', 'text/html')]) == html_headers
    assert headers_to_scrapy([{'name': 'Content-Type', 'value': 'text/html'}]) == html_headers

def test_identity_should_return_first_argument(text, integer):
    assert identity(text) is text
    assert identity(integer) is integer

def test_eq(text):
    assert eq(text, text)
    assert eq(text)(text)

def test_fold(integer, text, boolean):

    dictionary = {'key': 'value'}

    assert First(text).fold(identity) is text
    assert All(boolean).fold(identity) is boolean
    assert Sum(integers).fold(identity) is integers
    assert Map(dictionary).fold(identity) is dictionary

def test_write(self, text):
        fake_stderr = FakeOutput()
        fake_stdout = FakeOutput()
        output = Console(output=fake_stdout, error=fake_stderr)

        output.write(*text)
        assert fake_stdout.last_received == " ".join(text) + "\n"

    # Asserts that write_debug only writes when output.debug is True.

def test_write_if_debug(self, info, debug):
        fake_stderr = FakeOutput()
        fake_stdout = FakeOutput()
        output = Console(output=fake_stdout, error=fake_stderr)

        output.write(info)
        output.write_debug(debug)
        assert fake_stdout.last_received == info + "\n"

        output.debug = True
        output.write(info)
        output.write_debug(debug)
        assert fake_stdout.last_received == "DEBUG: %s\n" % debug

    # Asserts that the text is written to stderr with "ERROR: ".

def test_write_stderr(self, text):
        fake_stderr = FakeOutput()
        fake_stdout = FakeOutput()
        output = Console(output=fake_stdout, error=fake_stderr)

        output.write_error(*text)
        assert fake_stderr.last_received == "ERROR: %s\n" % " ".join(text)

    # Asserts that an end is properly appended to all written text.

def test_supply_end(self, text, end):
        fake_stderr = FakeOutput()
        fake_stdout = FakeOutput()
        output = Console(output=fake_stdout, error=fake_stderr)

        output.debug = True
        output.write(text, end=end)
        assert fake_stdout.last_received == text + end

        output.write_debug(text, end=end)
        assert fake_stdout.last_received == "DEBUG: " + text + end

        output.write_error(text, end=end)
        assert fake_stderr.last_received == "ERROR: " + text + end

def test_to_long_fail(self, fail_val):
        """
        to_long() with incompatible params:
        :param fail_val: random data of known incompatible types (floats, text)
        """
        self.force_fail(fail_val, to_long, TypeError)