Changeset 1066

Timestamp:

09/14/07 00:31:34 (15 months ago)

Author:

wichert

Message:

More server tests.

Location:

trunk

Files:

• CHANGES.txt (modified) (1 diff)
• pyrad/tests/mock.py (modified) (1 diff)
• pyrad/tests/testClient.py (modified) (1 diff)
• pyrad/tests/testServer.py (modified) (2 diffs)

trunk/CHANGES.txt

@@ -1 +1 @@
 Pyrad 1.0
 =========
+
+* The default dictionary for Server instances was shared between all instances,
+  possibly leading to unwanted data pollution. Each Server now gets its own
+  dict instance if none is passed in to the constructor.
 
 * Fixed a timeout handling problem in the client: after receiving an

trunk/pyrad/tests/mock.py

@@ -85 +85 @@
 
 
+class MockPoll:
+    def __init__(self):
+        self.registry=[]
+
+
+    def register(self, fd, options):
+        self.registry.append((fd, options))
+

trunk/pyrad/tests/testClient.py

@@ -7 +7 @@
 from pyrad.packet import AccessRequest
 from pyrad.packet import AccountingRequest
-from pyrad.packet.tests.mock import MockPacket
-from pyrad.packet.tests.mock import MockSocket
+from pyrad.tests.mock import MockPacket
+from pyrad.tests.mock import MockSocket
 
 BIND_IP = "127.0.0.1"

trunk/pyrad/tests/testServer.py

@@ +1 @@
+import select
+import socket
 import unittest
 from pyrad.server import RemoteHost
 from pyrad.server import Server
+from pyrad.server import PacketError
+from pyrad.tests.mock import MockPoll
+from pyrad.tests.mock import MockSocket
+from pyrad.packet import AccessRequest
+from pyrad.packet import AccountingRequest
+
+
+class TrivialObject:
+    """dummy object"""
+
+
+class MockFd:
+    data = object()
+    source = object()
+
+    def __init__(self, fd=0):
+        self.fd=fd
+
+    def fileno(self):
+        return self.fd
+
+    def recvfrom(self, size):
+        self.size=size
+        return (self.data, self.source)
+
+
 
 class RemoteHostTests(unittest.TestCase):
@@ -44 +72 @@
     def testBindDuringConstruction(self):
         def BindToAddress(self, addr):
-            self.bound=addr
+            self.bound.append(addr)
         bta=Server.BindToAddress
         Server.BindToAddress=BindToAddress
 
+        Server.bound=[]
         server=Server(["one", "two", "three"])
         self.assertEqual(server.bound, ["one", "two", "three"])
+        del Server.bound
 
         Server.BindToAddress=bta
 
 
