Commit a842f40c authored by Michal 'vorner' Vaner's avatar Michal 'vorner' Vaner
Browse files

[1454] The handle_incoming request

Reads the request and passes it to be handled. If it fails, the
connection is closed.
parent b6aa3ff5
......@@ -138,11 +138,32 @@ class DDNSServer:
session =
self._socket_sessions[fileno] = (socket, session)
def handle_request(self, request):
This is called whenever a new DDNS request comes. Here the magic
happens, the rest is either subroutines of the magic or the accounting
around it that calls it from time to time.
It is called with the request being session as received from
SocketSessionReceiver, eg. tupple
(socket, local_address, remote_address, data).
# TODO: Implement the magic
def handle_incoming(self, fileno):
Handle incoming event on the socket with given fileno.
(socket, session) = self._socket_sessions[fileno]
request = session.pop()
# No matter why this failed, the connection is in unknown, possibly
# broken state. So, we close the socket and remove the session.
del self._socket_sessions[fileno]
def run(self):
......@@ -217,6 +217,50 @@ class TestDDNSServer(unittest.TestCase):
self.assertEqual(3, self.__hook_called)
def test_handle_incoming_ok(self):
Test the handle_incoming pops the session and calls handle_request
when everything is OK.
socket = FakeSocket(3)
session = FakeSession(socket)
# It doesn't really matter what data we use here, it is only passed
# through the code
param = (FakeSocket(4), ('', 1234), ('', 1235),
'Some data')
def pop():
return param
# Prepare data into the session
session.pop = pop
self.ddns_server._socket_sessions = {3: (socket, session)}
self.ddns_server.handle_request = self.__hook
# Call it
# The popped data are passed into the handle_request
self.assertEqual(param, self.__hook_called)
# The sessions are kept the same
self.assertEqual({3: (socket, session)},
def test_handle_incoming_fail(self):
Test the handle_incoming removes (and closes) the socket and session
when the session complains.
socket = FakeSocket(3)
session = FakeSession(socket)
def pop():
raise'Test error')
session.pop = pop
socket.close = self.__hook
self.__hook_called = False
self.ddns_server._socket_sessions = {3: (socket, session)}
# The "dead" session is removed
self.assertEqual({}, self.ddns_server._socket_sessions)
# Close is called with no parameter, so the default None
class TestMain(unittest.TestCase):
def setUp(self):
self._server = MyDDNSServer()
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment