commit db0651bb31fe78bca33e29553d12098b318d62c9 Author: Arturo Filastò art@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() -