Python werkzeug.test 模块，Client() 实例源码

def test_handle():
    s = HostHttpServer(host)
    c = Client(s, Response)

    s.start()

    r = c.get('/')
    assert r.status == '403 FORBIDDEN'

    r = c.get('/?ticket=%s' % s.ticket_create())
    assert r.status == '200 OK'

    r = c.get('/foo/bar/?ticket=%s' % s.ticket_create())
    assert r.status == '500 INTERNAL SERVER ERROR'

    s.stop()

def __init__(self, url, service_impl_cls):
        self.url = url_endswith_slash(url)
        self.scheme, self.host, self.path, _, _ = urllib.parse.urlsplit(
            self.url
        )

        def null_app(environ, start_response):
            start_response('404 Not Found', [('Content-Type', 'text/plain')])
            return ['Not Found']

        wsgi_app = WsgiApp(service_impl_cls())
        if self.path != '/':
            self.wsgi_app = DispatcherMiddleware(
                null_app, {self.path[:-1]: wsgi_app}
            )
        else:
            self.wsgi_app = wsgi_app
        self.wsgi_test_client = Client(self.wsgi_app, Response)

def client(wsgi_app):
    app = StandardEndpointMiddleware(wsgi_app)
    return Client(app, BaseResponse)

def test_base_request(self):
        client = Client(request_demo_app, RequestTestResponse)

        # get requests
        response = client.get('/?foo=bar&foo=hehe')
        self.assert_strict_equal(response['args'], MultiDict([('foo', u'bar'), ('foo', u'hehe')]))
        self.assert_strict_equal(response['args_as_list'], [('foo', [u'bar', u'hehe'])])
        self.assert_strict_equal(response['form'], MultiDict())
        self.assert_strict_equal(response['form_as_list'], [])
        self.assert_strict_equal(response['data'], b'')
        self.assert_environ(response['environ'], 'GET')

        # post requests with form data
        response = client.post('/?blub=blah', data='foo=blub+hehe&blah=42',
                               content_type='application/x-www-form-urlencoded')
        self.assert_strict_equal(response['args'], MultiDict([('blub', u'blah')]))
        self.assert_strict_equal(response['args_as_list'], [('blub', [u'blah'])])
        self.assert_strict_equal(response['form'], MultiDict([('foo', u'blub hehe'), ('blah', u'42')]))
        self.assert_strict_equal(response['data'], b'')
        # currently we do not guarantee that the values are ordered correctly
        # for post data.
        ## self.assert_strict_equal(response['form_as_list'], [('foo', ['blub hehe']), ('blah', ['42'])])
        self.assert_environ(response['environ'], 'POST')

        # patch requests with form data
        response = client.patch('/?blub=blah', data='foo=blub+hehe&blah=42',
                                content_type='application/x-www-form-urlencoded')
        self.assert_strict_equal(response['args'], MultiDict([('blub', u'blah')]))
        self.assert_strict_equal(response['args_as_list'], [('blub', [u'blah'])])
        self.assert_strict_equal(response['form'],
                                 MultiDict([('foo', u'blub hehe'), ('blah', u'42')]))
        self.assert_strict_equal(response['data'], b'')
        self.assert_environ(response['environ'], 'PATCH')

        # post requests with json data
        json = b'{"foo": "bar", "blub": "blah"}'
        response = client.post('/?a=b', data=json, content_type='application/json')
        self.assert_strict_equal(response['data'], json)
        self.assert_strict_equal(response['args'], MultiDict([('a', u'b')]))
        self.assert_strict_equal(response['form'], MultiDict())

def test_ie7_unc_path(self):
        client = Client(form_data_consumer, Response)
        data_file = join(dirname(__file__), 'multipart', 'ie7_full_path_request.txt')
        data = get_contents(data_file)
        boundary = '---------------------------7da36d1b4a0164'
        response = client.post('/?object=cb_file_upload_multiple', data=data, content_type=
                                   'multipart/form-data; boundary="%s"' % boundary, content_length=len(data))
        lines = response.get_data().split(b'\n', 3)
        self.assert_strict_equal(lines[0],
                          repr(u'Sellersburg Town Council Meeting 02-22-2010doc.doc').encode('ascii'))

def test_responder(self):
        def foo(environ, start_response):
            return BaseResponse(b'Test')
        client = Client(wsgi.responder(foo), BaseResponse)
        response = client.get('/')
        self.assert_equal(response.status_code, 200)
        self.assert_equal(response.data, b'Test')

def test_append_slash_redirect(self):
        def app(env, sr):
            return utils.append_slash_redirect(env)(env, sr)
        client = Client(app, BaseResponse)
        response = client.get('foo', base_url='http://example.org/app')
        self.assert_equal(response.status_code, 301)
        self.assert_equal(response.headers['Location'], 'http://example.org/app/foo/')

