<commit_before>from __future__ import unicode_literals
import json
import socket
from uuid import uuid4
from time import sleep
from random import randrange
from unittest import TestCase
from contextlib import closing

import six
from six.moves.queue import Queue, Empty
from six.moves.urllib.parse import urlparse, urlencode, parse_qsl

import pytest
import cherrypy
import requests
from rpctools.jsonrpc import ServerProxy
from ws4py.server.cherrypyserver import WebSocketPlugin

import sideboard.websockets
from sideboard.lib import log, config, subscribes, notifies, services, cached_property, WebSocket
from sideboard.tests import service_patcher, config_patcher
from sideboard.tests.test_sa import Session


@pytest.mark.functional
class SideboardServerTest(TestCase):
    port = config['cherrypy']['server.socket_port']
    jsonrpc_url = 'http://127.0.0.1:{}/jsonrpc'.format(port)
    jsonrpc = ServerProxy(jsonrpc_url)

    rsess_username = 'unit_tests'

    @staticmethod
    def assert_port_open(port):
        with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            sock.bind(('0.0.0.0', port))

    @staticmethod
    def assert_can_connect_to_localhost(port):
        for i in range(50):
            try:
                socket.create_connection(('127.0.0.1', port)).close()
            except Exception as e:
                sleep(0.1)
            else:
                break
        else:
            raise e

    @classmethod
    def start_cherrypy(cls):
        class Root(object):
            @cherrypy.expose
            def index(self):
                cherrypy.session['username'] = cls.rsess_username
                return cls.rsess_username

        cherrypy.tree.apps.pop('/mock_login', None)
        cherrypy.tree.mount(Root(), '/mock_login')

        cls.assert_port_open(cls.port)
        cherrypy.config.update({'engine.autoreload_on': False})
        cherrypy.engine.start()
        cherrypy.engine.wait(cherrypy.engine.states.STARTED)
        cls.assert_can_connect_to_localhost(cls.port)

    @classmethod
    def stop_cherrypy(cls):
        cherrypy.engine.stop()
        cherrypy.engine.wait(cherrypy.engine.states.STOPPED)
        cherrypy.engine.state = cherrypy.engine.states.EXITING

        # ws4py does not support stopping and restarting CherryPy
        sideboard.websockets.websocket_plugin.unsubscribe()
        sideboard.websockets.websocket_plugin = WebSocketPlugin(cherrypy.engine)
        sideboard.websockets.websocket_plugin.subscribe()

    @classmethod
    def setUpClass(cls):
        super(SideboardServerTest, cls).setUpClass()
        cls.start_cherrypy()
        cls.ws = cls.patch_websocket(services.get_websocket())
        cls.ws.connect(max_wait=5)
        assert cls.ws.connected

    @classmethod
    def tearDownClass(cls):
        cls.stop_cherrypy()
        super(SideboardServerTest, cls).tearDownClass()

    @staticmethod
    def patch_websocket(ws):
        ws.q = Queue()
        ws.fallback = ws.q.put
        return ws

    def tearDown(self):
        while not self.ws.q.empty():
            q.get_nowait()

    def wait_for(self, func, *args, **kwargs):
        for i in range(50):
            cherrypy.engine.publish('main')  # since our unit tests don't call cherrypy.engine.block, we must publish this event manually
            try:
                result = func(*args, **kwargs)
                assert result or result is None
            except:
                sleep(0.1)
            else:
                break
        else:
            raise AssertionError('wait timed out')

    def wait_for_eq(self, target, func, *args, **kwargs):
        try:
            self.wait_for(lambda: target == func(*args, **kwargs))
        except:
            raise AssertionError('{!r} != {!r}'.format(target, func(*args, **kwargs)))

    def wait_for_ne(self, target, func, *args, **kwargs):
        try:
            self.wait_for(lambda: target != func(*args, **kwargs))
        except:
            raise AssertionError('{!r} == {!r}'.format(target, func(*args, **kwargs)))
<commit_msg>#62: fixed typo that somehow slipped through before pushing<commit_after>from __future__ import unicode_literals
import json
import socket
from uuid import uuid4
from time import sleep
from random import randrange
from unittest import TestCase
from contextlib import closing

