#!/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()