<commit_before>import traceback
import itertools
import logging
import time
import yaml
import json
import sys
import os
import re

from yaml.representer import *
from yaml.emitter import *
from yaml.serializer import *
from yaml.resolver import *
from twisted.python.util import untilConcludes
from twisted.trial import reporter
from twisted.internet import defer, reactor
from twisted.internet.error import ConnectionRefusedError
from twisted.python.failure import Failure

from ooni.utils import log

try:
    from scapy.packet import Packet
except ImportError:
    log.err("Scapy is not installed.")


from ooni.errors import InvalidOONIBCollectorAddress

from ooni import otime
from ooni.utils import geodata, pushFilenameStack
from ooni.utils.net import BodyReceiver, StringProducer, userAgents

from ooni import config

from ooni.tasks import ReportEntry, TaskTimedOut

class ReporterException(Exception):
    pass

def createPacketReport(packet_list):
    """
    Takes as input a packet a list.

    Returns a dict containing a dict with the packet
    summary and the raw packet.
    """
    report = []
    for packet in packet_list:
        report.append({'raw_packet': str(packet),
            'summary': str(packet.summary())})
    return report

class OSafeRepresenter(SafeRepresenter):
    """
    This is a custom YAML representer that allows us to represent reports
    safely.
    It extends the SafeRepresenter to be able to also represent complex
    numbers and scapy packet.
    """
    def represent_data(self, data):
        """
        This is very hackish. There is for sure a better way either by using
        the add_multi_representer or add_representer, the issue though lies in
        the fact that Scapy packets are metaclasses that leads to
        yaml.representer.get_classobj_bases to not be able to properly get the
        base of class of a Scapy packet.
        XXX fully debug this problem
        """
        if isinstance(data, Packet):
            data = createPacketReport(data)
        return SafeRepresenter.represent_data(self, data)

    def represent_complex(self, data):
        if data.imag == 0.0:
            data = u'%r' % data.real
        elif data.real == 0.0:
            data = u'%rj' % data.imag
        elif data.imag > 0:
            data = u'%r+%rj' % (data.real, data.imag)
        else:
            data = u'%r%rj' % (data.real, data.imag)
        return self.represent_scalar(u'tag:yaml.org,2002:python/complex', data)

OSafeRepresenter.add_representer(complex,
                                 OSafeRepresenter.represent_complex)

class OSafeDumper(Emitter, Serializer, OSafeRepresenter, Resolver):
    """
    This is a modification of the YAML Safe Dumper to use our own Safe
    Representer that supports complex numbers.
    """
    def __init__(self, stream,
            default_style=None, default_flow_style=None,
            canonical=None, indent=None, width=None,
            allow_unicode=None, line_break=None,
            encoding=None, explicit_start=None, explicit_end=None,
            version=None, tags=None):
        Emitter.__init__(self, stream, canonical=canonical,
                indent=indent, width=width,
                allow_unicode=allow_unicode, line_break=line_break)
        Serializer.__init__(self, encoding=encoding,
                explicit_start=explicit_start, explicit_end=explicit_end,
                version=version, tags=tags)
        OSafeRepresenter.__init__(self, default_style=default_style,
                default_flow_style=default_flow_style)
        Resolver.__init__(self)

class NoTestIDSpecified(Exception):
    pass

def safe_dump(data, stream=None, **kw):
    """
    Safely dump to a yaml file the specified data.
    """
    return yaml.dump_all([data], stream, Dumper=OSafeDumper, **kw)

class OReporter(object):
    def __init__(self, test_details):
        self.created = defer.Deferred()
        self.testDetails = test_details

    def createReport(self):
        """
        Override this with your own logic to implement tests.
        """
        raise NotImplemented

    def writeReportEntry(self, entry):
        """
        Takes as input an entry and writes a report for it.
        """
        raise NotImplemented

    def finish(self):
        pass

    def testDone(self, test, test_name):
        # XXX put this inside of Report.close
        # or perhaps put something like this inside of netTestDone
        log.msg("Finished running %s" % test_name)
        test_report = dict(test.report)

        if isinstance(test.input, Packet):
            test_input = createPacketReport(test.input)
        else:
            test_input = test.input

        test_started = test._start_time
        test_runtime = time.time() - test_started

        report = {'input': test_input,
                'test_name': test_name,
                'test_started': test_started,
                'test_runtime': test_runtime,
                'report': test_report}
        return defer.maybeDeferred(self.writeReportEntry, report)

class InvalidDestination(ReporterException):
    pass

