<commit_before># rewind-client talks to rewind, an event store server.
#
# Copyright (C) 2012  <NAME>
#
# This program is distributed under the MIT License. See the file LICENSE.txt
# for details.

"""Test overall Rewind execution."""
from __future__ import print_function
import threading
import time
import unittest
import uuid
import re

import mock
import zmq

import rewind.client as clients
import rewind.server.rewind as rewind


class _RewindRunnerThread(threading.Thread):

    """A thread that runs a rewind instance.

    While the thread is given command line arguments, Rewind is started as
    thread rather than external process. This makes it possible to check code
    coverage and track exit codes etc.

    """

    _EXIT_CODE = b'EXIT'

    def __init__(self, cmdline_args, exit_addr=None):
        """Constructor.

        Parameters:
        cmdline_args -- command line arguments used to execute the rewind.
        exit_addr    -- the ZeroMQ address used to send the exit message to.

        """
        thread = self

        assert '--exit-codeword' not in cmdline_args, \
            "'--exit-codeword' is added by _RewindRunnerThread. Not elsewhere"
        cmdline_args = (['--exit-codeword',
                         _RewindRunnerThread._EXIT_CODE.decode()] +
                        cmdline_args)

        def exitcode_runner(*args, **kwargs):
            try:
                thread.exit_code = rewind.main(*args, **kwargs)
            except SystemExit as e:
                thread.exit_code = e.code
            else:
                # If SystemExit is never thrown Python would have exitted with
                # exit code 0
                thread.exit_code = 0
        super(_RewindRunnerThread, self).__init__(target=exitcode_runner,
                                                  name="test-rewind",
                                                  args=(cmdline_args,))
        self._exit_addr = exit_addr

    def stop(self, context=None):
        """Send a stop message to the event thread."""
        assert self._exit_addr is not None

        if context is None:
            context = zmq.Context(1)
        socket = context.socket(zmq.REQ)
        socket.setsockopt(zmq.LINGER, 1000)
        socket.connect(self._exit_addr)
        socket.send(_RewindRunnerThread._EXIT_CODE)
        resp = socket.recv()
        assert resp == b'QUIT'
        assert not socket.getsockopt(zmq.RCVMORE)
        time.sleep(0.5)  # Acceptable exit time
        assert not self.isAlive()
        socket.close()


