Adding a cross-task eventing server.

testing/legion/unittests/event_server_test.py

Index: testing/legion/unittests/event_server_test.py
diff --git a/testing/legion/unittests/event_server_test.py b/testing/legion/unittests/event_server_test.py
new file mode 100644
index 0000000000000000000000000000000000000000..96253eee7a7c5a86a2d48b22236d70e7639da3dc
--- /dev/null
+++ b/testing/legion/unittests/event_server_test.py
@@ -0,0 +1,46 @@
+# Copyright 2015 The Chromium Authors. All rights reserved.
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+"""Unittest for legion.lib.event_server."""
+
+import httplib
+
+# pylint: disable=relative-import
+import legion_unittest
+
+from legion.lib import event_server
+
+
+class EventServerTest(legion_unittest.TestCase):
+
+  def setUp(self):

[M-A Ruel, 2016/01/15 14:07:13]

super(EventServerTest, self).setUp()

[Mike Meade, 2016/01/18 20:25:51]

The super calls really aren't needed for either method since the base class's
setUp and tearDown are no-ops, but I added them anyways.

+    self.server = event_server.ThreadedServer()
+    self.server.start()
+
+  def tearDown(self):
+    self.server.shutdown()

[M-A Ruel, 2016/01/15 14:07:13]

try:
  self.server.shutdown()
finally:
  super(EventServerTest, self).tearDown()

[Mike Meade, 2016/01/18 20:25:51]

Done.

+
+  def Connect(self, verb, path):
+    conn = httplib.HTTPConnection('localhost', self.server.port)
+    conn.request(verb, path)

[M-A Ruel, 2016/01/15 14:07:13]

make lines 26-28 a single line

[Mike Meade, 2016/01/18 20:25:51]

Changed in both locations.

+    resp = conn.getresponse()
+    return resp.status
+
+  def testSettingGettingAndClearingAnEvent(self):
+    self.assertEquals(self.Connect('GET', '/events/event1'), 404)
+    self.assertEquals(self.Connect('POST', '/events/event1'), 200)
+    self.assertEquals(self.Connect('GET', '/events/event1'), 200)
+    self.assertEquals(self.Connect('DELETE', '/events/event1'), 200)
+    self.assertEquals(self.Connect('DELETE', '/events/event1'), 404)
+    self.assertEquals(self.Connect('GET', '/events/event1'), 404)
+
+  def testErrors(self):
+    for verb in ['GET', 'POST', 'DELETE']:
+      self.assertEquals(self.Connect(verb, '/'), 403)
+      self.assertEquals(self.Connect(verb, '/foobar'), 501)
+      self.assertEquals(self.Connect(verb, '/events'), 501)
+
+
+if __name__ == '__main__':
+  legion_unittest.main()