#!/usr/bin/env python

# Copyright (c) 2013 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import at_channel

import fcntl
import functools
import glib
import logging
import mox
import os
import tempfile
import unittest

import task_loop

class ATChannelTestCase(unittest.TestCase):
    """
    Test fixture for ATChannel class.

    """

    def setUp(self):
        self.mox = mox.Mox()

        master, slave = os.openpty()
        self._at_channel = at_channel.ATChannel(
                self._recieve_command_local_callback, slave, 'test')

        # Replace the channel inside _at_channel with a tempfile
        # We will use the tempfile to simulate a tty pair.
        os.close(master)
        os.close(slave)
        self._channel_file = tempfile.TemporaryFile(mode = 'w+')
        # These properties are a copy of the properties set in ATChannel for the
        # tty pair.
        flags = fcntl.fcntl(self._channel_file.fileno(), fcntl.F_GETFL)
        flags = flags | os.O_NONBLOCK
        fcntl.fcntl(self._channel_file.fileno(), fcntl.F_SETFL, flags)
        self._at_channel._channel = self._channel_file.fileno()
        # We need to seek() to the beginning of the file to simulate tty read.
        # So remember the head of the file.
        self._channel_file_head = self._channel_file.tell()

        # Also mock out the task_loop
        self._mox_task_loop = self.mox.CreateMock(task_loop.TaskLoop)
        self._at_channel._task_loop = self._mox_task_loop


    def tearDown(self):
        self._channel_file.close()

    # ##########################################################################
    # Tests

    def test_successful_send(self):
        """
        Test that a single AT command can be sent on the channel.

        """
        payload = 'A not so huge AT+CEREG command.'
        self._at_channel.send(payload)
        received_command = self._recieve_command_remote()
        self.assertTrue(received_command.endswith('\r\n'))
        self.assertEqual(payload.strip(), received_command.strip())

        # Change the AT command guard strings and check again.
        self._at_channel.at_prefix = '$$'
        self._at_channel.at_suffix = '##'
        payload = 'A not so huge AT+CEREG command.'
        self._at_channel.send(payload)
        received_command = self._recieve_command_remote()
        self.assertTrue(received_command.startswith('$$'))
        self.assertTrue(received_command.endswith('##'))
        self.assertEqual(payload.strip(),
                         received_command.strip('$$').strip('##'))


    def test_recieve_single_at_command(self):
        """
        Test that a single AT command can be received together on the channel.

        """
        payload = 'We send you our AT+good wishes too!\r\n'
        callback = lambda channel, payload: None
        self._at_channel._receiver_callback = callback
        self._mox_task_loop.post_task(callback, payload.strip())
        self.mox.ReplayAll()
        self._send_command_remote(payload)
        self._at_channel._handle_channel_cb(self._channel_file.fileno(),
                                          glib.IO_IN)
        self.mox.VerifyAll()


    def test_receive_at_commands_differet_terminators(self):
        """
        Test that AT commands are recieved correctly when different supported
        termination strings are being used.

        """
        # ; is a continuation marker. AT1;2 == AT1\r\nAT2
        payloads = ['AT1\r\nA', 'T2\rA', 'T3\nA', 'T4;', '5\r\n']
        callback = lambda channel, payload: None
        self._at_channel._receiver_callback = callback
        self._mox_task_loop.post_task(callback, 'AT1')
        self._mox_task_loop.post_task(callback, 'AT2')
        self._mox_task_loop.post_task(callback, 'AT3')
        self._mox_task_loop.post_task(callback, 'AT4')
        self._mox_task_loop.post_task(callback, 'AT5')

        self.mox.ReplayAll()
        for payload in payloads:
            self._send_command_remote(payload)
            self._at_channel._handle_channel_cb(self._channel_file.fileno(),
                                                glib.IO_IN)
        self.mox.VerifyAll()


    def test_recieve_at_commands_in_parts(self):
        """
        Test that a multiple AT commands can be received in parts on the
        channel.

        """
        payloads = ['AT1', '11\r\n', '\r\nAT22', '2\r\nAT333', '\r\n']
        callback = lambda channel, payload: None
        self._at_channel._receiver_callback = callback
        self._mox_task_loop.post_task(callback, 'AT111')
        self._mox_task_loop.post_task(callback, 'AT222')
        self._mox_task_loop.post_task(callback, 'AT333')

        self.mox.ReplayAll()
        for payload in payloads:
            self._send_command_remote(payload)
            self._at_channel._handle_channel_cb(self._channel_file.fileno(),
                                                glib.IO_IN)
        self.mox.VerifyAll()


    def test_recieve_long_at_commands(self):
        """
        Test that a multiple AT commands can be received in parts on the
        channel.

        """
        payloads = ['AT1+',
                    '123456789\r\nAT2+123456789\r\nAT3+1234567',
                    '89\r\n']
        callback = lambda channel, payload: None
        self._at_channel._receiver_callback = callback
        self._mox_task_loop.post_task(callback, 'AT1+123456789')
        self._mox_task_loop.post_task(callback, 'AT2+123456789')
        self._mox_task_loop.post_task(callback, 'AT3+123456789')

        self.mox.ReplayAll()
        at_channel.CHANNEL_READ_CHUNK_SIZE = 4
        for payload in payloads:
            self._send_command_remote(payload)
            self._at_channel._handle_channel_cb(self._channel_file.fileno(),
                                                glib.IO_IN)
        self.mox.VerifyAll()

    # ##########################################################################
    # Helper functions

    def _clean_channel_file(self):
        """
        Clean the tempfile used to simulate tty, and reset the r/w head.

        """
        self._channel_file.truncate(0)
        self._channel_file_head = self._channel_file.tell()


    def _send_command_remote(self, payload):
        """
        Simulate a command being sent from the remote tty port.

        @param payload: The command to send.

        """
        self._clean_channel_file()
        self._channel_file.write(payload)
        self._channel_file.flush()
        self._channel_file.seek(self._channel_file_head)


    def _recieve_command_remote(self):
        """
        Simluate a command being received at the remote tty port.

        """
        self._channel_file.flush()
        self._channel_file.seek(self._channel_file_head)
        payload_list = []
        for buf in iter(functools.partial(self._channel_file.read, 128), ''):
            payload_list.append(buf)
        self._clean_channel_file()
        return ''.join(payload_list)


    def _recieve_command_local_callback(self, payload):
        pass


if __name__ == '__main__':
    logging.basicConfig(level=logging.DEBUG)
    unittest.main()