

def testValetServiceBottleStreamSecure(self):
    '\n        Test Valet WSGI service request response stream sse\n        '
    console.terse('{0}\n'.format(self.testValetServiceBottleStreamSecure.__doc__))
    try:
        import bottle
    except ImportError as ex:
        console.terse('Bottle not available.\n')
        return
    store = storing.Store(stamp=0.0)
    app = bottle.default_app()

    @app.get('/stream')
    def streamGet():
        '\n            Create test server sent event stream that sends count events\n            '
        timer = StoreTimer(store, duration=2.0)
        bottle.response.set_header('Content-Type', 'text/event-stream')
        bottle.response.set_header('Cache-Control', 'no-cache')
        (yield 'retry: 1000\n\n')
        i = 0
        (yield 'id: {0}\n'.format(i))
        i += 1
        (yield 'data: START\n\n')
        n = 1
        while (not timer.expired):
            (yield 'id: {0}\n'.format(i))
            i += 1
            (yield 'data: {0}\n\n'.format(n))
            n += 1
        (yield 'data: END\n\n')
    console.terse('{0}\n'.format('Building Valet ...\n'))
    wireLogAlpha = wiring.WireLog(buffify=True, same=True)
    result = wireLogAlpha.reopen()
    serverCertCommonName = 'localhost'
    serverKeypath = '/etc/pki/tls/certs/server_key.pem'
    serverCertpath = '/etc/pki/tls/certs/server_cert.pem'
    clientCafilepath = '/etc/pki/tls/certs/client.pem'
    alpha = serving.Valet(port=6101, bufsize=131072, wlog=wireLogAlpha, store=store, app=app, scheme='https', keypath=serverKeypath, certpath=serverCertpath, cafilepath=clientCafilepath)
    self.assertIs(alpha.servant.reopen(), True)
    self.assertEqual(alpha.servant.ha, ('0.0.0.0', 6101))
    self.assertEqual(alpha.servant.eha, ('127.0.0.1', 6101))
    console.terse('{0}\n'.format('Building Patron ...\n'))
    wireLogBeta = wiring.WireLog(buffify=True, same=True)
    result = wireLogBeta.reopen()
    clientKeypath = '/etc/pki/tls/certs/client_key.pem'
    clientCertpath = '/etc/pki/tls/certs/client_cert.pem'
    serverCafilepath = '/etc/pki/tls/certs/server.pem'
    path = 'https://{0}:{1}/'.format('localhost', alpha.servant.eha[1])
    beta = clienting.Patron(bufsize=131072, wlog=wireLogBeta, store=store, path=path, reconnectable=True, scheme='https', certedhost=serverCertCommonName, keypath=clientKeypath, certpath=clientCertpath, cafilepath=serverCafilepath)
    self.assertIs(beta.connector.reopen(), True)
    self.assertIs(beta.connector.accepted, False)
    self.assertIs(beta.connector.connected, False)
    self.assertIs(beta.connector.cutoff, False)
    request = odict([('method', 'GET'), ('path', '/stream'), ('qargs', odict()), ('fragment', ''), ('headers', odict([('Accept', 'application/json'), ('Content-Length', 0)])), ('body', None)])
    beta.requests.append(request)
    timer = StoreTimer(store, duration=1.0)
    while (not timer.expired):
        alpha.serviceAll()
        time.sleep(0.05)
        beta.serviceAll()
        time.sleep(0.05)
        store.advanceStamp(0.1)
    self.assertIs(beta.connector.accepted, True)
    self.assertIs(beta.connector.connected, True)
    self.assertIs(beta.connector.cutoff, False)
    self.assertEqual(len(alpha.servant.ixes), 1)
    self.assertEqual(len(alpha.reqs), 1)
    self.assertEqual(len(alpha.reps), 1)
    requestant = alpha.reqs.values()[0]
    self.assertEqual(requestant.method, request['method'])
    self.assertEqual(requestant.url, request['path'])
    self.assertEqual(requestant.headers, {
        'accept': 'application/json',
        'accept-encoding': 'identity',
        'content-length': '0',
        'host': 'localhost:6101',
    })
    self.assertIs(beta.waited, True)
    self.assertIs(beta.respondent.ended, False)
    self.assertEqual(len(beta.responses), 0)
    self.assertIn('content-type', beta.respondent.headers)
    self.assertEqual(beta.respondent.headers['content-type'], 'text/event-stream')
    self.assertIn('transfer-encoding', beta.respondent.headers)
    self.assertEqual(beta.respondent.headers['transfer-encoding'], 'chunked')
    self.assertTrue((len(beta.events) >= 3))
    self.assertEqual(beta.respondent.retry, 1000)
    self.assertTrue((int(beta.respondent.leid) >= 2))
    event = beta.events.popleft()
    self.assertEqual(event, {
        'id': '0',
        'name': '',
        'data': 'START',
    })
    event = beta.events.popleft()
    self.assertEqual(event, {
        'id': '1',
        'name': '',
        'data': '1',
    })
    event = beta.events.popleft()
    self.assertEqual(event, {
        'id': '2',
        'name': '',
        'data': '2',
    })
    serverCafilepath.events.clear()
    timer.restart(duration=1.5)
    while (not timer.expired):
        alpha.serviceAll()
        time.sleep(0.05)
        beta.serviceAll()
        time.sleep(0.05)
        store.advanceStamp(0.1)
    self.assertTrue((len(beta.events) >= 3))
    self.assertEqual(beta.respondent.leid, '9')
    self.assertEqual(beta.events[(- 2)], {
        'id': '9',
        'name': '',
        'data': '9',
    })
    self.assertEqual(beta.events[(- 1)], {
        'id': '9',
        'name': '',
        'data': 'END',
    })
    beta.events.clear()
    alpha.servant.closeAll()
    beta.connector.close()
    wireLogAlpha.close()
    wireLogBeta.close()

dataset/ETHPy150Open ioflo/ioflo/ioflo/aio/http/test/test_serving.py BasicTestCase.testValetServiceBottleStreamSecure/VarMisuse@1041/1607 `beta`->`serverCafilepath`