1"""
2Test various ways in which connections can fail.
3"""
4
5from twisted.words.xish import domish
6from twisted.words.protocols.jabber import xmlstream
7
8import dbus
9
10from servicetest import assertEquals
11from gabbletest import exec_test, GabbleAuthenticator
12import constants as cs
13import ns
14
15def test_network_error(q, bus, conn, stream):
16    conn.Connect()
17    q.expect('dbus-signal', signal='StatusChanged',
18        args=[cs.CONN_STATUS_CONNECTING, cs.CSR_REQUESTED])
19
20    # FIXME: this is G_IO_ERROR_FAILED, which we can't really map to anything
21    # better than NetworkError. The debug message says "Connection refused",
22    # so something, somewhere, ought to be able to do better, and give us
23    # enough information to produce cs.CONNECTION_REFUSED.
24    new = q.expect('dbus-signal', signal='ConnectionError')
25    assertEquals(cs.NETWORK_ERROR, new.args[0])
26
27    q.expect('dbus-signal', signal='StatusChanged',
28        args=[cs.CONN_STATUS_DISCONNECTED, cs.CSR_NETWORK_ERROR])
29
30def test_conflict_after_connect(q, bus, conn, stream):
31    stream.send_stream_error('conflict')
32
33    new = q.expect('dbus-signal', signal='ConnectionError')
34    assertEquals(cs.CONNECTION_REPLACED, new.args[0])
35
36    q.expect('dbus-signal', signal='StatusChanged',
37        args=[cs.CONN_STATUS_DISCONNECTED, cs.CSR_NAME_IN_USE])
38
39class StreamErrorAuthenticator(GabbleAuthenticator):
40    def __init__(self, stream_error):
41        GabbleAuthenticator.__init__(self, username='n/a', password='n/a')
42        self.__stream_error = stream_error
43
44    def streamStarted(self, root=None):
45        if root:
46            self.xmlstream.sid = root.getAttribute('id')
47
48        self.xmlstream.sendHeader()
49
50        no = domish.Element((xmlstream.NS_STREAMS, 'error'))
51        no.addElement((ns.STREAMS, self.__stream_error))
52        self.xmlstream.send(no)
53
54def test_stream_conflict_during_connect(q, bus, conn, stream):
55    conn.Connect()
56
57    q.expect('dbus-signal', signal='StatusChanged',
58        args=[cs.CONN_STATUS_CONNECTING, cs.CSR_REQUESTED])
59
60    new = q.expect('dbus-signal', signal='ConnectionError')
61    assertEquals(cs.ALREADY_CONNECTED, new.args[0])
62
63    old = q.expect('dbus-signal', signal='StatusChanged')
64    status, reason = old.args
65    assertEquals(cs.CONN_STATUS_DISCONNECTED, status)
66    assertEquals(cs.CSR_NAME_IN_USE, reason)
67
68def test_host_unknown(q, bus, conn, stream):
69    conn.Connect()
70
71    q.expect('dbus-signal', signal='StatusChanged',
72        args=[cs.CONN_STATUS_CONNECTING, cs.CSR_REQUESTED])
73
74    new = q.expect('dbus-signal', signal='ConnectionError')
75    assertEquals(cs.AUTHENTICATION_FAILED, new.args[0])
76
77    old = q.expect('dbus-signal', signal='StatusChanged')
78    status, reason = old.args
79    assertEquals(cs.CONN_STATUS_DISCONNECTED, status)
80    assertEquals(cs.CSR_AUTHENTICATION_FAILED, reason)
81
82if __name__ == '__main__':
83    exec_test(test_network_error, {'port': dbus.UInt32(4243)},
84              do_connect=False)
85    exec_test(test_conflict_after_connect)
86    exec_test(test_stream_conflict_during_connect,
87            authenticator=StreamErrorAuthenticator('conflict'), do_connect=False)
88    exec_test(test_host_unknown, {'server': 'localhost',
89                     'account': 'test@example.org',
90                    }, authenticator=StreamErrorAuthenticator('host-unknown'),
91              do_connect=False)
92