+
+class SocketTests(unittest.TestCase):
+    def setUp(self):
+        self.orgsocket=socket.socket
+        socket.socket=MockSocket
+        self.server=Server()
+
+
+    def tearDown(self):
+        socket.socket=self.orgsocket
+
+
+    def testBind(self):
+        self.server.BindToAddress("192.168.13.13")
+        self.assertEqual(len(self.server.authfds), 1)
+        self.assertEqual(self.server.authfds[0].address,
+                ("192.168.13.13", 1812))
+
+        self.assertEqual(len(self.server.acctfds), 1)
+        self.assertEqual(self.server.acctfds[0].address,
+                ("192.168.13.13", 1813))
+
+
+    def testGrabPacket(self):
+        def gen(data):
+            res=TrivialObject()
+            res.data=data
+            return res
+
+
+        fd=MockFd()
+        fd.source=object()
+        pkt=self.server._GrabPacket(gen, fd)
+        self.failUnless(isinstance(pkt, TrivialObject))
+        self.failUnless(pkt.fd is fd)
+        self.failUnless(pkt.source is fd.source)
+        self.failUnless(pkt.data is fd.data)
+
+
+    def testPrepareSocketNoFds(self):
+        self.server._poll=MockPoll()
+        self.server._PrepareSockets()
+
+        self.assertEqual(self.server._poll.registry, [])
+        self.assertEqual(self.server._realauthfds, [])
+        self.assertEqual(self.server._realacctfds, [])
+
+
+    def testPrepareSocketAuthFds(self):
+        self.server._poll=MockPoll()
+        self.server._fdmap={}
+        self.server.authfds=[MockFd(12), MockFd(14)]
+        self.server._PrepareSockets()
+
+        self.assertEqual(self.server._fdmap.keys(), [12, 14])
+        self.assertEqual(self.server._poll.registry,
+                [(12, select.POLLIN|select.POLLPRI|select.POLLERR),
+                 (14, select.POLLIN|select.POLLPRI|select.POLLERR)])
+
+
+    def testPrepareSocketAcctFds(self):
+        self.server._poll=MockPoll()
+        self.server._fdmap={}
+        self.server.acctfds=[MockFd(12), MockFd(14)]
+        self.server._PrepareSockets()
+
+        self.assertEqual(self.server._fdmap.keys(), [12, 14])
+        self.assertEqual(self.server._poll.registry,
+                [(12, select.POLLIN|select.POLLPRI|select.POLLERR),
+                 (14, select.POLLIN|select.POLLPRI|select.POLLERR)])
+
+
+class AuthPacketHandlingTests(unittest.TestCase):
+    def setUp(self):
+        self.server=Server()
+        self.server.hosts["host"]=TrivialObject()
+        self.server.hosts["host"].secret="supersecret"
+        self.packet=TrivialObject()
+        self.packet.code=AccessRequest
+        self.packet.source=("host", "port")
+
+
+    def testHandleAuthPacketUnknownHost(self):
+        self.packet.source=("stranger", "port")
+        try:
+            self.server._HandleAuthPacket(self.packet)
+        except PacketError, e:
+            self.failUnless("unknown host" in str(e))
+        else:
+            self.fail()
+
+
+    def testHandleAuthPacketWrongPort(self):
+        self.packet.code=AccountingRequest
+        try:
+            self.server._HandleAuthPacket(self.packet)
+        except PacketError, e:
+            self.failUnless("port" in str(e))
+        else:
+            self.fail()
+
+
+    def testHandleAuthPacket(self):
+        def HandleAuthPacket(self, pkt):
+            self.handled=pkt
+        hap=Server.HandleAuthPacket
+        Server.HandleAuthPacket=HandleAuthPacket
+
+        self.server._HandleAuthPacket(self.packet)
+        self.failUnless(self.server.handled is self.packet)
+
+        Server.HandleAuthPacket=hap
+
+
+
+class AcctPacketHandlingTests(unittest.TestCase):
+    def setUp(self):
+        self.server=Server()
+        self.server.hosts["host"]=TrivialObject()
+        self.server.hosts["host"].secret="supersecret"
+        self.packet=TrivialObject()
+        self.packet.code=AccountingRequest
+        self.packet.source=("host", "port")
+
+
+    def testHandleAcctPacketUnknownHost(self):
+        self.packet.source=("stranger", "port")
+        try:
+            self.server._HandleAcctPacket(self.packet)
+        except PacketError, e:
+            self.failUnless("unknown host" in str(e))
+        else:
+            self.fail()
+
+
+    def testHandleAcctPacketWrongPort(self):
+        self.packet.code=AccessRequest
+        try:
+            self.server._HandleAcctPacket(self.packet)
+        except PacketError, e:
+            self.failUnless("port" in str(e))
+        else:
+            self.fail()
+
+
+    def testHandleAcctPacket(self):
+        def HandleAcctPacket(self, pkt):
+            self.handled=pkt
+        hap=Server.HandleAcctPacket
+        Server.HandleAcctPacket=HandleAcctPacket
+
+        self.server._HandleAcctPacket(self.packet)
+        self.failUnless(self.server.handled is self.packet)
+
+        Server.HandleAcctPacket=hap
+