<commit_before># -*- coding: utf-8 -*-
'''Module containing helpers for testing the Chevah server.'''
from __future__ import with_statement

from select import error as SelectError
from StringIO import StringIO
from threading import currentThread, Thread, Timer
import base64
import BaseHTTPServer
import logging
import os
import random
import string
import urllib
import urllib2

from mock import Mock
from OpenSSL import SSL, crypto
from twisted.internet import address, defer, interfaces as internet_interfaces
from twisted.web import (
    http,
    http_headers,
    server as web_server,
    resource as web_resource,
    )
from zope.interface import implements

from chevah.compat import DefaultAvatar, system_users
from chevah.empirical.filesystem import LocalTestFilesystem
from chevah.empirical.constants import (
    TEST_NAME_MARKER,
    )
from chevah.utils.avatar import ApplicationAvatar, OSAvatar
from chevah.utils.configuration_file import (
    FileConfigurationProxy,
    )
from chevah.utils.constants import (
    LOG_SECTION_DEFAULTS,
    )
from chevah.utils.credentials import (
    PasswordCredentials,
    FTPPasswordCredentials,
    FTPSPasswordCredentials,
    HTTPBasicAuthCredentials,
    HTTPSBasicAuthCredentials,
    SSHKeyCredentials,
    SSHPasswordCredentials,
    SSLCertificateCredentials,
    )
from chevah.utils.event import (
    EventDefinition,
    EventGroupDefinition,
    EventsDefinition,
    )
from chevah.utils.json_file import JSONFile
from chevah.utils.json_rpc import JSONRPCResource
from chevah.utils.logger import (
    _Logger,
    LogEntry,
    )
from chevah.utils.log_configuration_section import (
    LogConfigurationSection,
    )


class MockHTTPServer(object):
    '''Mock HTTP server used for testing simple HTTP requests.

    This is a context manager class.
    '''
    # Time in second after the sever will be forced to shutdown.
    TIMEOUT = 4.0

    def __init__(self, responses=None, ip='127.0.0.1', port=0, debug=False):
        '''Initialize a new MockHTTPServer.

         * ip - IP to listen. Leave empty to listen to any interface.
         * port - Port to listen. Leave 0 to pick a random port.
         * responses - A list of MockHTTPResponse defining the behavior of
                        this server.

        After the server is started the ip and port are available in the
        context management instance.
        >>> responses = [
        ...     MockHTTPResponse(url='/hello.html', response='Hello!)]
        >>> with MochHTTPServer(responses=responses) as httpd:
        ...     print 'Listening at %s:%d' % (httpd.id, httpd.port)
        ...     self.assertEqual('Hello!', your_get())

        >>> responses = [
        ...     MockHTTPResponse(
        ...         url='/hello.php', request='user=John',
        ...         response_content='Hello John!, response_code=202)]
        >>> with MockHTTPServer(responses=responses) as httpd:
        ...     self.assertEqual(
        ...         'Hello John!',
        ...         get_you_post(url='hello.php', data='user=John'))
        '''

        MockRequestHandler.debug = debug
        if responses is None:
            MockRequestHandler.valid_responses = []
        else:
            MockRequestHandler.valid_responses = responses
        self._server = BaseHTTPServer.HTTPServer(
            (ip, port), MockRequestHandler)
        (self.ip, self.port) = self._server.server_address
        self._stopped = False
        self._server_process = Thread(target=self._serve_page)
        self._server_timeout = Timer(self.TIMEOUT, self._force_close)

    def _serve_page(self):
        '''Server a request while the server should still run.'''
        while not self._stopped:
            try:
                self._server.handle_request()
            except SelectError, error:
                pass

    def _force_close(self):
        '''Do what it takes to close and clear the server.'''
        try:
            urllib2.urlopen('http://%s:%d' % (self.ip, self.port))
        except:
            pass
        if currentThread is not None:
            self._server_process.join(0.1)
<commit_msg>Fix lint.<commit_after># -*- coding: utf-8 -*-
'''Module containing helpers for testing the Chevah server.'''
from __future__ import with_statement

