Source code
Revision control
Copy as Markdown
Other Tools
Test Info: Warnings
- This test gets skipped with pattern: os == 'win' && msix
- Manifest: netwerk/test/unit/xpcshell.toml
/* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this
"use strict";
// End-to-end tests for fetch() streaming upload bodies (ReadableStream + duplex)
// against every transport from plaintext HTTP/1.1 through HTTP/3, using the
// shared NodeServer helpers (and the prebuilt http3server binary for HTTP/3).
//
// Streaming uploads are only supported on HTTP/2 and HTTP/3; sending one over
// HTTP/1.x is rejected with a TypeError. That restriction is not mandated by
// the Fetch spec, but it matches other browsers and is asserted by WPT
// (fetch/api/basic/request-upload.any.js, "Streaming upload shouldn't work on
// Http/1.1.").
const { NodeHTTPServer, NodeHTTPSServer, NodeHTTP2Server, HTTP3Server } =
const { TestUtils } = ChromeUtils.importESModule(
);
const ECHO_HANDLER = function echoHandler(req, resp) {
let chunks = [];
req.on("data", c => chunks.push(c));
req.on("end", () => {
let body = Buffer.concat(chunks);
resp.writeHead(200, { "Content-Type": "application/octet-stream" });
resp.end(body);
});
};
// Counts what actually reached the server, so a rejected upload can be shown
// to have been refused before anything was sent.
const COUNTING_HANDLER = function countingHandler(req, resp) {
global.requestCount = (global.requestCount || 0) + 1;
let chunks = [];
req.on("data", c => chunks.push(c));
req.on("end", () => {
global.bodyBytes = (global.bodyBytes || 0) + Buffer.concat(chunks).length;
resp.writeHead(200, { "Content-Type": "application/octet-stream" });
resp.end("ok");
});
};
function makeStream(...chunks) {
return new ReadableStream({
start(controller) {
const encoder = new TextEncoder();
for (const chunk of chunks) {
controller.enqueue(
typeof chunk === "string" ? encoder.encode(chunk) : chunk
);
}
controller.close();
},
});
}
// Async producer: enqueues each chunk lazily via the pull callback so the
// body genuinely arrives over time rather than all upfront. Lets us verify
// the channel keeps the request open and forwards data as it shows up.
function makeAsyncStream(...chunks) {
let i = 0;
return new ReadableStream({
async pull(controller) {
await Promise.resolve();
if (i >= chunks.length) {
controller.close();
return;
}
const chunk = chunks[i++];
const encoder = new TextEncoder();
controller.enqueue(
typeof chunk === "string" ? encoder.encode(chunk) : chunk
);
},
});
}
const STREAM_TESTS = [
{
name: "empty stream",
body: () => makeStream(),
expected: "",
},
{
name: "single small chunk",
body: () => makeStream("Test"),
expected: "Test",
},
{
name: "multiple chunks (sync)",
body: () => makeStream("Hello", " ", "world"),
expected: "Hello world",
},
{
name: "multiple chunks (async)",
body: () => makeAsyncStream("alpha", "-", "beta", "-", "gamma"),
expected: "alpha-beta-gamma",
},
{
name: "large body across chunks",
body: () => makeAsyncStream("a".repeat(4096), "b".repeat(4096)),
expected: "a".repeat(4096) + "b".repeat(4096),
},
];
async function fetchStream(url, body) {
return fetch(url, { method: "POST", body, duplex: "half" });
}
async function runStreamingShouldSucceed(url, label) {
for (const t of STREAM_TESTS) {
info(`${label}: ${t.name}`);
let resp = await fetchStream(url, t.body());
Assert.equal(resp.status, 200, `${label} ${t.name}: status`);
let text = await resp.text();
Assert.equal(text.length, t.expected.length, `${label} ${t.name}: length`);
Assert.equal(text, t.expected, `${label} ${t.name}: body`);
}
}
// An HTTP/1.x streaming upload must be refused before the request goes out,
// not after the server has already received the whole body.
async function runNothingIsUploaded(server, label) {
await server.execute("global.requestCount = 0; global.bodyBytes = 0;");
let url = `${server.origin()}/count`;
await Assert.rejects(
fetchStream(url, makeStream("should", "never", "arrive")),
TypeError,
`${label}: streaming upload must reject`
);
Assert.equal(
await server.execute("global.bodyBytes"),
0,
`${label}: no body bytes should have reached the server`
);
Assert.equal(
await server.execute("global.requestCount"),
0,
`${label}: the request should not have been sent at all`
);
}
async function runStreamingShouldFail(url, label) {
for (const t of STREAM_TESTS) {
info(`${label} (must reject): ${t.name}`);
let threw = false;
try {
await fetchStream(url, t.body());
} catch (e) {
threw = true;
Assert.ok(
e instanceof TypeError,
`${label} ${t.name}: expected TypeError, got ${e}`
);
}
Assert.ok(
threw,
`${label} ${t.name}: streaming upload over HTTP/1.x must reject`
);
}
}
async function withServer(ServerClass, fn) {
let server = new ServerClass();
await server.start();
try {
await server.registerPathHandler("/echo", ECHO_HANDLER);
await server.registerPathHandler("/count", COUNTING_HANDLER);
await fn(server);
} finally {
await server.stop();
}
}
add_setup(async function setup() {
do_get_profile(); // installCert needs a profile.
Services.prefs.setBoolPref("network.http.http3.enable", true);
registerCleanupFunction(() => {
Services.prefs.clearUserPref("network.http.http3.enable");
});
});
add_task(async function test_http1_plain() {
await withServer(NodeHTTPServer, async server => {
await runStreamingShouldFail(`${server.origin()}/echo`, "HTTP/1.1 plain");
await runNothingIsUploaded(server, "HTTP/1.1 plain");
});
});
add_task(async function test_http1_tls() {
await withServer(NodeHTTPSServer, async server => {
await runStreamingShouldFail(`${server.origin()}/echo`, "HTTP/1.1 TLS");
await runNothingIsUploaded(server, "HTTP/1.1 TLS");
});
});
add_task(async function test_http2() {
await withServer(NodeHTTP2Server, async server => {
await runStreamingShouldSucceed(`${server.origin()}/echo`, "HTTP/2");
});
});
// REQUEST_BODY_SENT is driven by NS_NET_STATUS_WAITING_FOR, which HTTP/2 fires
// off the request body byte count a streaming upload never has.
add_task(async function test_http2_request_body_sent_is_reported() {
const BODY_SENT =
Ci.nsIHttpActivityObserver.ACTIVITY_SUBTYPE_REQUEST_BODY_SENT;
await withServer(NodeHTTP2Server, async server => {
let url = `${server.origin()}/echo`;
let sequence = [];
let distributor = Cc[
"@mozilla.org/network/http-activity-distributor;1"
].getService(Ci.nsIHttpActivityDistributor);
let observer = {
observeActivity(aChannel, aActivityType, aActivitySubtype) {
if (
aActivityType !==
Ci.nsIHttpActivityObserver.ACTIVITY_TYPE_HTTP_TRANSACTION
) {
return;
}
try {
if (aChannel.QueryInterface(Ci.nsIChannel).URI.spec === url) {
sequence.push(aActivitySubtype);
}
} catch (e) {}
},
};
distributor.addObserver(observer);
try {
let resp = await fetchStream(url, makeAsyncStream("alpha", "-", "beta"));
Assert.equal(resp.status, 200, "status");
Assert.equal(await resp.text(), "alpha-beta", "body echoed back");
// The notifications are marshalled to the main thread, so they are still
// in flight when the fetch resolves.
await TestUtils.waitForCondition(
() =>
sequence.includes(
Ci.nsIHttpActivityObserver.ACTIVITY_SUBTYPE_TRANSACTION_CLOSE
),
"the transaction was reported closed"
);
info(`activity: ${sequence.join(",")}`);
Assert.ok(
sequence.includes(BODY_SENT),
"the streaming request body is reported sent"
);
} finally {
distributor.removeObserver(observer);
}
});
});
// The http3server's /post endpoint reads the entire request body and
// replies with a constant body plus an x-data-received-length header
// telling us exactly how many body bytes it saw.
async function setupHttp3Server() {
const server = new HTTP3Server();
await server.start(
Services.env.get("MOZ_HTTP3_SERVER_PATH"),
Services.env.get("MOZ_HTTP3_CERT_DB_PATH")
);
// http3server uses the same CA as the HTTP/2 node server.
let certdb = Cc["@mozilla.org/security/x509certdb;1"].getService(
Ci.nsIX509CertDB
);
addCertFromFile(certdb, "../unit/http2-ca.pem", "CTu,u,u");
Services.prefs.setBoolPref("network.dns.disableIPv6", true);
Services.prefs.setCharPref("network.dns.localDomains", "foo.example.com");
Services.prefs.setCharPref(
"network.http.http3.alt-svc-mapping-for-testing",
`foo.example.com;h3=:${server.port()}`
);
// Prime alt-svc. The first request lands on regular HTTPS (which fails
// because only the H3 port is listening), but the alt-svc pref above
// is enough for the next request to go straight over H3. Retry the
// warm-up until nsHttpChannel reports that the connection actually used
// the advertised alternate.
let h3Route = `foo.example.com:${server.port()}`;
const maxAttempts = 20;
await new Promise((resolve, reject) => {
let attempts = 0;
let attempt = () => {
if (++attempts > maxAttempts) {
reject(
new Error(
`alt-svc warm-up did not route to ${h3Route} after ${maxAttempts} attempts`
)
);
return;
}
chan.asyncOpen({
onStartRequest() {},
onDataAvailable(request, stream, offset, count) {
read_stream(stream, count);
},
onStopRequest(request) {
let routed = "NA";
try {
routed = request.getRequestHeader("Alt-Used");
} catch (e) {
info(`Alt-Used not present yet: ${e}`);
}
if (routed == h3Route) {
resolve();
} else {
attempt();
}
},
});
};
attempt();
});
return server;
}
add_task(
{
skip_if: () =>
mozinfo.os == "android" ||
!Services.env.get("MOZ_HTTP3_SERVER_PATH") ||
!Services.env.get("MOZ_HTTP3_CERT_DB_PATH"),
},
async function test_http3() {
Services.prefs.setBoolPref("network.http.http3.enable", true);
let server = await setupHttp3Server();
registerCleanupFunction(async () => {
await server.stop();
Services.prefs.clearUserPref("network.dns.disableIPv6");
Services.prefs.clearUserPref("network.dns.localDomains");
Services.prefs.clearUserPref(
"network.http.http3.alt-svc-mapping-for-testing"
);
});
for (const t of STREAM_TESTS) {
info(`HTTP/3: ${t.name}`);
let resp = await fetchStream(url, t.body());
Assert.equal(resp.status, 200, `HTTP/3 ${t.name}: status`);
Assert.equal(
resp.headers.get("x-data-received-length"),
String(t.expected.length),
`HTTP/3 ${t.name}: server received correct number of bytes`
);
// Drain the response.
await resp.text();
}
}
);