class YAMLReporter(OReporter):
    """
    These are useful functions for reporting to YAML format.

    report_destination:
        the destination directory of the report

    """
    def __init__(self, test_details, report_destination='.'):
        self.reportDestination = report_destination

        if not os.path.isdir(report_destination):
            raise InvalidDestination

        report_filename = "report-" + \
                test_details['test_name'] + "-" + \
                otime.timestamp() + ".yamloo"

        report_path = os.path.join(self.reportDestination, report_filename)

        if os.path.exists(report_path):
            log.msg("Report already exists with filename %s" % report_path)
            pushFilenameStack(report_path)

        log.debug("Creating %s" % report_path)
        self._stream = open(report_path, 'w+')

        OReporter.__init__(self, test_details)

    def _writeln(self, line):
        self._write("%s\n" % line)

    def _write(self, format_string, *args):
        s = str(format_string)
        assert isinstance(s, type(''))
        if args:
            self._stream.write(s % args)
        else:
            self._stream.write(s)
        untilConcludes(self._stream.flush)

    def writeReportEntry(self, entry):
        log.debug("Writing report with YAML reporter")
        self._write('---\n')
        if isinstance(entry, Failure):
            self._write(entry.value)
        else:
            self._write(safe_dump(entry))
        self._write('...\n')

    def createReport(self):
        """
        Writes the report header and fire callbacks on self.created
        """
        self._writeln("###########################################")

        self._writeln("# OONI Probe Report for %s (%s)" % (self.testDetails['test_name'],
                    self.testDetails['test_version']))
        self._writeln("# %s" % otime.prettyDateNow())
        self._writeln("###########################################")

        self.writeReportEntry(self.testDetails)

    def finish(self):
        self._stream.close()

class OONIBReportError(Exception):
    pass

class OONIBReportUpdateError(OONIBReportError):
    pass

class OONIBReportCreationError(OONIBReportError):
    pass

class OONIBTestDetailsLookupError(OONIBReportError):
    pass

class OONIBReporter(OReporter):
    def __init__(self, test_details, collector_address):
        self.collectorAddress = collector_address
        self.validateCollectorAddress()

        self.reportID = None

        OReporter.__init__(self, test_details)

    def validateCollectorAddress(self):
        """
        Will raise :class:ooni.errors.InvalidOONIBCollectorAddress an exception
        if the oonib reporter is not valid.
        """
        regexp = '^(http|httpo):\/\/[a-zA-Z0-9\-\.]+(:\d+)?$'
        if not re.match(regexp, self.collectorAddress):
            raise InvalidOONIBCollectorAddress

    @defer.inlineCallbacks
    def writeReportEntry(self, entry):
        log.debug("Writing report with OONIB reporter")
        content = '---\n'
        content += safe_dump(entry)
        content += '...\n'

        url = self.collectorAddress + '/report'

        request = {'report_id': self.reportID,
                'content': content}

        log.debug("Updating report with id %s (%s)" % (self.reportID, url))
        request_json = json.dumps(request)
        log.debug("Sending %s" % request_json)

        bodyProducer = StringProducer(json.dumps(request))

        try:
            response = yield self.agent.request("PUT", url,
                                bodyProducer=bodyProducer)
        except:
            # XXX we must trap this in the runner and make sure to report the
            # data later.
            log.err("Error in writing report entry")
            raise OONIBReportUpdateError

    @defer.inlineCallbacks
    def createReport(self):
        """
        Creates a report on the oonib collector.
        """
        # XXX we should probably be setting this inside of the constructor,
        # however config.tor.socks_port is not set until Tor is started and the
        # reporter is instantiated before Tor is started. We probably want to
        # do this with some deferred kung foo or instantiate the reporter after
        # tor is started.

        from ooni.utils.txagentwithsocks import Agent
        from twisted.internet import reactor
        try:
            self.agent = Agent(reactor, sockshost="127.0.0.1",
                socksport=int(config.tor.socks_port))
        except Exception, e:
            log.exception(e)

        url = self.collectorAddress + '/report'

        content = '---\n'
        content += safe_dump(self.testDetails)
        content += '...\n'

        request = {'software_name': self.testDetails['software_name'],
            'software_version': self.testDetails['software_version'],
            'probe_asn': self.testDetails['probe_asn'],
            'test_name': self.testDetails['test_name'],
            'test_version': self.testDetails['test_version'],
            # XXX there is a bunch of redundancy in the arguments getting sent
            # to the backend. This may need to get changed in the client and the
            # backend.
            'content': content
        }

        log.msg("Reporting %s" % url)
        request_json = json.dumps(request)
        log.debug("Sending %s" % request_json)

        bodyProducer = StringProducer(json.dumps(request))

        log.msg("Creating report with OONIB Reporter. Please be patient.")
        log.msg("This may take up to 1-2 minutes...")

        try:
            response = yield self.agent.request("POST", url,
                                bodyProducer=bodyProducer)
        except ConnectionRefusedError:
            log.err("Connection to reporting backend failed (ConnectionRefusedError)")
            raise OONIBReportCreationError

        except Exception, e:
            log.exception(e)
            raise OONIBReportCreationError

        # This is a little trix to allow us to unspool the response. We create
        # a deferred and call yield on it.
        response_body = defer.Deferred()
        response.deliverBody(BodyReceiver(response_body))

        backend_response = yield response_body

        try:
            parsed_response = json.loads(backend_response)
        except Exception, e:
            log.exception(e)
            raise OONIBReportCreationError

        self.reportID = parsed_response['report_id']
        self.backendVersion = parsed_response['backend_version']
        log.debug("Created report with id %s" % parsed_response['report_id'])