import six
from six.moves.queue import Queue, Empty
from six.moves.urllib.parse import urlparse, urlencode, parse_qsl

import pytest
import cherrypy
import requests
from rpctools.jsonrpc import ServerProxy
from ws4py.server.cherrypyserver import WebSocketPlugin

import sideboard.websockets
from sideboard.lib import log, config, subscribes, notifies, services, cached_property, WebSocket
from sideboard.tests import service_patcher, config_patcher
from sideboard.tests.test_sa import Session


@pytest.mark.functional
class SideboardServerTest(TestCase):
    port = config['cherrypy']['server.socket_port']
    jsonrpc_url = 'http://127.0.0.1:{}/jsonrpc'.format(port)
    jsonrpc = ServerProxy(jsonrpc_url)

    rsess_username = 'unit_tests'

    @staticmethod
    def assert_port_open(port):
        with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            sock.bind(('0.0.0.0', port))

    @staticmethod
    def assert_can_connect_to_localhost(port):
        for i in range(50):
            try:
                socket.create_connection(('127.0.0.1', port)).close()
            except Exception as e:
                sleep(0.1)
            else:
                break
        else:
            raise e

    @classmethod
    def start_cherrypy(cls):
        class Root(object):
            @cherrypy.expose
            def index(self):
                cherrypy.session['username'] = cls.rsess_username
                return cls.rsess_username

        cherrypy.tree.apps.pop('/mock_login', None)
        cherrypy.tree.mount(Root(), '/mock_login')

        cls.assert_port_open(cls.port)
        cherrypy.config.update({'engine.autoreload_on': False})
        cherrypy.engine.start()
        cherrypy.engine.wait(cherrypy.engine.states.STARTED)
        cls.assert_can_connect_to_localhost(cls.port)

    @classmethod
    def stop_cherrypy(cls):
        cherrypy.engine.stop()
        cherrypy.engine.wait(cherrypy.engine.states.STOPPED)
        cherrypy.engine.state = cherrypy.engine.states.EXITING

        # ws4py does not support stopping and restarting CherryPy
        sideboard.websockets.websocket_plugin.unsubscribe()
        sideboard.websockets.websocket_plugin = WebSocketPlugin(cherrypy.engine)
        sideboard.websockets.websocket_plugin.subscribe()

    @classmethod
    def setUpClass(cls):
        super(SideboardServerTest, cls).setUpClass()
        cls.start_cherrypy()
        cls.ws = cls.patch_websocket(services.get_websocket())
        cls.ws.connect(max_wait=5)
        assert cls.ws.connected

    @classmethod
    def tearDownClass(cls):
        cls.stop_cherrypy()
        super(SideboardServerTest, cls).tearDownClass()

    @staticmethod
    def patch_websocket(ws):
        ws.q = Queue()
        ws.fallback = ws.q.put
        return ws

    def tearDown(self):
        while not self.ws.q.empty():
            self.ws.q.get_nowait()

    def wait_for(self, func, *args, **kwargs):
        for i in range(50):
            cherrypy.engine.publish('main')  # since our unit tests don't call cherrypy.engine.block, we must publish this event manually
            try:
                result = func(*args, **kwargs)
                assert result or result is None
            except:
                sleep(0.1)
            else:
                break
        else:
            raise AssertionError('wait timed out')

    def wait_for_eq(self, target, func, *args, **kwargs):
        try:
            self.wait_for(lambda: target == func(*args, **kwargs))
        except:
            raise AssertionError('{!r} != {!r}'.format(target, func(*args, **kwargs)))

    def wait_for_ne(self, target, func, *args, **kwargs):
        try:
            self.wait_for(lambda: target != func(*args, **kwargs))
        except:
            raise AssertionError('{!r} == {!r}'.format(target, func(*args, **kwargs)))

{'fraction_non_alphanumeric': 0.10263157894736842, 'fraction_numerical': 0.007177033492822967, 'mean_word_length': 2.624187256176853, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 6, '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': '6694661', 'n_tokens_mistral': 2589, 'n_tokens_neox': 2485, 'n_words': 546}