# test_server.py
# -*- coding: utf8 -*-
# vim:fileencoding=utf8 ai ts=4 sts=4 et sw=4
# Copyright 2009 SKA South Africa (http://ska.ac.za/)
# BSD license - see COPYING for details

"""Tests for the server module.
   """

import unittest2 as unittest
import socket
import errno
import time
import logging
import threading

import katcp
import mock

from collections import defaultdict
from functools import partial

from katcp.testutils import (
    TestLogHandler, BlockingTestClient, DeviceTestServer, TestUtilMixin,
    start_thread_with_cleanup, WaitingMock, ClientConnectionTest, mock_req)
from katcp.core import FailReply

log_handler = TestLogHandler()
logging.getLogger("katcp").addHandler(log_handler)

NO_HELP_MESSAGES = 16       # Number of requests on DeviceTestServer

class test_ClientConnectionTCP(unittest.TestCase):
    def test_init(self):
        # Test that the ClientConnection methods are correctly bound to the
        # server methods
        server = mock.Mock()
        raw_socket = 'raw_socket'
        DUT = katcp.server.ClientConnectionTCP(server, raw_socket)
        DUT.inform('inf_arg')
        server.tcp_inform.assert_called_once_with(raw_socket, 'inf_arg')
        DUT.reply_inform('rif_arg')
        server.tcp_reply_inform.assert_called_once_with(raw_socket, 'rif_arg')
        DUT.reply('rep_arg')
        server.tcp_reply.assert_called_once_with(raw_socket, 'rep_arg')


class test_ClientRequestConnection(unittest.TestCase):
    def setUp(self):
        self.client_connection = mock.Mock()
        self.req_msg = katcp.Message.request(
            'test-request', 'parm1', 'parm2', mid=42)
        self.DUT = katcp.server.ClientRequestConnection(
            self.client_connection, self.req_msg)

    def test_inform(self):
        arguments = ('inf1', 'inf2')
        self.DUT.inform(*arguments)
        self.assertEqual(self.client_connection.inform.call_count, 1)
        (inf_msg,), kwargs = self.client_connection.inform.call_args
        self.assertSequenceEqual(inf_msg.arguments, arguments)
        self.assertEqual(inf_msg.name, 'test-request')
        self.assertEqual(inf_msg.mid, '42')
        self.assertEqual(inf_msg.mtype, katcp.Message.INFORM)

    def test_reply(self):
        arguments = ('inf1', 'inf2')
        self.DUT.reply(*arguments)
        self.assertEqual(self.client_connection.reply.call_count, 1)
        (rep_msg, req_msg), kwargs = self.client_connection.reply.call_args
        self.assertIs(req_msg, self.req_msg)
        self.assertSequenceEqual(rep_msg.arguments, arguments)
        self.assertEqual(rep_msg.name, 'test-request')
        self.assertEqual(rep_msg.mid, '42')
        self.assertEqual(rep_msg.mtype, katcp.Message.REPLY)
        # Test that we can't reply twice
        with self.assertRaises(RuntimeError):
            self.DUT.reply(*arguments)

    def test_reply_with_msg(self):
        wrong_rep_msg = katcp.Message.reply('wrong-request', 'inf1', 'inf2')
        with self.assertRaises(AssertionError):
            self.DUT.reply_with_message(wrong_rep_msg)
        rep_msg = katcp.Message.reply('test-request', 'inf1', 'inf2')
        self.DUT.reply_with_message(rep_msg.copy())
        self.assertEqual(self.client_connection.reply.call_count, 1)
        (actual_rep_msg, req_msg), kwargs = self.client_connection.reply.call_args
        self.assertIs(req_msg, self.req_msg)
        self.assertEqual(actual_rep_msg, rep_msg)
        # Test that we can't reply twice
        with self.assertRaises(RuntimeError):
            self.DUT.reply_with_message(rep_msg)

    def test_reply_message(self):
        arguments = ('inf1', 'inf2')
        rep_msg = self.DUT.make_reply(*arguments)
        self.assertSequenceEqual(rep_msg.arguments, arguments)
        self.assertEqual(rep_msg.name, 'test-request')
        self.assertEqual(rep_msg.mid, '42')
        self.assertEqual(rep_msg.mtype, katcp.Message.REPLY)

class TestDeviceServerV4(unittest.TestCase, TestUtilMixin):

    class DeviceTestServerV4(DeviceTestServer):
        ## Protocol versions and flags for a katcp v4 server
        PROTOCOL_INFO = katcp.ProtocolFlags(4, 0, set([
            katcp.ProtocolFlags.MULTI_CLIENT,
            ]))

    def setUp(self):
        self.server = self.DeviceTestServerV4('', 0)

    def test_log(self):
        self.server.mass_inform = mock.Mock()
        self.server.log.warn('A warning', timestamp=1234)
        self.assertEqual(self.server.mass_inform.call_count, 1)
        (msg, ), _ = self.server.mass_inform.call_args
        level, timestamp, name, log_message = msg.arguments
        self.assertEqual(msg.name, 'log')
        self.assertIs(msg.mid, None)
        # Timestamp should be in miliseconds
        self.assertEqual(timestamp, '1234000')
        self.assertIn('A warning', log_message)

    def test_on_client_connect(self):
        conn = katcp.server.ClientConnectionTCP(self.server, 'fake-sock')
        m_sm = self.server._send_message = mock.Mock()
        self.server.BUILD_INFO = ('buildy', 1, 2, 'g')
        self.server.VERSION_INFO = ('deviceapi', 5, 6)
        self.server.on_client_connect(conn)
        # we are expecting 2 inform messages
        no_msgs = 2
        self.assertEqual(m_sm.call_count, no_msgs)
        # Check that calls were syntactically valid
        self.assertEqual(m_sm.call_args_list,
                         [mock.call('fake-sock', mock.ANY)]*no_msgs)
        # Get all the messages sent to _send_message
        msgs = [str(call[0][1]) for call in m_sm.call_args_list]
        self._assert_msgs_equal(msgs, (
            r'#version deviceapi-5.6',
            r'#build-state buildy-1.2g') )

    def test_sensor_sampling(self):
        start_thread_with_cleanup(self, self.server)
        s = katcp.Sensor.boolean('a-sens')
        s.set(1234, katcp.Sensor.NOMINAL, True)
        self.server.add_sensor(s)
        self.server._send_message = WaitingMock()
        self.server.wait_running(timeout=1.)
        self.assertTrue(self.server.running())
        self.server._strategies = defaultdict(lambda : {})
        req = mock_req('sensor-sampling', 'a-sens', 'event')
        self.server.request_sensor_sampling(req, req.msg)
        inf = req.client_connection.inform
        inf.assert_wait_call_count(count=1)
        (inf_msg, ) = inf.call_args[0]
        self._assert_msgs_equal([inf_msg], (
            r'#sensor-status 1234000 1 a-sens nominal 1',))
        req = mock_req('sensor-sampling', 'a-sens', 'period', 1000)
        self.server.request_sensor_sampling(req, req.msg)
        client = req.client_connection
        strat = self.server._strategies[client][s]
        # Test that the periodic update period is converted to seconds
        self.assertEqual(strat._period, 1.)
        # test that parameters returned by the request matches v4 format.
        #
        # We need to pass in the same client_conn as used by the previous
        # request since strategies are bound to specific connections
        req = mock_req('sensor-sampling', 'a-sens', client_conn=client)
        reply = self.server.request_sensor_sampling(req, req.msg)
        self._assert_msgs_equal([reply],
                                ['!sensor-sampling ok a-sens period 1000'])
        # event-rate is not an allowed v4 strategy
        with self.assertRaises(FailReply):
            self.server.request_sensor_sampling(req, katcp.Message.request(
                'sensor-sampling', 'a-sens', 'event-rate', 1000, 2000))
        # differential-rate is not an allowed v4 strategy
        with self.assertRaises(FailReply):
            self.server.request_sensor_sampling(req, katcp.Message.request(
             'sensor-sampling', 'a-sens', 'differential-rate', 1, 1000, 2000))


    def test_sensor_value(self):
        s = katcp.Sensor.boolean('a-sens')
        s.set(1234, katcp.Sensor.NOMINAL, True)
        self.server.add_sensor(s)
        client_conn = ClientConnectionTest()
        self.server.handle_message(client_conn, katcp.Message.request(
            'sensor-value', 'a-sens'))
        self._assert_msgs_equal(client_conn.messages, [
            '#sensor-value 1234000 1 a-sens nominal 1',
            '!sensor-value ok 1'])

class TestVersionCompatibility(unittest.TestCase):
    def test_wrong_version(self):
        class DeviceTestServerWrong(DeviceTestServer):
            ## Protocol versions and flags for a katcp v4 server
            PROTOCOL_INFO = katcp.ProtocolFlags(3, 0, set([
                katcp.ProtocolFlags.MULTI_CLIENT,
                ]))

        # Only major versions 4 and 5 are supported
        with self.assertRaises(ValueError):
            DeviceTestServerWrong('', 0)
        DeviceTestServerWrong.PROTOCOL_INFO.major = 6
        with self.assertRaises(ValueError):
            DeviceTestServerWrong('', 0)


