describe("http.stream_common", function()
	local h1_connection = require "http.h1_connection"
	local new_headers = require "http.headers".new
	local cqueues = require "cqueues"
	local ca = require "cqueues.auxlib"
	local cs = require "cqueues.socket"
	local function new_pair(version)
		local s, c = ca.assert(cs.pair())
		s = h1_connection.new(s, "server", version)
		c = h1_connection.new(c, "client", version)
		return s, c
	end
	local function new_request_headers()
		local headers = new_headers()
		headers:append(":method", "GET")
		headers:append(":scheme", "http")
		headers:append(":authority", "myauthority")
		headers:append(":path", "/")
		return headers
	end
	it("Can read a number of characters", function()
		local server, client = new_pair(1.1)
		local cq = cqueues.new()
		cq:wrap(function()
			local stream = client:new_stream()
			assert(stream:write_headers(new_request_headers(), false))
			assert(stream:write_chunk("foo", false))
			assert(stream:write_chunk("\nb", false))
			assert(stream:write_chunk("ar\n", true))
		end)
		cq:wrap(function()
			local stream = server:get_next_incoming_stream()
			-- same size as next chunk
			assert.same("foo", stream:get_body_chars(3))
			-- less than chunk
			assert.same("\n", stream:get_body_chars(1))
			-- crossing chunks
			assert.same("bar", stream:get_body_chars(3))
			-- more than available
			assert.same("\n", stream:get_body_chars(8))
			-- when none available
			assert.same(nil, stream:get_body_chars(8))
		end)
		assert_loop(cq, TEST_TIMEOUT)
		assert.truthy(cq:empty())
		client:close()
		server:close()
	end)
	it("Can read a line", function()
		local server, client = new_pair(1.1)
		local cq = cqueues.new()
		cq:wrap(function()
			local stream = client:new_stream()
			assert(stream:write_headers(new_request_headers(), false))
			assert(stream:write_chunk("foo", false))
			assert(stream:write_chunk("\nb", false))
			assert(stream:write_chunk("ar\n", true))
		end)
		cq:wrap(function()
			local stream = server:get_next_incoming_stream()
			assert.same("foo", stream:get_body_until("\n", true, false))
			assert.same("bar", stream:get_body_until("\n", true, false))
			assert.same(nil, stream:get_body_until("\n", true, false))
		end)
		assert_loop(cq, TEST_TIMEOUT)
		assert.truthy(cq:empty())
		client:close()
		server:close()
	end)
	it("can read into a temporary file", function()
		local server, client = new_pair(1.1)
		local cq = cqueues.new()
		cq:wrap(function()
			local stream = client:new_stream()
			assert(stream:write_headers(new_request_headers(), false))
			assert(stream:write_chunk("hello world!", true))
		end)
		cq:wrap(function()
			local stream = assert(server:get_next_incoming_stream())
			local file = assert(stream:get_body_as_file())
			assert.same("hello world!", file:read"*a")
		end)
		assert_loop(cq, TEST_TIMEOUT)
		assert.truthy(cq:empty())
		client:close()
		server:close()
	end)
	describe("write_body_from_file", function()
		it("works with a temporary file", function()
			local server, client = new_pair(1.1)
			local cq = cqueues.new()
			cq:wrap(function()
				local file = io.tmpfile()
				assert(file:write("hello world!"))
				assert(file:seek("set"))
				local stream = client:new_stream()
				assert(stream:write_headers(new_request_headers(), false))
				assert(stream:write_body_from_file(file))
			end)
			cq:wrap(function()
				local stream = assert(server:get_next_incoming_stream())
				assert.same("hello world!", assert(stream:get_body_as_string()))
			end)
			assert_loop(cq, TEST_TIMEOUT)
			assert.truthy(cq:empty())
			client:close()
			server:close()
		end)
		it("works using the options form", function()
			local server, client = new_pair(1.1)
			local cq = cqueues.new()
			cq:wrap(function()
				local file = io.tmpfile()
				assert(file:write("hello world!"))
				assert(file:seek("set"))
				local stream = client:new_stream()
				assert(stream:write_headers(new_request_headers(), false))
				assert(stream:write_body_from_file({
					file = file;
				}))
			end)
			cq:wrap(function()
				local stream = assert(server:get_next_incoming_stream())
				assert.same("hello world!", assert(stream:get_body_as_string()))
			end)
			assert_loop(cq, TEST_TIMEOUT)
			assert.truthy(cq:empty())
			client:close()
			server:close()
		end)
		it("validates .count option", function()
			local server, client = new_pair(1.1)
			local cq = cqueues.new()
			cq:wrap(function()
				local stream = client:new_stream()
				assert(stream:write_headers(new_request_headers(), false))
				assert.has_error(function()
					stream:write_body_from_file({
						file = io.tmpfile();
						count = "invalid count field";
					})
				end)
			end)
			cq:wrap(function()
				assert(server:get_next_incoming_stream())
			end)
			assert_loop(cq, TEST_TIMEOUT)
			assert.truthy(cq:empty())
			client:close()
			server:close()
		end)
		it("limits number of bytes when using .count option", function()
			local server, client = new_pair(1.1)
			local cq = cqueues.new()
			cq:wrap(function()
				local file = io.tmpfile()
				assert(file:write("hello world!"))
				assert(file:seek("set"))
				local stream = client:new_stream()
				assert(stream:write_headers(new_request_headers(), false))
				assert(stream:write_body_from_file({
					file = file;
					count = 5;
				}))
			end)
			cq:wrap(function()
				local stream = assert(server:get_next_incoming_stream())
				assert.same("hello", assert(stream:get_body_as_string()))
			end)
			assert_loop(cq, TEST_TIMEOUT)
			assert.truthy(cq:empty())
			client:close()
			server:close()
		end)
		it("reports an error on early EOF", function()
			local server, client = new_pair(1.1)
			local cq = cqueues.new()
			cq:wrap(function()
				local file = io.tmpfile()
				assert(file:write("hello world!"))
				assert(file:seek("set"))
				local stream = client:new_stream()
				assert(stream:write_headers(new_request_headers(), false))
				assert.has_error(function()
					assert(stream:write_body_from_file({
						file = file;
						count = 50; -- longer than the file
					}))
				end)
			end)
			cq:wrap(function()
				assert(server:get_next_incoming_stream())
			end)
			assert_loop(cq, TEST_TIMEOUT)
			assert.truthy(cq:empty())
			client:close()
			server:close()
		end)
	end)
end)
