XiaoMo/anyproxy
1
0
Fork 0
mirror of https://github.com/alibaba/anyproxy.git synced 2025-05-10 14:58:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: update test code for websocket

Browse Source
This commit is contained in:
xiaofeng.mxf 2021-08-30 00:27:17 +08:00
parent 42e43fca7b
commit 570acb6519
5 changed files with 127 additions and 371 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
lib/handlers/wsHandler.js
View File

@ -84,7 +84,7 @@ function handleWs(userRule, recorder, wsClient, wsReq) {
const clientMsgQueue = []; const clientMsgQueue = [];
const serverInfo = getWsReqInfo(wsReq); const serverInfo = getWsReqInfo(wsReq);
// proxy-layer websocket client // proxy-layer websocket client
const proxyWs = new WebSocket(serverInfo.url, '', { const proxyWs = new WebSocket(serverInfo.url, {
rejectUnauthorized: !self.dangerouslyIgnoreUnauthorized, rejectUnauthorized: !self.dangerouslyIgnoreUnauthorized,
headers: serverInfo.noWsHeaders headers: serverInfo.noWsHeaders
}); });
@ -106,13 +106,8 @@ function handleWs(userRule, recorder, wsClient, wsReq) {
*/ */
const sendProxyMessage = (finalMsg) => { const sendProxyMessage = (finalMsg) => {
const message = finalMsg.data; const message = finalMsg.data;
if (proxyWs.readyState === 1) { if (proxyWs.readyState === 1 && clientMsgQueue.length === 0) {
// if there still are msg queue consuming, keep it going proxyWs.send(message);
if (clientMsgQueue.length > 0) {
clientMsgQueue.push(message);
} else {
proxyWs.send(message);
}
} else { } else {
clientMsgQueue.push(message); clientMsgQueue.push(message);
} }
@ -120,7 +115,6 @@ function handleWs(userRule, recorder, wsClient, wsReq) {
/** /**
* consume the message in queue when the proxy ws is not ready yet * consume the message in queue when the proxy ws is not ready yet
* will handle them from the first one-by-one
*/ */
const consumeMsgQueue = () => { const consumeMsgQueue = () => {
while (clientMsgQueue.length > 0) { while (clientMsgQueue.length > 0) {

2
package.json
View File

@ -37,7 +37,7 @@
"svg-inline-react": "^1.0.2", "svg-inline-react": "^1.0.2",
"thunkify": "^2.1.2", "thunkify": "^2.1.2",
"whatwg-fetch": "^1.0.0", "whatwg-fetch": "^1.0.0",
"ws": "^5.1.0" "ws": "^8.2.1"
}, },
"devDependencies": { "devDependencies": {
"@babel/core": "^7.8.3", "@babel/core": "^7.8.3",

236
test/basic.spec.js
View File

@ -3,9 +3,7 @@ process.env.NODE_TLS_REJECT_UNAUTHORIZED = '0';
const path = require('path'); const path = require('path');
const fs = require('fs'); const fs = require('fs');
const urllib = require('urllib'); const urllib = require('urllib');
const request = require('request');
const { basicProxyRequest, proxyServerWithRule, } = require('./util.js'); const { basicProxyRequest, proxyServerWithRule, } = require('./util.js');
const http = require('http');
const WebSocket = require('ws'); const WebSocket = require('ws');
const tunnel = require('tunnel'); const tunnel = require('tunnel');
@ -26,34 +24,7 @@ afterAll(() => {
return proxyServer && proxyServer.close(); return proxyServer && proxyServer.close();
}); });
function doProxyWebSocket(url, headers = {}) { ['http'].forEach(protocol => {
let agent = new tunnel.httpOverHttp({
proxy: {
hostname: 'localhost',
port: proxyPort,
}
})
if (url.indexOf('wss') === 0) {
agent = new tunnel.httpsOverHttp({
rejectUnauthorized: false,
proxy: {
hostname: 'localhost',
port: proxyPort,
}
})
}
const ws = new WebSocket(url, {
agent,
rejectUnauthorized: false,
headers
});
return ws;
}
['http', 'https'].forEach(protocol => {
describe.only(`${protocol} - HTTP verbs`, () => { describe.only(`${protocol} - HTTP verbs`, () => {
const assertProxyRes = (result) => { const assertProxyRes = (result) => {
const proxyRes = result.response; const proxyRes = result.response;
@ -103,26 +74,65 @@ function doProxyWebSocket(url, headers = {}) {
await basicProxyRequest(proxyHost, 'PATCH', url).then(assertProxyRes); await basicProxyRequest(proxyHost, 'PATCH', url).then(assertProxyRes);
}); });
it.only('Websocket', async () => { describe.only('websocket', () => {
const expectEcho = (ws) => { const WS_PORT = 8012;
return new Promise((resolve, reject) => { let wsEchoServer;
const wsMsg = Buffer.alloc(100 * 1024, 'a').toString(); // 100kb beforeAll(async () => {
wsEchoServer = new WebSocket.WebSocketServer({ port: WS_PORT });
wsEchoServer.on('connection', (ws) => {
ws.on('message', (message) => {
ws.send(message);
});
});
});
afterAll(async () => {
wsEchoServer.close();
});
it('Websocket', async () => {
const wsUrl = `${protocol === 'https' ? 'wss' : 'ws'}://127.0.0.1:${WS_PORT}`;
let agent;
if (wsUrl.indexOf('wss') === 0) {
agent = new tunnel.httpsOverHttp({
rejectUnauthorized: false,
proxy: {
hostname: 'localhost',
port: proxyPort,
}
});
} else {
agent = new tunnel.httpOverHttp({
proxy: {
hostname: 'localhost',
port: proxyPort,
}
});
}
agent.on('free', (a, b, c) => {
console.log('agent on Free', a, b, c);
})
const ws = new WebSocket(wsUrl, {
agent,
rejectUnauthorized: false,
headers: {},
});
await new Promise((resolve, reject) => {
const wsMsg = Buffer.alloc(100 * 1024, 'a');
ws.on('open', () => { ws.on('open', () => {
ws.send(wsMsg); ws.send(wsMsg);
}); });
ws.on('message', (msg) => { ws.on('message', (msg) => {
expect(msg).toBe(wsMsg); expect(msg.equals(wsMsg));
ws.close(); ws.close();
resolve(); setTimeout(resolve, 300); // some clean up job
}); });
}); });
}; });
const wsUrl = `${protocol === 'https' ? 'wss' : 'ws'}://echo.websocket.org`;
const ws = doProxyWebSocket(wsUrl, {});
await expectEcho(ws);
}); });
}); });
}); });
@ -160,83 +170,83 @@ describe('response data formats', () => {
}); });
}); });
describe('big files', () => { // describe('big files', () => {
const BIG_FILE_SIZE = 100 * 1024 * 1024 - 1; // 100 mb // const BIG_FILE_SIZE = 100 * 1024 * 1024 - 1; // 100 mb
const BUFFER_FILL = 'a'; // const BUFFER_FILL = 'a';
let server; // let server;
beforeAll(() => { // beforeAll(() => {
server = http.createServer({}, (req, res) => { // server = http.createServer({}, (req, res) => {
if (/download/.test(req.url)) { // if (/download/.test(req.url)) {
const bufferContent = Buffer.alloc(BIG_FILE_SIZE, BUFFER_FILL); // const bufferContent = Buffer.alloc(BIG_FILE_SIZE, BUFFER_FILL);
res.write(bufferContent); // res.write(bufferContent);
res.end(); // res.end();
} else if (/upload/.test(req.url)) { // } else if (/upload/.test(req.url)) {
let reqPayloadSize = 0; // let reqPayloadSize = 0;
req.on('data', (data) => { // req.on('data', (data) => {
const bufferLength = data.length; // const bufferLength = data.length;
reqPayloadSize += bufferLength; // reqPayloadSize += bufferLength;
const expectBufferContent = Buffer.alloc(bufferLength, BUFFER_FILL); // const expectBufferContent = Buffer.alloc(bufferLength, BUFFER_FILL);
if (!expectBufferContent.equals(data)) { // if (!expectBufferContent.equals(data)) {
res.statusCode = 500; // res.statusCode = 500;
res.write('content not match'); // res.write('content not match');
} // }
}).on('end', () => { // }).on('end', () => {
if (res.statusCode === 500 || reqPayloadSize !== BIG_FILE_SIZE) { // if (res.statusCode === 500 || reqPayloadSize !== BIG_FILE_SIZE) {
res.statusCode = 500; // res.statusCode = 500;
} else { // } else {
res.statusCode = 200; // res.statusCode = 200;
} // }
res.end(); // res.end();
}); // });
} // }
}); // });
server.listen(3000); // server.listen(3000);
}); // });
afterAll((done) => { // afterAll((done) => {
server && server.close(done); // server && server.close(done);
}); // });
it('download big file', (done) => { // it('download big file', (done) => {
let responseSizeCount = 0; // let responseSizeCount = 0;
request({ // request({
url: 'http://127.0.0.1:3000/download', // url: 'http://127.0.0.1:3000/download',
proxy: proxyHost, // proxy: proxyHost,
}).on('data', (data) => { // }).on('data', (data) => {
const bufferLength = data.length; // const bufferLength = data.length;
responseSizeCount += bufferLength; // responseSizeCount += bufferLength;
const expectBufferContent = Buffer.alloc(bufferLength, BUFFER_FILL); // const expectBufferContent = Buffer.alloc(bufferLength, BUFFER_FILL);
if (!expectBufferContent.equals(data)) { // if (!expectBufferContent.equals(data)) {
return done(new Error('download content not match')); // return done(new Error('download content not match'));
} // }
}).on('end', () => { // }).on('end', () => {
if (responseSizeCount !== BIG_FILE_SIZE) { // if (responseSizeCount !== BIG_FILE_SIZE) {
return done(new Error('file size not match')); // return done(new Error('file size not match'));
} // }
done(); // done();
}); // });
}, 120 * 1000); // }, 120 * 1000);
it('upload big file', (done) => { // it('upload big file', (done) => {
const bufferContent = Buffer.alloc(BIG_FILE_SIZE, BUFFER_FILL); // const bufferContent = Buffer.alloc(BIG_FILE_SIZE, BUFFER_FILL);
const req = request({ // const req = request({
url: 'http://127.0.0.1:3000/upload', // url: 'http://127.0.0.1:3000/upload',
method: 'POST', // method: 'POST',
proxy: proxyHost, // proxy: proxyHost,
}, (err, response, body) => { // }, (err, response, body) => {
if (err) { // if (err) {
return done(err); // return done(err);
} else if (response.statusCode !== 200) { // } else if (response.statusCode !== 200) {
return done(new Error('upload failed ' + body)); // return done(new Error('upload failed ' + body));
} // }
done(); // done();
}); // });
req.write(bufferContent); // req.write(bufferContent);
req.end(); // req.end();
}, 120 * 1000); // }, 120 * 1000);
}); // });
describe('web interface', () => { describe('web interface', () => {
it('should be available', async () => { it('should be available', async () => {

247
test/basic.spec.js~78858bf... fix: replace upload fixture image.png with a smaller upload.txt (#563)
View File

@ -1,247 +0,0 @@
process.env.NODE_TLS_REJECT_UNAUTHORIZED = '0';
const path = require('path');
const fs = require('fs');
const urllib = require('urllib');
const request = require('request');
const { basicProxyRequest, proxyServerWithRule, } = require('./util.js');
const http = require('http');
const WebSocket = require('ws');
const tunnel = require('tunnel');
let proxyServer;
let proxyPort;
let proxyHost;
let proxyWebInterfaceHost;
beforeAll(async () => {
jest.DEFAULT_TIMEOUT_INTERVAL = 20 * 1000;
proxyServer = await proxyServerWithRule({}, {});
proxyPort = proxyServer.proxyPort;
proxyHost = `http://localhost:${proxyPort}`;
const proxyWebInterfacePort = proxyServer.webServerInstance.webPort;
proxyWebInterfaceHost = `http://localhost:${proxyWebInterfacePort}`;
});
afterAll(() => {
return proxyServer && proxyServer.close();
});
function doProxyWebSocket(url, headers = {}) {
let agent = new tunnel.httpOverHttp({
proxy: {
hostname: 'localhost',
port: proxyPort,
}
})
if (url.indexOf('wss') === 0) {
agent = new tunnel.httpsOverHttp({
rejectUnauthorized: false,
proxy: {
hostname: 'localhost',
port: proxyPort,
}
})
}
const ws = new WebSocket(url, {
agent,
rejectUnauthorized: false,
headers
});
return ws;
}
['http', 'https'].forEach(protocol => {
describe(`${protocol} - HTTP verbs`, () => {
const assertProxyRes = (result) => {
const proxyRes = result.response;
const body = JSON.parse(result.body);
expect(proxyRes.statusCode).toBe(200);
expect(body.args).toMatchSnapshot('args');
expect(body.data).toMatchSnapshot('data');
return body;
};
it('GET', async () => {
const url = `${protocol}://httpbin.org/get`;
const getParam = {
param: 'param_value'
};
await basicProxyRequest(proxyHost, 'GET', url, {}, getParam).then(assertProxyRes);
});
it('POST body and header', async () => {
const url = `${protocol}://httpbin.org/post`;
const payloadStream = fs.createReadStream(path.resolve(__dirname, './fixtures/upload.txt'));
const postHeaders = {
anyproxy_header: 'header_value',
};
const body = await basicProxyRequest(proxyHost, 'POST', url, postHeaders, {}, payloadStream).then(assertProxyRes);
expect(body.headers['Anyproxy-Header']).toBe(postHeaders.anyproxy_header);
});
it('PUT', async () => {
const url = `${protocol}://httpbin.org/put`;
const payloadStream = fs.createReadStream(path.resolve(__dirname, './fixtures/upload.txt'));
await basicProxyRequest(proxyHost, 'PUT', url, {}, undefined, payloadStream).then(assertProxyRes);
});
it('DELETE', async () => {
const url = `${protocol}://httpbin.org/delete`;
const param = {
foo: 'bar',
};
await basicProxyRequest(proxyHost, 'DELETE', url, {}, param).then(assertProxyRes);
});
it('PATCH', async () => {
const url = `${protocol}://httpbin.org/patch`;
await basicProxyRequest(proxyHost, 'PATCH', url).then(assertProxyRes);
});
it('Websocket', async () => {
const expectEcho = (ws) => {
return new Promise((resolve, reject) => {
const wsMsg = Buffer.alloc(100 * 1024, 'a').toString(); // 100kb
ws.on('open', () => {
ws.send(wsMsg);
});
ws.on('message', (msg) => {
expect(msg).toBe(wsMsg);
ws.close();
resolve();
});
});
};
const wsUrl = `${protocol === 'https' ? 'wss' : 'ws'}://echo.websocket.org`;
const ws = doProxyWebSocket(wsUrl, {});
await expectEcho(ws);
});
});
});
describe('status code and headers', () => {
[302, 404, 500].forEach(statusCode => {
it(`GET ${statusCode}`, async () => {
const status = statusCode;
const url = `http://httpbin.org/status/${status}`;
const result = await basicProxyRequest(proxyHost, 'GET', url, {}, {});
const proxyRes = result.response;
expect(proxyRes.statusCode).toBe(statusCode);
});
it(`PUT ${statusCode}`, async () => {
const status = statusCode;
const url = `http://httpbin.org/status/${status}`;
const result = await basicProxyRequest(proxyHost, 'PUT', url, {}, {});
const proxyRes = result.response;
expect(proxyRes.statusCode).toBe(statusCode);
});
});
});
describe('response data formats', () => {
['brotli', 'deflate', 'gzip'].forEach(encoding => {
it(`GET ${encoding}`, async () => {
const url = `http://httpbin.org/${encoding}`;
const result = await basicProxyRequest(proxyHost, 'GET', url);
const headers = result.response.headers;
const body = JSON.parse(result.body);
expect(headers['content-encoding']).toBeUndefined(); // should be removed by anyproxy
expect(body.brotli || body.deflated || body.gzipped).toBeTruthy();
});
});
});
describe('big files', () => {
const BIG_FILE_SIZE = 100 * 1024 * 1024 - 1; // 100 mb
const BUFFER_FILL = 'a';
let server;
beforeAll(() => {
server = http.createServer({}, (req, res) => {
if (/download/.test(req.url)) {
const bufferContent = Buffer.alloc(BIG_FILE_SIZE, BUFFER_FILL);
res.write(bufferContent);
res.end();
} else if (/upload/.test(req.url)) {
let reqPayloadSize = 0;
req.on('data', (data) => {
const bufferLength = data.length;
reqPayloadSize += bufferLength;
const expectBufferContent = Buffer.alloc(bufferLength, BUFFER_FILL);
if (!expectBufferContent.equals(data)) {
res.statusCode = 500;
res.write('content not match');
}
}).on('end', () => {
if (res.statusCode === 500 || reqPayloadSize !== BIG_FILE_SIZE) {
res.statusCode = 500;
} else {
res.statusCode = 200;
}
res.end();
});
}
});
server.listen(3000);
});
afterAll((done) => {
server && server.close(done);
});
it('download big file', (done) => {
let responseSizeCount = 0;
request({
url: 'http://127.0.0.1:3000/download',
proxy: proxyHost,
}).on('data', (data) => {
const bufferLength = data.length;
responseSizeCount += bufferLength;
const expectBufferContent = Buffer.alloc(bufferLength, BUFFER_FILL);
if (!expectBufferContent.equals(data)) {
return done(new Error('download content not match'));
}
}).on('end', () => {
if (responseSizeCount !== BIG_FILE_SIZE) {
return done(new Error('file size not match'));
}
done();
});
}, 120 * 1000);
it('upload big file', (done) => {
const bufferContent = Buffer.alloc(BIG_FILE_SIZE, BUFFER_FILL);
const req = request({
url: 'http://127.0.0.1:3000/upload',
method: 'POST',
proxy: proxyHost,
}, (err, response, body) => {
if (err) {
return done(err);
} else if (response.statusCode !== 200) {
return done(new Error('upload failed ' + body));
}
done();
});
req.write(bufferContent);
req.end();
}, 120 * 1000);
});
describe('web interface', () => {
it('should be available', async () => {
await urllib.request(proxyWebInterfaceHost).then((result) => {
expect(result.status).toBe(200);
});
});
});

1
test/spec_rule/rule_replace_ws_message_spec.js → test/spec_rule/rule_replace_ws_message.spec.js
View File

@ -10,7 +10,6 @@ describe('Rule to replace the websocket message', () => {
let proxyServer = null; let proxyServer = null;
beforeAll((done) => { beforeAll((done) => {
jasmine.DEFAULT_TIMEOUT_INTERVAL = 50000;
printLog('Start server for rule_replace_ws_message_spec'); printLog('Start server for rule_replace_ws_message_spec');
testServer = new TestServer(); testServer = new TestServer();