commit db0651bb31fe78bca33e29553d12098b318d62c9
Author: Arturo Filastò <art(a)fuffa.org>
Date: Fri Mar 29 18:15:56 2013 -0700
Remove test_runner, move tests into ooni/
---
ooni/tests/mocks.py | 168 +++++++++++++++++++++++
ooni/tests/test-class-design.py | 101 ++++++++++++++
ooni/tests/test_director.py | 58 ++++++++
ooni/tests/test_dns.py | 24 ++++
ooni/tests/test_inputunit.py | 29 ++++
ooni/tests/test_managers.py | 215 ++++++++++++++++++++++++++++++
ooni/tests/test_mutate.py | 15 +++
ooni/tests/test_nettest.py | 264 +++++++++++++++++++++++++++++++++++++
ooni/tests/test_otime.py | 15 +++
ooni/tests/test_reporter.py | 238 +++++++++++++++++++++++++++++++++
ooni/tests/test_safe_represent.py | 14 ++
ooni/tests/test_trueheaders.py | 41 ++++++
ooni/tests/test_utils.py | 20 +++
tests/mocks.py | 168 -----------------------
tests/test-class-design.py | 101 --------------
tests/test_director.py | 58 --------
tests/test_dns.py | 24 ----
tests/test_inputunit.py | 29 ----
tests/test_managers.py | 215 ------------------------------
tests/test_mutate.py | 15 ---
tests/test_nettest.py | 264 -------------------------------------
tests/test_otime.py | 15 ---
tests/test_reporter.py | 238 ---------------------------------
tests/test_runner.py | 80 -----------
tests/test_safe_represent.py | 14 --
tests/test_trueheaders.py | 41 ------
tests/test_utils.py | 20 ---
27 files changed, 1202 insertions(+), 1282 deletions(-)
diff --git a/ooni/tests/__init__.py b/ooni/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/ooni/tests/mocks.py b/ooni/tests/mocks.py
new file mode 100644
index 0000000..fed683e
--- /dev/null
+++ b/ooni/tests/mocks.py
@@ -0,0 +1,168 @@
+from ooni.tasks import BaseTask, TaskWithTimeout
+from twisted.python import failure
+from ooni.nettest import NetTest
+from ooni.managers import TaskManager
+from twisted.internet import defer
+
+class MockMeasurementFailOnce(BaseTask):
+ def run(self):
+ f = open('dummyTaskFailOnce.txt', 'w')
+ f.write('fail')
+ f.close()
+ if self.failure >= 1:
+ return defer.succeed(self)
+ else:
+ return defer.fail(failure.Failure)
+
+class MockMeasurementManager(TaskManager):
+ def __init__(self):
+ self.successes = []
+ TaskManager.__init__(self)
+
+ def failed(self, failure, task):
+ pass
+
+ def succeeded(self, result, task):
+ self.successes.append((result, task))
+
+class MockReporter(object):
+ def __init__(self):
+ self.created = defer.Deferred()
+
+ def writeReportEntry(self, entry):
+ pass
+
+ def createReport(self):
+ self.created.callback(self)
+
+ def finish(self):
+ pass
+
+class MockFailure(Exception):
+ pass
+
+## from test_managers
+mockFailure = failure.Failure(MockFailure('mock'))
+
+class MockSuccessTask(BaseTask):
+ def run(self):
+ return defer.succeed(42)
+
+class MockFailTask(BaseTask):
+ def run(self):
+ return defer.fail(mockFailure)
+
+class MockFailOnceTask(BaseTask):
+ def run(self):
+ if self.failures >= 1:
+ return defer.succeed(42)
+ else:
+ return defer.fail(mockFailure)
+
+class MockSuccessTaskWithTimeout(TaskWithTimeout):
+ def run(self):
+ return defer.succeed(42)
+
+class MockFailTaskThatTimesOut(TaskWithTimeout):
+ def run(self):
+ return defer.Deferred()
+
+class MockTimeoutOnceTask(TaskWithTimeout):
+ def run(self):
+ if self.failures >= 1:
+ return defer.succeed(42)
+ else:
+ return defer.Deferred()
+
+class MockFailTaskWithTimeout(TaskWithTimeout):
+ def run(self):
+ return defer.fail(mockFailure)
+
+
+class MockNetTest(object):
+ def __init__(self):
+ self.successes = []
+
+ def succeeded(self, measurement):
+ self.successes.append(measurement)
+
+class MockMeasurement(TaskWithTimeout):
+ def __init__(self, net_test):
+ TaskWithTimeout.__init__(self)
+ self.netTest = net_test
+
+ def succeeded(self, result):
+ return self.netTest.succeeded(42)
+
+class MockSuccessMeasurement(MockMeasurement):
+ def run(self):
+ return defer.succeed(42)
+
+class MockFailMeasurement(MockMeasurement):
+ def run(self):
+ return defer.fail(mockFailure)
+
+class MockFailOnceMeasurement(MockMeasurement):
+ def run(self):
+ if self.failures >= 1:
+ return defer.succeed(42)
+ else:
+ return defer.fail(mockFailure)
+
+class MockDirector(object):
+ def __init__(self):
+ self.successes = []
+
+ def measurementFailed(self, failure, measurement):
+ pass
+
+ def measurementSucceeded(self, measurement):
+ self.successes.append(measurement)
+
+## from test_reporter.py
+class MockOReporter(object):
+ def __init__(self):
+ self.created = defer.Deferred()
+
+ def writeReportEntry(self, entry):
+ return defer.succeed(42)
+
+ def finish(self):
+ pass
+
+ def createReport(self):
+ from ooni.utils import log
+ log.debug("Creating report with %s" % self)
+ self.created.callback(self)
+
+class MockOReporterThatFailsWrite(MockOReporter):
+ def writeReportEntry(self, entry):
+ raise MockFailure
+
+class MockOReporterThatFailsOpen(MockOReporter):
+ def createReport(self):
+ self.created.errback(failure.Failure(MockFailure()))
+
+class MockOReporterThatFailsWriteOnce(MockOReporter):
+ def __init__(self):
+ self.failure = 0
+ MockOReporter.__init__(self)
+
+ def writeReportEntry(self, entry):
+ if self.failure >= 1:
+ return defer.succeed(42)
+ else:
+ self.failure += 1
+ raise MockFailure
+
+class MockTaskManager(TaskManager):
+ def __init__(self):
+ self.successes = []
+ TaskManager.__init__(self)
+
+ def failed(self, failure, task):
+ pass
+
+ def succeeded(self, result, task):
+ self.successes.append((result, task))
+
diff --git a/ooni/tests/test-class-design.py b/ooni/tests/test-class-design.py
new file mode 100644
index 0000000..bb80cd3
--- /dev/null
+++ b/ooni/tests/test-class-design.py
@@ -0,0 +1,101 @@
+#!/usr/bin/env python
+#
+# testing classes to test multiple inheritance.
+# these are not meant to be run by trial, though they could be made to be so.
+# i didn't know where to put them. --isis
+
+import abc
+from pprint import pprint
+from inspect import classify_class_attrs
+
+class PluginBase(object):
+ __metaclass__ = abc.ABCMeta
+
+ @abc.abstractproperty
+ def name(self):
+ return 'you should not see this'
+
+ @name.setter
+ def name(self, value):
+ return 'you should not set this'
+
+ @name.deleter
+ def name(self):
+ return 'you should not del this'
+
+ @abc.abstractmethod
+ def inputParser(self, line):
+ """Do something to parse something."""
+ return
+
+class Foo(object):
+ woo = "this class has some shit in it"
+ def bar(self):
+ print "i'm a Foo.bar()!"
+ print woo
+
+class KwargTest(Foo):
+ _name = "isis"
+
+ #def __new__(cls, *a, **kw):
+ # return super(KwargTest, cls).__new__(cls, *a, **kw)
+
+ @property
+ def name(self):
+ return self._name
+
+ @name.setter
+ def name(self, value):
+ self._name = value
+
+ def __init__(self, *a, **kw):
+ super(KwargTest, self).__init__()
+
+ ## this causes the instantion args to override the class attrs
+ for key, value in kw.items():
+ setattr(self.__class__, key, value)
+
+ print "%s.__init__(): self.__dict__ = %s" \
+ % (type(self), pprint(type(self).__dict__))
+
+ for attr in classify_class_attrs(self):
+ print attr
+
+ @classmethod
+ def sayname(cls):
+ print cls.name
+
+class KwargTestChild(KwargTest):
+ name = "arturo"
+ def __init__(self):
+ super(KwargTestChild, self).__init__()
+ print self.name
+
+class KwargTestChildOther(KwargTest):
+ def __init__(self, name="robot", does="lasers"):
+ super(KwargTestChildOther, self).__init__()
+ print self.name
+
+
+if __name__ == "__main__":
+ print "class KwargTest attr name: %s" % KwargTest.name
+ kwargtest = KwargTest()
+ print "KwargTest instantiated wo args"
+ print "kwargtest.name: %s" % kwargtest.name
+ print "kwargtest.sayname(): %s" % kwargtest.sayname()
+ kwargtest2 = KwargTest(name="lovecruft", does="hacking")
+ print "KwargTest instantiated with name args"
+ print "kwargtest.name: %s" % kwargtest2.name
+ print "kwargtest.sayname(): %s" % kwargtest2.sayname()
+
+ print "class KwargTestChild attr name: %s" % KwargTestChild.name
+ kwargtestchild = KwargTestChild()
+ print "KwargTestChild instantiated wo args"
+ print "kwargtestchild.name: %s" % kwargtestchild.name
+ print "kwargtestchild.sayname(): %s" % kwargtestchild.sayname()
+
+ print "class KwargTestChildOther attr name: %s" % KwargTestChildOther.name
+ kwargtestchildother = KwargTestChildOther()
+ print "KwargTestChildOther instantiated wo args"
+ print "kwargtestchildother.name: %s" % kwargtestchildother.name
+ print "kwargtestchildother.sayname(): %s" % kwargtestchildother.sayname()
diff --git a/ooni/tests/test_director.py b/ooni/tests/test_director.py
new file mode 100644
index 0000000..79a1e96
--- /dev/null
+++ b/ooni/tests/test_director.py
@@ -0,0 +1,58 @@
+from twisted.internet import defer, base
+from twisted.trial import unittest
+
+from ooni.director import Director
+from ooni.nettest import NetTestLoader
+from tests.mocks import MockReporter
+base.DelayedCall.debug = True
+
+net_test_string = """
+from twisted.python import usage
+from ooni.nettest import NetTestCase
+
+class UsageOptions(usage.Options):
+ optParameters = [['spam', 's', None, 'ham']]
+
+class DummyTestCase(NetTestCase):
+ inputFile = ['file', 'f', None, 'The input File']
+
+ usageOptions = UsageOptions
+
+ def test_a(self):
+ self.report['bar'] = 'bar'
+
+ def test_b(self):
+ self.report['foo'] = 'foo'
+"""
+
+
+dummyArgs = ('--spam', 1, '--file', 'dummyInputFile.txt')
+
+class TestDirector(unittest.TestCase):
+ timeout = 1
+ def setUp(self):
+ with open('dummyInputFile.txt', 'w') as f:
+ for i in range(10):
+ f.write("%s\n" % i)
+
+ self.reporters = [MockReporter()]
+ self.director = Director()
+
+ def tearDown(self):
+ pass
+
+ def test_start_net_test(self):
+ options = {'test':net_test_string, 'subargs':dummyArgs}
+ net_test_loader = NetTestLoader(options)
+ net_test_loader.checkOptions()
+ d = self.director.startNetTest('', net_test_loader, self.reporters)
+
+ @d.addCallback
+ def done(result):
+ self.assertEqual(self.director.successfulMeasurements, 20)
+
+ return d
+
+ def test_stop_net_test(self):
+ pass
+
diff --git a/ooni/tests/test_dns.py b/ooni/tests/test_dns.py
new file mode 100644
index 0000000..e9bb524
--- /dev/null
+++ b/ooni/tests/test_dns.py
@@ -0,0 +1,24 @@
+#
+# This unittest is to verify that our usage of the twisted DNS resolver does
+# not break with new versions of twisted.
+
+import pdb
+from twisted.trial import unittest
+
+from twisted.internet import reactor
+
+from twisted.names import dns
+from twisted.names.client import Resolver
+
+class DNSTest(unittest.TestCase):
+ def test_a_lookup_ooni_query(self):
+ def done_query(message, *arg):
+ answer = message.answers[0]
+ self.assertEqual(answer.type, 1)
+
+ dns_query = [dns.Query('ooni.nu', type=dns.A)]
+ resolver = Resolver(servers=[('8.8.8.8', 53)])
+ d = resolver.queryUDP(dns_query)
+ d.addCallback(done_query)
+ return d
+
diff --git a/ooni/tests/test_inputunit.py b/ooni/tests/test_inputunit.py
new file mode 100644
index 0000000..1f9043c
--- /dev/null
+++ b/ooni/tests/test_inputunit.py
@@ -0,0 +1,29 @@
+import unittest
+from ooni.inputunit import InputUnit, InputUnitFactory
+
+def dummyGenerator():
+ for x in range(100):
+ yield x
+
+class TestInputUnit(unittest.TestCase):
+ def test_input_unit_factory(self):
+ inputUnit = InputUnitFactory(range(100))
+ for i in inputUnit:
+ self.assertEqual(len(list(i)), inputUnit.inputUnitSize)
+
+ def test_input_unit(self):
+ inputs = range(100)
+ inputUnit = InputUnit(inputs)
+ idx = 0
+ for i in inputUnit:
+ idx += 1
+
+ self.assertEqual(idx, 100)
+
+ def test_input_unit_factory_length(self):
+ inputUnitFactory = InputUnitFactory(range(100))
+ l1 = len(inputUnitFactory)
+ l2 = sum(1 for _ in inputUnitFactory)
+ self.assertEqual(l1, 10)
+ self.assertEqual(l2, 10)
+
diff --git a/ooni/tests/test_managers.py b/ooni/tests/test_managers.py
new file mode 100644
index 0000000..39f0881
--- /dev/null
+++ b/ooni/tests/test_managers.py
@@ -0,0 +1,215 @@
+from twisted.trial import unittest
+from twisted.python import failure
+from twisted.internet import defer, task
+
+from ooni.tasks import BaseTask, TaskWithTimeout, TaskTimedOut
+from ooni.managers import TaskManager, MeasurementManager
+
+from tests.mocks import MockSuccessTask, MockFailTask, MockFailOnceTask, MockFailure
+from tests.mocks import MockSuccessTaskWithTimeout, MockFailTaskThatTimesOut
+from tests.mocks import MockTimeoutOnceTask, MockFailTaskWithTimeout
+from tests.mocks import MockTaskManager, mockFailure, MockDirector
+from tests.mocks import MockNetTest, MockMeasurement, MockSuccessMeasurement
+from tests.mocks import MockFailMeasurement, MockFailOnceMeasurement
+
+class TestTaskManager(unittest.TestCase):
+ timeout = 1
+ def setUp(self):
+ self.measurementManager = MockTaskManager()
+ self.measurementManager.concurrency = 20
+ self.measurementManager.retries = 2
+
+ self.measurementManager.start()
+
+ self.clock = task.Clock()
+
+ def schedule_successful_tasks(self, task_type, number=1):
+ all_done = []
+ for x in range(number):
+ mock_task = task_type()
+ all_done.append(mock_task.done)
+ self.measurementManager.schedule(mock_task)
+
+ d = defer.DeferredList(all_done)
+ @d.addCallback
+ def done(res):
+ for task_result, task_instance in self.measurementManager.successes:
+ self.assertEqual(task_result, 42)
+ self.assertIsInstance(task_instance, task_type)
+
+ return d
+
+ def schedule_failing_tasks(self, task_type, number=1):
+ all_done = []
+ for x in range(number):
+ mock_task = task_type()
+ all_done.append(mock_task.done)
+ self.measurementManager.schedule(mock_task)
+
+ d = defer.DeferredList(all_done)
+ @d.addCallback
+ def done(res):
+ # 10*2 because 2 is the number of retries
+ self.assertEqual(len(self.measurementManager.failures), number*3)
+ for task_result, task_instance in self.measurementManager.failures:
+ self.assertEqual(task_result, mockFailure)
+ self.assertIsInstance(task_instance, task_type)
+
+ return d
+
+ def test_schedule_failing_with_mock_failure_task(self):
+ mock_task = MockFailTask()
+ self.measurementManager.schedule(mock_task)
+ self.assertFailure(mock_task.done, MockFailure)
+ return mock_task.done
+
+ def test_schedule_successful_one_task(self):
+ return self.schedule_successful_tasks(MockSuccessTask)
+
+ def test_schedule_successful_one_task_with_timeout(self):
+ return self.schedule_successful_tasks(MockSuccessTaskWithTimeout)
+
+ def test_schedule_failing_tasks_that_timesout(self):
+ self.measurementManager.retries = 0
+
+ task_type = MockFailTaskThatTimesOut
+ task_timeout = 5
+
+ mock_task = task_type()
+ mock_task.timeout = task_timeout
+ mock_task.clock = self.clock
+
+ self.measurementManager.schedule(mock_task)
+
+ self.clock.advance(task_timeout)
+
+ @mock_task.done.addBoth
+ def done(res):
+ self.assertEqual(len(self.measurementManager.failures), 1)
+ for task_result, task_instance in self.measurementManager.failures:
+ self.assertIsInstance(task_instance, task_type)
+
+ return mock_task.done
+
+ def test_schedule_time_out_once(self):
+ task_type = MockTimeoutOnceTask
+ task_timeout = 5
+
+ mock_task = task_type()
+ mock_task.timeout = task_timeout
+ mock_task.clock = self.clock
+
+ self.measurementManager.schedule(mock_task)
+
+ self.clock.advance(task_timeout)
+
+ @mock_task.done.addBoth
+ def done(res):
+ self.assertEqual(len(self.measurementManager.failures), 1)
+ for task_result, task_instance in self.measurementManager.failures:
+ self.assertIsInstance(task_instance, task_type)
+
+ for task_result, task_instance in self.measurementManager.successes:
+ self.assertEqual(task_result, 42)
+ self.assertIsInstance(task_instance, task_type)
+
+ return mock_task.done
+
+
+ def test_schedule_failing_one_task(self):
+ return self.schedule_failing_tasks(MockFailTask)
+
+ def test_schedule_failing_one_task_with_timeout(self):
+ return self.schedule_failing_tasks(MockFailTaskWithTimeout)
+
+ def test_schedule_successful_ten_tasks(self):
+ return self.schedule_successful_tasks(MockSuccessTask, number=10)
+
+ def test_schedule_failing_ten_tasks(self):
+ return self.schedule_failing_tasks(MockFailTask, number=10)
+
+ def test_schedule_successful_27_tasks(self):
+ return self.schedule_successful_tasks(MockSuccessTask, number=27)
+
+ def test_schedule_failing_27_tasks(self):
+ return self.schedule_failing_tasks(MockFailTask, number=27)
+
+ def test_task_retry_and_succeed(self):
+ mock_task = MockFailOnceTask()
+ self.measurementManager.schedule(mock_task)
+
+ @mock_task.done.addCallback
+ def done(res):
+ self.assertEqual(len(self.measurementManager.failures), 1)
+
+ self.assertEqual(self.measurementManager.failures,
+ [(mockFailure, mock_task)])
+ self.assertEqual(self.measurementManager.successes,
+ [(42, mock_task)])
+
+ return mock_task.done
+
+ def dd_test_task_retry_and_succeed_56_tasks(self):
+ """
+ XXX this test fails in a non-deterministic manner.
+ """
+ all_done = []
+ number = 56
+ for x in range(number):
+ mock_task = MockFailOnceTask()
+ all_done.append(mock_task.done)
+ self.measurementManager.schedule(mock_task)
+
+ d = defer.DeferredList(all_done)
+
+ @d.addCallback
+ def done(res):
+ self.assertEqual(len(self.measurementManager.failures), number)
+
+ for task_result, task_instance in self.measurementManager.successes:
+ self.assertEqual(task_result, 42)
+ self.assertIsInstance(task_instance, MockFailOnceTask)
+
+ return d
+
+class TestMeasurementManager(unittest.TestCase):
+ def setUp(self):
+ mock_director = MockDirector()
+
+ self.measurementManager = MeasurementManager()
+ self.measurementManager.director = mock_director
+
+ self.measurementManager.concurrency = 10
+ self.measurementManager.retries = 2
+
+ self.measurementManager.start()
+
+ self.mockNetTest = MockNetTest()
+
+ def test_schedule_and_net_test_notified(self, number=1):
+ # XXX we should probably be inheriting from the base test class
+ mock_task = MockSuccessMeasurement(self.mockNetTest)
+ self.measurementManager.schedule(mock_task)
+
+ @mock_task.done.addCallback
+ def done(res):
+ self.assertEqual(self.mockNetTest.successes,
+ [42])
+
+ self.assertEqual(len(self.mockNetTest.successes), 1)
+ return mock_task.done
+
+ def test_schedule_failing_one_measurement(self):
+ mock_task = MockFailMeasurement(self.mockNetTest)
+ self.measurementManager.schedule(mock_task)
+
+ @mock_task.done.addErrback
+ def done(failure):
+ self.assertEqual(len(self.measurementManager.failures), 3)
+
+ self.assertEqual(failure, mockFailure)
+ self.assertEqual(len(self.mockNetTest.successes), 0)
+
+ return mock_task.done
+
+
diff --git a/ooni/tests/test_mutate.py b/ooni/tests/test_mutate.py
new file mode 100644
index 0000000..7e30586
--- /dev/null
+++ b/ooni/tests/test_mutate.py
@@ -0,0 +1,15 @@
+import unittest
+from ooni.kit import daphn3
+
+class TestDaphn3(unittest.TestCase):
+ def test_mutate_string(self):
+ original_string = '\x00\x00\x00'
+ mutated = daphn3.daphn3MutateString(original_string, 1)
+ self.assertEqual(mutated, '\x00\x01\x00')
+ def test_mutate_daphn3(self):
+ original_dict = [{'client': '\x00\x00\x00'},
+ {'server': '\x00\x00\x00'}]
+ mutated_dict = daphn3.daphn3Mutate(original_dict, 1, 1)
+ self.assertEqual(mutated_dict, [{'client': '\x00\x00\x00'},
+ {'server': '\x00\x01\x00'}])
+
diff --git a/ooni/tests/test_nettest.py b/ooni/tests/test_nettest.py
new file mode 100644
index 0000000..a0ccb32
--- /dev/null
+++ b/ooni/tests/test_nettest.py
@@ -0,0 +1,264 @@
+import os
+from StringIO import StringIO
+from tempfile import TemporaryFile, mkstemp
+
+from twisted.trial import unittest
+from twisted.internet import defer, reactor
+from twisted.python.usage import UsageError
+
+from ooni.nettest import NetTest, InvalidOption, MissingRequiredOption
+from ooni.nettest import NetTestLoader, FailureToLoadNetTest
+from ooni.tasks import BaseTask
+from ooni.utils import NotRootError
+
+from ooni.director import Director
+
+from ooni.managers import TaskManager
+
+from tests.mocks import MockMeasurement, MockMeasurementFailOnce
+from tests.mocks import MockNetTest, MockDirector, MockReporter
+from tests.mocks import MockMeasurementManager
+defer.setDebugging(True)
+
+net_test_string = """
+from twisted.python import usage
+from ooni.nettest import NetTestCase
+
+class UsageOptions(usage.Options):
+ optParameters = [['spam', 's', None, 'ham']]
+
+class DummyTestCase(NetTestCase):
+
+ usageOptions = UsageOptions
+
+ def test_a(self):
+ self.report['bar'] = 'bar'
+
+ def test_b(self):
+ self.report['foo'] = 'foo'
+"""
+
+net_test_root_required = net_test_string+"""
+ requiresRoot = True
+"""
+
+net_test_string_with_file = """
+from twisted.python import usage
+from ooni.nettest import NetTestCase
+
+class UsageOptions(usage.Options):
+ optParameters = [['spam', 's', None, 'ham']]
+
+class DummyTestCase(NetTestCase):
+ inputFile = ['file', 'f', None, 'The input File']
+
+ usageOptions = UsageOptions
+
+ def test_a(self):
+ self.report['bar'] = 'bar'
+
+ def test_b(self):
+ self.report['foo'] = 'foo'
+"""
+
+net_test_with_required_option = """
+from twisted.python import usage
+from ooni.nettest import NetTestCase
+
+class UsageOptions(usage.Options):
+ optParameters = [['spam', 's', None, 'ham'],
+ ['foo', 'o', None, 'moo'],
+ ['bar', 'o', None, 'baz'],
+ ]
+
+class DummyTestCase(NetTestCase):
+ inputFile = ['file', 'f', None, 'The input File']
+
+ usageOptions = UsageOptions
+
+ def test_a(self):
+ self.report['bar'] = 'bar'
+
+ def test_b(self):
+ self.report['foo'] = 'foo'
+
+ requiredOptions = ['foo', 'bar']
+"""
+
+dummyInputs = range(1)
+dummyArgs = ('--spam', 'notham')
+dummyOptions = {'spam':'notham'}
+dummyInvalidArgs = ('--cram', 'jam')
+dummyInvalidOptions= {'cram':'jam'}
+dummyArgsWithRequiredOptions = ('--foo', 'moo', '--bar', 'baz')
+dummyRequiredOptions = {'foo':'moo', 'bar':'baz'}
+dummyArgsWithFile = ('--spam', 'notham', '--file', 'dummyInputFile.txt')
+
+class TestNetTest(unittest.TestCase):
+ timeout = 1
+ def setUp(self):
+ with open('dummyInputFile.txt', 'w') as f:
+ for i in range(10):
+ f.write("%s\n" % i)
+
+ def assertCallable(self, thing):
+ self.assertIn('__call__', dir(thing))
+
+ def verifyMethods(self, testCases):
+ uniq_test_methods = set()
+ for test_class, test_methods in testCases:
+ instance = test_class()
+ for test_method in test_methods:
+ c = getattr(instance, test_method)
+ self.assertCallable(c)
+ uniq_test_methods.add(test_method)
+ self.assertEqual(set(['test_a', 'test_b']), uniq_test_methods)
+
+ def test_load_net_test_from_file(self):
+ """
+ Given a file verify that the net test cases are properly
+ generated.
+ """
+ __, net_test_file = mkstemp()
+ with open(net_test_file, 'w') as f:
+ f.write(net_test_string)
+ f.close()
+
+ options = {'subargs':dummyArgs, 'test':net_test_file}
+ ntl = NetTestLoader(options)
+ self.verifyMethods(ntl.testCases)
+ os.unlink(net_test_file)
+
+ def test_load_net_test_from_str(self):
+ """
+ Given a file like object verify that the net test cases are properly
+ generated.
+ """
+ options = {'subargs':dummyArgs, 'test':net_test_string}
+ ntl = NetTestLoader(options)
+ self.verifyMethods(ntl.testCases)
+
+ def test_load_net_test_from_StringIO(self):
+ """
+ Given a file like object verify that the net test cases are properly
+ generated.
+ """
+ options = {'subargs':dummyArgs, 'test':StringIO(net_test_string)}
+ ntl = NetTestLoader(options)
+ self.verifyMethods(ntl.testCases)
+
+ def test_load_with_option(self):
+ options = {'subargs':dummyArgs, 'test':StringIO(net_test_string)}
+ ntl = NetTestLoader(options)
+ self.assertIsInstance(ntl, NetTestLoader)
+ for test_klass, test_meth in ntl.testCases:
+ for option in dummyOptions.keys():
+ self.assertIn(option, test_klass.usageOptions())
+
+ def test_load_with_invalid_option(self):
+ options = {'subargs':dummyInvalidArgs,
+ 'test':StringIO(net_test_string)}
+ try:
+ ntl = NetTestLoader(options)
+ ntl.checkOptions()
+ raise Exception
+ except UsageError:
+ pass
+
+ def test_load_with_required_option(self):
+ options = {'subargs':dummyArgsWithRequiredOptions,
+ 'test':StringIO(net_test_with_required_option)}
+ net_test = NetTestLoader(options)
+ self.assertIsInstance(net_test, NetTestLoader)
+
+ def test_load_with_missing_required_option(self):
+ options = {'subargs':dummyArgs,
+ 'test':StringIO(net_test_with_required_option)}
+ try:
+ net_test = NetTestLoader(options)
+ except MissingRequiredOption:
+ pass
+
+ def test_net_test_inputs(self):
+ options = {'subargs':dummyArgsWithFile,
+ 'test':StringIO(net_test_string_with_file)}
+ ntl = NetTestLoader(options)
+ ntl.checkOptions()
+
+ # XXX: if you use the same test_class twice you will have consumed all
+ # of its inputs!
+ tested = set([])
+ for test_class, test_method in ntl.testCases:
+ if test_class not in tested:
+ tested.update([test_class])
+ self.assertEqual(len(list(test_class.inputs)), 10)
+
+ def test_setup_local_options_in_test_cases(self):
+ options = {'subargs':dummyArgs, 'test':StringIO(net_test_string)}
+ ntl = NetTestLoader(options)
+ ntl.checkOptions()
+
+ for test_class, test_method in ntl.testCases:
+ self.assertEqual(test_class.localOptions, dummyOptions)
+
+ def test_generate_measurements_size(self):
+
+ options = {'subargs':dummyArgsWithFile,
+ 'test':StringIO(net_test_string_with_file)}
+ ntl = NetTestLoader(options)
+ ntl.checkOptions()
+ net_test = NetTest(ntl, None)
+
+ measurements = list(net_test.generateMeasurements())
+ self.assertEqual(len(measurements), 20)
+
+ def test_net_test_completed_callback(self):
+ options = {'subargs':dummyArgsWithFile,
+ 'test':StringIO(net_test_string_with_file)}
+ ntl = NetTestLoader(options)
+ ntl.checkOptions()
+ director = Director()
+
+ d = director.startNetTest('', ntl, [MockReporter()])
+
+ @d.addCallback
+ def complete(result):
+ #XXX: why is the return type (True, None) ?
+ self.assertEqual(result, [(True,None)])
+ self.assertEqual(director.successfulMeasurements, 20)
+
+ return d
+
+ def test_require_root_succeed(self):
+ #XXX: will require root to run
+ options = {'subargs':dummyArgs,
+ 'test':StringIO(net_test_root_required)}
+ ntl = NetTestLoader(options)
+ for test_class, method in ntl.testCases:
+ self.assertTrue(test_class.requiresRoot)
+
+ #def test_require_root_failed(self):
+ # #XXX: will fail if you run as root
+ # try:
+ # net_test = NetTestLoader(StringIO(net_test_root_required),
+ # dummyArgs)
+ # except NotRootError:
+ # pass
+
+ #def test_create_report_succeed(self):
+ # pass
+
+ #def test_create_report_failed(self):
+ # pass
+
+ #def test_run_all_test(self):
+ # raise NotImplementedError
+
+ #def test_resume_test(self):
+ # pass
+
+ #def test_progress(self):
+ # pass
+
+ #def test_time_out(self):
+ # raise NotImplementedError
diff --git a/ooni/tests/test_otime.py b/ooni/tests/test_otime.py
new file mode 100644
index 0000000..80979f2
--- /dev/null
+++ b/ooni/tests/test_otime.py
@@ -0,0 +1,15 @@
+import unittest
+from datetime import datetime
+from ooni import otime
+
+test_date = datetime(2002, 6, 26, 22, 45, 49)
+
+class TestOtime(unittest.TestCase):
+ def test_timestamp(self):
+ self.assertEqual(otime.timestamp(test_date), "2002-06-26T224549Z")
+
+ def test_fromTimestamp(self):
+ time_stamp = otime.timestamp(test_date)
+ self.assertEqual(test_date, otime.fromTimestamp(time_stamp))
+
+
diff --git a/ooni/tests/test_reporter.py b/ooni/tests/test_reporter.py
new file mode 100644
index 0000000..e21b7a1
--- /dev/null
+++ b/ooni/tests/test_reporter.py
@@ -0,0 +1,238 @@
+from twisted.internet import defer
+from twisted.trial import unittest
+
+from ooni.reporter import Report, YAMLReporter, OONIBReporter, safe_dump
+from ooni.managers import ReportEntryManager, TaskManager
+from ooni.nettest import NetTest, NetTestState
+from ooni.errors import ReportNotCreated, ReportAlreadyClosed
+
+from ooni.tasks import TaskWithTimeout
+from tests.mocks import MockOReporter, MockTaskManager
+from tests.mocks import MockMeasurement, MockNetTest
+from tests.mocks import MockOReporterThatFailsWrite
+from tests.mocks import MockOReporterThatFailsWriteOnce
+from tests.mocks import MockOReporterThatFailsOpen
+
+from twisted.python import failure
+import yaml
+
+class TestReport(unittest.TestCase):
+ def setUp(self):
+ pass
+ def tearDown(self):
+ pass
+ def test_create_report_with_no_reporter(self):
+ report = Report([],ReportEntryManager())
+ self.assertIsInstance(report, Report)
+
+ def test_create_report_with_single_reporter(self):
+ report = Report([MockOReporter()], ReportEntryManager())
+ self.assertIsInstance(report, Report)
+
+ def test_create_report_with_multiple_reporters(self):
+ report = Report([MockOReporter() for x in xrange(3)],
+ ReportEntryManager())
+ self.assertIsInstance(report, Report)
+
+ def test_report_open_with_single_reporter(self):
+ report = Report([MockOReporter()],ReportEntryManager())
+ d = report.open()
+ return d
+
+ def test_report_open_with_multiple_reporter(self):
+ report = Report([MockOReporter() for x in xrange(3)],
+ ReportEntryManager())
+ d = report.open()
+ return d
+
+ def test_fail_to_open_report_with_single_reporter(self):
+ report = Report([MockOReporterThatFailsOpen()],
+ ReportEntryManager())
+ d = report.open()
+ def f(x):
+ self.assertEquals(len(report.reporters), 0)
+ d.addCallback(f)
+ return d
+
+ def test_fail_to_open_single_report_with_multiple_reporter(self):
+ report = Report([MockOReporterThatFailsOpen(), MockOReporter(),
+ MockOReporter()], ReportEntryManager())
+ d = report.open()
+ def f(x):
+ self.assertEquals(len(report.reporters),2)
+ d.addCallback(f)
+ return d
+
+ def test_fail_to_open_all_reports_with_multiple_reporter(self):
+ report = Report([MockOReporterThatFailsOpen() for x in xrange(3)],
+ ReportEntryManager())
+ d = report.open()
+ def f(x):
+ self.assertEquals(len(report.reporters),0)
+ d.addCallback(f)
+ return d
+
+ def test_write_report_with_single_reporter_and_succeed(self):
+ #XXX: verify that the MockOReporter writeReportEntry succeeds
+ report = Report([MockOReporter()], ReportEntryManager())
+ report.open()
+ d = report.write(MockMeasurement(MockNetTest()))
+ return d
+
+ def test_write_report_with_single_reporter_and_fail_after_timeout(self):
+ report = Report([MockOReporterThatFailsWrite()], ReportEntryManager())
+ report.open()
+ d = report.write(MockMeasurement(MockNetTest()))
+ def f(err):
+ self.assertEquals(len(report.reporters),0)
+ d.addBoth(f)
+ return d
+
+ def test_write_report_with_single_reporter_and_succeed_after_timeout(self):
+ report = Report([MockOReporterThatFailsWriteOnce()], ReportEntryManager())
+ report.open()
+ d = report.write(MockMeasurement(MockNetTest()))
+ return d
+
+ def test_write_report_with_multiple_reporter_and_succeed(self):
+ report = Report([MockOReporter() for x in xrange(3)], ReportEntryManager())
+ report.open()
+ d = report.write(MockMeasurement(MockNetTest()))
+ return d
+
+ def test_write_report_with_multiple_reporter_and_fail_a_single_reporter(self):
+ report = Report([MockOReporter(), MockOReporter(), MockOReporterThatFailsWrite()], ReportEntryManager())
+ d = report.open()
+
+ self.assertEquals(len(report.reporters),3)
+ d = report.write(MockMeasurement(MockNetTest()))
+
+ def f(x):
+ # one of the reporters should have been removed
+ self.assertEquals(len(report.reporters), 2)
+ d.addBoth(f)
+ return d
+
+ def test_write_report_with_multiple_reporter_and_fail_all_reporter(self):
+ report = Report([MockOReporterThatFailsWrite() for x in xrange(3)], ReportEntryManager())
+ report.open()
+ d = report.write(MockMeasurement(MockNetTest()))
+ def f(err):
+ self.assertEquals(len(report.reporters),0)
+ d.addErrback(f)
+ return d
+
+class TestYAMLReporter(unittest.TestCase):
+ def setUp(self):
+ self.testDetails = {'software_name': 'ooniprobe', 'options':
+ {'pcapfile': None, 'help': 0, 'subargs': ['-f', 'alexa_10'], 'resume':
+ 0, 'parallelism': '10', 'no-default-reporter': 0, 'testdeck': None,
+ 'test': 'nettests/blocking/http_requests.py', 'logfile': None,
+ 'collector': None, 'reportfile': None}, 'test_version': '0.2.3',
+ 'software_version': '0.0.10', 'test_name': 'http_requests_test',
+ 'start_time': 1362054343.0, 'probe_asn': 'AS0', 'probe_ip':
+ '127.0.0.1', 'probe_cc': 'US'}
+
+ def tearDown(self):
+ pass
+ def test_create_yaml_reporter(self):
+ self.assertIsInstance(YAMLReporter(self.testDetails),
+ YAMLReporter)
+
+ def test_open_yaml_report_and_succeed(self):
+ r = YAMLReporter(self.testDetails)
+ r.createReport()
+ # verify that testDetails was written to report properly
+ def f(r):
+ r._stream.seek(0)
+ details, = yaml.safe_load_all(r._stream)
+ self.assertEqual(details, self.testDetails)
+ r.created.addCallback(f)
+ return r.created
+
+ #def test_open_yaml_report_and_fail(self):
+ # #XXX: YAMLReporter does not handle failures of this type
+ # pass
+
+ def test_write_yaml_report_entry(self):
+ r = YAMLReporter(self.testDetails)
+ r.createReport()
+
+ report_entry = {'foo':'bar', 'bin':'baz'}
+ r.writeReportEntry(report_entry)
+
+ # verify that details and entry were written to report
+ def f(r):
+ r._stream.seek(0)
+ report = yaml.safe_load_all(r._stream)
+ details, entry = report
+ self.assertEqual(details, self.testDetails)
+ self.assertEqual(entry, report_entry)
+ r.created.addCallback(f)
+ return r.created
+
+ def test_write_multiple_yaml_report_entry(self):
+ r = YAMLReporter(self.testDetails)
+ r.createReport()
+ def reportEntry():
+ for x in xrange(10):
+ yield {'foo':'bar', 'bin':'baz', 'item':x}
+ for entry in reportEntry():
+ r.writeReportEntry(entry)
+ # verify that details and multiple entries were written to report
+ def f(r):
+ r._stream.seek(0)
+ report = yaml.safe_load_all(r._stream)
+ details = report.next()
+ self.assertEqual(details, self.testDetails)
+ self.assertEqual([r for r in report], [r for r in reportEntry()])
+ r.created.addCallback(f)
+ return r.created
+
+ def test_close_yaml_report(self):
+ r = YAMLReporter(self.testDetails)
+ r.createReport()
+ r.finish()
+ self.assertTrue(r._stream.closed)
+
+ def test_write_yaml_report_after_close(self):
+ r = YAMLReporter(self.testDetails)
+ r.createReport()
+ r.finish()
+ def f(r):
+ r.writeReportEntry("foo")
+ r.created.addCallback(f)
+ self.assertFailure(r.created, ReportAlreadyClosed)
+
+ def test_write_yaml_report_before_open(self):
+ r = YAMLReporter(self.testDetails)
+ def f(r):
+ r.writeReportEntry("foo")
+ r.created.addCallback(f)
+ self.assertFailure(r.created, ReportNotCreated)
+
+#class TestOONIBReporter(unittest.TestCase):
+# def setUp(self):
+# pass
+# def tearDown(self):
+# pass
+# def test_create_oonib_reporter(self):
+# raise NotImplementedError
+# def test_open_oonib_report_and_succeed(self):
+# raise NotImplementedError
+# def test_open_oonib_report_and_fail(self):
+# raise NotImplementedError
+# def test_write_oonib_report_entry_and_succeed(self):
+# raise NotImplementedError
+# def test_write_oonib_report_entry_and_succeed_after_timeout(self):
+# raise NotImplementedError
+# def test_write_oonib_report_entry_and_fail_after_timeout(self):
+# raise NotImplementedError
+# def test_write_oonib_report_after_close(self):
+# raise NotImplementedError
+# def test_write_oonib_report_before_open(self):
+# raise NotImplementedError
+# def test_close_oonib_report_and_succeed(self):
+# raise NotImplementedError
+# def test_close_oonib_report_and_fail(self):
+# raise NotImplementedError
diff --git a/ooni/tests/test_safe_represent.py b/ooni/tests/test_safe_represent.py
new file mode 100644
index 0000000..82a5196
--- /dev/null
+++ b/ooni/tests/test_safe_represent.py
@@ -0,0 +1,14 @@
+import yaml
+
+from twisted.trial import unittest
+
+from ooni.reporter import OSafeDumper
+
+from scapy.all import IP, UDP
+
+class TestScapyRepresent(unittest.TestCase):
+ def test_represent_scapy(self):
+ data = IP()/UDP()
+ yaml.dump_all([data], Dumper=OSafeDumper)
+
+
diff --git a/ooni/tests/test_trueheaders.py b/ooni/tests/test_trueheaders.py
new file mode 100644
index 0000000..9ac0a27
--- /dev/null
+++ b/ooni/tests/test_trueheaders.py
@@ -0,0 +1,41 @@
+from twisted.trial import unittest
+
+from ooni.utils.txagentwithsocks import TrueHeaders
+
+dummy_headers_dict = {
+ 'Header1': ['Value1', 'Value2'],
+ 'Header2': ['ValueA', 'ValueB']
+}
+
+dummy_headers_dict2 = {
+ 'Header1': ['Value1', 'Value2'],
+ 'Header2': ['ValueA', 'ValueB'],
+ 'Header3': ['ValueA', 'ValueB'],
+}
+
+dummy_headers_dict3 = {
+ 'Header1': ['Value1', 'Value2'],
+ 'Header2': ['ValueA', 'ValueB'],
+ 'Header4': ['ValueA', 'ValueB'],
+}
+
+
+class TestTrueHeaders(unittest.TestCase):
+ def test_names_match(self):
+ th = TrueHeaders(dummy_headers_dict)
+ self.assertEqual(th.getDiff(TrueHeaders(dummy_headers_dict)), set())
+
+ def test_names_not_match(self):
+ th = TrueHeaders(dummy_headers_dict)
+ self.assertEqual(th.getDiff(TrueHeaders(dummy_headers_dict2)), set(['Header3']))
+
+ th = TrueHeaders(dummy_headers_dict3)
+ self.assertEqual(th.getDiff(TrueHeaders(dummy_headers_dict2)), set(['Header3', 'Header4']))
+
+ def test_names_match_expect_ignore(self):
+ th = TrueHeaders(dummy_headers_dict)
+ self.assertEqual(th.getDiff(TrueHeaders(dummy_headers_dict2), ignore=['Header3']), set())
+
+
+
+
diff --git a/ooni/tests/test_utils.py b/ooni/tests/test_utils.py
new file mode 100644
index 0000000..cc648e0
--- /dev/null
+++ b/ooni/tests/test_utils.py
@@ -0,0 +1,20 @@
+import unittest
+from ooni.utils import pushFilenameStack
+
+class TestUtils(unittest.TestCase):
+ def test_pushFilenameStack(self):
+ f = open("dummyfile", "w+")
+ f.write("0\n")
+ f.close()
+ for i in xrange(1, 5):
+ f = open("dummyfile.%s" % i, "w+")
+ f.write("%s\n" % i)
+ f.close()
+
+ pushFilenameStack("dummyfile")
+ for i in xrange(1, 5):
+ f = open("dummyfile.%s" % i)
+ c = f.readlines()[0].strip()
+ self.assertEqual(str(i-1), str(c))
+ f.close()
+
diff --git a/tests/__init__.py b/tests/__init__.py
deleted file mode 100644
index e69de29..0000000
diff --git a/tests/mocks.py b/tests/mocks.py
deleted file mode 100644
index fed683e..0000000
--- a/tests/mocks.py
+++ /dev/null
@@ -1,168 +0,0 @@
-from ooni.tasks import BaseTask, TaskWithTimeout
-from twisted.python import failure
-from ooni.nettest import NetTest
-from ooni.managers import TaskManager
-from twisted.internet import defer
-
-class MockMeasurementFailOnce(BaseTask):
- def run(self):
- f = open('dummyTaskFailOnce.txt', 'w')
- f.write('fail')
- f.close()
- if self.failure >= 1:
- return defer.succeed(self)
- else:
- return defer.fail(failure.Failure)
-
-class MockMeasurementManager(TaskManager):
- def __init__(self):
- self.successes = []
- TaskManager.__init__(self)
-
- def failed(self, failure, task):
- pass
-
- def succeeded(self, result, task):
- self.successes.append((result, task))
-
-class MockReporter(object):
- def __init__(self):
- self.created = defer.Deferred()
-
- def writeReportEntry(self, entry):
- pass
-
- def createReport(self):
- self.created.callback(self)
-
- def finish(self):
- pass
-
-class MockFailure(Exception):
- pass
-
-## from test_managers
-mockFailure = failure.Failure(MockFailure('mock'))
-
-class MockSuccessTask(BaseTask):
- def run(self):
- return defer.succeed(42)
-
-class MockFailTask(BaseTask):
- def run(self):
- return defer.fail(mockFailure)
-
-class MockFailOnceTask(BaseTask):
- def run(self):
- if self.failures >= 1:
- return defer.succeed(42)
- else:
- return defer.fail(mockFailure)
-
-class MockSuccessTaskWithTimeout(TaskWithTimeout):
- def run(self):
- return defer.succeed(42)
-
-class MockFailTaskThatTimesOut(TaskWithTimeout):
- def run(self):
- return defer.Deferred()
-
-class MockTimeoutOnceTask(TaskWithTimeout):
- def run(self):
- if self.failures >= 1:
- return defer.succeed(42)
- else:
- return defer.Deferred()
-
-class MockFailTaskWithTimeout(TaskWithTimeout):
- def run(self):
- return defer.fail(mockFailure)
-
-
-class MockNetTest(object):
- def __init__(self):
- self.successes = []
-
- def succeeded(self, measurement):
- self.successes.append(measurement)
-
-class MockMeasurement(TaskWithTimeout):
- def __init__(self, net_test):
- TaskWithTimeout.__init__(self)
- self.netTest = net_test
-
- def succeeded(self, result):
- return self.netTest.succeeded(42)
-
-class MockSuccessMeasurement(MockMeasurement):
- def run(self):
- return defer.succeed(42)
-
-class MockFailMeasurement(MockMeasurement):
- def run(self):
- return defer.fail(mockFailure)
-
-class MockFailOnceMeasurement(MockMeasurement):
- def run(self):
- if self.failures >= 1:
- return defer.succeed(42)
- else:
- return defer.fail(mockFailure)
-
-class MockDirector(object):
- def __init__(self):
- self.successes = []
-
- def measurementFailed(self, failure, measurement):
- pass
-
- def measurementSucceeded(self, measurement):
- self.successes.append(measurement)
-
-## from test_reporter.py
-class MockOReporter(object):
- def __init__(self):
- self.created = defer.Deferred()
-
- def writeReportEntry(self, entry):
- return defer.succeed(42)
-
- def finish(self):
- pass
-
- def createReport(self):
- from ooni.utils import log
- log.debug("Creating report with %s" % self)
- self.created.callback(self)
-
-class MockOReporterThatFailsWrite(MockOReporter):
- def writeReportEntry(self, entry):
- raise MockFailure
-
-class MockOReporterThatFailsOpen(MockOReporter):
- def createReport(self):
- self.created.errback(failure.Failure(MockFailure()))
-
-class MockOReporterThatFailsWriteOnce(MockOReporter):
- def __init__(self):
- self.failure = 0
- MockOReporter.__init__(self)
-
- def writeReportEntry(self, entry):
- if self.failure >= 1:
- return defer.succeed(42)
- else:
- self.failure += 1
- raise MockFailure
-
-class MockTaskManager(TaskManager):
- def __init__(self):
- self.successes = []
- TaskManager.__init__(self)
-
- def failed(self, failure, task):
- pass
-
- def succeeded(self, result, task):
- self.successes.append((result, task))
-
diff --git a/tests/test-class-design.py b/tests/test-class-design.py
deleted file mode 100644
index bb80cd3..0000000
--- a/tests/test-class-design.py
+++ /dev/null
@@ -1,101 +0,0 @@
-#!/usr/bin/env python
-#
-# testing classes to test multiple inheritance.
-# these are not meant to be run by trial, though they could be made to be so.
-# i didn't know where to put them. --isis
-
-import abc
-from pprint import pprint
-from inspect import classify_class_attrs
-
-class PluginBase(object):
- __metaclass__ = abc.ABCMeta
-
- @abc.abstractproperty
- def name(self):
- return 'you should not see this'
-
- @name.setter
- def name(self, value):
- return 'you should not set this'
-
- @name.deleter
- def name(self):
- return 'you should not del this'
-
- @abc.abstractmethod
- def inputParser(self, line):
- """Do something to parse something."""
- return
-
-class Foo(object):
- woo = "this class has some shit in it"
- def bar(self):
- print "i'm a Foo.bar()!"
- print woo
-
-class KwargTest(Foo):
- _name = "isis"
-
- #def __new__(cls, *a, **kw):
- # return super(KwargTest, cls).__new__(cls, *a, **kw)
-
- @property
- def name(self):
- return self._name
-
- @name.setter
- def name(self, value):
- self._name = value
-
- def __init__(self, *a, **kw):
- super(KwargTest, self).__init__()
-
- ## this causes the instantion args to override the class attrs
- for key, value in kw.items():
- setattr(self.__class__, key, value)
-
- print "%s.__init__(): self.__dict__ = %s" \
- % (type(self), pprint(type(self).__dict__))
-
- for attr in classify_class_attrs(self):
- print attr
-
- @classmethod
- def sayname(cls):
- print cls.name
-
-class KwargTestChild(KwargTest):
- name = "arturo"
- def __init__(self):
- super(KwargTestChild, self).__init__()
- print self.name
-
-class KwargTestChildOther(KwargTest):
- def __init__(self, name="robot", does="lasers"):
- super(KwargTestChildOther, self).__init__()
- print self.name
-
-
-if __name__ == "__main__":
- print "class KwargTest attr name: %s" % KwargTest.name
- kwargtest = KwargTest()
- print "KwargTest instantiated wo args"
- print "kwargtest.name: %s" % kwargtest.name
- print "kwargtest.sayname(): %s" % kwargtest.sayname()
- kwargtest2 = KwargTest(name="lovecruft", does="hacking")
- print "KwargTest instantiated with name args"
- print "kwargtest.name: %s" % kwargtest2.name
- print "kwargtest.sayname(): %s" % kwargtest2.sayname()
-
- print "class KwargTestChild attr name: %s" % KwargTestChild.name
- kwargtestchild = KwargTestChild()
- print "KwargTestChild instantiated wo args"
- print "kwargtestchild.name: %s" % kwargtestchild.name
- print "kwargtestchild.sayname(): %s" % kwargtestchild.sayname()
-
- print "class KwargTestChildOther attr name: %s" % KwargTestChildOther.name
- kwargtestchildother = KwargTestChildOther()
- print "KwargTestChildOther instantiated wo args"
- print "kwargtestchildother.name: %s" % kwargtestchildother.name
- print "kwargtestchildother.sayname(): %s" % kwargtestchildother.sayname()
diff --git a/tests/test_director.py b/tests/test_director.py
deleted file mode 100644
index 79a1e96..0000000
--- a/tests/test_director.py
+++ /dev/null
@@ -1,58 +0,0 @@
-from twisted.internet import defer, base
-from twisted.trial import unittest
-
-from ooni.director import Director
-from ooni.nettest import NetTestLoader
-from tests.mocks import MockReporter
-base.DelayedCall.debug = True
-
-net_test_string = """
-from twisted.python import usage
-from ooni.nettest import NetTestCase
-
-class UsageOptions(usage.Options):
- optParameters = [['spam', 's', None, 'ham']]
-
-class DummyTestCase(NetTestCase):
- inputFile = ['file', 'f', None, 'The input File']
-
- usageOptions = UsageOptions
-
- def test_a(self):
- self.report['bar'] = 'bar'
-
- def test_b(self):
- self.report['foo'] = 'foo'
-"""
-
-
-dummyArgs = ('--spam', 1, '--file', 'dummyInputFile.txt')
-
-class TestDirector(unittest.TestCase):
- timeout = 1
- def setUp(self):
- with open('dummyInputFile.txt', 'w') as f:
- for i in range(10):
- f.write("%s\n" % i)
-
- self.reporters = [MockReporter()]
- self.director = Director()
-
- def tearDown(self):
- pass
-
- def test_start_net_test(self):
- options = {'test':net_test_string, 'subargs':dummyArgs}
- net_test_loader = NetTestLoader(options)
- net_test_loader.checkOptions()
- d = self.director.startNetTest('', net_test_loader, self.reporters)
-
- @d.addCallback
- def done(result):
- self.assertEqual(self.director.successfulMeasurements, 20)
-
- return d
-
- def test_stop_net_test(self):
- pass
-
diff --git a/tests/test_dns.py b/tests/test_dns.py
deleted file mode 100644
index e9bb524..0000000
--- a/tests/test_dns.py
+++ /dev/null
@@ -1,24 +0,0 @@
-#
-# This unittest is to verify that our usage of the twisted DNS resolver does
-# not break with new versions of twisted.
-
-import pdb
-from twisted.trial import unittest
-
-from twisted.internet import reactor
-
-from twisted.names import dns
-from twisted.names.client import Resolver
-
-class DNSTest(unittest.TestCase):
- def test_a_lookup_ooni_query(self):
- def done_query(message, *arg):
- answer = message.answers[0]
- self.assertEqual(answer.type, 1)
-
- dns_query = [dns.Query('ooni.nu', type=dns.A)]
- resolver = Resolver(servers=[('8.8.8.8', 53)])
- d = resolver.queryUDP(dns_query)
- d.addCallback(done_query)
- return d
-
diff --git a/tests/test_inputunit.py b/tests/test_inputunit.py
deleted file mode 100644
index 1f9043c..0000000
--- a/tests/test_inputunit.py
+++ /dev/null
@@ -1,29 +0,0 @@
-import unittest
-from ooni.inputunit import InputUnit, InputUnitFactory
-
-def dummyGenerator():
- for x in range(100):
- yield x
-
-class TestInputUnit(unittest.TestCase):
- def test_input_unit_factory(self):
- inputUnit = InputUnitFactory(range(100))
- for i in inputUnit:
- self.assertEqual(len(list(i)), inputUnit.inputUnitSize)
-
- def test_input_unit(self):
- inputs = range(100)
- inputUnit = InputUnit(inputs)
- idx = 0
- for i in inputUnit:
- idx += 1
-
- self.assertEqual(idx, 100)
-
- def test_input_unit_factory_length(self):
- inputUnitFactory = InputUnitFactory(range(100))
- l1 = len(inputUnitFactory)
- l2 = sum(1 for _ in inputUnitFactory)
- self.assertEqual(l1, 10)
- self.assertEqual(l2, 10)
-
diff --git a/tests/test_managers.py b/tests/test_managers.py
deleted file mode 100644
index 39f0881..0000000
--- a/tests/test_managers.py
+++ /dev/null
@@ -1,215 +0,0 @@
-from twisted.trial import unittest
-from twisted.python import failure
-from twisted.internet import defer, task
-
-from ooni.tasks import BaseTask, TaskWithTimeout, TaskTimedOut
-from ooni.managers import TaskManager, MeasurementManager
-
-from tests.mocks import MockSuccessTask, MockFailTask, MockFailOnceTask, MockFailure
-from tests.mocks import MockSuccessTaskWithTimeout, MockFailTaskThatTimesOut
-from tests.mocks import MockTimeoutOnceTask, MockFailTaskWithTimeout
-from tests.mocks import MockTaskManager, mockFailure, MockDirector
-from tests.mocks import MockNetTest, MockMeasurement, MockSuccessMeasurement
-from tests.mocks import MockFailMeasurement, MockFailOnceMeasurement
-
-class TestTaskManager(unittest.TestCase):
- timeout = 1
- def setUp(self):
- self.measurementManager = MockTaskManager()
- self.measurementManager.concurrency = 20
- self.measurementManager.retries = 2
-
- self.measurementManager.start()
-
- self.clock = task.Clock()
-
- def schedule_successful_tasks(self, task_type, number=1):
- all_done = []
- for x in range(number):
- mock_task = task_type()
- all_done.append(mock_task.done)
- self.measurementManager.schedule(mock_task)
-
- d = defer.DeferredList(all_done)
- @d.addCallback
- def done(res):
- for task_result, task_instance in self.measurementManager.successes:
- self.assertEqual(task_result, 42)
- self.assertIsInstance(task_instance, task_type)
-
- return d
-
- def schedule_failing_tasks(self, task_type, number=1):
- all_done = []
- for x in range(number):
- mock_task = task_type()
- all_done.append(mock_task.done)
- self.measurementManager.schedule(mock_task)
-
- d = defer.DeferredList(all_done)
- @d.addCallback
- def done(res):
- # 10*2 because 2 is the number of retries
- self.assertEqual(len(self.measurementManager.failures), number*3)
- for task_result, task_instance in self.measurementManager.failures:
- self.assertEqual(task_result, mockFailure)
- self.assertIsInstance(task_instance, task_type)
-
- return d
-
- def test_schedule_failing_with_mock_failure_task(self):
- mock_task = MockFailTask()
- self.measurementManager.schedule(mock_task)
- self.assertFailure(mock_task.done, MockFailure)
- return mock_task.done
-
- def test_schedule_successful_one_task(self):
- return self.schedule_successful_tasks(MockSuccessTask)
-
- def test_schedule_successful_one_task_with_timeout(self):
- return self.schedule_successful_tasks(MockSuccessTaskWithTimeout)
-
- def test_schedule_failing_tasks_that_timesout(self):
- self.measurementManager.retries = 0
-
- task_type = MockFailTaskThatTimesOut
- task_timeout = 5
-
- mock_task = task_type()
- mock_task.timeout = task_timeout
- mock_task.clock = self.clock
-
- self.measurementManager.schedule(mock_task)
-
- self.clock.advance(task_timeout)
-
- @mock_task.done.addBoth
- def done(res):
- self.assertEqual(len(self.measurementManager.failures), 1)
- for task_result, task_instance in self.measurementManager.failures:
- self.assertIsInstance(task_instance, task_type)
-
- return mock_task.done
-
- def test_schedule_time_out_once(self):
- task_type = MockTimeoutOnceTask
- task_timeout = 5
-
- mock_task = task_type()
- mock_task.timeout = task_timeout
- mock_task.clock = self.clock
-
- self.measurementManager.schedule(mock_task)
-
- self.clock.advance(task_timeout)
-
- @mock_task.done.addBoth
- def done(res):
- self.assertEqual(len(self.measurementManager.failures), 1)
- for task_result, task_instance in self.measurementManager.failures:
- self.assertIsInstance(task_instance, task_type)
-
- for task_result, task_instance in self.measurementManager.successes:
- self.assertEqual(task_result, 42)
- self.assertIsInstance(task_instance, task_type)
-
- return mock_task.done
-
-
- def test_schedule_failing_one_task(self):
- return self.schedule_failing_tasks(MockFailTask)
-
- def test_schedule_failing_one_task_with_timeout(self):
- return self.schedule_failing_tasks(MockFailTaskWithTimeout)
-
- def test_schedule_successful_ten_tasks(self):
- return self.schedule_successful_tasks(MockSuccessTask, number=10)
-
- def test_schedule_failing_ten_tasks(self):
- return self.schedule_failing_tasks(MockFailTask, number=10)
-
- def test_schedule_successful_27_tasks(self):
- return self.schedule_successful_tasks(MockSuccessTask, number=27)
-
- def test_schedule_failing_27_tasks(self):
- return self.schedule_failing_tasks(MockFailTask, number=27)
-
- def test_task_retry_and_succeed(self):
- mock_task = MockFailOnceTask()
- self.measurementManager.schedule(mock_task)
-
- @mock_task.done.addCallback
- def done(res):
- self.assertEqual(len(self.measurementManager.failures), 1)
-
- self.assertEqual(self.measurementManager.failures,
- [(mockFailure, mock_task)])
- self.assertEqual(self.measurementManager.successes,
- [(42, mock_task)])
-
- return mock_task.done
-
- def dd_test_task_retry_and_succeed_56_tasks(self):
- """
- XXX this test fails in a non-deterministic manner.
- """
- all_done = []
- number = 56
- for x in range(number):
- mock_task = MockFailOnceTask()
- all_done.append(mock_task.done)
- self.measurementManager.schedule(mock_task)
-
- d = defer.DeferredList(all_done)
-
- @d.addCallback
- def done(res):
- self.assertEqual(len(self.measurementManager.failures), number)
-
- for task_result, task_instance in self.measurementManager.successes:
- self.assertEqual(task_result, 42)
- self.assertIsInstance(task_instance, MockFailOnceTask)
-
- return d
-
-class TestMeasurementManager(unittest.TestCase):
- def setUp(self):
- mock_director = MockDirector()
-
- self.measurementManager = MeasurementManager()
- self.measurementManager.director = mock_director
-
- self.measurementManager.concurrency = 10
- self.measurementManager.retries = 2
-
- self.measurementManager.start()
-
- self.mockNetTest = MockNetTest()
-
- def test_schedule_and_net_test_notified(self, number=1):
- # XXX we should probably be inheriting from the base test class
- mock_task = MockSuccessMeasurement(self.mockNetTest)
- self.measurementManager.schedule(mock_task)
-
- @mock_task.done.addCallback
- def done(res):
- self.assertEqual(self.mockNetTest.successes,
- [42])
-
- self.assertEqual(len(self.mockNetTest.successes), 1)
- return mock_task.done
-
- def test_schedule_failing_one_measurement(self):
- mock_task = MockFailMeasurement(self.mockNetTest)
- self.measurementManager.schedule(mock_task)
-
- @mock_task.done.addErrback
- def done(failure):
- self.assertEqual(len(self.measurementManager.failures), 3)
-
- self.assertEqual(failure, mockFailure)
- self.assertEqual(len(self.mockNetTest.successes), 0)
-
- return mock_task.done
-
-
diff --git a/tests/test_mutate.py b/tests/test_mutate.py
deleted file mode 100644
index 7e30586..0000000
--- a/tests/test_mutate.py
+++ /dev/null
@@ -1,15 +0,0 @@
-import unittest
-from ooni.kit import daphn3
-
-class TestDaphn3(unittest.TestCase):
- def test_mutate_string(self):
- original_string = '\x00\x00\x00'
- mutated = daphn3.daphn3MutateString(original_string, 1)
- self.assertEqual(mutated, '\x00\x01\x00')
- def test_mutate_daphn3(self):
- original_dict = [{'client': '\x00\x00\x00'},
- {'server': '\x00\x00\x00'}]
- mutated_dict = daphn3.daphn3Mutate(original_dict, 1, 1)
- self.assertEqual(mutated_dict, [{'client': '\x00\x00\x00'},
- {'server': '\x00\x01\x00'}])
-
diff --git a/tests/test_nettest.py b/tests/test_nettest.py
deleted file mode 100644
index a0ccb32..0000000
--- a/tests/test_nettest.py
+++ /dev/null
@@ -1,264 +0,0 @@
-import os
-from StringIO import StringIO
-from tempfile import TemporaryFile, mkstemp
-
-from twisted.trial import unittest
-from twisted.internet import defer, reactor
-from twisted.python.usage import UsageError
-
-from ooni.nettest import NetTest, InvalidOption, MissingRequiredOption
-from ooni.nettest import NetTestLoader, FailureToLoadNetTest
-from ooni.tasks import BaseTask
-from ooni.utils import NotRootError
-
-from ooni.director import Director
-
-from ooni.managers import TaskManager
-
-from tests.mocks import MockMeasurement, MockMeasurementFailOnce
-from tests.mocks import MockNetTest, MockDirector, MockReporter
-from tests.mocks import MockMeasurementManager
-defer.setDebugging(True)
-
-net_test_string = """
-from twisted.python import usage
-from ooni.nettest import NetTestCase
-
-class UsageOptions(usage.Options):
- optParameters = [['spam', 's', None, 'ham']]
-
-class DummyTestCase(NetTestCase):
-
- usageOptions = UsageOptions
-
- def test_a(self):
- self.report['bar'] = 'bar'
-
- def test_b(self):
- self.report['foo'] = 'foo'
-"""
-
-net_test_root_required = net_test_string+"""
- requiresRoot = True
-"""
-
-net_test_string_with_file = """
-from twisted.python import usage
-from ooni.nettest import NetTestCase
-
-class UsageOptions(usage.Options):
- optParameters = [['spam', 's', None, 'ham']]
-
-class DummyTestCase(NetTestCase):
- inputFile = ['file', 'f', None, 'The input File']
-
- usageOptions = UsageOptions
-
- def test_a(self):
- self.report['bar'] = 'bar'
-
- def test_b(self):
- self.report['foo'] = 'foo'
-"""
-
-net_test_with_required_option = """
-from twisted.python import usage
-from ooni.nettest import NetTestCase
-
-class UsageOptions(usage.Options):
- optParameters = [['spam', 's', None, 'ham'],
- ['foo', 'o', None, 'moo'],
- ['bar', 'o', None, 'baz'],
- ]
-
-class DummyTestCase(NetTestCase):
- inputFile = ['file', 'f', None, 'The input File']
-
- usageOptions = UsageOptions
-
- def test_a(self):
- self.report['bar'] = 'bar'
-
- def test_b(self):
- self.report['foo'] = 'foo'
-
- requiredOptions = ['foo', 'bar']
-"""
-
-dummyInputs = range(1)
-dummyArgs = ('--spam', 'notham')
-dummyOptions = {'spam':'notham'}
-dummyInvalidArgs = ('--cram', 'jam')
-dummyInvalidOptions= {'cram':'jam'}
-dummyArgsWithRequiredOptions = ('--foo', 'moo', '--bar', 'baz')
-dummyRequiredOptions = {'foo':'moo', 'bar':'baz'}
-dummyArgsWithFile = ('--spam', 'notham', '--file', 'dummyInputFile.txt')
-
-class TestNetTest(unittest.TestCase):
- timeout = 1
- def setUp(self):
- with open('dummyInputFile.txt', 'w') as f:
- for i in range(10):
- f.write("%s\n" % i)
-
- def assertCallable(self, thing):
- self.assertIn('__call__', dir(thing))
-
- def verifyMethods(self, testCases):
- uniq_test_methods = set()
- for test_class, test_methods in testCases:
- instance = test_class()
- for test_method in test_methods:
- c = getattr(instance, test_method)
- self.assertCallable(c)
- uniq_test_methods.add(test_method)
- self.assertEqual(set(['test_a', 'test_b']), uniq_test_methods)
-
- def test_load_net_test_from_file(self):
- """
- Given a file verify that the net test cases are properly
- generated.
- """
- __, net_test_file = mkstemp()
- with open(net_test_file, 'w') as f:
- f.write(net_test_string)
- f.close()
-
- options = {'subargs':dummyArgs, 'test':net_test_file}
- ntl = NetTestLoader(options)
- self.verifyMethods(ntl.testCases)
- os.unlink(net_test_file)
-
- def test_load_net_test_from_str(self):
- """
- Given a file like object verify that the net test cases are properly
- generated.
- """
- options = {'subargs':dummyArgs, 'test':net_test_string}
- ntl = NetTestLoader(options)
- self.verifyMethods(ntl.testCases)
-
- def test_load_net_test_from_StringIO(self):
- """
- Given a file like object verify that the net test cases are properly
- generated.
- """
- options = {'subargs':dummyArgs, 'test':StringIO(net_test_string)}
- ntl = NetTestLoader(options)
- self.verifyMethods(ntl.testCases)
-
- def test_load_with_option(self):
- options = {'subargs':dummyArgs, 'test':StringIO(net_test_string)}
- ntl = NetTestLoader(options)
- self.assertIsInstance(ntl, NetTestLoader)
- for test_klass, test_meth in ntl.testCases:
- for option in dummyOptions.keys():
- self.assertIn(option, test_klass.usageOptions())
-
- def test_load_with_invalid_option(self):
- options = {'subargs':dummyInvalidArgs,
- 'test':StringIO(net_test_string)}
- try:
- ntl = NetTestLoader(options)
- ntl.checkOptions()
- raise Exception
- except UsageError:
- pass
-
- def test_load_with_required_option(self):
- options = {'subargs':dummyArgsWithRequiredOptions,
- 'test':StringIO(net_test_with_required_option)}
- net_test = NetTestLoader(options)
- self.assertIsInstance(net_test, NetTestLoader)
-
- def test_load_with_missing_required_option(self):
- options = {'subargs':dummyArgs,
- 'test':StringIO(net_test_with_required_option)}
- try:
- net_test = NetTestLoader(options)
- except MissingRequiredOption:
- pass
-
- def test_net_test_inputs(self):
- options = {'subargs':dummyArgsWithFile,
- 'test':StringIO(net_test_string_with_file)}
- ntl = NetTestLoader(options)
- ntl.checkOptions()
-
- # XXX: if you use the same test_class twice you will have consumed all
- # of its inputs!
- tested = set([])
- for test_class, test_method in ntl.testCases:
- if test_class not in tested:
- tested.update([test_class])
- self.assertEqual(len(list(test_class.inputs)), 10)
-
- def test_setup_local_options_in_test_cases(self):
- options = {'subargs':dummyArgs, 'test':StringIO(net_test_string)}
- ntl = NetTestLoader(options)
- ntl.checkOptions()
-
- for test_class, test_method in ntl.testCases:
- self.assertEqual(test_class.localOptions, dummyOptions)
-
- def test_generate_measurements_size(self):
-
- options = {'subargs':dummyArgsWithFile,
- 'test':StringIO(net_test_string_with_file)}
- ntl = NetTestLoader(options)
- ntl.checkOptions()
- net_test = NetTest(ntl, None)
-
- measurements = list(net_test.generateMeasurements())
- self.assertEqual(len(measurements), 20)
-
- def test_net_test_completed_callback(self):
- options = {'subargs':dummyArgsWithFile,
- 'test':StringIO(net_test_string_with_file)}
- ntl = NetTestLoader(options)
- ntl.checkOptions()
- director = Director()
-
- d = director.startNetTest('', ntl, [MockReporter()])
-
- @d.addCallback
- def complete(result):
- #XXX: why is the return type (True, None) ?
- self.assertEqual(result, [(True,None)])
- self.assertEqual(director.successfulMeasurements, 20)
-
- return d
-
- def test_require_root_succeed(self):
- #XXX: will require root to run
- options = {'subargs':dummyArgs,
- 'test':StringIO(net_test_root_required)}
- ntl = NetTestLoader(options)
- for test_class, method in ntl.testCases:
- self.assertTrue(test_class.requiresRoot)
-
- #def test_require_root_failed(self):
- # #XXX: will fail if you run as root
- # try:
- # net_test = NetTestLoader(StringIO(net_test_root_required),
- # dummyArgs)
- # except NotRootError:
- # pass
-
- #def test_create_report_succeed(self):
- # pass
-
- #def test_create_report_failed(self):
- # pass
-
- #def test_run_all_test(self):
- # raise NotImplementedError
-
- #def test_resume_test(self):
- # pass
-
- #def test_progress(self):
- # pass
-
- #def test_time_out(self):
- # raise NotImplementedError
diff --git a/tests/test_otime.py b/tests/test_otime.py
deleted file mode 100644
index 80979f2..0000000
--- a/tests/test_otime.py
+++ /dev/null
@@ -1,15 +0,0 @@
-import unittest
-from datetime import datetime
-from ooni import otime
-
-test_date = datetime(2002, 6, 26, 22, 45, 49)
-
-class TestOtime(unittest.TestCase):
- def test_timestamp(self):
- self.assertEqual(otime.timestamp(test_date), "2002-06-26T224549Z")
-
- def test_fromTimestamp(self):
- time_stamp = otime.timestamp(test_date)
- self.assertEqual(test_date, otime.fromTimestamp(time_stamp))
-
-
diff --git a/tests/test_reporter.py b/tests/test_reporter.py
deleted file mode 100644
index e21b7a1..0000000
--- a/tests/test_reporter.py
+++ /dev/null
@@ -1,238 +0,0 @@
-from twisted.internet import defer
-from twisted.trial import unittest
-
-from ooni.reporter import Report, YAMLReporter, OONIBReporter, safe_dump
-from ooni.managers import ReportEntryManager, TaskManager
-from ooni.nettest import NetTest, NetTestState
-from ooni.errors import ReportNotCreated, ReportAlreadyClosed
-
-from ooni.tasks import TaskWithTimeout
-from tests.mocks import MockOReporter, MockTaskManager
-from tests.mocks import MockMeasurement, MockNetTest
-from tests.mocks import MockOReporterThatFailsWrite
-from tests.mocks import MockOReporterThatFailsWriteOnce
-from tests.mocks import MockOReporterThatFailsOpen
-
-from twisted.python import failure
-import yaml
-
-class TestReport(unittest.TestCase):
- def setUp(self):
- pass
- def tearDown(self):
- pass
- def test_create_report_with_no_reporter(self):
- report = Report([],ReportEntryManager())
- self.assertIsInstance(report, Report)
-
- def test_create_report_with_single_reporter(self):
- report = Report([MockOReporter()], ReportEntryManager())
- self.assertIsInstance(report, Report)
-
- def test_create_report_with_multiple_reporters(self):
- report = Report([MockOReporter() for x in xrange(3)],
- ReportEntryManager())
- self.assertIsInstance(report, Report)
-
- def test_report_open_with_single_reporter(self):
- report = Report([MockOReporter()],ReportEntryManager())
- d = report.open()
- return d
-
- def test_report_open_with_multiple_reporter(self):
- report = Report([MockOReporter() for x in xrange(3)],
- ReportEntryManager())
- d = report.open()
- return d
-
- def test_fail_to_open_report_with_single_reporter(self):
- report = Report([MockOReporterThatFailsOpen()],
- ReportEntryManager())
- d = report.open()
- def f(x):
- self.assertEquals(len(report.reporters), 0)
- d.addCallback(f)
- return d
-
- def test_fail_to_open_single_report_with_multiple_reporter(self):
- report = Report([MockOReporterThatFailsOpen(), MockOReporter(),
- MockOReporter()], ReportEntryManager())
- d = report.open()
- def f(x):
- self.assertEquals(len(report.reporters),2)
- d.addCallback(f)
- return d
-
- def test_fail_to_open_all_reports_with_multiple_reporter(self):
- report = Report([MockOReporterThatFailsOpen() for x in xrange(3)],
- ReportEntryManager())
- d = report.open()
- def f(x):
- self.assertEquals(len(report.reporters),0)
- d.addCallback(f)
- return d
-
- def test_write_report_with_single_reporter_and_succeed(self):
- #XXX: verify that the MockOReporter writeReportEntry succeeds
- report = Report([MockOReporter()], ReportEntryManager())
- report.open()
- d = report.write(MockMeasurement(MockNetTest()))
- return d
-
- def test_write_report_with_single_reporter_and_fail_after_timeout(self):
- report = Report([MockOReporterThatFailsWrite()], ReportEntryManager())
- report.open()
- d = report.write(MockMeasurement(MockNetTest()))
- def f(err):
- self.assertEquals(len(report.reporters),0)
- d.addBoth(f)
- return d
-
- def test_write_report_with_single_reporter_and_succeed_after_timeout(self):
- report = Report([MockOReporterThatFailsWriteOnce()], ReportEntryManager())
- report.open()
- d = report.write(MockMeasurement(MockNetTest()))
- return d
-
- def test_write_report_with_multiple_reporter_and_succeed(self):
- report = Report([MockOReporter() for x in xrange(3)], ReportEntryManager())
- report.open()
- d = report.write(MockMeasurement(MockNetTest()))
- return d
-
- def test_write_report_with_multiple_reporter_and_fail_a_single_reporter(self):
- report = Report([MockOReporter(), MockOReporter(), MockOReporterThatFailsWrite()], ReportEntryManager())
- d = report.open()
-
- self.assertEquals(len(report.reporters),3)
- d = report.write(MockMeasurement(MockNetTest()))
-
- def f(x):
- # one of the reporters should have been removed
- self.assertEquals(len(report.reporters), 2)
- d.addBoth(f)
- return d
-
- def test_write_report_with_multiple_reporter_and_fail_all_reporter(self):
- report = Report([MockOReporterThatFailsWrite() for x in xrange(3)], ReportEntryManager())
- report.open()
- d = report.write(MockMeasurement(MockNetTest()))
- def f(err):
- self.assertEquals(len(report.reporters),0)
- d.addErrback(f)
- return d
-
-class TestYAMLReporter(unittest.TestCase):
- def setUp(self):
- self.testDetails = {'software_name': 'ooniprobe', 'options':
- {'pcapfile': None, 'help': 0, 'subargs': ['-f', 'alexa_10'], 'resume':
- 0, 'parallelism': '10', 'no-default-reporter': 0, 'testdeck': None,
- 'test': 'nettests/blocking/http_requests.py', 'logfile': None,
- 'collector': None, 'reportfile': None}, 'test_version': '0.2.3',
- 'software_version': '0.0.10', 'test_name': 'http_requests_test',
- 'start_time': 1362054343.0, 'probe_asn': 'AS0', 'probe_ip':
- '127.0.0.1', 'probe_cc': 'US'}
-
- def tearDown(self):
- pass
- def test_create_yaml_reporter(self):
- self.assertIsInstance(YAMLReporter(self.testDetails),
- YAMLReporter)
-
- def test_open_yaml_report_and_succeed(self):
- r = YAMLReporter(self.testDetails)
- r.createReport()
- # verify that testDetails was written to report properly
- def f(r):
- r._stream.seek(0)
- details, = yaml.safe_load_all(r._stream)
- self.assertEqual(details, self.testDetails)
- r.created.addCallback(f)
- return r.created
-
- #def test_open_yaml_report_and_fail(self):
- # #XXX: YAMLReporter does not handle failures of this type
- # pass
-
- def test_write_yaml_report_entry(self):
- r = YAMLReporter(self.testDetails)
- r.createReport()
-
- report_entry = {'foo':'bar', 'bin':'baz'}
- r.writeReportEntry(report_entry)
-
- # verify that details and entry were written to report
- def f(r):
- r._stream.seek(0)
- report = yaml.safe_load_all(r._stream)
- details, entry = report
- self.assertEqual(details, self.testDetails)
- self.assertEqual(entry, report_entry)
- r.created.addCallback(f)
- return r.created
-
- def test_write_multiple_yaml_report_entry(self):
- r = YAMLReporter(self.testDetails)
- r.createReport()
- def reportEntry():
- for x in xrange(10):
- yield {'foo':'bar', 'bin':'baz', 'item':x}
- for entry in reportEntry():
- r.writeReportEntry(entry)
- # verify that details and multiple entries were written to report
- def f(r):
- r._stream.seek(0)
- report = yaml.safe_load_all(r._stream)
- details = report.next()
- self.assertEqual(details, self.testDetails)
- self.assertEqual([r for r in report], [r for r in reportEntry()])
- r.created.addCallback(f)
- return r.created
-
- def test_close_yaml_report(self):
- r = YAMLReporter(self.testDetails)
- r.createReport()
- r.finish()
- self.assertTrue(r._stream.closed)
-
- def test_write_yaml_report_after_close(self):
- r = YAMLReporter(self.testDetails)
- r.createReport()
- r.finish()
- def f(r):
- r.writeReportEntry("foo")
- r.created.addCallback(f)
- self.assertFailure(r.created, ReportAlreadyClosed)
-
- def test_write_yaml_report_before_open(self):
- r = YAMLReporter(self.testDetails)
- def f(r):
- r.writeReportEntry("foo")
- r.created.addCallback(f)
- self.assertFailure(r.created, ReportNotCreated)
-
-#class TestOONIBReporter(unittest.TestCase):
-# def setUp(self):
-# pass
-# def tearDown(self):
-# pass
-# def test_create_oonib_reporter(self):
-# raise NotImplementedError
-# def test_open_oonib_report_and_succeed(self):
-# raise NotImplementedError
-# def test_open_oonib_report_and_fail(self):
-# raise NotImplementedError
-# def test_write_oonib_report_entry_and_succeed(self):
-# raise NotImplementedError
-# def test_write_oonib_report_entry_and_succeed_after_timeout(self):
-# raise NotImplementedError
-# def test_write_oonib_report_entry_and_fail_after_timeout(self):
-# raise NotImplementedError
-# def test_write_oonib_report_after_close(self):
-# raise NotImplementedError
-# def test_write_oonib_report_before_open(self):
-# raise NotImplementedError
-# def test_close_oonib_report_and_succeed(self):
-# raise NotImplementedError
-# def test_close_oonib_report_and_fail(self):
-# raise NotImplementedError
diff --git a/tests/test_runner.py b/tests/test_runner.py
deleted file mode 100644
index 0b148e1..0000000
--- a/tests/test_runner.py
+++ /dev/null
@@ -1,80 +0,0 @@
-from twisted.trial import unittest
-
-
-from ooni.inputunit import InputUnit
-from ooni.nettest import NetTestCase
-from ooni.reporter import OReporter
-
-from ooni.runner import loadTestsAndOptions, runTestCasesWithInputUnit
-
-class DummyTestCase(NetTestCase):
- def test_a(self):
- self.report['bar'] = 'bar'
- def test_b(self):
- self.report['foo'] = 'foo'
-
-class DummyTestCasePP(DummyTestCase):
- def postProcessor(self, report):
- self.report['antani'] = 'sblinda'
-
-class DummyReporter(OReporter):
- dummy_report = []
- def createReport(self, options):
- pass
-
- def writeReportEntry(self, entry):
- self.dummy_report.append(entry)
-
-class TestRunner(unittest.TestCase):
- def test_load_test_and_options(self):
- input_unit = InputUnit([0,1,2,3,4])
- cmd_line_options = {}
- test_cases, options = loadTestsAndOptions([DummyTestCase],
- cmd_line_options)
- self.assertEqual(test_cases[0][1], 'test_b')
- self.assertEqual(test_cases[1][1], 'test_a')
-
- def test_run_testcase_with_input_unit(self):
- def done(result):
- report = oreporter.dummy_report
- self.assertEqual(len(report), 10*2)
- # XXX debug why this is failing
- # for idx, entry in enumerate(oreporter.dummy_report):
- # if idx % 2 == 0:
- # self.assertEqual(entry['report']['foo'], 'foo')
- # else:
- # self.assertEqual(entry['report']['bar'], 'bar')
-
- input_unit = InputUnit([0,1,2,3,4])
- cmd_line_options = {'collector': None}
-
- oreporter = DummyReporter(cmd_line_options)
- oreporter.dummy_report = []
-
- test_cases, options = loadTestsAndOptions([DummyTestCase],
- cmd_line_options)
-
- d = runTestCasesWithInputUnit(test_cases, input_unit, oreporter, oreporter)
- d.addBoth(done)
- return d
-
- def test_with_post_processing(self):
- def done(result):
- report = oreporter.dummy_report
- self.assertEqual(len(report), 6)
- for entry in report:
- if entry['test_name'] == 'summary':
- self.assertEqual(entry['report'], {'antani': 'sblinda'})
-
- input_unit = InputUnit([None])
- cmd_line_options = {'collector': None}
-
- oreporter = DummyReporter(cmd_line_options)
- oreporter.dummy_report = []
-
- test_cases, options = loadTestsAndOptions([DummyTestCasePP],
- cmd_line_options)
-
- d = runTestCasesWithInputUnit(test_cases, input_unit, oreporter, oreporter)
- d.addBoth(done)
- return d
diff --git a/tests/test_safe_represent.py b/tests/test_safe_represent.py
deleted file mode 100644
index 82a5196..0000000
--- a/tests/test_safe_represent.py
+++ /dev/null
@@ -1,14 +0,0 @@
-import yaml
-
-from twisted.trial import unittest
-
-from ooni.reporter import OSafeDumper
-
-from scapy.all import IP, UDP
-
-class TestScapyRepresent(unittest.TestCase):
- def test_represent_scapy(self):
- data = IP()/UDP()
- yaml.dump_all([data], Dumper=OSafeDumper)
-
-
diff --git a/tests/test_trueheaders.py b/tests/test_trueheaders.py
deleted file mode 100644
index 9ac0a27..0000000
--- a/tests/test_trueheaders.py
+++ /dev/null
@@ -1,41 +0,0 @@
-from twisted.trial import unittest
-
-from ooni.utils.txagentwithsocks import TrueHeaders
-
-dummy_headers_dict = {
- 'Header1': ['Value1', 'Value2'],
- 'Header2': ['ValueA', 'ValueB']
-}
-
-dummy_headers_dict2 = {
- 'Header1': ['Value1', 'Value2'],
- 'Header2': ['ValueA', 'ValueB'],
- 'Header3': ['ValueA', 'ValueB'],
-}
-
-dummy_headers_dict3 = {
- 'Header1': ['Value1', 'Value2'],
- 'Header2': ['ValueA', 'ValueB'],
- 'Header4': ['ValueA', 'ValueB'],
-}
-
-
-class TestTrueHeaders(unittest.TestCase):
- def test_names_match(self):
- th = TrueHeaders(dummy_headers_dict)
- self.assertEqual(th.getDiff(TrueHeaders(dummy_headers_dict)), set())
-
- def test_names_not_match(self):
- th = TrueHeaders(dummy_headers_dict)
- self.assertEqual(th.getDiff(TrueHeaders(dummy_headers_dict2)), set(['Header3']))
-
- th = TrueHeaders(dummy_headers_dict3)
- self.assertEqual(th.getDiff(TrueHeaders(dummy_headers_dict2)), set(['Header3', 'Header4']))
-
- def test_names_match_expect_ignore(self):
- th = TrueHeaders(dummy_headers_dict)
- self.assertEqual(th.getDiff(TrueHeaders(dummy_headers_dict2), ignore=['Header3']), set())
-
-
-
-
diff --git a/tests/test_utils.py b/tests/test_utils.py
deleted file mode 100644
index cc648e0..0000000
--- a/tests/test_utils.py
+++ /dev/null
@@ -1,20 +0,0 @@
-import unittest
-from ooni.utils import pushFilenameStack
-
-class TestUtils(unittest.TestCase):
- def test_pushFilenameStack(self):
- f = open("dummyfile", "w+")
- f.write("0\n")
- f.close()
- for i in xrange(1, 5):
- f = open("dummyfile.%s" % i, "w+")
- f.write("%s\n" % i)
- f.close()
-
- pushFilenameStack("dummyfile")
- for i in xrange(1, 5):
- f = open("dummyfile.%s" % i)
- c = f.readlines()[0].strip()
- self.assertEqual(str(i-1), str(c))
- f.close()
-