class ReportClosed(Exception):
    pass

class Report(object):
    def __init__(self, reporters, reportEntryManager):
        """
        This is an abstraction layer on top of all the configured reporters.

        It allows to lazily write to the reporters that are to be used.

        Args:

            reporters:
                a list of :class:ooni.reporter.OReporter instances

            reportEntryManager:
                an instance of :class:ooni.tasks.ReportEntryManager
        """
        self.reporters = reporters

        self.done = defer.Deferred()
        #self.done.addCallback(self.close)

        self.reportEntryManager = reportEntryManager

    def open(self):
        """
        This will create all the reports that need to be created and fires the
        created callback of the reporter whose report got created.
        """
        for reporter in self.reporters:
            reporter.created.addErrback(self.failedOpeningReport, reporter)
            d = defer.maybeDeferred(reporter.createReport)
            d.addCallback(reporter.created.callback)
            d.addErrback(reporter.created.callback)


    def failedOpeningReport(self, failure, reporter):
        """
        This errback get's called every time we fail to create a report.
        By fail we mean that the number of retries has exceeded.
        Once a report has failed to be created with a reporter we give up and
        remove the reporter from the list of reporters to write to.
        """
        log.err("Failed to open %s reporter, giving up..." % reporter)
        self.reporters.remove(reporter)
        failure.reporter = reporter
        return failure

    def write(self, measurement):
        """
        This is a lazy call that will write to all the reporters by waiting on
        them to be created.

        Will return a deferred that will fire once the report for the specified
        measurement have been written to all the reporters.

        Args:

            measurement:
                an instance of :class:ooni.tasks.Measurement

        Returns:
            a deferred list that will fire once all the report entries have
            been written.
        """
        l = []
        for reporter in self.reporters:
            def writeReportEntry(result):
                report_write_task = ReportEntry(reporter, measurement)
                self.reportEntryManager.schedule(report_write_task)
                return report_write_task.done

            d = reporter.created.addBoth(writeReportEntry)
            # Give up on writing to the reporter if the task fails X times
            d.addErrback(self.failedOpeningReport, reporter)
            l.append(d)

        dl = defer.DeferredList(l)
        return dl

    def close(self, _):
        """
        Close the report by calling it's finish method.

        Returns:
            a :class:twisted.internet.defer.DeferredList that will fire when
            all the reports have been closed.

        """
        l = []
        for reporter in self.reporters:
            d = defer.maybeDeferred(reporter.finish)
            l.append(d)
        dl = defer.DeferredList(l)
        return dl
<commit_msg>Fix race condition bug in reporter<commit_after>import traceback
import itertools
import logging
import time
import yaml
import json
import sys
import os
import re

from yaml.representer import *
from yaml.emitter import *
from yaml.serializer import *
from yaml.resolver import *
from twisted.python.util import untilConcludes
from twisted.trial import reporter
from twisted.internet import defer, reactor
from twisted.internet.error import ConnectionRefusedError
from twisted.python.failure import Failure

from ooni.utils import log

try:
    from scapy.packet import Packet
except ImportError:
    log.err("Scapy is not installed.")


from ooni.errors import InvalidOONIBCollectorAddress

from ooni import otime
from ooni.utils import geodata, pushFilenameStack
from ooni.utils.net import BodyReceiver, StringProducer, userAgents