class test_DeviceServer(unittest.TestCase, TestUtilMixin):
    def setUp(self):
        self.server = DeviceTestServer('', 0)

    def test_on_client_connect(self):
        conn = katcp.server.ClientConnectionTCP(self.server, 'fake-sock')
        m_sm = self.server._send_message = mock.Mock()
        self.server.BUILD_INFO = ('buildy', 1, 2, 'g')
        self.server.VERSION_INFO = ('deviceapi', 5, 6)
        self.server.on_client_connect(conn)
        # we are expecting 3 inform messages
        no_msgs = 3
        self.assertEqual(m_sm.call_count, no_msgs)
        # Check that calls were syntactically valid
        self.assertEqual(m_sm.call_args_list,
                         [mock.call('fake-sock', mock.ANY)]*no_msgs)
        # Get all the messages sent to _send_message
        msgs = [str(call[0][1]) for call in m_sm.call_args_list]
        self._assert_msgs_equal(msgs, (
            r'#version-connect katcp-protocol 5.0-IM',
            # Will have to be updated for every library version bump
            r'#version-connect katcp-library katcp-python-0.5.4',
            r'#version-connect katcp-device deviceapi-5.6 buildy-1.2g') )

    def test_request_sensor_sampling_clear(self):
        self.server.clear_strategies = mock.Mock()
        client_connection = ClientConnectionTest()
        self.server.handle_message(
            client_connection, katcp.Message.request('sensor-sampling-clear'))
        self._assert_msgs_equal(client_connection.messages, [
            '!sensor-sampling-clear ok'])
        self.server.clear_strategies.assert_called_once_with(client_connection)

    def test_has_sensor(self):
        self.assertFalse(self.server.has_sensor('blaah'))
        self.server.add_sensor(katcp.Sensor.boolean('blaah', 'blaah sens'))
        self.assertTrue(self.server.has_sensor('blaah'))


