/usr/share/pyshared/pyrad/tests/testClient.py is in python-pyrad 2.0-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 | import socket
import unittest
import six
from pyrad.client import Client
from pyrad.client import Timeout
from pyrad.packet import AuthPacket
from pyrad.packet import AcctPacket
from pyrad.packet import AccessRequest
from pyrad.packet import AccountingRequest
from pyrad.tests.mock import MockPacket
from pyrad.tests.mock import MockSocket
BIND_IP = "127.0.0.1"
BIND_PORT = 53535
class ConstructionTests(unittest.TestCase):
def setUp(self):
self.server = object()
def testSimpleConstruction(self):
client = Client(self.server)
self.failUnless(client.server is self.server)
self.assertEqual(client.authport, 1812)
self.assertEqual(client.acctport, 1813)
self.assertEqual(client.secret, six.b(''))
self.assertEqual(client.retries, 3)
self.assertEqual(client.timeout, 5)
self.failUnless(client.dict is None)
def testParameterOrder(self):
marker = object()
client = Client(self.server, 123, 456, "secret", marker)
self.failUnless(client.server is self.server)
self.assertEqual(client.authport, 123)
self.assertEqual(client.acctport, 456)
self.assertEqual(client.secret, "secret")
self.failUnless(client.dict is marker)
def testNamedParameters(self):
marker = object()
client = Client(server=self.server, authport=123, acctport=456,
secret="secret", dict=marker)
self.failUnless(client.server is self.server)
self.assertEqual(client.authport, 123)
self.assertEqual(client.acctport, 456)
self.assertEqual(client.secret, "secret")
self.failUnless(client.dict is marker)
class SocketTests(unittest.TestCase):
def setUp(self):
self.server = object()
self.client = Client(self.server)
self.orgsocket = socket.socket
socket.socket = MockSocket
def tearDown(self):
socket.socket = self.orgsocket
def testReopen(self):
self.client._SocketOpen()
sock = self.client._socket
self.client._SocketOpen()
self.failUnless(sock is self.client._socket)
def testBind(self):
self.client.bind((BIND_IP, BIND_PORT))
self.assertEqual(self.client._socket.address, (BIND_IP, BIND_PORT))
self.assertEqual(self.client._socket.options,
[(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)])
def testBindClosesSocket(self):
s = MockSocket(socket.AF_INET, socket.SOCK_DGRAM)
self.client._socket = s
self.client.bind((BIND_IP, BIND_PORT))
self.assertEqual(s.closed, True)
def testSendPacket(self):
def MockSend(self, pkt, port):
self._mock_pkt = pkt
self._mock_port = port
_SendPacket = Client._SendPacket
Client._SendPacket = MockSend
self.client.SendPacket(AuthPacket())
self.assertEqual(self.client._mock_port, self.client.authport)
self.client.SendPacket(AcctPacket())
self.assertEqual(self.client._mock_port, self.client.acctport)
Client._SendPacket = _SendPacket
def testNoRetries(self):
self.client.retries = 0
self.assertRaises(Timeout, self.client._SendPacket, None, None)
def testSingleRetry(self):
self.client.retries = 1
self.client.timeout = 0
packet = MockPacket(AccessRequest)
self.assertRaises(Timeout, self.client._SendPacket, packet, 432)
self.assertEqual(self.client._socket.output,
[("request packet", (self.server, 432))])
def testDoubleRetry(self):
self.client.retries = 2
self.client.timeout = 0
packet = MockPacket(AccessRequest)
self.assertRaises(Timeout, self.client._SendPacket, packet, 432)
self.assertEqual(self.client._socket.output,
[("request packet", (self.server, 432)),
("request packet", (self.server, 432))])
def testAuthDelay(self):
self.client.retries = 2
self.client.timeout = 1
packet = MockPacket(AccessRequest)
self.assertRaises(Timeout, self.client._SendPacket, packet, 432)
self.failIf("Acct-Delay-Time" in packet)
def testSingleAccountDelay(self):
self.client.retries = 2
self.client.timeout = 1
packet = MockPacket(AccountingRequest)
self.assertRaises(Timeout, self.client._SendPacket, packet, 432)
self.assertEqual(packet["Acct-Delay-Time"], [1])
def testDoubleAccountDelay(self):
self.client.retries = 3
self.client.timeout = 1
packet = MockPacket(AccountingRequest)
self.assertRaises(Timeout, self.client._SendPacket, packet, 432)
self.assertEqual(packet["Acct-Delay-Time"], [2])
def testIgnorePacketError(self):
self.client.retries = 1
self.client.timeout = 1
self.client._socket = MockSocket(1, 2, six.b("valid reply"))
packet = MockPacket(AccountingRequest, verify=True, error=True)
self.assertRaises(Timeout, self.client._SendPacket, packet, 432)
def testValidReply(self):
self.client.retries = 1
self.client.timeout = 1
self.client._socket = MockSocket(1, 2, six.b("valid reply"))
packet = MockPacket(AccountingRequest, verify=True)
reply = self.client._SendPacket(packet, 432)
self.failUnless(reply is packet.reply)
def testInvalidReply(self):
self.client.retries = 1
self.client.timeout = 1
self.client._socket = MockSocket(1, 2, six.b("invalid reply"))
packet = MockPacket(AccountingRequest, verify=False)
self.assertRaises(Timeout, self.client._SendPacket, packet, 432)
class OtherTests(unittest.TestCase):
def setUp(self):
self.server = object()
self.client = Client(self.server, secret=six.b('zeer geheim'))
def testCreateAuthPacket(self):
packet = self.client.CreateAuthPacket(id=15)
self.failUnless(isinstance(packet, AuthPacket))
self.failUnless(packet.dict is self.client.dict)
self.assertEqual(packet.id, 15)
self.assertEqual(packet.secret, six.b('zeer geheim'))
def testCreateAcctPacket(self):
packet = self.client.CreateAcctPacket(id=15)
self.failUnless(isinstance(packet, AcctPacket))
self.failUnless(packet.dict is self.client.dict)
self.assertEqual(packet.id, 15)
self.assertEqual(packet.secret, six.b('zeer geheim'))
|