Adding a cross-task eventing server.

testing/legion/examples/events/task.py

Index: testing/legion/examples/events/task.py
diff --git a/testing/legion/examples/events/task.py b/testing/legion/examples/events/task.py
new file mode 100644
index 0000000000000000000000000000000000000000..4fb071497894d157dba2ab22c32a66d756361bc4
--- /dev/null
+++ b/testing/legion/examples/events/task.py
@@ -0,0 +1,47 @@
+# Copyright 2016 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.
+
+"""Task-based unittest for the Legion event server."""
+
+import argparse
+import httplib
+import sys
+import unittest
+
+# pylint: disable=global-variable-not-assigned
+args = None

[M-A Ruel, 2016/01/18 21:57:37]

Make it a class member instead, e.g. around line 17

[Mike Meade, 2016/01/19 21:53:23]

Done.

+
+
+class EventServerTest(unittest.TestCase):
+
+  def Connect(self, verb, path):
+    global args
+    conn = httplib.HTTPConnection(args.address, args.port)
+    conn.request(verb, path)
+    return conn.getresponse().status
+
+  def testSettingGettingAndClearingAnEvent(self):
+    self.assertEquals(self.Connect('GET', '/events/event1'), 404)
+    self.assertEquals(self.Connect('PUT', '/events/event1'), 200)
+    self.assertEquals(self.Connect('GET', '/events/event1'), 200)
+    self.assertEquals(self.Connect('GET', '/events/event2'), 404)
+    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', 'PUT', 'DELETE']:
+      self.assertEquals(self.Connect(verb, '/'), 403)
+      self.assertEquals(self.Connect(verb, '/foobar'), 501)
+      self.assertEquals(self.Connect(verb, '/events'), 501)
+
+
+if __name__ == '__main__':
+  parser = argparse.ArgumentParser()
+  parser.add_argument('--address')
+  parser.add_argument('--port', type=int)
+  parser.add_argument('unittest_args', nargs='*')
+  args = parser.parse_args()

[M-A Ruel, 2016/01/18 21:57:37]

Actually, you should use parse_known_args() here.

[Mike Meade, 2016/01/19 21:53:23]

If I use parse_known_args and no other args are passed in (aside from the ones
we have defined here) then unittest_args isn't defined and we get an
AttributeError when trying to access it. Using parse_args results in an empty
list.

The big problem trying to be solved here is that unittest defines its own flags
and doesn't accept any other args on the command line. Passing additional args
on the command line results in unknown cmd line arg error. So we are parsing,
pulling out address and port, and passing all of the other args to unittest via
unittest_args and sys.arg. 

+  sys.argv[1:] = args.unittest_args
+  unittest.main()