class TestDeviceServerClientIntegrated(unittest.TestCase, TestUtilMixin):

    BLACKLIST = ("version-connect", "version", "build-state")

    def setUp(self):
        self.server = DeviceTestServer('', 0)
        self.server.start(timeout=0.1)
        host, port = self.server._sock.getsockname()
        self.server_addr = (host, port)

        self.client = BlockingTestClient(self, host, port)
        self.client.start(timeout=0.1)
        self.assertTrue(self.client.wait_protocol(timeout=0.1))

    def tearDown(self):
        if self.client.running():
            self.client.stop()
            self.client.join()
        if self.server.running():
            self.server.stop()
            self.server.join()

    def test_log(self):
        get_msgs = self.client.message_recorder(
                blacklist=self.BLACKLIST,
                replies=True)

        with mock.patch('katcp.server.time.time') as m_time:
            m_time.return_value = 1234
            self.server.log.error('An error')
        get_msgs.wait_number(1)
        self._assert_msgs_equal(
            get_msgs(), [r"#log error 1234.000000 root An\_error"])

    def test_simple_connect(self):
        """Test a simple server setup and teardown with client connect."""
        get_msgs = self.client.message_recorder(
                blacklist=self.BLACKLIST,
                replies=True)
        # basic send
        self.client.request(katcp.Message.request("foo"), use_mid=False)

        # pipe-lined send
        self.client.raw_send("?bar-boom\r\n?baz\r")

        # broken up sends
        self.client.raw_send("?boo")
        self.client.raw_send("m arg1 arg2")
        self.client.raw_send("\n")

        time.sleep(0.1)

        self._assert_msgs_equal(get_msgs(), [
            r"!foo invalid Unknown\_request.",
            r"!bar-boom invalid Unknown\_request.",
            r"!baz invalid Unknown\_request.",
            r"!boom invalid Unknown\_request.",
        ])

    def test_bad_requests(self):
        """Test request failure paths in device server."""
        get_msgs = self.client.message_recorder(
                blacklist=self.BLACKLIST, replies=True)
        self.client.raw_send("bad msg\n")
        # wait for reply
        self.client.blocking_request(
            katcp.Message.request("watchdog"), use_mid=False)

        self._assert_msgs_like(get_msgs(), [
            (r"#log error", "KatcpSyntaxError:"
                            "\_Bad\_type\_character\_'b'.\\n"),
            (r"!watchdog ok", ""),
        ])

    def test_slow_client(self):
        # Test that server does not choke sending messages to slow clients
        self.server.send_timeout = 0.1    # Set a short sending timeout

        self.client.wait_protocol(1)
        slow_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        slow_sock.connect(self.server_addr)
        slow_sock.settimeout(0.1)
        # Send a bunch of request to the server, but don't read anything from
        # the server
        try:
            slow_sock.sendall('?help\n'*100000)
        except socket.timeout:
            pass

        t0 = time.time()
        # Request should not have taken a very long time.
        self.client.assert_request_succeeds('help', informs_count=NO_HELP_MESSAGES)
        self.assertTrue(time.time() - t0 < 1)


    def test_server_ignores_informs_and_replies(self):
        """Test server ignores informs and replies."""
        get_msgs = self.client.message_recorder(
                blacklist=self.BLACKLIST, replies=True)
        self.client.raw_send("#some inform\n")
        self.client.raw_send("!some reply\n")

        time.sleep(0.1)

        self.assertFalse(get_msgs())

    def test_standard_requests(self):
        """Test standard request and replies."""
        get_msgs = self.client.message_recorder(
                blacklist=self.BLACKLIST, replies=True)
        nomid_req = partial(self.client.request, use_mid=False)
        nomid_req(katcp.Message.request("watchdog"), use_mid=False)
        nomid_req(katcp.Message.request("restart"))
        nomid_req(katcp.Message.request("log-level"))
        nomid_req(katcp.Message.request("log-level", "trace"))
        nomid_req(katcp.Message.request("log-level", "unknown"))
        nomid_req(katcp.Message.request("help"))
        nomid_req(katcp.Message.request("help", "watchdog"))
        nomid_req(katcp.Message.request("help", "unknown-request"))
        nomid_req(katcp.Message.request("client-list"))
        nomid_req(katcp.Message.request("version-list"))
        nomid_req(katcp.Message.request("sensor-list"))
        nomid_req(katcp.Message.request("sensor-list", "an.int"))
        nomid_req(katcp.Message.request("sensor-list", "an.unknown"))
        nomid_req(katcp.Message.request("sensor-value"))
        nomid_req(katcp.Message.request("sensor-value", "an.int"))
        nomid_req(katcp.Message.request("sensor-value",
                                                  "an.unknown"))
        nomid_req(katcp.Message.request("sensor-sampling", "an.int"))
        nomid_req(katcp.Message.request("sensor-sampling", "an.int",
                                                  "differential", "2"))
        nomid_req(katcp.Message.request("sensor-sampling", "an.int",
                                                  "event-rate", "2", "3"))
        nomid_req(katcp.Message.request("sensor-sampling"))
        nomid_req(katcp.Message.request("sensor-sampling",
                                                  "an.unknown", "auto"))
        self.client.blocking_request(katcp.Message.request(
            "sensor-sampling", "an.int", "unknown"), use_mid=False)

        time.sleep(0.1)

        self.server.log.trace("trace-msg")
        self.server.log.debug("debug-msg")
        self.server.log.info("info-msg")
        self.server.log.warn("warn-msg")
        self.server.log.error("error-msg")
        self.server.log.fatal("fatal-msg")

        time.sleep(0.1)

        self.assertEqual(self.server.restart_queue.get_nowait(), self.server)
        self._assert_msgs_like(get_msgs(), [
            (r"!watchdog ok", ""),
            (r"!restart ok", ""),
            (r"!log-level ok warn", ""),
            (r"!log-level ok trace", ""),
            (r"!log-level fail Unknown\_logging\_level\_name\_'unknown'", ""),
            (r"#help cancel-slow-command Cancel\_slow\_command\_request,\_"
             "resulting\_in\_it\_replying\_immediately", ""),
            (r"#help client-list", ""),
            (r"#help halt", ""),
            (r"#help help", ""),
            (r"#help log-level", ""),
            (r"#help new-command", ""),
            (r"#help raise-exception", ""),
            (r"#help raise-fail", ""),
            (r"#help restart", ""),
            (r"#help sensor-list", ""),
            (r"#help sensor-sampling", ""),
            (r"#help sensor-sampling-clear", ""),
            (r"#help sensor-value", ""),
            (r"#help slow-command", ""),
            (r"#help version-list", ""),
            (r"#help watchdog", ""),
            (r"!help ok %d" % NO_HELP_MESSAGES, ""),
            (r"#help watchdog", ""),
            (r"!help ok 1", ""),
            (r"!help fail", ""),
            (r"#client-list", ""),
            (r"!client-list ok 1", ""),
            (r"#version-list katcp-protocol", ""),
            (r"#version-list katcp-library", ""),
            (r"#version-list katcp-device", ""),
            (r"!version-list ok 3", ""),
            (r"#sensor-list an.int An\_Integer. count integer -5 5", ""),
            (r"!sensor-list ok 1", ""),
            (r"#sensor-list an.int An\_Integer. count integer -5 5", ""),
            (r"!sensor-list ok 1", ""),
            (r"!sensor-list fail", ""),
            (r"#sensor-value 12345.000000 1 an.int nominal 3", ""),
            (r"!sensor-value ok 1", ""),
            (r"#sensor-value 12345.000000 1 an.int nominal 3", ""),
            (r"!sensor-value ok 1", ""),
            (r"!sensor-value fail", ""),
            (r"!sensor-sampling ok an.int none", ""),
            (r"#sensor-status 12345.000000 1 an.int nominal 3", ""),
            (r"!sensor-sampling ok an.int differential 2", ""),
            (r"#sensor-status 12345.000000 1 an.int nominal 3", ""),
            (r"!sensor-sampling ok an.int event-rate 2 3", ""),
            (r"!sensor-sampling fail No\_sensor\_name\_given.", ""),
            (r"!sensor-sampling fail Unknown\_sensor\_name:\_an.unknown.", ""),
            (r"!sensor-sampling fail Unknown\_strategy\_name:\_unknown.", ""),
            (r"#log trace", r"root trace-msg"),
            (r"#log debug", r"root debug-msg"),
            (r"#log info", r"root info-msg"),
            (r"#log warn", r"root warn-msg"),
            (r"#log error", r"root error-msg"),
            (r"#log fatal", r"root fatal-msg"),
        ])

    def test_standard_requests_with_ids(self):
        """Test standard request and replies with message ids."""
        get_msgs = self.client.message_recorder(
                blacklist=self.BLACKLIST, replies=True)

        current_id = [0]

        def mid():
            current_id[0] += 1
            return str(current_id[0])

        def mid_req(*args):
            return katcp.Message.request(*args, mid=mid())


        self.client.request(mid_req("watchdog"))
        self.client.request(mid_req("restart"))
        self.client.request(mid_req("log-level"))
        self.client.request(mid_req("log-level", "trace"))
        self.client.request(mid_req("log-level", "unknown"))
        self.client.request(mid_req("help"))
        self.client.request(mid_req("help", "watchdog"))
        self.client.request(mid_req("help", "unknown-request"))
        self.client.request(mid_req("client-list"))
        self.client.request(mid_req("version-list"))
        self.client.request(mid_req("sensor-list"))
        self.client.request(mid_req("sensor-list", "an.int"))
        self.client.request(mid_req("sensor-list", "an.unknown"))
        self.client.request(mid_req("sensor-value"))
        self.client.request(mid_req("sensor-value", "an.int"))
        self.client.request(mid_req("sensor-value", "an.unknown"))
        self.client._next_id = mid  # mock our mid generator for testing
        self.client.blocking_request(mid_req("sensor-sampling", "an.int"))
        self.client.blocking_request(mid_req(
            "sensor-sampling", "an.int", "differential", "2"))
        self.client.blocking_request(mid_req(
            "sensor-sampling", "an.int", "event-rate", "2", "3")),
        self.client.blocking_request(mid_req("sensor-sampling"))
        self.client.blocking_request(mid_req(
            "sensor-sampling", "an.unknown", "auto"))
        self.client.blocking_request(mid_req(
            "sensor-sampling", "an.int", "unknown"))

        self.server.log.trace("trace-msg")
        self.server.log.debug("debug-msg")
        self.server.log.info("info-msg")
        self.server.log.warn("warn-msg")
        self.server.log.error("error-msg")
        self.server.log.fatal("fatal-msg")

        time.sleep(0.1)

        self.assertEqual(self.server.restart_queue.get_nowait(), self.server)
        self._assert_msgs_like(get_msgs(), [
            (r"!watchdog[1] ok", ""),
            (r"!restart[2] ok", ""),
            (r"!log-level[3] ok warn", ""),
            (r"!log-level[4] ok trace", ""),
            (r"!log-level[5] fail Unknown\_logging\_level\_name\_'unknown'",
             ""),
            (r"#help[6] cancel-slow-command Cancel\_slow\_command\_request,\_"
             "resulting\_in\_it\_replying\_immediately", ""),
            (r"#help[6] client-list", ""),
            (r"#help[6] halt", ""),
            (r"#help[6] help", ""),
            (r"#help[6] log-level", ""),
            (r"#help[6] new-command", ""),
            (r"#help[6] raise-exception", ""),
            (r"#help[6] raise-fail", ""),
            (r"#help[6] restart", ""),
            (r"#help[6] sensor-list", ""),
            (r"#help[6] sensor-sampling", ""),
            (r"#help[6] sensor-sampling-clear", ""),
            (r"#help[6] sensor-value", ""),
            (r"#help[6] slow-command", ""),
            (r"#help[6] version-list", ""),
            (r"#help[6] watchdog", ""),
            (r"!help[6] ok %d" % NO_HELP_MESSAGES, ""),
            (r"#help[7] watchdog", ""),
            (r"!help[7] ok 1", ""),
            (r"!help[8] fail", ""),
            (r"#client-list[9]", ""),
            (r"!client-list[9] ok 1", ""),
            (r"#version-list[10] katcp-protocol", ""),
            (r"#version-list[10] katcp-library", ""),
            (r"#version-list[10] katcp-device", ""),
            (r"!version-list[10] ok 3", ""),
            (r"#sensor-list[11] an.int An\_Integer. count integer -5 5", ""),
            (r"!sensor-list[11] ok 1", ""),
            (r"#sensor-list[12] an.int An\_Integer. count integer -5 5", ""),
            (r"!sensor-list[12] ok 1", ""),
            (r"!sensor-list[13] fail", ""),
            (r"#sensor-value[14] 12345.000000 1 an.int nominal 3", ""),
            (r"!sensor-value[14] ok 1", ""),
            (r"#sensor-value[15] 12345.000000 1 an.int nominal 3", ""),
            (r"!sensor-value[15] ok 1", ""),
            (r"!sensor-value[16] fail", ""),
            (r"!sensor-sampling[17] ok an.int none", ""),
            (r"#sensor-status 12345.000000 1 an.int nominal 3", ""),
            (r"!sensor-sampling[18] ok an.int differential 2", ""),
            (r"#sensor-status 12345.000000 1 an.int nominal 3", ""),
            (r"!sensor-sampling[19] ok an.int event-rate 2 3", ""),
            (r"!sensor-sampling[20] fail No\_sensor\_name\_given.", ""),
            (r"!sensor-sampling[21] fail Unknown\_sensor\_name:\_an.unknown.", ""),
            (r"!sensor-sampling[22] fail Unknown\_strategy\_name:\_unknown.", ""),
            (r"#log trace", r"root trace-msg"),
            (r"#log debug", r"root debug-msg"),
            (r"#log info", r"root info-msg"),
            (r"#log warn", r"root warn-msg"),
            (r"#log error", r"root error-msg"),
            (r"#log fatal", r"root fatal-msg"),
        ])

    def test_sensor_list_regex(self):
        reply, informs = self.client.blocking_request(katcp.Message.request(
                "sensor-list", "/a.*/"), use_mid=False)
        self._assert_msgs_equal(informs + [reply], [
            r"#sensor-list an.int An\_Integer. count integer -5 5",
            r"!sensor-list ok 1",
        ])

        reply, informs = self.client.blocking_request(katcp.Message.request(
                "sensor-list", "//"), use_mid=False)
        self._assert_msgs_equal(informs + [reply], [
            r"#sensor-list an.int An\_Integer. count integer -5 5",
            r"!sensor-list ok 1",
        ])

        reply, informs = self.client.blocking_request(katcp.Message.request(
                "sensor-list", "/^int/"), use_mid=False)
        self._assert_msgs_equal(informs + [reply], [
            r"!sensor-list ok 0",
        ])

    def test_sensor_value_regex(self):
        reply, informs = self.client.blocking_request(katcp.Message.request(
                "sensor-value", "/a.*/"), use_mid=False)
        self._assert_msgs_equal(informs + [reply], [
            r"#sensor-value 12345.000000 1 an.int nominal 3",
            r"!sensor-value ok 1",
        ])

        reply, informs = self.client.blocking_request(katcp.Message.request(
                "sensor-value", "//"), use_mid=False)
        self._assert_msgs_equal(informs + [reply], [
            r"#sensor-value 12345.000000 1 an.int nominal 3",
            r"!sensor-value ok 1",
        ])

        reply, informs = self.client.blocking_request(katcp.Message.request(
                "sensor-value", "/^int/"), use_mid=False)
        self._assert_msgs_equal(informs + [reply], [
            r"!sensor-value ok 0",
        ])

    def test_client_list(self):
        reply, informs = self.client.blocking_request(
            katcp.Message.request('client-list'), use_mid=False)
        self.assertEqual(str(reply), '!client-list ok 1')
        self.assertEqual(len(informs), 1)
        inform = str(informs[0])
        self.assertTrue(inform.startswith('#client-list 127.0.0.1:'))
        _, addr = inform.split()
        host, port = addr.split(':')
        port = int(port)
        self.assertEqual((host, port), self.client._sock.getsockname())

    def test_halt_request(self):
        """Test halt request."""
        get_msgs = self.client.message_recorder(
                blacklist=self.BLACKLIST, replies=True)
        self.client.request(katcp.Message.request("halt"))
        # hack to hide re-connect exception
        self.client.connect = lambda: None
        self.server.join()
        time.sleep(0.1)

        self._assert_msgs_equal(get_msgs(), [
            r"!halt[1] ok",
            r"#disconnect Device\_server\_shutting\_down.",
        ])

    def test_bad_handlers(self):
        """Test that bad request and inform handlers are picked up."""
        try:

            class BadServer(katcp.DeviceServer):
                def request_baz(self, req, msg):
                    pass

        except AssertionError:
            pass
        else:
            self.fail("Server metaclass accepted missing request_ docstring.")

        try:

            class BadServer(katcp.DeviceServer):
                def inform_baz(self, req, msg):
                    pass

        except AssertionError:
            pass
        else:
            self.fail("Server metaclass accepted missing inform_ docstring.")

        class SortOfOkayServer(katcp.DeviceServer):
            request_bar = 1
            inform_baz = 2

        assert("bar" not in SortOfOkayServer._request_handlers)
        assert("baz" not in SortOfOkayServer._inform_handlers)

    def test_handler_exceptions(self):
        """Test handling of failure replies and other exceptions."""
        get_msgs = self.client.message_recorder(
                blacklist=self.BLACKLIST, replies=True)
        self.assertTrue(self.client.wait_protocol(timeout=1))

        self.client.request(katcp.Message.request("raise-exception"))
        self.client.request(katcp.Message.request("raise-fail"))

        time.sleep(0.1)

        self._assert_msgs_like(get_msgs(), [
            (r"!raise-exception[1] fail Traceback", ""),
            (r"!raise-fail[2] fail There\_was\_a\_problem\_with\_your\_request.",
             ""),
        ])

    def test_stop_and_restart(self):
        """Test stopping and restarting the device server."""
        self.server.stop(timeout=0.1)
        self.server.join(timeout=1.0)
        self.assertEqual(self.server._thread, None)
        self.assertFalse(self.server._running.isSet())
        self.server.start(timeout=1.0)

    def test_bad_client_socket(self):
        """Test what happens when select is called on a dead client socket."""
        # wait for client to arrive
        time.sleep(0.1)

        # close socket while the server isn't looking
        # then wait for the server to notice
        sock = self.server._socks[0]
        sock.close()
        time.sleep(0.75)

        # check that client was removed
        self.assertTrue(sock not in self.server._socks,
                        "Expected %r to not be in %r" %
                        (sock, self.server._socks))

    def test_bad_server_socket(self):
        """Test what happens when select is called on a dead server socket."""
        # wait for client to arrive
        time.sleep(0.1)

        # close socket while the server isn't looking
        # then wait for the server to notice
        sock = self.server._sock
        sockname = sock.getsockname()
        sock.close()
        time.sleep(0.75)

        # check that server restarted
        self.assertTrue(sock is not self.server._sock,
                        "Expected %r to not be %r" % (sock, self.server._sock))
        self.assertEqual(sockname, self.server._sock.getsockname())

    def test_daemon_value(self):
        """Test passing in a daemon value to server start method."""
        self.server.stop(timeout=0.1)
        self.server.join(timeout=1.0)

        self.server.start(timeout=0.1, daemon=True)
        self.assertTrue(self.server._thread.isDaemon())

    def test_excepthook(self):
        """Test passing in an excepthook to server start method."""
        exceptions = []
        except_event = threading.Event()

        def excepthook(etype, value, traceback):
            """Keep track of exceptions."""
            exceptions.append(etype)
            except_event.set()

        self.server.stop(timeout=0.1)
        self.server.join(timeout=1.5)

        self.server.start(timeout=0.1, excepthook=excepthook)
        # force exception by deleteing _running
        old_running = self.server._running
        try:
            del self.server._running
            except_event.wait(1.5)
            self.assertEqual(exceptions, [AttributeError])
        finally:
            self.server._running = old_running

        # close socket -- server didn't shut down correctly
        self.server._sock.close()
        self.server.stop(timeout=0.1)
        self.server.join(timeout=1.5)

        except_event.clear()
        del exceptions[:]
        self.server.start(timeout=0.1, excepthook=excepthook)
        # force exception in sample reactor and check that it makes
        # it back up
        reactor = self.server._reactor
        old_stop = reactor._stopEvent
        try:
            del reactor._stopEvent
            reactor._wakeEvent.set()
            except_event.wait(0.1)
            self.assertEqual(exceptions, [AttributeError])
        finally:
            reactor._stopEvent = old_stop

        # close socket -- server didn't shut down correctly
        self.server._sock.close()

    def test_sampling(self):
        """Test sensor sampling."""
        get_msgs = self.client.message_recorder(
                blacklist=self.BLACKLIST, replies=True)
        self.client.wait_protocol(timeout=1)
        self.client.request(katcp.Message.request(
            "sensor-sampling", "an.int", "period", 1/32.))

        # Wait for the request reply and for the sensor update messages to
        # arrive. We expect update one the moment the sensor-sampling request is
        # made, then four more over 4/32. of a second, resutling in 6
        # messages. Wait half a period longer just to be sure we get everything.

        self.assertTrue(get_msgs.wait_number(6, timeout=4.5/32.))
        self.client.assert_request_succeeds("sensor-sampling", "an.int", "none")
        # Wait for reply to above request
        get_msgs.wait_number(7)
        msgs = get_msgs()
        updates = [x for x in msgs if x.name == "sensor-status"]
        others = [x for x in msgs if x.name != "sensor-status"]
        self.assertTrue(abs(len(updates) - 5) < 2,
                        "Expected 5 informs, saw %d." % len(updates))

        self._assert_msgs_equal(others, [
            r"!sensor-sampling[1] ok an.int period %s" % (1/32.),
            r"!sensor-sampling[2] ok an.int none",
        ])

        self.assertEqual(updates[0].arguments[1:],
                         ["1", "an.int", "nominal", "3"])

        ## Now clear the strategies on this sensor
        # There should only be on connection to the server, so it should be
        # the test client
        client_conn = self.server._sock_connections.values()[0]
        self.server.clear_strategies(client_conn)
        self.client.assert_request_succeeds("sensor-sampling", "an.int",
                                            args_equal=["an.int", "none"])

        # Check that we did not accidentally clobber the strategy datastructure
        # in the proccess
        self.client.assert_request_succeeds(
            "sensor-sampling", "an.int", "period", 0.125)


    def test_add_remove_sensors(self):
        """Test adding and removing sensors from a running device."""
        an_int = self.server._sensors["an.int"]
        self.server.remove_sensor(an_int)
        self.server.add_sensor(an_int)
        self.test_sampling()

