commit 1d2d684e3975e0e55369485b3d75155e90ea6aef Author: Damian Johnson <atagar@torproject.org> Date: Mon Nov 5 08:36:05 2012 -0800 BW event unit tests Adding a few simple unit tests for the BandwidthEvent class. --- stem/response/events.py | 8 ++++++-- test/unit/response/events.py | 38 +++++++++++++++++++++++++++++++++++--- 2 files changed, 41 insertions(+), 5 deletions(-) diff --git a/stem/response/events.py b/stem/response/events.py index 7a2f9da..fa451c7 100644 --- a/stem/response/events.py +++ b/stem/response/events.py @@ -81,8 +81,12 @@ class BandwidthEvent(Event): _POSITIONAL_ARGS = ("read", "written") def _parse(self): - if (self.read and not self.read.isdigit()) or (self.written and not self.written.isdigit()): - raise stem.socket.ProtocolError("A BW event's bytes sent and received values should be numeric, received: %s" % self) + if not self.read: + raise stem.socket.ProtocolError("BW event is missing its read value") + elif not self.written: + raise stem.socket.ProtocolError("BW event is missing its written value") + elif not self.read.isdigit() or not self.written.isdigit(): + raise stem.socket.ProtocolError("A BW event's bytes sent and received should be a positive numeric value, received: %s" % self) self.read = long(self.read) self.written = long(self.written) diff --git a/test/unit/response/events.py b/test/unit/response/events.py index b1111bc..1dea7c7 100644 --- a/test/unit/response/events.py +++ b/test/unit/response/events.py @@ -6,8 +6,16 @@ import threading import unittest import stem.response +import stem.response.events import test.mocking as mocking +from stem.socket import ProtocolError + +def _get_event(content): + controller_event = mocking.get_message(content) + stem.response.convert("EVENT", controller_event, arrived_at = 25) + return controller_event + class TestEvents(unittest.TestCase): def test_example(self): """ @@ -23,9 +31,7 @@ class TestEvents(unittest.TestCase): def event_sender(): for i in xrange(3): - controller_event = mocking.get_message("650 BW 15 25") - stem.response.convert("EVENT", controller_event, arrived_at = 25) - print_bw(controller_event) + print_bw(_get_event("650 BW 15 25")) time.sleep(0.05) controller = mocking.get_object(Controller, { @@ -40,4 +46,30 @@ class TestEvents(unittest.TestCase): events_thread.start() time.sleep(0.2) events_thread.join() + + def test_bw_event(self): + event = _get_event("650 BW 15 25") + + self.assertTrue(isinstance(event, stem.response.events.BandwidthEvent)) + self.assertEqual(15, event.read) + self.assertEqual(25, event.written) + + event = _get_event("650 BW 0 0") + self.assertEqual(0, event.read) + self.assertEqual(0, event.written) + + # BW events are documented as possibly having various keywords including + # DIR, OR, EXIT, and APP in the future. This is kinda a pointless note + # since tor doesn't actually do it yet (and likely never will), but might + # as well sanity test that it'll be ok. + + event = _get_event("650 BW 10 20 OR=5 EXIT=500") + self.assertEqual(10, event.read) + self.assertEqual(20, event.written) + self.assertEqual({'OR': '5', 'EXIT': '500'}, event.keyword_args) + + self.assertRaises(ProtocolError, _get_event, "650 BW 15") + self.assertRaises(ProtocolError, _get_event, "650 BW -15 25") + self.assertRaises(ProtocolError, _get_event, "650 BW 15 -25") + self.assertRaises(ProtocolError, _get_event, "650 BW x 25")