nim-libp2p/tests/testmplex.nim

329 lines
12 KiB
Nim
Raw Normal View History

2019-09-09 17:33:32 +00:00
import unittest, sequtils, sugar, strformat, options, strformat
import chronos, nimcrypto/utils, chronicles
2019-09-04 03:08:51 +00:00
import ../libp2p/connection,
../libp2p/stream/lpstream,
2019-09-04 06:40:11 +00:00
../libp2p/stream/bufferstream,
2019-09-06 07:13:03 +00:00
../libp2p/transports/tcptransport,
../libp2p/transports/transport,
../libp2p/protocols/identify,
2019-09-04 03:08:51 +00:00
../libp2p/multiaddress,
../libp2p/muxers/mplex/mplex,
../libp2p/muxers/mplex/coder,
../libp2p/muxers/mplex/types,
2019-09-09 17:33:32 +00:00
../libp2p/muxers/mplex/channel
2019-09-03 20:40:51 +00:00
suite "Mplex":
test "encode header with channel id 0":
proc testEncodeHeader(): Future[bool] {.async.} =
proc encHandler(msg: seq[byte]) {.async.} =
check msg == fromHex("000873747265616d2031")
let stream = newBufferStream(encHandler)
let conn = newConnection(stream)
2019-09-12 02:10:38 +00:00
await conn.writeMsg(0, MessageType.New, cast[seq[byte]]("stream 1"))
result = true
check:
waitFor(testEncodeHeader()) == true
test "encode header with channel id other than 0":
2019-09-04 01:43:57 +00:00
proc testEncodeHeader(): Future[bool] {.async.} =
proc encHandler(msg: seq[byte]) {.async.} =
check msg == fromHex("88010873747265616d2031")
2019-09-03 20:40:51 +00:00
let stream = newBufferStream(encHandler)
let conn = newConnection(stream)
2019-09-12 02:10:38 +00:00
await conn.writeMsg(17, MessageType.New, cast[seq[byte]]("stream 1"))
2019-09-04 01:43:57 +00:00
result = true
2019-09-03 20:40:51 +00:00
2019-09-04 01:43:57 +00:00
check:
waitFor(testEncodeHeader()) == true
2019-09-03 20:40:51 +00:00
test "encode header and body with channel id 0":
proc testEncodeHeaderBody(): Future[bool] {.async.} =
var step = 0
proc encHandler(msg: seq[byte]) {.async.} =
check msg == fromHex("020873747265616d2031")
let stream = newBufferStream(encHandler)
let conn = newConnection(stream)
2019-09-12 02:10:38 +00:00
await conn.writeMsg(0, MessageType.MsgOut, cast[seq[byte]]("stream 1"))
result = true
check:
waitFor(testEncodeHeaderBody()) == true
test "encode header and body with channel id other than 0":
proc testEncodeHeaderBody(): Future[bool] {.async.} =
var step = 0
proc encHandler(msg: seq[byte]) {.async.} =
check msg == fromHex("8a010873747265616d2031")
let stream = newBufferStream(encHandler)
let conn = newConnection(stream)
2019-09-12 02:10:38 +00:00
await conn.writeMsg(17, MessageType.MsgOut, cast[seq[byte]]("stream 1"))
await conn.close()
result = true
check:
waitFor(testEncodeHeaderBody()) == true
test "decode header with channel id 0":
2019-09-04 01:43:57 +00:00
proc testDecodeHeader(): Future[bool] {.async.} =
proc encHandler(msg: seq[byte]) {.async.} = discard
let stream = newBufferStream(encHandler)
let conn = newConnection(stream)
await stream.pushTo(fromHex("000873747265616d2031"))
2019-09-08 06:32:41 +00:00
let msg = await conn.readMsg()
2019-09-03 20:40:51 +00:00
2019-09-08 06:32:41 +00:00
if msg.isSome:
check msg.get().id == 0
check msg.get().msgType == MessageType.New
result = true
2019-09-03 20:40:51 +00:00
2019-09-04 01:43:57 +00:00
check:
waitFor(testDecodeHeader()) == true
test "decode header and body with channel id 0":
proc testDecodeHeader(): Future[bool] {.async.} =
proc encHandler(msg: seq[byte]) {.async.} = discard
let stream = newBufferStream(encHandler)
let conn = newConnection(stream)
await stream.pushTo(fromHex("021668656C6C6F2066726F6D206368616E6E656C20302121"))
2019-09-08 06:32:41 +00:00
let msg = await conn.readMsg()
2019-09-08 06:32:41 +00:00
if msg.isSome:
check msg.get().id == 0
check msg.get().msgType == MessageType.MsgOut
check cast[string](msg.get().data) == "hello from channel 0!!"
result = true
check:
waitFor(testDecodeHeader()) == true
test "decode header and body with channel id other than 0":
proc testDecodeHeader(): Future[bool] {.async.} =
proc encHandler(msg: seq[byte]) {.async.} = discard
let stream = newBufferStream(encHandler)
let conn = newConnection(stream)
await stream.pushTo(fromHex("8a011668656C6C6F2066726F6D206368616E6E656C20302121"))
2019-09-08 06:32:41 +00:00
let msg = await conn.readMsg()
2019-09-08 06:32:41 +00:00
if msg.isSome:
check msg.get().id == 17
check msg.get().msgType == MessageType.MsgOut
check cast[string](msg.get().data) == "hello from channel 0!!"
result = true
check:
waitFor(testDecodeHeader()) == true
2019-09-04 06:40:11 +00:00
test "e2e - read/write initiator":
proc testNewStream(): Future[bool] {.async.} =
let ma: MultiAddress = Multiaddress.init("/ip4/127.0.0.1/tcp/53380")
proc connHandler(conn: Connection) {.async, gcsafe.} =
proc handleMplexListen(stream: Connection) {.async, gcsafe.} =
await stream.writeLp("Hello from stream!")
await stream.close()
let mplexListen = newMplex(conn)
mplexListen.streamHandler = handleMplexListen
2019-09-04 06:40:11 +00:00
await mplexListen.handle()
let transport1: TcpTransport = newTransport(TcpTransport)
await transport1.listen(ma, connHandler)
let transport2: TcpTransport = newTransport(TcpTransport)
let conn = await transport2.dial(ma)
let mplexDial = newMplex(conn)
2019-09-04 06:40:11 +00:00
let dialFut = mplexDial.handle()
let stream = await mplexDial.newStream("DIALER")
let msg = cast[string](await stream.readLp())
check msg == "Hello from stream!"
2019-09-04 06:40:11 +00:00
await conn.close()
# await dialFut
2019-09-04 06:40:11 +00:00
result = true
check:
waitFor(testNewStream()) == true
test "e2e - read/write receiver":
proc testNewStream(): Future[bool] {.async.} =
let ma: MultiAddress = Multiaddress.init("/ip4/127.0.0.1/tcp/53381")
proc connHandler(conn: Connection) {.async, gcsafe.} =
proc handleMplexListen(stream: Connection) {.async, gcsafe.} =
let msg = await stream.readLp()
check cast[string](msg) == "Hello from stream!"
await stream.close()
let mplexListen = newMplex(conn)
mplexListen.streamHandler = handleMplexListen
2019-09-04 06:40:11 +00:00
await mplexListen.handle()
let transport1: TcpTransport = newTransport(TcpTransport)
await transport1.listen(ma, connHandler)
let transport2: TcpTransport = newTransport(TcpTransport)
let conn = await transport2.dial(ma)
let mplexDial = newMplex(conn)
2019-09-04 06:40:11 +00:00
let stream = await mplexDial.newStream()
await stream.writeLp("Hello from stream!")
await conn.close()
result = true
check:
waitFor(testNewStream()) == true
2019-09-05 03:02:22 +00:00
test "e2e - multiple streams":
proc testNewStream(): Future[bool] {.async.} =
let ma: MultiAddress = Multiaddress.init("/ip4/127.0.0.1/tcp/53382")
var count = 1
2019-09-09 17:52:08 +00:00
var listenConn: Connection
2019-09-05 03:02:22 +00:00
proc connHandler(conn: Connection) {.async, gcsafe.} =
proc handleMplexListen(stream: Connection) {.async, gcsafe.} =
let msg = await stream.readLp()
check cast[string](msg) == &"stream {count}!"
2019-09-05 03:02:22 +00:00
count.inc
await stream.close()
2019-09-09 17:52:08 +00:00
listenConn = conn
2019-09-05 03:02:22 +00:00
let mplexListen = newMplex(conn)
mplexListen.streamHandler = handleMplexListen
2019-09-09 17:52:08 +00:00
mplexListen.handle()
.addCallback(proc(udata: pointer) =
debug "handle completed")
2019-09-05 03:02:22 +00:00
let transport1: TcpTransport = newTransport(TcpTransport)
await transport1.listen(ma, connHandler)
let transport2: TcpTransport = newTransport(TcpTransport)
let conn = await transport2.dial(ma)
let mplexDial = newMplex(conn)
for i in 1..<10:
2019-09-05 03:02:22 +00:00
let stream = await mplexDial.newStream()
await stream.writeLp(&"stream {i}!")
await stream.close()
2019-09-05 03:02:22 +00:00
await conn.close()
2019-09-09 17:52:08 +00:00
await listenConn.close()
2019-09-05 03:02:22 +00:00
result = true
check:
waitFor(testNewStream()) == true
2019-09-08 06:32:41 +00:00
test "e2e - multiple read/write streams":
proc testNewStream(): Future[bool] {.async.} =
let ma: MultiAddress = Multiaddress.init("/ip4/127.0.0.1/tcp/53383")
var count = 1
var listenFut: Future[void]
2019-09-09 17:52:08 +00:00
var listenConn: Connection
2019-09-08 06:32:41 +00:00
proc connHandler(conn: Connection) {.async, gcsafe.} =
2019-09-09 17:52:08 +00:00
listenConn = conn
2019-09-08 06:32:41 +00:00
proc handleMplexListen(stream: Connection) {.async, gcsafe.} =
let msg = await stream.readLp()
check cast[string](msg) == &"stream {count} from dialer!"
await stream.writeLp(&"stream {count} from listener!")
count.inc
await stream.close()
let mplexListen = newMplex(conn)
mplexListen.streamHandler = handleMplexListen
listenFut = mplexListen.handle()
listenFut.addCallback(proc(udata: pointer) {.gcsafe.}
= debug "completed listener")
let transport1: TcpTransport = newTransport(TcpTransport)
await transport1.listen(ma, connHandler)
let transport2: TcpTransport = newTransport(TcpTransport)
let conn = await transport2.dial(ma)
let mplexDial = newMplex(conn)
let dialFut = mplexDial.handle()
dialFut.addCallback(proc(udata: pointer = nil) {.gcsafe.}
= debug "completed dialer")
for i in 1..10:
let stream = await mplexDial.newStream("dialer stream")
await stream.writeLp(&"stream {i} from dialer!")
let msg = await stream.readLp()
check cast[string](msg) == &"stream {i} from listener!"
await stream.close()
await conn.close()
2019-09-09 17:52:08 +00:00
await listenConn.close()
await allFutures(dialFut, listenFut)
2019-09-08 06:32:41 +00:00
result = true
check:
waitFor(testNewStream()) == true
2019-09-04 06:40:11 +00:00
test "half closed - channel should close for write":
proc testClosedForWrite(): Future[void] {.async.} =
proc writeHandler(data: seq[byte]) {.async, gcsafe.} = discard
let chann = newChannel(1, newConnection(newBufferStream(writeHandler)), true)
2019-09-04 06:40:11 +00:00
await chann.close()
await chann.write("Hello")
expect LPStreamClosedError:
waitFor(testClosedForWrite())
test "half closed - channel should close for read":
proc testClosedForRead(): Future[void] {.async.} =
proc writeHandler(data: seq[byte]) {.async, gcsafe.} = discard
let chann = newChannel(1, newConnection(newBufferStream(writeHandler)), true)
2019-09-04 16:40:05 +00:00
await chann.closedByRemote()
2019-09-04 06:40:11 +00:00
asyncDiscard chann.read()
expect LPStreamClosedError:
waitFor(testClosedForRead())
test "half closed - channel should close for read after eof":
proc testClosedForRead(): Future[void] {.async.} =
proc writeHandler(data: seq[byte]) {.async, gcsafe.} = discard
let chann = newChannel(1, newConnection(newBufferStream(writeHandler)), true)
2019-09-04 06:40:11 +00:00
2019-09-12 02:10:38 +00:00
await chann.pushTo(cast[seq[byte]]("Hello!"))
2019-09-04 06:40:11 +00:00
await chann.close()
let msg = await chann.read()
asyncDiscard chann.read()
expect LPStreamClosedError:
waitFor(testClosedForRead())
test "reset - channel should fail reading":
proc testResetRead(): Future[void] {.async.} =
proc writeHandler(data: seq[byte]) {.async, gcsafe.} = discard
let chann = newChannel(1, newConnection(newBufferStream(writeHandler)), true)
2019-09-04 06:40:11 +00:00
await chann.reset()
asyncDiscard chann.read()
expect LPStreamClosedError:
waitFor(testResetRead())
test "reset - channel should fail writing":
proc testResetWrite(): Future[void] {.async.} =
proc writeHandler(data: seq[byte]) {.async, gcsafe.} = discard
let chann = newChannel(1, newConnection(newBufferStream(writeHandler)), true)
2019-09-04 06:40:11 +00:00
await chann.reset()
asyncDiscard chann.read()
expect LPStreamClosedError:
waitFor(testResetWrite())
2019-09-04 06:51:16 +00:00
test "should not allow pushing data to channel when remote end closed":
proc testResetWrite(): Future[void] {.async.} =
proc writeHandler(data: seq[byte]) {.async, gcsafe.} = discard
let chann = newChannel(1, newConnection(newBufferStream(writeHandler)), true)
2019-09-04 16:40:05 +00:00
await chann.closedByRemote()
2019-09-04 06:51:16 +00:00
await chann.pushTo(@[byte(1)])
expect LPStreamClosedError:
waitFor(testResetWrite())