# test_server.py
# -*- coding: utf8 -*-
# vim:fileencoding=utf8 ai ts=4 sts=4 et sw=4
# Copyright 2009 SKA South Africa (http://ska.ac.za/)
# BSD license - see COPYING for details

"""Tests for the server module.
   """

import unittest2 as unittest
import socket
import errno
import time
import logging
import threading

import katcp
import mock

from collections import defaultdict
from functools import partial

from katcp.testutils import (
    TestLogHandler, BlockingTestClient, DeviceTestServer, TestUtilMixin,
    start_thread_with_cleanup, WaitingMock, ClientConnectionTest, mock_req)
from katcp.core import FailReply

log_handler = TestLogHandler()
logging.getLogger("katcp").addHandler(log_handler)

NO_HELP_MESSAGES = 16       # Number of requests on DeviceTestServer

class test_ClientConnectionTCP(unittest.TestCase):
    def test_init(self):
        # Test that the ClientConnection methods are correctly bound to the
        # server methods
        server = mock.Mock()
        raw_socket = 'raw_socket'
        DUT = katcp.server.ClientConnectionTCP(server, raw_socket)
        DUT.inform('inf_arg')
        server.tcp_inform.assert_called_once_with(raw_socket, 'inf_arg')
        DUT.reply_inform('rif_arg')
        server.tcp_reply_inform.assert_called_once_with(raw_socket, 'rif_arg')
        DUT.reply('rep_arg')
        server.tcp_reply.assert_called_once_with(raw_socket, 'rep_arg')