def test(self, worker_cls, publish_cls):
        mock_worker = worker_cls.return_value = Mock()
        mock_publisher = publish_cls.return_value = Mock()

        app = make_app_from_environ()
        c = Client(app, BaseResponse)
        sent_data = {
            'url': '/url/',
            'referrer': '/referrer/',
            'user_id': 1,
        }

        # /page/ endpoint.
        resp = c.post('/page/', data=json.dumps(sent_data))
        assert resp.status_code == 201
        assert mock_worker.queue.call_count == 1
        row = mock_worker.queue.call_args[0][0]
        for key in sent_data.keys() + ['id', 'received_at', 'context', 'sent_at']:
            assert key in row

        # /events/ endpoint.
        sent_data['event_name'] = 'click'
        resp = c.post('/event/', data=json.dumps(sent_data))
        assert resp.status_code == 201
        assert mock_publisher.publish.call_count == 1
        row = mock_publisher.publish.call_args[1]['data']
        # Make sure the UUID format is valid.
        UUID(bytes=b64decode(row['uuid']))
        for key in ('timestamp', 'type', 'data'):
            assert key in row
        data = row['data']
        for key in sent_data.keys() + ['received_at', 'context', 'sent_at']:
            if key != 'event_name':
                assert key in data

def setUp(self):
    self.logdir = self.get_temp_dir()
    self.addCleanup(shutil.rmtree, self.logdir)
    self._generate_test_data(run_name='run1')
    self.multiplexer = event_multiplexer.EventMultiplexer(
        size_guidance=application.DEFAULT_SIZE_GUIDANCE,
        purge_orphaned_data=True)
    self._context = base_plugin.TBContext(
        assets_zip_provider=get_test_assets_zip_provider(),
        logdir=self.logdir,
        multiplexer=self.multiplexer)
    self.plugin = core_plugin.CorePlugin(self._context)
    app = application.TensorBoardWSGIApp(
        self.logdir, [self.plugin], self.multiplexer, 0, path_prefix='')
    self.server = werkzeug_test.Client(app, wrappers.BaseResponse)

def _SetupWSGIApp(self):
    multiplexer = event_multiplexer.EventMultiplexer(
        size_guidance=application.DEFAULT_SIZE_GUIDANCE,
        purge_orphaned_data=True)
    context = base_plugin.TBContext(
        logdir=self.log_dir, multiplexer=multiplexer)
    self.plugin = projector_plugin.ProjectorPlugin(context)
    wsgi_app = application.TensorBoardWSGIApp(
        self.log_dir, [self.plugin], multiplexer, reload_interval=0,
        path_prefix='')
    self.server = werkzeug_test.Client(wsgi_app, wrappers.BaseResponse)

def setUp(self):
    plugins = [
        FakePlugin(
            None, plugin_name='foo', is_active_value=True, routes_mapping={}),
        FakePlugin(
            None, plugin_name='bar', is_active_value=False, routes_mapping={}),
    ]
    app = application.TensorBoardWSGI(plugins)
    self.server = werkzeug_test.Client(app, wrappers.BaseResponse)

def test_basic(self):
        resources = join(dirname(__file__), 'multipart')
        client = Client(form_data_consumer, Response)

        repository = [
            ('firefox3-2png1txt', '---------------------------186454651713519341951581030105', [
                (u'anchor.png', 'file1', 'image/png', 'file1.png'),
                (u'application_edit.png', 'file2', 'image/png', 'file2.png')
            ], u'example text'),
            ('firefox3-2pnglongtext', '---------------------------14904044739787191031754711748', [
                (u'accept.png', 'file1', 'image/png', 'file1.png'),
                (u'add.png', 'file2', 'image/png', 'file2.png')
            ], u'--long text\r\n--with boundary\r\n--lookalikes--'),
            ('opera8-2png1txt', '----------zEO9jQKmLc2Cq88c23Dx19', [
                (u'arrow_branch.png', 'file1', 'image/png', 'file1.png'),
                (u'award_star_bronze_1.png', 'file2', 'image/png', 'file2.png')
            ], u'blafasel öäü'),
            ('webkit3-2png1txt', '----WebKitFormBoundaryjdSFhcARk8fyGNy6', [
                (u'gtk-apply.png', 'file1', 'image/png', 'file1.png'),
                (u'gtk-no.png', 'file2', 'image/png', 'file2.png')
            ], u'this is another text with ümläüts'),
            ('ie6-2png1txt', '---------------------------7d91b03a20128', [
                (u'file1.png', 'file1', 'image/x-png', 'file1.png'),
                (u'file2.png', 'file2', 'image/x-png', 'file2.png')
            ], u'ie6 sucks :-/')
        ]

        for name, boundary, files, text in repository:
            folder = join(resources, name)
            data = get_contents(join(folder, 'request.txt'))
            for filename, field, content_type, fsname in files:
                response = client.post('/?object=' + field, data=data, content_type=
                                       'multipart/form-data; boundary="%s"' % boundary,
                                       content_length=len(data))
                lines = response.get_data().split(b'\n', 3)
                self.assert_strict_equal(lines[0], repr(filename).encode('ascii'))
                self.assert_strict_equal(lines[1], repr(field).encode('ascii'))
                self.assert_strict_equal(lines[2], repr(content_type).encode('ascii'))
                self.assert_strict_equal(lines[3], get_contents(join(folder, fsname)))
            response = client.post('/?object=text', data=data, content_type=
                                   'multipart/form-data; boundary="%s"' % boundary,
                                   content_length=len(data))
            self.assert_strict_equal(response.get_data(), repr(text).encode('utf-8'))