from ooni import config

from ooni.tasks import ReportEntry, TaskTimedOut

class ReporterException(Exception):
    pass

def createPacketReport(packet_list):
    """
    Takes as input a packet a list.

    Returns a dict containing a dict with the packet
    summary and the raw packet.
    """
    report = []
    for packet in packet_list:
        report.append({'raw_packet': str(packet),
            'summary': str(packet.summary())})
    return report

class OSafeRepresenter(SafeRepresenter):
    """
    This is a custom YAML representer that allows us to represent reports
    safely.
    It extends the SafeRepresenter to be able to also represent complex
    numbers and scapy packet.
    """
    def represent_data(self, data):
        """
        This is very hackish. There is for sure a better way either by using
        the add_multi_representer or add_representer, the issue though lies in
        the fact that Scapy packets are metaclasses that leads to
        yaml.representer.get_classobj_bases to not be able to properly get the
        base of class of a Scapy packet.
        XXX fully debug this problem
        """
        if isinstance(data, Packet):
            data = createPacketReport(data)
        return SafeRepresenter.represent_data(self, data)

    def represent_complex(self, data):
        if data.imag == 0.0:
            data = u'%r' % data.real
        elif data.real == 0.0:
            data = u'%rj' % data.imag
        elif data.imag > 0:
            data = u'%r+%rj' % (data.real, data.imag)
        else:
            data = u'%r%rj' % (data.real, data.imag)
        return self.represent_scalar(u'tag:yaml.org,2002:python/complex', data)

OSafeRepresenter.add_representer(complex,
                                 OSafeRepresenter.represent_complex)

class OSafeDumper(Emitter, Serializer, OSafeRepresenter, Resolver):
    """
    This is a modification of the YAML Safe Dumper to use our own Safe
    Representer that supports complex numbers.
    """
    def __init__(self, stream,
            default_style=None, default_flow_style=None,
            canonical=None, indent=None, width=None,
            allow_unicode=None, line_break=None,
            encoding=None, explicit_start=None, explicit_end=None,
            version=None, tags=None):
        Emitter.__init__(self, stream, canonical=canonical,
                indent=indent, width=width,
                allow_unicode=allow_unicode, line_break=line_break)
        Serializer.__init__(self, encoding=encoding,
                explicit_start=explicit_start, explicit_end=explicit_end,
                version=version, tags=tags)
        OSafeRepresenter.__init__(self, default_style=default_style,
                default_flow_style=default_flow_style)
        Resolver.__init__(self)

class NoTestIDSpecified(Exception):
    pass

def safe_dump(data, stream=None, **kw):
    """
    Safely dump to a yaml file the specified data.
    """
    return yaml.dump_all([data], stream, Dumper=OSafeDumper, **kw)

class OReporter(object):
    def __init__(self, test_details):
        self.created = defer.Deferred()
        self.testDetails = test_details

    def createReport(self):
        """
        Override this with your own logic to implement tests.
        """
        raise NotImplemented

    def writeReportEntry(self, entry):
        """
        Takes as input an entry and writes a report for it.
        """
        raise NotImplemented

    def finish(self):
        pass

    def testDone(self, test, test_name):
        # XXX put this inside of Report.close
        # or perhaps put something like this inside of netTestDone
        log.msg("Finished running %s" % test_name)
        test_report = dict(test.report)

        if isinstance(test.input, Packet):
            test_input = createPacketReport(test.input)
        else:
            test_input = test.input

        test_started = test._start_time
        test_runtime = time.time() - test_started

        report = {'input': test_input,
                'test_name': test_name,
                'test_started': test_started,
                'test_runtime': test_runtime,
                'report': test_report}
        return defer.maybeDeferred(self.writeReportEntry, report)

class InvalidDestination(ReporterException):
    pass