class test_ClientRequestConnection(unittest.TestCase):
    def setUp(self):
        self.client_connection = mock.Mock()
        self.req_msg = katcp.Message.request(
            'test-request', 'parm1', 'parm2', mid=42)
        self.DUT = katcp.server.ClientRequestConnection(
            self.client_connection, self.req_msg)

    def test_inform(self):
        arguments = ('inf1', 'inf2')
        self.DUT.inform(*arguments)
        self.assertEqual(self.client_connection.inform.call_count, 1)
        (inf_msg,), kwargs = self.client_connection.inform.call_args
        self.assertSequenceEqual(inf_msg.arguments, arguments)
        self.assertEqual(inf_msg.name, 'test-request')
        self.assertEqual(inf_msg.mid, '42')
        self.assertEqual(inf_msg.mtype, katcp.Message.INFORM)

    def test_reply(self):
        arguments = ('inf1', 'inf2')
        self.DUT.reply(*arguments)
        self.assertEqual(self.client_connection.reply.call_count, 1)
        (rep_msg, req_msg), kwargs = self.client_connection.reply.call_args
        self.assertIs(req_msg, self.req_msg)
        self.assertSequenceEqual(rep_msg.arguments, arguments)
        self.assertEqual(rep_msg.name, 'test-request')
        self.assertEqual(rep_msg.mid, '42')
        self.assertEqual(rep_msg.mtype, katcp.Message.REPLY)
        # Test that we can't reply twice
        with self.assertRaises(RuntimeError):
            self.DUT.reply(*arguments)

    def test_reply_with_msg(self):
        wrong_rep_msg = katcp.Message.reply('wrong-request', 'inf1', 'inf2')
        with self.assertRaises(AssertionError):
            self.DUT.reply_with_message(wrong_rep_msg)
        rep_msg = katcp.Message.reply('test-request', 'inf1', 'inf2')
        self.DUT.reply_with_message(rep_msg.copy())
        self.assertEqual(self.client_connection.reply.call_count, 1)
        (actual_rep_msg, req_msg), kwargs = self.client_connection.reply.call_args
        self.assertIs(req_msg, self.req_msg)
        self.assertEqual(actual_rep_msg, rep_msg)
        # Test that we can't reply twice
        with self.assertRaises(RuntimeError):
            self.DUT.reply_with_message(rep_msg)

    def test_reply_message(self):
        arguments = ('inf1', 'inf2')
        rep_msg = self.DUT.make_reply(*arguments)
        self.assertSequenceEqual(rep_msg.arguments, arguments)
        self.assertEqual(rep_msg.name, 'test-request')
        self.assertEqual(rep_msg.mid, '42')
        self.assertEqual(rep_msg.mtype, katcp.Message.REPLY)

class TestDeviceServerV4(unittest.TestCase, TestUtilMixin):

    class DeviceTestServerV4(DeviceTestServer):
        ## Protocol versions and flags for a katcp v4 server
        PROTOCOL_INFO = katcp.ProtocolFlags(4, 0, set([
            katcp.ProtocolFlags.MULTI_CLIENT,
            ]))

    def setUp(self):
        self.server = self.DeviceTestServerV4('', 0)

    def test_log(self):
        self.server.mass_inform = mock.Mock()
        self.server.log.warn('A warning', timestamp=1234)
        self.assertEqual(self.server.mass_inform.call_count, 1)
        (msg, ), _ = self.server.mass_inform.call_args
        level, timestamp, name, log_message = msg.arguments
        self.assertEqual(msg.name, 'log')
        self.assertIs(msg.mid, None)
        # Timestamp should be in miliseconds
        self.assertEqual(timestamp, '1234000')
        self.assertIn('A warning', log_message)

    def test_on_client_connect(self):
        conn = katcp.server.ClientConnectionTCP(self.server, 'fake-sock')
        m_sm = self.server._send_message = mock.Mock()
        self.server.BUILD_INFO = ('buildy', 1, 2, 'g')
        self.server.VERSION_INFO = ('deviceapi', 5, 6)
        self.server.on_client_connect(conn)
        # we are expecting 2 inform messages
        no_msgs = 2
        self.assertEqual(m_sm.call_count, no_msgs)
        # Check that calls were syntactically valid
        self.assertEqual(m_sm.call_args_list,
                         [mock.call('fake-sock', mock.ANY)]*no_msgs)
        # Get all the messages sent to _send_message
        msgs = [str(call[0][1]) for call in m_sm.call_args_list]
        self._assert_msgs_equal(msgs, (
            r'#version deviceapi-5.6',
            r'#build-state buildy-1.2g') )

    def test_sensor_sampling(self):
        start_thread_with_cleanup(self, self.server)
        s = katcp.Sensor.boolean('a-sens')
        s.set(1234, katcp.Sensor.NOMINAL, True)
        self.server.add_sensor(s)
        self.server._send_message = WaitingMock()
        self.server.wait_running(timeout=1.)
        self.assertTrue(self.server.running())
        self.server._strategies = defaultdict(lambda : {})
        req = mock_req('sensor-sampling', 'a-sens', 'event')
        self.server.request_sensor_sampling(req, req.msg)
        inf = req.client_connection.inform
        inf.assert_wait_call_count(count=1)
        (inf_msg, ) = inf.call_args[0]
        self._assert_msgs_equal([inf_msg], (
            r'#sensor-status 1234000 1 a-sens nominal 1',))
        req = mock_req('sensor-sampling', 'a-sens', 'period', 1000)
        self.server.request_sensor_sampling(req, req.msg)
        client = req.client_connection
        strat = self.server._strategies[client][s]
        # Test that the periodic update period is converted to seconds
        self.assertEqual(strat._period, 1.)
        # test that parameters returned by the request matches v4 format.
        #
        # We need to pass in the same client_conn as used by the previous
        # request since strategies are bound to specific connections
        req = mock_req('sensor-sampling', 'a-sens', client_conn=client)
        reply = self.server.request_sensor_sampling(req, req.msg)
        self._assert_msgs_equal([reply],
                                ['!sensor-sampling ok a-sens period 1000'])
        # event-rate is not an allowed v4 strategy
        with self.assertRaises(FailReply):
            self.server.request_sensor_sampling(req, katcp.Message.request(
                'sensor-sampling', 'a-sens', 'event-rate', 1000, 2000))
        # differential-rate is not an allowed v4 strategy
        with self.assertRaises(FailReply):
            self.server.request_sensor_sampling(req, katcp.Message.request(
             'sensor-sampling', 'a-sens', 'differential-rate', 1, 1000, 2000))


    def test_sensor_value(self):
        s = katcp.Sensor.boolean('a-sens')
        s.set(1234, katcp.Sensor.NOMINAL, True)
        self.server.add_sensor(s)
        client_conn = ClientConnectionTest()
        self.server.handle_message(client_conn, katcp.Message.request(
            'sensor-value', 'a-sens'))
        self._assert_msgs_equal(client_conn.messages, [
            '#sensor-value 1234000 1 a-sens nominal 1',
            '!sensor-value ok 1'])

class TestVersionCompatibility(unittest.TestCase):
    def test_wrong_version(self):
        class DeviceTestServerWrong(DeviceTestServer):
            ## Protocol versions and flags for a katcp v4 server
            PROTOCOL_INFO = katcp.ProtocolFlags(3, 0, set([
                katcp.ProtocolFlags.MULTI_CLIENT,
                ]))

        # Only major versions 4 and 5 are supported
        with self.assertRaises(ValueError):
            DeviceTestServerWrong('', 0)
        DeviceTestServerWrong.PROTOCOL_INFO.major = 6
        with self.assertRaises(ValueError):
            DeviceTestServerWrong('', 0)


class test_DeviceServer(unittest.TestCase, TestUtilMixin):
    def setUp(self):
        self.server = DeviceTestServer('', 0)

    def test_on_client_connect(self):
        conn = katcp.server.ClientConnectionTCP(self.server, 'fake-sock')
        m_sm = self.server._send_message = mock.Mock()
        self.server.BUILD_INFO = ('buildy', 1, 2, 'g')
        self.server.VERSION_INFO = ('deviceapi', 5, 6)
        self.server.on_client_connect(conn)
        # we are expecting 3 inform messages
        no_msgs = 3
        self.assertEqual(m_sm.call_count, no_msgs)
        # Check that calls were syntactically valid
        self.assertEqual(m_sm.call_args_list,
                         [mock.call('fake-sock', mock.ANY)]*no_msgs)
        # Get all the messages sent to _send_message
        msgs = [str(call[0][1]) for call in m_sm.call_args_list]
        self._assert_msgs_equal(msgs, (
            r'#version-connect katcp-protocol 5.0-IM',
            # Will have to be updated for every library version bump
            r'#version-connect katcp-library katcp-python-0.5.4',
            r'#version-connect katcp-device deviceapi-5.6 buildy-1.2g') )

    def test_request_sensor_sampling_clear(self):
        self.server.clear_strategies = mock.Mock()
        client_connection = ClientConnectionTest()
        self.server.handle_message(
            client_connection, katcp.Message.request('sensor-sampling-clear'))
        self._assert_msgs_equal(client_connection.messages, [
            '!sensor-sampling-clear ok'])
        self.server.clear_strategies.assert_called_once_with(client_connection)

    def test_has_sensor(self):
        self.assertFalse(self.server.has_sensor('blaah'))
        self.server.add_sensor(katcp.Sensor.boolean('blaah', 'blaah sens'))
        self.assertTrue(self.server.has_sensor('blaah'))


