ExchangeTest.pudu

Pudu169 lines10.1 KB

Copy pathGitHub ↗

1/** @Test.Transport.Exchange.Suite — response framing read from scripted raw connections */2module PuduLangHttpClient.Transport.ExchangeTest34import Std.Bytes as ByteSeq5import Std.Concurrent.Cancel as Cancel6import Std.Concurrent as Concurrent7import Std.Http as Http8import Std.Io as Io9import Std.Net as Net10import Std.Result as Result11import Std.Sync as Sync12import Std.Test as Test13import PuduLangHttpClient.Content as Content14import PuduLangHttpClient as HttpClient15import PuduLangHttpClient.Request as Request16import PuduLangHttpClient.Response as Response17import PuduLangHttpClient.Transport as Transport18import Support.Loopback as Loopback1920/// A raw server answering every request on every connection with the same text, keeping the21/// connection open for the next request.22fn answering(reply: Str) -> Loopback.Scripted {23  Loopback.scripted(fn(connection: Net.Connection, _number: Int) -> () {24      var going = true25      while going {26        if Loopback.readRequest(&connection).isEmpty() { going = false } else { Loopback.write(&connection, reply) }27      }28    })29}3031/// The response a transport gets for a `GET` of the server's root within two seconds.32fn fetched(transport: &Transport.Transport, server: &Loopback.Scripted) -> HttpClient.Outcome[Response.Response] {33  Transport.send(transport)(Request.get(server.base + "/"), Cancel.expiring(2000))34}3536/// The status, body, and trailers of an outcome.37fn shape(outcome: &HttpClient.Outcome[Response.Response]) -> Result[(Int, Str, Array[(Str, Str)]), HttpClient.Failure] {38  Result.map(*outcome, |response: Response.Response| (response.status.code, Result.unwrapOr(Response.text(&response), ""), response.trailers))39}4041/// Runs the suite.42fn main() -> Int {43  let chunkedServer = answering("HTTP/1.1 200 OK\r\ntransfer-encoding: chunked\r\n\r\n5\r\nhello\r\n6;ext=1\r\n world\r\n0\r\nx-sum: 11\r\n\r\n")44  let chunkedTransport = Transport.create(Transport.defaults())45  let chunked = shape(&fetched(&chunkedTransport, &chunkedServer))46  let chunkedAgain = shape(&fetched(&chunkedTransport, &chunkedServer))47  let chunkedStats = Transport.statistics(&chunkedTransport)48  let tooBigChunks = fetched(&Transport.create(Transport.defaults().withMaxResponseContentLength(4)), &chunkedServer)4950  let interim = answering("HTTP/1.1 100 Continue\r\n\r\nHTTP/1.1 200 OK\r\ncontent-length: 2\r\n\r\nok")51  let continued = shape(&fetched(&Transport.create(Transport.defaults()), &interim))5253  let closing = Loopback.scripted(fn(connection: Net.Connection, _number: Int) -> () {54      let _read = Loopback.readRequest(&connection)55      Loopback.write(&connection, "HTTP/1.0 200 OK\r\n\r\nuntil the end")56      Loopback.hangUp(&connection)57    })58  let closingTransport = Transport.create(Transport.defaults())59  let untilClose = shape(&fetched(&closingTransport, &closing))60  let closeStats = Transport.statistics(&closingTransport)6162  let staleServer = Loopback.scripted(fn(connection: Net.Connection, _number: Int) -> () {63      let _read = Loopback.readRequest(&connection)64      Loopback.write(&connection, "HTTP/1.1 200 OK\r\ncontent-length: 5\r\n\r\nfresh")65      Loopback.hangUp(&connection)66    })67  let staleTransport = Transport.create(Transport.defaults())68  let beforeStale = shape(&fetched(&staleTransport, &staleServer))69  let afterStale = shape(&fetched(&staleTransport, &staleServer))7071  let streamedServer = answering("HTTP/1.1 200 OK\r\ncontent-length: 11\r\n\r\nhello world")72  let pieces = Sync.cell(ByteSeq.empty())73  let streamed = Transport.send(&Transport.create(Transport.defaults()))(Request.get(streamedServer.base + "/").withReceiver(fn(piece: Bytes) -> Bool {74        let held = Result.unwrapOr(Sync.get(&pieces), ByteSeq.empty())75        let _kept = Sync.set(&pieces, held.concat(piece))76        true77      }), Cancel.expiring(2000))78  let stoppedTransport = Transport.create(Transport.defaults())79  let stopped = Transport.send(&stoppedTransport)(Request.get(chunkedServer.base + "/").withReceiver(|_piece: Bytes| false), Cancel.expiring(2000))8081  let garbage = fetched(&Transport.create(Transport.defaults()), &answering("NOT HTTP\r\n\r\n"))82  let badChunk = fetched(&Transport.create(Transport.defaults()), &answering("HTTP/1.1 200 OK\r\ntransfer-encoding: chunked\r\n\r\nzz\r\n"))83  let cut = Loopback.scripted(fn(connection: Net.Connection, _number: Int) -> () {84      let _read = Loopback.readRequest(&connection)85      Loopback.write(&connection, "HTTP/1.1 200 OK\r\ncontent-length: 10\r\n\r\nshort")86      Loopback.hangUp(&connection)87    })88  let ended = fetched(&Transport.create(Transport.defaults()), &cut)89  let silent = Loopback.scripted(fn(connection: Net.Connection, _number: Int) -> () {90      let _read = Loopback.readRequest(&connection)91      let _held = Loopback.readRequest(&connection)92    })93  let unanswered = Transport.send(&Transport.create(Transport.defaults()))(Request.get(silent.base + "/"), Cancel.expiring(100))94  let stopper = Cancel.token()95  let abortTransport = Transport.create(Transport.defaults())96  let _canceller = Concurrent.start(fn() -> () {97      let _waited = Concurrent.sleep(60)98      let _cancelled = Cancel.cancel(&stopper, "user left")99    })100  let abortBegan = clock()101  let aborted = Transport.send(&abortTransport)(Request.get(silent.base + "/"), stopper)102  let abortTook = clock() - abortBegan103  let headOnly = Transport.send(&Transport.create(Transport.defaults()))(Request.head(streamedServer.base + "/"), Cancel.expiring(2000))104105  let uploadServer = Loopback.scripted(fn(connection: Net.Connection, _number: Int) -> () {106      var received = ""107      var going = true108      while going && !received.contains("\r\n0\r\n\r\n") {109        match Net.receiveWithin(&connection, 65536, 2000) {110          case Ok(Some(piece)) => { received = received + Result.unwrapOr(ByteSeq.toText(&piece), "") }111          case _ => { going = false }112        }113      }114      let status = if received.startsWith("POST /moved") { "307 Temporary Redirect\r\nlocation: /elsewhere" } else { "200 OK" }115      let body = received.split("\r\n\r\n").slice(1, 9).join("\r\n\r\n")116      Loopback.write(&connection, "HTTP/1.1 " + status + "\r\ncontent-length: " + show(body.toBytes().length()) + "\r\nx-chunked: " + show(received.toLower().contains("transfer-encoding: chunked")) + "\r\n\r\n" + body)117      Loopback.hangUp(&connection)118    })119  let pieces2 = Sync.cell(["ab", "", "cde"])120  let producer = fn() -> Option[Bytes] {121    match Result.unwrapOr(Sync.get(&pieces2), []) {122      case [first, ..rest] => {123        let _kept = Sync.set(&pieces2, rest)124        Some(first.toBytes())125      }126      case _ => None127    }128  }129  let uploaded = Transport.send(&Transport.create(Transport.defaults()))(Request.post(uploadServer.base + "/up", Content.streamed("text/plain", producer)), Cancel.expiring(2000))130  let movedUpload = Transport.send(&Transport.create(Transport.defaults()))(Request.post(uploadServer.base + "/moved", Content.streamed("text/plain", || None)), Cancel.expiring(2000))131  Loopback.close(&uploadServer)132  for server in [chunkedServer, interim, closing, staleServer, streamedServer, cut, silent] { Loopback.close(&server) }133134  let checks = Test.suite("Transport.Exchange", &[135      Test.equals("a chunked body is joined and its trailers kept", &chunked, &Ok((200, "hello world", [("x-sum", "11")]))),136      Test.equals("a chunked connection is reused", &(chunkedAgain, chunkedStats.opened, chunkedStats.reused), &(Ok((200, "hello world", [("x-sum", "11")])), 1, 1)),137      Test.equals("a chunked body over the limit is refused", &Result.map(tooBigChunks, |r: Response.Response| r.status.code), &Err(HttpClient.ContentTooLarge(4))),138      Test.equals("an interim response is skipped", &continued, &Ok((200, "ok", []))),139      Test.equals("a body ending with the connection is read to its end", &untilClose, &Ok((200, "until the end", []))),140      Test.equals("a connection that ended a body is not pooled", &(closeStats.idle, closeStats.closed), &(0, 1)),141      Test.equals("the first request on a connection the server then closes answers", &beforeStale, &Ok((200, "fresh", []))),142      Test.equals("a pooled connection the server closed is replaced and the request sent again", &(afterStale, Loopback.connections(&staleServer)), &(Ok((200, "fresh", [])), 2)),143      Test.equals("a receiver gets the body and the response keeps none", &(Result.map(streamed, |r: Response.Response| Response.bytes(&r).length()), Result.unwrapOr(Sync.get(&pieces), ByteSeq.empty()).toText()), &(Ok(0), Some("hello world"))),144      Test.equals("a receiver that stops reading closes the connection", &(Result.map(stopped, |r: Response.Response| r.status.code), Transport.statistics(&stoppedTransport).idle), &(Ok(200), 0)),145      Test.that("a response that is not HTTP is a protocol failure", match garbage {146          case Err(HttpClient.Protocol(_)) => true147          case _ => false148        }),149      Test.that("a malformed chunk is a protocol failure", match badChunk {150          case Err(HttpClient.Protocol(_)) => true151          case _ => false152        }),153      Test.equals("a body cut short ends the response", &Result.map(ended, |r: Response.Response| r.status.code), &Err(HttpClient.ResponseEnded)),154      Test.that("a server that never answers times out", match unanswered {155          case Err(HttpClient.TimedOut(_)) => true156          case _ => false157        }),158      Test.equals("a streamed body is sent in chunks", &Result.map(uploaded, |r: Response.Response| (Result.unwrapOr(Response.text(&r), ""), Response.header(&r, "x-chunked"))), &Ok(("2\r\nab\r\n3\r\ncde\r\n0\r\n\r\n", Some("true")))),159      Test.equals("a redirect that would send a streamed body again is answered as it is", &Result.map(movedUpload, |r: Response.Response| r.status.code), &Ok(307)),160      Test.equals("a request cancelled while waiting for its response stops at once and keeps no connection", &(Result.map(aborted, |r: Response.Response| r.status.code), abortTook < 1000, Transport.statistics(&abortTransport).idle), &(Err(HttpClient.Cancelled("user left")), true, 0)),161      Test.equals("a head request reads no body", &Result.map(headOnly, |r: Response.Response| (r.status.code, r.request.method, Response.bytes(&r).length())), &Ok((200, Http.Head, 0)))162    ])163  let ran = Test.run(&checks)164  for failure in Test.failuresOf(&ran) {165    let _reported = Io.writeErrorLine(failure)166  }167  Test.report(&ran)168}169