michael@0: # This Source Code Form is subject to the terms of the Mozilla Public
michael@0: # License, v. 2.0. If a copy of the MPL was not distributed with this
michael@0: # file, You can obtain one at http://mozilla.org/MPL/2.0/.
michael@0: 
michael@0: import re
michael@0: import socket
michael@0: from time import strptime
michael@0: 
michael@0: from dmunit import DeviceManagerTestCase, heartbeat_port
michael@0: 
michael@0: class DataChannelTestCase(DeviceManagerTestCase):
michael@0: 
michael@0:     runs_on_test_device = False
michael@0: 
michael@0:     def runTest(self):
michael@0:         """This tests the heartbeat and the data channel.
michael@0:         """
michael@0:         ip = self.dm.host
michael@0: 
michael@0:         # Let's connect
michael@0:         self._datasock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
michael@0:         # Assume 60 seconds between heartbeats
michael@0:         self._datasock.settimeout(float(60 * 2))
michael@0:         self._datasock.connect((ip, heartbeat_port))
michael@0:         self._connected = True
michael@0: 
michael@0:         # Let's listen
michael@0:         numbeats = 0
michael@0:         capturedHeader = False
michael@0:         while numbeats < 3:
michael@0:             data = self._datasock.recv(1024)
michael@0:             print data
michael@0:             self.assertNotEqual(len(data), 0)
michael@0: 
michael@0:             # Check for the header
michael@0:             if not capturedHeader:
michael@0:                 m = re.match(r"(.*?) trace output", data)
michael@0:                 self.assertNotEqual(m, None,
michael@0:                     'trace output line does not match. The line: ' + str(data))
michael@0:                 capturedHeader = True
michael@0: 
michael@0:             # Check for standard heartbeat messsage
michael@0:             m = re.match(r"(.*?) Thump thump - (.*)", data)
michael@0:             if m == None:
michael@0:                 # This isn't an error, it usually means we've obtained some
michael@0:                 # unexpected data from the device
michael@0:                 continue
michael@0: 
michael@0:             # Ensure it matches our format
michael@0:             mHeartbeatTime = m.group(1)
michael@0:             mHeartbeatTime = strptime(mHeartbeatTime, "%Y%m%d-%H:%M:%S")
michael@0:             numbeats = numbeats + 1