class TestDeviceServerClientIntegrated(unittest.TestCase, TestUtilMixin):

    BLACKLIST = ("version-connect", "version", "build-state")

    def setUp(self):
        self.server = DeviceTestServer('', 0)
        self.server.start(timeout=0.1)
        host, port = self.server._sock.getsockname()
        self.server_addr = (host, port)

        self.client = BlockingTestClient(self, host, port)
        self.client.start(timeout=0.1)
        self.assertTrue(self.client.wait_protocol(timeout=0.1))

    def tearDown(self):
        if self.client.running():
            self.client.stop()
            self.client.join()
        if self.server.running():
            self.server.stop()
            self.server.join()

    def test_log(self):
        get_msgs = self.client.message_recorder(
                blacklist=self.BLACKLIST,
                replies=True)

        with mock.patch('katcp.server.time.time') as m_time:
            m_time.return_value = 1234
            self.server.log.error('An error')
        get_msgs.wait_number(1)
        self._assert_msgs_equal(
            get_msgs(), [r"#log error 1234.000000 root An\_error"])

    def test_simple_connect(self):
        """Test a simple server setup and teardown with client connect."""
        get_msgs = self.client.message_recorder(
                blacklist=self.BLACKLIST,
                replies=True)
        # basic send
        self.client.request(katcp.Message.request("foo"), use_mid=False)

        # pipe-lined send
        self.client.raw_send("?bar-boom\r\n?baz\r")

        # broken up sends
        self.client.raw_send("?boo")
        self.client.raw_send("m arg1 arg2")
        self.client.raw_send("\n")

        time.sleep(0.1)

        self._assert_msgs_equal(get_msgs(), [
            r"!foo invalid Unknown\_request.",
            r"!bar-boom invalid Unknown\_request.",
            r"!baz invalid Unknown\_request.",
            r"!boom invalid Unknown\_request.",
        ])

    def test_bad_requests(self):
        """Test request failure paths in device server."""
        get_msgs = self.client.message_recorder(
                blacklist=self.BLACKLIST, replies=True)
        self.client.raw_send("bad msg\n")
        # wait for reply
        self.client.blocking_request(
            katcp.Message.request("watchdog"), use_mid=False)

        self._assert_msgs_like(get_msgs(), [
            (r"#log error", "KatcpSyntaxError:"
                            "\_Bad\_type\_character\_'b'.\\n"),
            (r"!watchdog ok", ""),
        ])

    def test_slow_client(self):
        # Test that server does not choke sending messages to slow clients
        self.server.send_timeout = 0.1    # Set a short sending timeout

        self.client.wait_protocol(1)
        slow_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        slow_sock.connect(self.server_addr)
        slow_sock.settimeout(0.1)
        # Send a bunch of request to the server, but don't read anything from
        # the server
        try:
            slow_sock.sendall('?help\n'*100000)
        except socket.timeout:
            pass

        t0 = time.time()
        # Request should not have taken a very long time.
        self.client.assert_request_succeeds('help', informs_count=NO_HELP_MESSAGES)
        self.assertTrue(time.time() - t0 < 1)


    def test_server_ignores_informs_and_replies(self):
        """Test server ignores informs and replies."""
        get_msgs = self.client.message_recorder(
                blacklist=self.BLACKLIST, replies=True)
        self.client.raw_send("#some inform\n")
        self.client.raw_send("!some reply\n")

        time.sleep(0.1)

        self.assertFalse(get_msgs())

    def test_standard_requests(self):
        """Test standard request and replies."""
        get_msgs = self.client.message_recorder(
                blacklist=self.BLACKLIST, replies=True)
        nomid_req = partial(self.client.request, use_mid=False)
        nomid_req(katcp.Message.request("watchdog"), use_mid=False)
        nomid_req(katcp.Message.request("restart"))
        nomid_req(katcp.Message.request("log-level"))
        nomid_req(katcp.Message.request("log-level", "trace"))
        nomid_req(katcp.Message.request("log-level", "unknown"))
        nomid_req(katcp.Message.request("help"))
        nomid_req(katcp.Message.request("help", "watchdog"))
        nomid_req(katcp.Message.request("help", "unknown-request"))
        nomid_req(katcp.Message.request("client-list"))
        nomid_req(katcp.Message.request("version-list"))
        nomid_req(katcp.Message.request("sensor-list"))
        nomid_req(katcp.Message.request("sensor-list", "an.int"))
        nomid_req(katcp.Message.request("sensor-list", "an.unknown"))
        nomid_req(katcp.Message.request("sensor-value"))
        nomid_req(katcp.Message.request("sensor-value", "an.int"))
        nomid_req(katcp.Message.request("sensor-value",
                                                  "an.unknown"))
        nomid_req(katcp.Message.request("sensor-sampling", "an.int"))
        nomid_req(katcp.Message.request("sensor-sampling", "an.int",
                                                  "differential", "2"))
        nomid_req(katcp.Message.request("sensor-sampling", "an.int",
                                                  "event-rate", "2", "3"))
        nomid_req(katcp.Message.request("sensor-sampling"))
        nomid_req(katcp.Message.request("sensor-sampling",
                                                  "an.unknown", "auto"))
        self.client.blocking_request(katcp.Message.request(
            "sensor-sampling", "an.int", "unknown"), use_mid=False)

        time.sleep(0.1)

        self.server.log.trace("trace-msg")
        self.server.log.debug("debug-msg")
        self.server.log.info("info-msg")
        self.server.log.warn("warn-msg")
        self.server.log.error("error-msg")
        self.server.log.fatal("fatal-msg")

        time.sleep(0.1)

        self.assertEqual(self.server.restart_queue.get_nowait(), self.server)
        self._assert_msgs_like(get_msgs(), [
            (r"!watchdog ok", ""),
            (r"!restart ok", ""),
            (r"!log-level ok warn", ""),
            (r"!log-level ok trace", ""),
            (r"!log-level fail Unknown\_logging\_level\_name\_'unknown'", ""),
            (r"#help cancel-slow-command Cancel\_slow\_command\_request,\_"
             "resulting\_in\_it\_replying\_immediately", ""),
            (r"#help client-list", ""),
            (r"#help halt", ""),
            (r"#help help", ""),
            (r"#help log-level", ""),
            (r"#help new-command", ""),
            (r"#help raise-exception", ""),
            (r"#help raise-fail", ""),
            (r"#help restart", ""),
            (r"#help sensor-list", ""),
            (r"#help sensor-sampling", ""),
            (r"#help sensor-sampling-clear", ""),
            (r"#help sensor-value", ""),
            (r"#help slow-command", ""),
            (r"#help version-list", ""),
            (r"#help watchdog", ""),
            (r"!help ok %d" % NO_HELP_MESSAGES, ""),
            (r"#help watchdog", ""),
            (r"!help ok 1", ""),
            (r"!help fail", ""),
            (r"#client-list", ""),
            (r"!client-list ok 1", ""),
            (r"#version-list katcp-protocol", ""),
            (r"#version-list katcp-library", ""),
            (r"#version-list katcp-device", ""),
            (r"!version-list ok 3", ""),
            (r"#sensor-list an.int An\_Integer. count integer -5 5", ""),
            (r"!sensor-list ok 1", ""),
            (r"#sensor-list an.int An\_Integer. count integer -5 5", ""),
            (r"!sensor-list ok 1", ""),
            (r"!sensor-list fail", ""),
            (r"#sensor-value 12345.000000 1 an.int nominal 3", ""),
            (r"!sensor-value ok 1", ""),
            (r"#sensor-value 12345.000000 1 an.int nominal 3", ""),
            (r"!sensor-value ok 1", ""),
            (r"!sensor-value fail", ""),
            (r"!sensor-sampling ok an.int none", ""),
            (r"#sensor-status 12345.000000 1 an.int nominal 3", ""),
            (r"!sensor-sampling ok an.int differential 2", ""),
            (r"#sensor-status 12345.000000 1 an.int nominal 3", ""),
            (r"!sensor-sampling ok an.int event-rate 2 3", ""),
            (r"!sensor-sampling fail No\_sensor\_name\_given.", ""),
            (r"!sensor-sampling fail Unknown\_sensor\_name:\_an.unknown.", ""),
            (r"!sensor-sampling fail Unknown\_strategy\_name:\_unknown.", ""),
            (r"#log trace", r"root trace-msg"),
            (r"#log debug", r"root debug-msg"),
            (r"#log info", r"root info-msg"),
            (r"#log warn", r"root warn-msg"),
            (r"#log error", r"root error-msg"),
            (r"#log fatal", r"root fatal-msg"),
        ])

    def test_standard_requests_with_ids(self):
        """Test standard request and replies with message ids."""
        get_msgs = self.client.message_recorder(
                blacklist=self.BLACKLIST, replies=True)

        current_id = [0]

        def mid():
            current_id[0] += 1
            return str(current_id[0])

        def mid_req(*args):
            return katcp.Message.request(*args, mid=mid())


        self.client.request(mid_req("watchdog"))
        self.client.request(mid_req("restart"))
        self.client.request(mid_req("log-level"))
        self.client.request(mid_req("log-level", "trace"))
        self.client.request(mid_req("log-level", "unknown"))
        self.client.request(mid_req("help"))
        self.client.request(mid_req("help", "watchdog"))
        self.client.request(mid_req("help", "unknown-request"))
        self.client.request(mid_req("client-list"))
        self.client.request(mid_req("version-list"))
        self.client.request(mid_req("sensor-list"))
        self.client.request(mid_req("sensor-list", "an.int"))
        self.client.request(mid_req("sensor-list", "an.unknown"))
        self.client.request(mid_req("sensor-value"))
        self.client.request(mid_req("sensor-value", "an.int"))
        self.client.request(mid_req("sensor-value", "an.unknown"))
        self.client._next_id = mid  # mock our mid generator for testing
        self.client.blocking_request(mid_req("sensor-sampling", "an.int"))
        self.client.blocking_request(mid_req(
            "sensor-sampling", "an.int", "differential", "2"))
        self.client.blocking_request(mid_req(
            "sensor-sampling", "an.int", "event-rate", "2", "3")),
        self.client.blocking_request(mid_req("sensor-sampling"))
        self.client.blocking_request(mid_req(
            "sensor-sampling", "an.unknown", "auto"))
        self.client.blocking_request(mid_req(
            "sensor-sampling", "an.int", "unknown"))

        self.server.log.trace("trace-msg")
        self.server.log.debug("debug-msg")
        self.server.log.info("info-msg")
        self.server.log.warn("warn-msg")
        self.server.log.error("error-msg")
        self.server.log.fatal("fatal-msg")

        time.sleep(0.1)

        self.assertEqual(self.server.restart_queue.get_nowait(), self.server)
        self._assert_msgs_like(get_msgs(), [
            (r"!watchdog[1] ok", ""),
            (r"!restart[2] ok", ""),
            (r"!log-level[3] ok warn", ""),
            (r"!log-level[4] ok trace", ""),
            (r"!log-level[5] fail Unknown\_logging\_level\_name\_'unknown'",
             ""),
            (r"#help[6] cancel-slow-command Cancel\_slow\_command\_request,\_"
             "resulting\_in\_it\_replying\_immediately", ""),
            (r"#help[6] client-list", ""),
            (r"#help[6] halt", ""),
            (r"#help[6] help", ""),
            (r"#help[6] log-level", ""),
            (r"#help[6] new-command", ""),
            (r"#help[6] raise-exception", ""),
            (r"#help[6] raise-fail", ""),
            (r"#help[6] restart", ""),
            (r"#help[6] sensor-list", ""),
            (r"#help[6] sensor-sampling", ""),
            (r"#help[6] sensor-sampling-clear", ""),
            (r"#help[6] sensor-value", ""),
            (r"#help[6] slow-command", ""),
            (r"#help[6] version-list", ""),
            (r"#help[6] watchdog", ""),
            (r"!help[6] ok %d" % NO_HELP_MESSAGES, ""),
            (r"#help[7] watchdog", ""),
            (r"!help[7] ok 1", ""),
            (r"!help[8] fail", ""),
            (r"#client-list[9]", ""),
            (r"!client-list[9] ok 1", ""),
            (r"#version-list[10] katcp-protocol", ""),
            (r"#version-list[10] katcp-library", ""),
            (r"#version-list[10] katcp-device", ""),
            (r"!version-list[10] ok 3", ""),
            (r"#sensor-list[11] an.int An\_Integer. count integer -5 5", ""),
            (r"!sensor-list[11] ok 1", ""),
            (r"#sensor-list[12] an.int An\_Integer. count integer -5 5", ""),
            (r"!sensor-list[12] ok 1", ""),
            (r"!sensor-list[13] fail", ""),
            (r"#sensor-value[14] 12345.000000 1 an.int nominal 3", ""),
            (r"!sensor-value[14] ok 1", ""),
            (r"#sensor-value[15] 12345.000000 1 an.int nominal 3", ""),
            (r"!sensor-value[15] ok 1", ""),
            (r"!sensor-value[16] fail", ""),
            (r"!sensor-sampling[17] ok an.int none", ""),
            (r"#sensor-status 12345.000000 1 an.int nominal 3", ""),
            (r"!sensor-sampling[18] ok an.int differential 2", ""),
            (r"#sensor-status 12345.000000 1 an.int nominal 3", ""),
            (r"!sensor-sampling[19] ok an.int event-rate 2 3", ""),
            (r"!sensor-sampling[20] fail No\_sensor\_name\_given.", ""),
            (r"!sensor-sampling[21] fail Unknown\_sensor\_name:\_an.unknown.", ""),
            (r"!sensor-sampling[22] fail Unknown\_strategy\_name:\_unknown.", ""),
            (r"#log trace", r"root trace-msg"),
            (r"#log debug", r"root debug-msg"),
            (r"#log info", r"root info-msg"),
            (r"#log warn", r"root warn-msg"),
            (r"#log error", r"root error-msg"),
            (r"#log fatal", r"root fatal-msg"),
        ])

    def test_sensor_list_regex(self):
        reply, informs = self.client.blocking_request(katcp.Message.request(
                "sensor-list", "/a.*/"), use_mid=False)
        self._assert_msgs_equal(informs + [reply], [
            r"#sensor-list an.int An\_Integer. count integer -5 5",
            r"!sensor-list ok 1",
        ])

        reply, informs = self.client.blocking_request(katcp.Message.request(
                "sensor-list", "//"), use_mid=False)
        self._assert_msgs_equal(informs + [reply], [
            r"#sensor-list an.int An\_Integer. count integer -5 5",
            r"!sensor-list ok 1",
        ])

        reply, informs = self.client.blocking_request(katcp.Message.request(
                "sensor-list", "/^int/"), use_mid=False)
        self._assert_msgs_equal(informs + [reply], [
            r"!sensor-list ok 0",
        ])

    def test_sensor_value_regex(self):
        reply, informs = self.client.blocking_request(katcp.Message.request(
                "sensor-value", "/a.*/"), use_mid=False)
        self._assert_msgs_equal(informs + [reply], [
            r"#sensor-value 12345.000000 1 an.int nominal 3",
            r"!sensor-value ok 1",
        ])

        reply, informs = self.client.blocking_request(katcp.Message.request(
                "sensor-value", "//"), use_mid=False)
        self._assert_msgs_equal(informs + [reply], [
            r"#sensor-value 12345.000000 1 an.int nominal 3",
            r"!sensor-value ok 1",
        ])

        reply, informs = self.client.blocking_request(katcp.Message.request(
                "sensor-value", "/^int/"), use_mid=False)
        self._assert_msgs_equal(informs + [reply], [
            r"!sensor-value ok 0",
        ])

    def test_client_list(self):
        reply, informs = self.client.blocking_request(
            katcp.Message.request('client-list'), use_mid=False)
        self.assertEqual(str(reply), '!client-list ok 1')
        self.assertEqual(len(informs), 1)
        inform = str(informs[0])
        self.assertTrue(inform.startswith('#client-list 127.0.0.1:'))
        _, addr = inform.split()
        host, port = addr.split(':')
        port = int(port)
        self.assertEqual((host, port), self.client._sock.getsockname())

    def test_halt_request(self):
        """Test halt request."""
        get_msgs = self.client.message_recorder(
                blacklist=self.BLACKLIST, replies=True)
        self.client.request(katcp.Message.request("halt"))
        # hack to hide re-connect exception
        self.client.connect = lambda: None
        self.server.join()
        time.sleep(0.1)

        self._assert_msgs_equal(get_msgs(), [
            r"!halt[1] ok",
            r"#disconnect Device\_server\_shutting\_down.",
        ])

    def test_bad_handlers(self):
        """Test that bad request and inform handlers are picked up."""
        try:

            class BadServer(katcp.DeviceServer):
                def request_baz(self, req, msg):
                    pass

        except AssertionError:
            pass
        else:
            self.fail("Server metaclass accepted missing request_ docstring.")

        try:

            class BadServer(katcp.DeviceServer):
                def inform_baz(self, req, msg):
                    pass

        except AssertionError:
            pass
        else:
            self.fail("Server metaclass accepted missing inform_ docstring.")

        class SortOfOkayServer(katcp.DeviceServer):
            request_bar = 1
            inform_baz = 2

        assert("bar" not in SortOfOkayServer._request_handlers)
        assert("baz" not in SortOfOkayServer._inform_handlers)

    def test_handler_exceptions(self):
        """Test handling of failure replies and other exceptions."""
        get_msgs = self.client.message_recorder(
                blacklist=self.BLACKLIST, replies=True)
        self.assertTrue(self.client.wait_protocol(timeout=1))

        self.client.request(katcp.Message.request("raise-exception"))
        self.client.request(katcp.Message.request("raise-fail"))

        time.sleep(0.1)

        self._assert_msgs_like(get_msgs(), [
            (r"!raise-exception[1] fail Traceback", ""),
            (r"!raise-fail[2] fail There\_was\_a\_problem\_with\_your\_request.",
             ""),
        ])

    def test_stop_and_restart(self):
        """Test stopping and restarting the device server."""
        self.server.stop(timeout=0.1)
        self.server.join(timeout=1.0)
        self.assertEqual(self.server._thread, None)
        self.assertFalse(self.server._running.isSet())
        self.server.start(timeout=1.0)

    def test_bad_client_socket(self):
        """Test what happens when select is called on a dead client socket."""
        # wait for client to arrive
        time.sleep(0.1)

        # close socket while the server isn't looking
        # then wait for the server to notice
        sock = self.server._socks[0]
        sock.close()
        time.sleep(0.75)

        # check that client was removed
        self.assertTrue(sock not in self.server._socks,
                        "Expected %r to not be in %r" %
                        (sock, self.server._socks))

    def test_bad_server_socket(self):
        """Test what happens when select is called on a dead server socket."""
        # wait for client to arrive
        time.sleep(0.1)

        # close socket while the server isn't looking
        # then wait for the server to notice
        sock = self.server._sock
        sockname = sock.getsockname()
        sock.close()
        time.sleep(0.75)

        # check that server restarted
        self.assertTrue(sock is not self.server._sock,
                        "Expected %r to not be %r" % (sock, self.server._sock))
        self.assertEqual(sockname, self.server._sock.getsockname())

    def test_daemon_value(self):
        """Test passing in a daemon value to server start method."""
        self.server.stop(timeout=0.1)
        self.server.join(timeout=1.0)

        self.server.start(timeout=0.1, daemon=True)
        self.assertTrue(self.server._thread.isDaemon())

    def test_excepthook(self):
        """Test passing in an excepthook to server start method."""
        exceptions = []
        except_event = threading.Event()

        def excepthook(etype, value, traceback):
            """Keep track of exceptions."""
            exceptions.append(etype)
            except_event.set()

        self.server.stop(timeout=0.1)
        self.server.join(timeout=1.5)

        self.server.start(timeout=0.1, excepthook=excepthook)
        # force exception by deleteing _running
        old_running = self.server._running
        try:
            del self.server._running
            except_event.wait(1.5)
            self.assertEqual(exceptions, [AttributeError])
        finally:
            self.server._running = old_running

        # close socket -- server didn't shut down correctly
        self.server._sock.close()
        self.server.stop(timeout=0.1)
        self.server.join(timeout=1.5)

        except_event.clear()
        del exceptions[:]
        self.server.start(timeout=0.1, excepthook=excepthook)
        # force exception in sample reactor and check that it makes
        # it back up
        reactor = self.server._reactor
        old_stop = reactor._stopEvent
        try:
            del reactor._stopEvent
            reactor._wakeEvent.set()
            except_event.wait(0.1)
            self.assertEqual(exceptions, [AttributeError])
        finally:
            reactor._stopEvent = old_stop

        # close socket -- server didn't shut down correctly
        self.server._sock.close()

    def test_sampling(self):
        """Test sensor sampling."""
        get_msgs = self.client.message_recorder(
                blacklist=self.BLACKLIST, replies=True)
        self.client.wait_protocol(timeout=1)
        self.client.request(katcp.Message.request(
            "sensor-sampling", "an.int", "period", 1/32.))

        # Wait for the request reply and for the sensor update messages to
        # arrive. We expect update one the moment the sensor-sampling request is
        # made, then four more over 4/32. of a second, resutling in 6
        # messages. Wait half a period longer just to be sure we get everything.

        self.assertTrue(get_msgs.wait_number(6, timeout=4.5/32.))
        self.client.assert_request_succeeds("sensor-sampling", "an.int", "none")
        # Wait for reply to above request
        get_msgs.wait_number(7)
        msgs = get_msgs()
        updates = [x for x in msgs if x.name == "sensor-status"]
        others = [x for x in msgs if x.name != "sensor-status"]
        self.assertTrue(abs(len(updates) - 5) < 2,
                        "Expected 5 informs, saw %d." % len(updates))

        self._assert_msgs_equal(others, [
            r"!sensor-sampling[1] ok an.int period %s" % (1/32.),
            r"!sensor-sampling[2] ok an.int none",
        ])

        self.assertEqual(updates[0].arguments[1:],
                         ["1", "an.int", "nominal", "3"])

        ## Now clear the strategies on this sensor
        # There should only be on connection to the server, so it should be
        # the test client
        client_conn = self.server._sock_connections.values()[0]
        self.server.clear_strategies(client_conn)
        self.client.assert_request_succeeds("sensor-sampling", "an.int",
                                            args_equal=["an.int", "none"])

        # Check that we did not accidentally clobber the strategy datastructure
        # in the proccess
        self.client.assert_request_succeeds(
            "sensor-sampling", "an.int", "period", 0.125)


    def test_add_remove_sensors(self):
        """Test adding and removing sensors from a running device."""
        an_int = self.server._sensors["an.int"]
        self.server.remove_sensor(an_int)
        self.server.add_sensor(an_int)
        self.test_sampling()

