Thu, 22 Jan 2015 13:21:57 +0100
Incorporate requested changes from Mozilla in review:
https://bugzilla.mozilla.org/show_bug.cgi?id=1123480#c6
1 # This Source Code Form is subject to the terms of the Mozilla Public
2 # License, v. 2.0. If a copy of the MPL was not distributed with this
3 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
5 import re
6 import socket
7 from time import strptime
9 from dmunit import DeviceManagerTestCase, heartbeat_port
11 class DataChannelTestCase(DeviceManagerTestCase):
13 runs_on_test_device = False
15 def runTest(self):
16 """This tests the heartbeat and the data channel.
17 """
18 ip = self.dm.host
20 # Let's connect
21 self._datasock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
22 # Assume 60 seconds between heartbeats
23 self._datasock.settimeout(float(60 * 2))
24 self._datasock.connect((ip, heartbeat_port))
25 self._connected = True
27 # Let's listen
28 numbeats = 0
29 capturedHeader = False
30 while numbeats < 3:
31 data = self._datasock.recv(1024)
32 print data
33 self.assertNotEqual(len(data), 0)
35 # Check for the header
36 if not capturedHeader:
37 m = re.match(r"(.*?) trace output", data)
38 self.assertNotEqual(m, None,
39 'trace output line does not match. The line: ' + str(data))
40 capturedHeader = True
42 # Check for standard heartbeat messsage
43 m = re.match(r"(.*?) Thump thump - (.*)", data)
44 if m == None:
45 # This isn't an error, it usually means we've obtained some
46 # unexpected data from the device
47 continue
49 # Ensure it matches our format
50 mHeartbeatTime = m.group(1)
51 mHeartbeatTime = strptime(mHeartbeatTime, "%Y%m%d-%H:%M:%S")
52 numbeats = numbeats + 1