class YAMLReporter(OReporter):
    """
    These are useful functions for reporting to YAML format.

    report_destination:
        the destination directory of the report

    """
    def __init__(self, test_details, report_destination='.'):
        self.reportDestination = report_destination

        if not os.path.isdir(report_destination):
            raise InvalidDestination

        report_filename = "report-" + \
                test_details['test_name'] + "-" + \
                otime.timestamp() + ".yamloo"

        report_path = os.path.join(self.reportDestination, report_filename)

        if os.path.exists(report_path):
            log.msg("Report already exists with filename %s" % report_path)
            pushFilenameStack(report_path)

        log.debug("Creating %s" % report_path)
        self._stream = open(report_path, 'w+')

        OReporter.__init__(self, test_details)

    def _writeln(self, line):
        self._write("%s\n" % line)

    def _write(self, format_string, *args):
        s = str(format_string)
        assert isinstance(s, type(''))
        if args:
            self._stream.write(s % args)
        else:
            self._stream.write(s)
        untilConcludes(self._stream.flush)

    def writeReportEntry(self, entry):
        log.debug("Writing report with YAML reporter")
        self._write('---\n')
        if isinstance(entry, Failure):
            self._write(entry.value)
        else:
            self._write(safe_dump(entry))
        self._write('...\n')

    def createReport(self):
        """
        Writes the report header and fire callbacks on self.created
        """
        self._writeln("###########################################")

        self._writeln("# OONI Probe Report for %s (%s)" % (self.testDetails['test_name'],
                    self.testDetails['test_version']))
        self._writeln("# %s" % otime.prettyDateNow())
        self._writeln("###########################################")

        self.writeReportEntry(self.testDetails)
        self.created.callback(self)

    def finish(self):
        self._stream.close()

class OONIBReportError(Exception):
    pass

class OONIBReportUpdateError(OONIBReportError):
    pass

class OONIBReportCreationError(OONIBReportError):
    pass

class OONIBTestDetailsLookupError(OONIBReportError):
    pass

class OONIBReporter(OReporter):
    def __init__(self, test_details, collector_address):
        self.collectorAddress = collector_address
        self.validateCollectorAddress()

        self.reportID = None

        OReporter.__init__(self, test_details)

    def validateCollectorAddress(self):
        """
        Will raise :class:ooni.errors.InvalidOONIBCollectorAddress an exception
        if the oonib reporter is not valid.
        """
        regexp = '^(http|httpo):\/\/[a-zA-Z0-9\-\.]+(:\d+)?$'
        if not re.match(regexp, self.collectorAddress):
            raise InvalidOONIBCollectorAddress

    @defer.inlineCallbacks
    def writeReportEntry(self, entry):
        log.debug("Writing report with OONIB reporter")
        content = '---\n'
        content += safe_dump(entry)
        content += '...\n'

        url = self.collectorAddress + '/report'

        request = {'report_id': self.reportID,
                'content': content}

        log.debug("Updating report with id %s (%s)" % (self.reportID, url))
        request_json = json.dumps(request)
        log.debug("Sending %s" % request_json)

        bodyProducer = StringProducer(json.dumps(request))

        try:
            response = yield self.agent.request("PUT", url,
                                bodyProducer=bodyProducer)
        except:
            # XXX we must trap this in the runner and make sure to report the
            # data later.
            log.err("Error in writing report entry")
            raise OONIBReportUpdateError

    @defer.inlineCallbacks
    def createReport(self):
        """
        Creates a report on the oonib collector.
        """
        # XXX we should probably be setting this inside of the constructor,
        # however config.tor.socks_port is not set until Tor is started and the
        # reporter is instantiated before Tor is started. We probably want to
        # do this with some deferred kung foo or instantiate the reporter after
        # tor is started.

        from ooni.utils.txagentwithsocks import Agent
        from twisted.internet import reactor
        try:
            self.agent = Agent(reactor, sockshost="127.0.0.1",
                socksport=int(config.tor.socks_port))
        except Exception, e:
            log.exception(e)

        url = self.collectorAddress + '/report'

        content = '---\n'
        content += safe_dump(self.testDetails)
        content += '...\n'

        request = {'software_name': self.testDetails['software_name'],
            'software_version': self.testDetails['software_version'],
            'probe_asn': self.testDetails['probe_asn'],
            'test_name': self.testDetails['test_name'],
            'test_version': self.testDetails['test_version'],
            # XXX there is a bunch of redundancy in the arguments getting sent
            # to the backend. This may need to get changed in the client and the
            # backend.
            'content': content
        }

        log.msg("Reporting %s" % url)
        request_json = json.dumps(request)
        log.debug("Sending %s" % request_json)

        bodyProducer = StringProducer(json.dumps(request))

        log.msg("Creating report with OONIB Reporter. Please be patient.")
        log.msg("This may take up to 1-2 minutes...")

        try:
            response = yield self.agent.request("POST", url,
                                bodyProducer=bodyProducer)
        except ConnectionRefusedError:
            log.err("Connection to reporting backend failed (ConnectionRefusedError)")
            raise OONIBReportCreationError

        except Exception, e:
            log.exception(e)
            raise OONIBReportCreationError

        # This is a little trix to allow us to unspool the response. We create
        # a deferred and call yield on it.
        response_body = defer.Deferred()
        response.deliverBody(BodyReceiver(response_body))

        backend_response = yield response_body

        try:
            parsed_response = json.loads(backend_response)
        except Exception, e:
            log.exception(e)
            raise OONIBReportCreationError

        self.reportID = parsed_response['report_id']
        self.backendVersion = parsed_response['backend_version']
        log.debug("Created report with id %s" % parsed_response['report_id'])
        self.created.callback(self)