def setUp(self):
    self.log_dir = tempfile.mkdtemp()

    # We use numpy.random to generate images. We seed to avoid non-determinism
    # in this test.
    numpy.random.seed(42)

    # Create old-style image summaries for run "foo".
    tf.reset_default_graph()
    sess = tf.Session()
    placeholder = tf.placeholder(tf.uint8)
    tf.summary.image(name="baz", tensor=placeholder)
    merged_summary_op = tf.summary.merge_all()
    foo_directory = os.path.join(self.log_dir, "foo")
    writer = tf.summary.FileWriter(foo_directory)
    writer.add_graph(sess.graph)
    for step in xrange(2):
      writer.add_summary(sess.run(merged_summary_op, feed_dict={
          placeholder: (numpy.random.rand(1, 16, 42, 3) * 255).astype(
              numpy.uint8)
      }), global_step=step)
    writer.close()

    # Create new-style image summaries for run bar.
    tf.reset_default_graph()
    sess = tf.Session()
    placeholder = tf.placeholder(tf.uint8)
    summary.op(name="quux", images=placeholder,
               description="how do you pronounce that, anyway?")
    merged_summary_op = tf.summary.merge_all()
    bar_directory = os.path.join(self.log_dir, "bar")
    writer = tf.summary.FileWriter(bar_directory)
    writer.add_graph(sess.graph)
    for step in xrange(2):
      writer.add_summary(sess.run(merged_summary_op, feed_dict={
          placeholder: (numpy.random.rand(1, 6, 8, 3) * 255).astype(
              numpy.uint8)
      }), global_step=step)
    writer.close()

    # Start a server with the plugin.
    multiplexer = event_multiplexer.EventMultiplexer({
        "foo": foo_directory,
        "bar": bar_directory,
    })
    context = base_plugin.TBContext(
        logdir=self.log_dir, multiplexer=multiplexer)
    plugin = images_plugin.ImagesPlugin(context)
    wsgi_app = application.TensorBoardWSGIApp(
        self.log_dir, [plugin], multiplexer, reload_interval=0, path_prefix='')
    self.server = werkzeug_test.Client(wsgi_app, wrappers.BaseResponse)
    self.routes = plugin.get_plugin_apps()

def setUp(self):
    self.log_dir = tempfile.mkdtemp()

    # We use numpy.random to generate audio. We seed to avoid non-determinism
    # in this test.
    numpy.random.seed(42)

    # Create old-style audio summaries for run "foo".
    tf.reset_default_graph()
    sess = tf.Session()
    placeholder = tf.placeholder(tf.float32)
    tf.summary.audio(name="baz", tensor=placeholder, sample_rate=44100)
    merged_summary_op = tf.summary.merge_all()
    foo_directory = os.path.join(self.log_dir, "foo")
    writer = tf.summary.FileWriter(foo_directory)
    writer.add_graph(sess.graph)
    for step in xrange(2):
      # The floats (sample data) range from -1 to 1.
      writer.add_summary(sess.run(merged_summary_op, feed_dict={
          placeholder: numpy.random.rand(42, 22050) * 2 - 1
      }), global_step=step)
    writer.close()

    # Create new-style audio summaries for run "bar".
    tf.reset_default_graph()
    sess = tf.Session()
    audio_placeholder = tf.placeholder(tf.float32)
    labels_placeholder = tf.placeholder(tf.string)
    summary.op("quux", audio_placeholder, sample_rate=44100,
               labels=labels_placeholder,
               description="how do you pronounce that, anyway?")
    merged_summary_op = tf.summary.merge_all()
    bar_directory = os.path.join(self.log_dir, "bar")
    writer = tf.summary.FileWriter(bar_directory)
    writer.add_graph(sess.graph)
    for step in xrange(2):
      # The floats (sample data) range from -1 to 1.
      writer.add_summary(sess.run(merged_summary_op, feed_dict={
          audio_placeholder: numpy.random.rand(42, 11025, 1) * 2 - 1,
          labels_placeholder: [
              tf.compat.as_bytes('step **%s**, sample %s' % (step, sample))
              for sample in xrange(42)
          ],
      }), global_step=step)
    writer.close()

    # Start a server with the plugin.
    multiplexer = event_multiplexer.EventMultiplexer({
        "foo": foo_directory,
        "bar": bar_directory,
    })
    context = base_plugin.TBContext(
        logdir=self.log_dir, multiplexer=multiplexer)
    self.plugin = audio_plugin.AudioPlugin(context)
    wsgi_app = application.TensorBoardWSGIApp(
        self.log_dir, [self.plugin], multiplexer, reload_interval=0,
        path_prefix='')
    self.server = werkzeug_test.Client(wsgi_app, wrappers.BaseResponse)