class TestReplication(unittest.TestCase):

    """Test high-level replication behaviour."""

    UUID_REGEXP = ("[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-"
                   "[0-9a-f]{12}")

    def setUp(self):
        """Starting a Rewind instance to test replication."""
        args = ['--query-bind-endpoint', 'tcp://127.0.0.1:8090',
                '--streaming-bind-endpoint', 'tcp://127.0.0.1:8091']
        self.rewind = _RewindRunnerThread(args, 'tcp://127.0.0.1:8090')
        self.rewind.start()

        self.context = zmq.Context(3)

        self.transmitter = self.context.socket(zmq.REQ)
        self.transmitter.connect('tcp://127.0.0.1:8090')

        # Making sure context.term() does not time out
        # Could be removed if this test works as expected
        self.transmitter.setsockopt(zmq.LINGER, 1000)

        self.receiver = self.context.socket(zmq.SUB)
        self.receiver.setsockopt(zmq.SUBSCRIBE, b'')
        self.receiver.connect('tcp://127.0.0.1:8091')

        # Time it takes to connect. This is particularly important so that the
        # receiver does not just receive the tail of the stream.
        time.sleep(0.5)

    def testBasicEventProxying(self):
        """Asserting a single event is proxied."""
        eventstring = b"THIS IS AN EVENT"

        clients.publish_event(self.transmitter, eventstring)

        received_id = self.receiver.recv().decode()
        self.assertTrue(bool(self.receiver.getsockopt(zmq.RCVMORE)))
        prev_received_id = self.receiver.recv()
        self.assertEquals(prev_received_id, b'')
        self.assertTrue(bool(self.receiver.getsockopt(zmq.RCVMORE)))
        received_string = self.receiver.recv()
        self.assertFalse(bool(self.receiver.getsockopt(zmq.RCVMORE)))

        self.assertIsNotNone(re.match(self.UUID_REGEXP, received_id))
        self.assertEqual(received_string, eventstring)

    def testProxyingABunchOfEvents(self):
        """Testing that a bunch of incoming messages processed correctly.

        That is, they are all being proxied and in order.

        """
        NMESSAGES = 200
        messages = []
        for id in range(NMESSAGES):
            eventstring = "THIS IS EVENT NUMBER {0}".format(id).encode()
            messages.append(eventstring)

        # Sending
        for msg in messages:
            clients.publish_event(self.transmitter, msg)

        # Receiving and asserting correct messages
        eventids = []
        received_messages = []
        previd = b''
        for msg in messages:
            received_id = self.receiver.recv().decode()
            self.assertTrue(bool(self.receiver.getsockopt(zmq.RCVMORE)))
            received_prev_id = self.receiver.recv()

            self.assertEquals(received_prev_id, previd)
            previd = received_id

            self.assertTrue(bool(self.receiver.getsockopt(zmq.RCVMORE)))
            received_string = self.receiver.recv()
            received_messages.append(received_string)
            self.assertFalse(bool(self.receiver.getsockopt(zmq.RCVMORE)))
            self.assertIsNotNone(re.match(self.UUID_REGEXP, received_id))
            eventids.append(received_id)
            self.assertEqual(received_string, msg)

        self.assertEqual(len(set(eventids)), len(eventids),
                         "Found duplicate event id!")
        self.assertEqual(messages, received_messages,
                         "Not all messages received")

    def tearDown(self):
        """Shutting down Rewind test instance."""
        self.transmitter.close()
        self.receiver.close()

        self.assertTrue(self.rewind.isAlive(),<commit_msg>Fixing a broken test in Python 3<commit_after># rewind-client talks to rewind, an event store server.
#
# Copyright (C) 2012  <NAME>
#
# This program is distributed under the MIT License. See the file LICENSE.txt
# for details.

"""Test overall Rewind execution."""
from __future__ import print_function
import threading
import time
import unittest
import uuid
import re

import mock
import zmq

import rewind.client as clients
import rewind.server.rewind as rewind


class _RewindRunnerThread(threading.Thread):

    """A thread that runs a rewind instance.

    While the thread is given command line arguments, Rewind is started as
    thread rather than external process. This makes it possible to check code
    coverage and track exit codes etc.

    """

    _EXIT_CODE = b'EXIT'

    def __init__(self, cmdline_args, exit_addr=None):
        """Constructor.

        Parameters:
        cmdline_args -- command line arguments used to execute the rewind.
        exit_addr    -- the ZeroMQ address used to send the exit message to.

        """
        thread = self

        assert '--exit-codeword' not in cmdline_args, \
            "'--exit-codeword' is added by _RewindRunnerThread. Not elsewhere"
        cmdline_args = (['--exit-codeword',
                         _RewindRunnerThread._EXIT_CODE.decode()] +
                        cmdline_args)

        def exitcode_runner(*args, **kwargs):
            try:
                thread.exit_code = rewind.main(*args, **kwargs)
            except SystemExit as e:
                thread.exit_code = e.code
            else:
                # If SystemExit is never thrown Python would have exitted with
                # exit code 0
                thread.exit_code = 0
        super(_RewindRunnerThread, self).__init__(target=exitcode_runner,
                                                  name="test-rewind",
                                                  args=(cmdline_args,))
        self._exit_addr = exit_addr

    def stop(self, context=None):
        """Send a stop message to the event thread."""
        assert self._exit_addr is not None

        if context is None:
            context = zmq.Context(1)
        socket = context.socket(zmq.REQ)
        socket.setsockopt(zmq.LINGER, 1000)
        socket.connect(self._exit_addr)
        socket.send(_RewindRunnerThread._EXIT_CODE)
        resp = socket.recv()
        assert resp == b'QUIT'
        assert not socket.getsockopt(zmq.RCVMORE)
        time.sleep(0.5)  # Acceptable exit time
        assert not self.isAlive()
        socket.close()


class TestReplication(unittest.TestCase):

    """Test high-level replication behaviour."""

    UUID_REGEXP = ("[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-"
                   "[0-9a-f]{12}")

    def setUp(self):
        """Starting a Rewind instance to test replication."""
        args = ['--query-bind-endpoint', 'tcp://127.0.0.1:8090',
                '--streaming-bind-endpoint', 'tcp://127.0.0.1:8091']
        self.rewind = _RewindRunnerThread(args, 'tcp://127.0.0.1:8090')
        self.rewind.start()

        self.context = zmq.Context(3)

        self.transmitter = self.context.socket(zmq.REQ)
        self.transmitter.connect('tcp://127.0.0.1:8090')

        # Making sure context.term() does not time out
        # Could be removed if this test works as expected
        self.transmitter.setsockopt(zmq.LINGER, 1000)

        self.receiver = self.context.socket(zmq.SUB)
        self.receiver.setsockopt(zmq.SUBSCRIBE, b'')
        self.receiver.connect('tcp://127.0.0.1:8091')

        # Time it takes to connect. This is particularly important so that the
        # receiver does not just receive the tail of the stream.
        time.sleep(0.5)

    def testBasicEventProxying(self):
        """Asserting a single event is proxied."""
        eventstring = b"THIS IS AN EVENT"

        clients.publish_event(self.transmitter, eventstring)

        received_id = self.receiver.recv().decode()
        self.assertTrue(bool(self.receiver.getsockopt(zmq.RCVMORE)))
        prev_received_id = self.receiver.recv()
        self.assertEquals(prev_received_id, b'')
        self.assertTrue(bool(self.receiver.getsockopt(zmq.RCVMORE)))
        received_string = self.receiver.recv()
        self.assertFalse(bool(self.receiver.getsockopt(zmq.RCVMORE)))

        self.assertIsNotNone(re.match(self.UUID_REGEXP, received_id))
        self.assertEqual(received_string, eventstring)

    def testProxyingABunchOfEvents(self):
        """Testing that a bunch of incoming messages processed correctly.

        That is, they are all being proxied and in order.

        """
        NMESSAGES = 200
        messages = []
        for id in range(NMESSAGES):
            eventstring = "THIS IS EVENT NUMBER {0}".format(id).encode()
            messages.append(eventstring)

        # Sending
        for msg in messages:
            clients.publish_event(self.transmitter, msg)

        # Receiving and asserting correct messages
        eventids = []
        received_messages = []
        previd = b''
        for msg in messages:
            received_id = self.receiver.recv()
            self.assertTrue(bool(self.receiver.getsockopt(zmq.RCVMORE)))
            received_prev_id = self.receiver.recv()

            self.assertEquals(received_prev_id, previd)
            previd = received_id

            self.assertTrue(bool(self.receiver.getsockopt(zmq.RCVMORE)))
            received_string = self.receiver.recv()
            received_messages.append(received_string)
            self.assertFalse(bool(self.receiver.getsockopt(zmq.RCVMORE)))
            self.assertIsNotNone(re.match(self.UUID_REGEXP,
                                          received_id.decode()))
            eventids.append(received_id)
            self.assertEqual(received_string, msg)

        self.assertEqual(len(set(eventids)), len(eventids),
                         "Found duplicate event id!")
        self.assertEqual(messages, received_messages,
                         "Not all messages received")

    def tearDown(self):
        """Shutting down Rewind test instance."""
        self.transmitter.close()
        self.receiver.close()

        self.assertTrue(self.rewind.isAlive(),
{'fraction_non_alphanumeric': 0.10335128184521418, 'fraction_numerical': 0.014546331270593908, 'mean_word_length': 2.216334970276557, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 10, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3810335', 'n_tokens_mistral': 3740, 'n_tokens_neox': 3514, 'n_words': 855}