class ReportClosed(Exception):
    pass

class Report(object):
    def __init__(self, reporters, reportEntryManager):
        """
        This is an abstraction layer on top of all the configured reporters.

        It allows to lazily write to the reporters that are to be used.

        Args:

            reporters:
                a list of :class:ooni.reporter.OReporter instances

            reportEntryManager:
                an instance of :class:ooni.tasks.ReportEntryManager
        """
        self.reporters = reporters

        self.done = defer.Deferred()
        self.reportEntryManager = reportEntryManager

    def open(self):
        """
        This will create all the reports that need to be created and fires the
        created callback of the reporter whose report got created.
        """
        l = []
        for reporter in self.reporters[:]:
            reporter.createReport()
            reporter.created.addErrback(self.failedOpeningReport, reporter)
            l.append(reporter.created)
        log.debug("Reporters created: %s" % l)
        # Should we consume errors silently?
        dl = defer.DeferredList(l)
        return dl

    def failedOpeningReport(self, failure, reporter):
        """
        This errback get's called every time we fail to create a report.
        By fail we mean that the number of retries has exceeded.
        Once a report has failed to be created with a reporter we give up and
        remove the reporter from the list of reporters to write to.
        """
        log.err("Failed to open %s reporter, giving up..." % reporter)
        log.err("Reporter %s failed, removing from report..." % reporter)
        self.reporters.remove(reporter)
        # Don't forward the exception unless there are no more reporters
        if len(self.reporters) == 0:
            log.err("Removed last reporter %s" % reporter)
            failure.reporter = reporter
            return failure

    def write(self, measurement):
        """
        This is a lazy call that will write to all the reporters by waiting on
        them to be created.

        Will return a deferred that will fire once the report for the specified
        measurement have been written to all the reporters.

        Args:

            measurement:
                an instance of :class:ooni.tasks.Measurement

        Returns:
            a deferred list that will fire once all the report entries have
            been written.
        """
        l = []
        for reporter in self.reporters[:]:
            report_write_task = ReportEntry(reporter, measurement)
            def scheduleWriteReportEntry(result):
                self.reportEntryManager.schedule(report_write_task)

            # delay scheduling the task until after the report is created
            log.debug("Adding report entry task %s" % report_write_task)
            reporter.created.addCallback(scheduleWriteReportEntry)

            # if the write task fails n times, kill the reporter
            report_write_task.done.addErrback(self.failedOpeningReport, reporter)
            l.append(report_write_task.done)

        # XXX: This seems a bit fragile.
        # failedOpeningReport will forward the errback if the remaining
        # reporter has failed. If we fireOnOneErrback, this means that
        # the caller of report.write is responsible for attaching an
        # errback to the returned deferred and handle this case. That
        # probably means stopping the net test.

        # Here, fireOnOneErrback means to call the deferredlists errback
        # as soon as any of the deferreds return a failure. consumeErrors
        # is used to prevent any uncaught failures from raising an
        # exception. Alternately we could attach a logger to the errback
        # of each deferred and it would have the same effect

        # Probably the better thing to do here would be to add a callback
        # to the deferredlist that checks to see if any reporters are left
        # and raise an exception if there are no remaining reporters
        dl = defer.DeferredList(l,fireOnOneErrback=True, consumeErrors=True)
        return dl

    def close(self, _):
        """
        Close the report by calling it's finish method.

        Returns:
            a :class:twisted.internet.defer.DeferredList that will fire when
            all the reports have been closed.

        """
        l = []
        for reporter in self.reporters[:]:
            d = defer.maybeDeferred(reporter.finish)
            l.append(d)
        dl = defer.DeferredList(l)
        return dl

{'fraction_non_alphanumeric': 0.08795088746103441, 'fraction_numerical': 0.001240536929830142, 'mean_word_length': 2.3474233390119252, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2435441', 'n_tokens_mistral': 8616, 'n_tokens_neox': 8229, 'n_words': 2530}