# test_server.py # -*- coding: utf8 -*- # vim:fileencoding=utf8 ai ts=4 sts=4 et sw=4 # Copyright 2009 SKA South Africa (http://ska.ac.za/) # BSD license - see COPYING for details Tests for the server module. # Number of requests on DeviceTestServer # Test that the ClientConnection methods are correctly bound to the # server methods # Test that we can't reply twice # Test that we can't reply twice ## Protocol versions and flags for a katcp v4 server # Timestamp should be in miliseconds # we are expecting 2 inform messages # Check that calls were syntactically valid # Get all the messages sent to _send_message # Test that the periodic update period is converted to seconds # test that parameters returned by the request matches v4 format. # # We need to pass in the same client_conn as used by the previous # request since strategies are bound to specific connections # event-rate is not an allowed v4 strategy # differential-rate is not an allowed v4 strategy ## Protocol versions and flags for a katcp v4 server # Only major versions 4 and 5 are supported # we are expecting 3 inform messages # Check that calls were syntactically valid # Get all the messages sent to _send_message # Will have to be updated for every library version bump Test a simple server setup and teardown with client connect. # basic send # pipe-lined send # broken up sends Test request failure paths in device server. # wait for reply # Test that server does not choke sending messages to slow clients # Set a short sending timeout # Send a bunch of request to the server, but don't read anything from # the server # Request should not have taken a very long time. Test server ignores informs and replies. Test standard request and replies. Test standard request and replies with message ids. # mock our mid generator for testing Test halt request. # hack to hide re-connect exception Test that bad request and inform handlers are picked up. Test handling of failure replies and other exceptions. Test stopping and restarting the device server. Test what happens when select is called on a dead client socket. # wait for client to arrive # close socket while the server isn't looking # then wait for the server to notice # check that client was removed Test what happens when select is called on a dead server socket. # wait for client to arrive # close socket while the server isn't looking # then wait for the server to notice # check that server restarted Test passing in a daemon value to server start method. Test passing in an excepthook to server start method. Keep track of exceptions. # force exception by deleteing _running # close socket -- server didn't shut down correctly # force exception in sample reactor and check that it makes # it back up # close socket -- server didn't shut down correctly Test sensor sampling. # Wait for the request reply and for the sensor update messages to # arrive. We expect update one the moment the sensor-sampling request is # made, then four more over 4/32. of a second, resutling in 6 # messages. Wait half a period longer just to be sure we get everything. # Wait for reply to above request ## Now clear the strategies on this sensor # There should only be on connection to the server, so it should be # the test client # Check that we did not accidentally clobber the strategy datastructure # in the proccess Test adding and removing sensors from a running device.