go-xmpp/client_internal_test.go

609 lines
17 KiB
Go
Raw Normal View History

package xmpp
import (
"bytes"
2020-03-06 15:44:01 +00:00
"encoding/xml"
"fmt"
"gosrc.io/xmpp/stanza"
"strconv"
"testing"
2020-03-06 15:44:01 +00:00
"time"
)
const (
streamManagementID = "test-stream_management-id"
)
func TestClient_Send(t *testing.T) {
buffer := bytes.NewBufferString("")
client := Client{}
data := []byte("https://da.wikipedia.org/wiki/J%C3%A6vnd%C3%B8gn")
if err := client.sendWithWriter(buffer, data); err != nil {
t.Errorf("Writing failed: %v", err)
}
if buffer.String() != string(data) {
t.Errorf("Incorrect value sent to buffer: '%s'", buffer.String())
}
}
2020-03-06 15:44:01 +00:00
// Stream management test.
// Connection is established, then the server sends supported features and so on.
// After the bind, client attempts a stream management enablement, and server replies in kind.
func Test_StreamManagement(t *testing.T) {
serverDone := make(chan struct{})
clientDone := make(chan struct{})
client, mock := initSrvCliForResumeTests(t, func(t *testing.T, sc *ServerConn) {
checkClientOpenStream(t, sc)
sendStreamFeatures(t, sc) // Send initial features
readAuth(t, sc.decoder)
sc.connection.Write([]byte("<success xmlns=\"urn:ietf:params:xml:ns:xmpp-sasl\"/>"))
checkClientOpenStream(t, sc) // Reset stream
sendFeaturesStreamManagment(t, sc) // Send post auth features
bind(t, sc)
enableStreamManagement(t, sc, false, true)
serverDone <- struct{}{}
}, testClientStreamManagement, true, true)
go func() {
var state SMState
var err error
// Client is ok, we now open XMPP session
if client.Session, err = NewSession(client, state); err != nil {
t.Fatalf("failed to open XMPP session: %s", err)
}
clientDone <- struct{}{}
}()
waitForEntity(t, clientDone)
waitForEntity(t, serverDone)
mock.Stop()
}
// Absence of stream management test.
// Connection is established, then the server sends supported features and so on.
// Client has stream management disabled in its config, and should not ask for it. Server is not set up to reply.
func Test_NoStreamManagement(t *testing.T) {
serverDone := make(chan struct{})
clientDone := make(chan struct{})
// Setup Mock server
client, mock := initSrvCliForResumeTests(t, func(t *testing.T, sc *ServerConn) {
checkClientOpenStream(t, sc)
sendStreamFeatures(t, sc) // Send initial features
readAuth(t, sc.decoder)
sc.connection.Write([]byte("<success xmlns=\"urn:ietf:params:xml:ns:xmpp-sasl\"/>"))
checkClientOpenStream(t, sc) // Reset stream
sendFeaturesNoStreamManagment(t, sc) // Send post auth features
bind(t, sc)
serverDone <- struct{}{}
}, testClientStreamManagement, true, false)
go func() {
var state SMState
// Client is ok, we now open XMPP session
var err error
if client.Session, err = NewSession(client, state); err != nil {
t.Fatalf("failed to open XMPP session: %s", err)
}
clientDone <- struct{}{}
}()
waitForEntity(t, clientDone)
waitForEntity(t, serverDone)
mock.Stop()
}
func Test_StreamManagementNotSupported(t *testing.T) {
serverDone := make(chan struct{})
clientDone := make(chan struct{})
client, mock := initSrvCliForResumeTests(t, func(t *testing.T, sc *ServerConn) {
checkClientOpenStream(t, sc)
sendStreamFeatures(t, sc) // Send initial features
readAuth(t, sc.decoder)
sc.connection.Write([]byte("<success xmlns=\"urn:ietf:params:xml:ns:xmpp-sasl\"/>"))
checkClientOpenStream(t, sc) // Reset stream
sendFeaturesNoStreamManagment(t, sc) // Send post auth features
bind(t, sc)
serverDone <- struct{}{}
}, testClientStreamManagement, true, true)
go func() {
var state SMState
var err error
// Client is ok, we now open XMPP session
if client.Session, err = NewSession(client, state); err != nil {
t.Fatalf("failed to open XMPP session: %s", err)
}
clientDone <- struct{}{}
}()
// Wait for client
waitForEntity(t, clientDone)
// Check if client got a positive stream management response from the server
if client.Session.Features.DoesStreamManagement() {
t.Fatalf("server does not provide stream management")
}
// Wait for server
waitForEntity(t, serverDone)
mock.Stop()
}
func Test_StreamManagementNoResume(t *testing.T) {
serverDone := make(chan struct{})
clientDone := make(chan struct{})
client, mock := initSrvCliForResumeTests(t, func(t *testing.T, sc *ServerConn) {
checkClientOpenStream(t, sc)
sendStreamFeatures(t, sc) // Send initial features
readAuth(t, sc.decoder)
sc.connection.Write([]byte("<success xmlns=\"urn:ietf:params:xml:ns:xmpp-sasl\"/>"))
checkClientOpenStream(t, sc) // Reset stream
sendFeaturesStreamManagment(t, sc) // Send post auth features
bind(t, sc)
enableStreamManagement(t, sc, false, false)
serverDone <- struct{}{}
}, testClientStreamManagement, true, true)
go func() {
var state SMState
var err error
// Client is ok, we now open XMPP session
if client.Session, err = NewSession(client, state); err != nil {
t.Fatalf("failed to open XMPP session: %s", err)
}
clientDone <- struct{}{}
}()
waitForEntity(t, clientDone)
if IsStreamResumable(client) {
t.Fatalf("server does not support resumption but client says stream is resumable")
}
waitForEntity(t, serverDone)
mock.Stop()
}
func Test_StreamManagementResume(t *testing.T) {
2020-03-09 09:55:22 +00:00
serverDone := make(chan struct{})
clientDone := make(chan struct{})
2020-03-06 15:44:01 +00:00
// Setup Mock server
mock := ServerMock{}
mock.Start(t, testXMPPAddress, func(t *testing.T, sc *ServerConn) {
checkClientOpenStream(t, sc)
sendStreamFeatures(t, sc) // Send initial features
readAuth(t, sc.decoder)
sc.connection.Write([]byte("<success xmlns=\"urn:ietf:params:xml:ns:xmpp-sasl\"/>"))
checkClientOpenStream(t, sc) // Reset stream
sendFeaturesStreamManagment(t, sc) // Send post auth features
bind(t, sc)
enableStreamManagement(t, sc, false, true)
discardPresence(t, sc)
2020-03-09 09:55:22 +00:00
serverDone <- struct{}{}
2020-03-06 15:44:01 +00:00
})
// Test / Check result
config := Config{
TransportConfiguration: TransportConfiguration{
Address: testXMPPAddress,
},
Jid: "test@localhost",
Credential: Password("test"),
Insecure: true,
StreamManagementEnable: true,
streamManagementResume: true} // Enable stream management
var client *Client
router := NewRouter()
client, err := NewClient(&config, router, clientDefaultErrorHandler)
if err != nil {
t.Errorf("connect create XMPP client: %s", err)
}
2020-03-09 09:55:22 +00:00
// =================================================================
// Connect client, then disconnect it so we can resume the session
go func() {
err = client.Connect()
if err != nil {
t.Fatalf("could not connect client to mock server: %s", err)
}
clientDone <- struct{}{}
}()
waitForEntity(t, clientDone)
2020-03-06 15:44:01 +00:00
2020-03-09 09:55:22 +00:00
// ===========================================================================================
// Check that the client correctly went into "disconnected" state, after being disconnected
2020-03-06 15:44:01 +00:00
statusCorrectChan := make(chan struct{})
kill := make(chan struct{})
transp, ok := client.transport.(*XMPPTransport)
if !ok {
t.Fatalf("problem with client transport ")
}
transp.conn.Close()
2020-03-09 09:55:22 +00:00
waitForEntity(t, serverDone)
2020-03-06 15:44:01 +00:00
mock.Stop()
go checkClientResumeStatus(client, statusCorrectChan, kill)
select {
case <-statusCorrectChan:
// Test passed
case <-time.After(5 * time.Second):
kill <- struct{}{}
t.Fatalf("Client is not in disconnected state while it should be. Timed out")
}
// Check if the client can have its connection resumed using its state but also its configuration
if !IsStreamResumable(client) {
t.Fatalf("should support resumption")
}
// Reboot server. We need to make a new one because (at least for now) the mock server can only have one handler
// and they should be different between a first connection and a stream resume since exchanged messages
// are different (See XEP-0198)
mock2 := ServerMock{}
mock2.Start(t, testXMPPAddress, func(t *testing.T, sc *ServerConn) {
// Reconnect
checkClientOpenStream(t, sc)
sendStreamFeatures(t, sc) // Send initial features
readAuth(t, sc.decoder)
sc.connection.Write([]byte("<success xmlns=\"urn:ietf:params:xml:ns:xmpp-sasl\"/>"))
checkClientOpenStream(t, sc) // Reset stream
sendFeaturesStreamManagment(t, sc) // Send post auth features
resumeStream(t, sc)
2020-03-09 09:55:22 +00:00
serverDone <- struct{}{}
2020-03-06 15:44:01 +00:00
})
// Reconnect
2020-03-09 09:55:22 +00:00
go func() {
err = client.Resume()
if err != nil {
t.Fatalf("could not connect client to mock server: %s", err)
}
clientDone <- struct{}{}
}()
2020-03-06 15:44:01 +00:00
2020-03-09 09:55:22 +00:00
waitForEntity(t, clientDone)
waitForEntity(t, serverDone)
2020-03-06 15:44:01 +00:00
2020-03-06 15:44:01 +00:00
mock2.Stop()
}
func Test_StreamManagementFail(t *testing.T) {
2020-03-09 09:55:22 +00:00
serverDone := make(chan struct{})
clientDone := make(chan struct{})
2020-03-06 15:44:01 +00:00
// Setup Mock server
mock := ServerMock{}
mock.Start(t, testXMPPAddress, func(t *testing.T, sc *ServerConn) {
checkClientOpenStream(t, sc)
sendStreamFeatures(t, sc) // Send initial features
readAuth(t, sc.decoder)
sc.connection.Write([]byte("<success xmlns=\"urn:ietf:params:xml:ns:xmpp-sasl\"/>"))
checkClientOpenStream(t, sc) // Reset stream
sendFeaturesStreamManagment(t, sc) // Send post auth features
bind(t, sc)
enableStreamManagement(t, sc, true, true)
2020-03-09 09:55:22 +00:00
serverDone <- struct{}{}
2020-03-06 15:44:01 +00:00
})
// Test / Check result
config := Config{
TransportConfiguration: TransportConfiguration{
Address: testXMPPAddress,
},
Jid: "test@localhost",
Credential: Password("test"),
Insecure: true,
StreamManagementEnable: true,
streamManagementResume: true} // Enable stream management
var client *Client
router := NewRouter()
client, err := NewClient(&config, router, clientDefaultErrorHandler)
if err != nil {
t.Errorf("connect create XMPP client: %s", err)
}
var state SMState
2020-03-09 09:55:22 +00:00
go func() {
_, err = client.transport.Connect()
if err != nil {
return
}
2020-03-06 15:44:01 +00:00
2020-03-09 09:55:22 +00:00
// Client is ok, we now open XMPP session
if client.Session, err = NewSession(client, state); err == nil {
t.Fatalf("test is supposed to err")
}
if client.Session.SMState.StreamErrorGroup == nil {
t.Fatalf("error was not stored correctly in session state")
}
clientDone <- struct{}{}
}()
waitForEntity(t, serverDone)
waitForEntity(t, clientDone)
2020-03-06 15:44:01 +00:00
mock.Stop()
}
func Test_SendStanzaQueueWithSM(t *testing.T) {
2020-03-09 09:55:22 +00:00
serverDone := make(chan struct{})
clientDone := make(chan struct{})
2020-03-06 15:44:01 +00:00
// Setup Mock server
mock := ServerMock{}
mock.Start(t, testXMPPAddress, func(t *testing.T, sc *ServerConn) {
checkClientOpenStream(t, sc)
sendStreamFeatures(t, sc) // Send initial features
readAuth(t, sc.decoder)
sc.connection.Write([]byte("<success xmlns=\"urn:ietf:params:xml:ns:xmpp-sasl\"/>"))
checkClientOpenStream(t, sc) // Reset stream
sendFeaturesStreamManagment(t, sc) // Send post auth features
bind(t, sc)
enableStreamManagement(t, sc, false, true)
// Ignore the initial presence sent to the server by the client so we can move on to the next packet.
discardPresence(t, sc)
// Used here to silently discard the IQ sent by the client, in order to later trigger a resend
skipPacket(t, sc)
// Respond to the client ACK request with a number of processed stanzas of 0. This should trigger a resend
// of previously ignored stanza to the server, which this handler element will be expecting.
2020-03-09 09:55:22 +00:00
respondWithAck(t, sc, 0)
serverDone <- struct{}{}
2020-03-06 15:44:01 +00:00
})
// Test / Check result
config := Config{
TransportConfiguration: TransportConfiguration{
Address: testXMPPAddress,
},
Jid: "test@localhost",
Credential: Password("test"),
Insecure: true,
StreamManagementEnable: true,
streamManagementResume: true} // Enable stream management
var client *Client
router := NewRouter()
client, err := NewClient(&config, router, clientDefaultErrorHandler)
if err != nil {
t.Errorf("connect create XMPP client: %s", err)
}
2020-03-09 09:55:22 +00:00
go func() {
err = client.Connect()
2020-03-06 15:44:01 +00:00
2020-03-09 09:55:22 +00:00
client.SendRaw(`<iq id='ls72g593' type='get'>
2020-03-06 15:44:01 +00:00
<query xmlns='jabber:iq:roster'/>
</iq>
`)
2020-03-09 09:55:22 +00:00
// Last stanza was discarded silently by the server. Let's ask an ack for it. This should trigger resend as the server
// will respond with an acknowledged number of stanzas of 0.
r := stanza.SMRequest{}
client.Send(r)
clientDone <- struct{}{}
}()
waitForEntity(t, serverDone)
waitForEntity(t, clientDone)
2020-03-06 15:44:01 +00:00
mock.Stop()
}
//========================================================================
// Helper functions for tests
func skipPacket(t *testing.T, sc *ServerConn) {
var p stanza.IQ
se, err := stanza.NextStart(sc.decoder)
if err != nil {
t.Fatalf("cannot read packet: %s", err)
return
}
if err := sc.decoder.DecodeElement(&p, &se); err != nil {
t.Fatalf("cannot decode packet: %s", err)
return
}
}
2020-03-09 09:55:22 +00:00
func respondWithAck(t *testing.T, sc *ServerConn, h int) {
2020-03-06 15:44:01 +00:00
// Mock server reads the ack request
var p stanza.SMRequest
se, err := stanza.NextStart(sc.decoder)
if err != nil {
t.Fatalf("cannot read packet: %s", err)
return
}
if err := sc.decoder.DecodeElement(&p, &se); err != nil {
t.Fatalf("cannot decode packet: %s", err)
return
}
// Mock server sends the ack response
a := stanza.SMAnswer{
H: uint(h),
}
data, err := xml.Marshal(a)
_, err = sc.connection.Write(data)
if err != nil {
t.Fatalf("failed to send response ack")
}
// Mock server reads the re-sent stanza that was previously discarded intentionally
var p2 stanza.IQ
nse, err := stanza.NextStart(sc.decoder)
if err != nil {
t.Fatalf("cannot read packet: %s", err)
return
}
if err := sc.decoder.DecodeElement(&p2, &nse); err != nil {
t.Fatalf("cannot decode packet: %s", err)
return
}
}
func sendFeaturesStreamManagment(t *testing.T, sc *ServerConn) {
// This is a basic server, supporting only 2 features after auth: stream management & session binding
features := `<stream:features>
<bind xmlns='urn:ietf:params:xml:ns:xmpp-bind'/>
<sm xmlns='urn:xmpp:sm:3'/>
</stream:features>`
if _, err := fmt.Fprintln(sc.connection, features); err != nil {
t.Fatalf("cannot send stream feature: %s", err)
}
}
func sendFeaturesNoStreamManagment(t *testing.T, sc *ServerConn) {
// This is a basic server, supporting only 2 features after auth: stream management & session binding
features := `<stream:features>
<bind xmlns='urn:ietf:params:xml:ns:xmpp-bind'/>
</stream:features>`
if _, err := fmt.Fprintln(sc.connection, features); err != nil {
t.Fatalf("cannot send stream feature: %s", err)
}
}
// enableStreamManagement is a function for the mock server that can either mock a successful session, or fail depending on
// the value of the "fail" boolean. True means the session should fail.
func enableStreamManagement(t *testing.T, sc *ServerConn, fail bool, resume bool) {
// Decode element into pointer storage
var ed stanza.SMEnable
se, err := stanza.NextStart(sc.decoder)
if err != nil {
t.Fatalf("cannot read stream management enable: %s", err)
return
}
if err := sc.decoder.DecodeElement(&ed, &se); err != nil {
t.Fatalf("cannot decode stream management enable: %s", err)
return
}
if fail {
f := stanza.SMFailed{
H: nil,
StreamErrorGroup: &stanza.UnexpectedRequest{},
}
data, err := xml.Marshal(f)
if err != nil {
t.Fatalf("failed to marshall error response: %s", err)
}
sc.connection.Write(data)
} else {
e := &stanza.SMEnabled{
Resume: strconv.FormatBool(resume),
Id: streamManagementID,
}
data, err := xml.Marshal(e)
if err != nil {
t.Fatalf("failed to marshall error response: %s", err)
}
sc.connection.Write(data)
}
}
func resumeStream(t *testing.T, sc *ServerConn) {
h := uint(0)
response := stanza.SMResumed{
PrevId: streamManagementID,
H: &h,
}
data, err := xml.Marshal(response)
if err != nil {
t.Fatalf("failed to marshall stream management enabled response : %s", err)
}
writtenChan := make(chan struct{})
go func() {
sc.connection.Write(data)
writtenChan <- struct{}{}
}()
select {
case <-writtenChan:
// We're done here
return
case <-time.After(defaultTimeout):
t.Fatalf("failed to write enabled nonza to client")
}
}
func checkClientResumeStatus(client *Client, statusCorrectChan chan struct{}, killChan chan struct{}) {
for {
if client.CurrentState.getState() == StateDisconnected {
statusCorrectChan <- struct{}{}
}
select {
case <-killChan:
return
case <-time.After(time.Millisecond * 10):
// Keep checking status value
}
}
}
func initSrvCliForResumeTests(t *testing.T, serverHandler func(*testing.T, *ServerConn), port int, StreamManagementEnable, StreamManagementResume bool) (*Client, *ServerMock) {
mock := &ServerMock{}
testServerAddress := fmt.Sprintf("%s:%d", testClientDomain, port)
mock.Start(t, testServerAddress, serverHandler)
config := Config{
TransportConfiguration: TransportConfiguration{
Address: testServerAddress,
},
Jid: "test@localhost",
Credential: Password("test"),
Insecure: true,
StreamManagementEnable: StreamManagementEnable,
streamManagementResume: StreamManagementResume}
var client *Client
var err error
router := NewRouter()
if client, err = NewClient(&config, router, clientDefaultErrorHandler); err != nil {
t.Fatalf("connect create XMPP client: %s", err)
}
if _, err = client.transport.Connect(); err != nil {
t.Fatalf("XMPP connection failed: %s", err)
}
return client, mock
}
func waitForEntity(t *testing.T, entityDone chan struct{}) {
select {
case <-entityDone:
case <-time.After(defaultTimeout):
t.Fatalf("test timed out")
}
}