from select import error as SelectError
from StringIO import StringIO
from threading import currentThread, Thread, Timer
import base64
import BaseHTTPServer
import logging
import os
import random
import string
import urllib
import urllib2

from mock import Mock
from OpenSSL import SSL, crypto
from twisted.internet import address, defer, interfaces as internet_interfaces
from twisted.web import (
    http,
    http_headers,
    server as web_server,
    resource as web_resource,
    )
from zope.interface import implements

from chevah.compat import DefaultAvatar, system_users
from chevah.empirical.filesystem import LocalTestFilesystem
from chevah.empirical.constants import (
    TEST_NAME_MARKER,
    )
from chevah.utils.avatar import ApplicationAvatar, OSAvatar
from chevah.utils.configuration_file import (
    FileConfigurationProxy,
    )
from chevah.utils.constants import (
    LOG_SECTION_DEFAULTS,
    )
from chevah.utils.credentials import (
    PasswordCredentials,
    FTPPasswordCredentials,
    FTPSPasswordCredentials,
    HTTPBasicAuthCredentials,
    HTTPSBasicAuthCredentials,
    SSHKeyCredentials,
    SSHPasswordCredentials,
    SSLCertificateCredentials,
    )
from chevah.utils.event import (
    EventDefinition,
    EventGroupDefinition,
    EventsDefinition,
    )
from chevah.utils.json_file import JSONFile
from chevah.utils.json_rpc import JSONRPCResource
from chevah.utils.logger import (
    _Logger,
    LogEntry,
    )
from chevah.utils.log_configuration_section import (
    LogConfigurationSection,
    )


class MockHTTPServer(object):
    '''Mock HTTP server used for testing simple HTTP requests.

    This is a context manager class.
    '''
    # Time in second after the sever will be forced to shutdown.
    TIMEOUT = 4.0

    def __init__(self, responses=None, ip='127.0.0.1', port=0, debug=False):
        '''Initialize a new MockHTTPServer.

         * ip - IP to listen. Leave empty to listen to any interface.
         * port - Port to listen. Leave 0 to pick a random port.
         * responses - A list of MockHTTPResponse defining the behavior of
                        this server.

        After the server is started the ip and port are available in the
        context management instance.
        >>> responses = [
        ...     MockHTTPResponse(url='/hello.html', response='Hello!)]
        >>> with MochHTTPServer(responses=responses) as httpd:
        ...     print 'Listening at %s:%d' % (httpd.id, httpd.port)
        ...     self.assertEqual('Hello!', your_get())

        >>> responses = [
        ...     MockHTTPResponse(
        ...         url='/hello.php', request='user=John',
        ...         response_content='Hello John!, response_code=202)]
        >>> with MockHTTPServer(responses=responses) as httpd:
        ...     self.assertEqual(
        ...         'Hello John!',
        ...         get_you_post(url='hello.php', data='user=John'))
        '''

        MockRequestHandler.debug = debug
        if responses is None:
            MockRequestHandler.valid_responses = []
        else:
            MockRequestHandler.valid_responses = responses
        self._server = BaseHTTPServer.HTTPServer(
            (ip, port), MockRequestHandler)
        (self.ip, self.port) = self._server.server_address
        self._stopped = False
        self._server_process = Thread(target=self._serve_page)
        self._server_timeout = Timer(self.TIMEOUT, self._force_close)

    def _serve_page(self):
        '''Server a request while the server should still run.'''
        while not self._stopped:
            try:
                self._server.handle_request()
            except SelectError:
                # I don't know why this error is raised on Python 2.7.
                # I will just ignore it for now.
                pass

    def _force_close(self):
        '''Do what it takes to close and clear the server.'''
        try:
            urllib2.urlopen('http://%s:%d' % (self.ip, self.port))
        except:
            pass
        if currentThread is not None:
            self._server_process.join(0.1)

{'fraction_non_alphanumeric': 0.08534986291572297, 'fraction_numerical': 0.005006556204553582, 'mean_word_length': 2.599313599313599, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6856308', 'n_tokens_mistral': 2320, 'n_tokens_neox': 2311, 'n_words': 674}