<commit_before>import socket
from threading import Thread, Event

from retask import Task
import pytest

from app.core.messaging import Sender, Receiver, RequiredFieldsMissing
from app.core.messaging import QueueNotFound, SchemaValidationFailed
from app.services.messaging import MessageService


CONFIG = {
    "redis_config": {},
    "queues": {
        "queues": [
            {
                "queue_name": "a.b.c",
                "queue_type": "dummy",
                "request_schema": {
                    "type": "object",
                    "properties": {
                        "foo": {
                            "type": "string"
                        }
                    },
                    "required": ["foo"]
                }
            }
        ]
    }
}


def send_raw(msg):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect(("localhost", Sender.PORT))
    wfile = sock.makefile('wb', Sender.WRITE_BUF_SIZE)
    rfile = sock.makefile('rb', Sender.READ_BUF_SIZE)

    wfile.write(msg.encode())
    wfile.flush()
    sock.shutdown(socket.SHUT_WR)

    resp = rfile.readline().decode("UTF-8").strip()

    sock.close()

    return resp


class TestMessagingService(object):
    @classmethod
    def setup_class(cls):
        event = Event()
        cls.service = MessageService(CONFIG)
        cls.service.notify_start = lambda: event.set()
        cls.service_thread = Thread(target=cls.service.on_service_start)
        cls.service_thread.start()
        event.wait()
        print("Server started.")

    @classmethod
    def teardown_class(cls):
        cls.service.on_service_stop()
        cls.service_thread.join()

    def test_connect_disconnect(self):
        assert send_raw('') == ''

    def test_bad_structure(self):
        assert send_raw("sdkhsds-ssgdlks!!3l") == "INVALID-MESSAGE-STRUCTURE"

    def test_required_fields_missing(self):
        assert send_raw("OP enqueue\nMSG blah\n\n") == "REQUIRED-FIELDS-MISSING"

    def test_bad_operation(self):
        assert send_raw('MSG {"a": "b"}\n'
                        'OP bad-operation\n'
                        'Q a.b.c\n\n') == "BAD-OPERATION"

    def test_bad_json(self):
        assert send_raw('MSG {a": "b"}\n'
                        'OP enqueue\n'
                        'Q a.b.c\n\n') == "SCHEMA-VALIDATION-FAILED"

    def test_enqueue_without_task(self):
        s = Sender("a.b.c")
        s.start()
        with pytest.raises(RequiredFieldsMissing):
            s.send(None)

    def test_enqueue_to_unknown_queue(self):
        s = Sender("unknown.queue")
        s.start()
        with pytest.raises(QueueNotFound):
            s.send(Task({"a": "b"}))

    def test_enqueue_with_bad_schema(self):
        s = Sender("a.b.c")
        s.start()
        with pytest.raises(SchemaValidationFailed):
            s.send(Task({"foo": [1, 2]}))

    def test_simple_enqueue_dequeue(self):
        msgs = []

        s = Sender("a.b.c")
        s.start()
        s.send(Task({"foo": "bar"}))

        r = Receiver("a.b.c")
        r.start()
        r.on_message = lambda msg: msgs.append(msg) or r.stop()
        thread = Thread(target=r.run)
        thread.start()

        thread.join()

        assert msgs == [{"foo": "bar"}]

    def test_multiple_enqueue_dequeue(self):
        obj = {"foo": "bar"}<commit_msg>Update test cases.<commit_after>import socket
from threading import Thread, Event

from retask import Task
import pytest

from app.core.messaging import Sender, Receiver, RequiredFieldsMissing
from app.core.messaging import QueueNotFound, SchemaValidationFailed
from app.core.messaging import InvalidMessageStructure, BadOperation
from app.core.messaging import read_message, ensure_ok_message
from app.services.messaging import MessageService


CONFIG = {
    "redis_config": {},
    "queues": {
        "queues": [
            {
                "queue_name": "a.b.c",
                "queue_type": "dummy",
                "request_schema": {
                    "type": "object",
                    "properties": {
                        "foo": {
                            "type": "string"
                        }
                    },
                    "required": ["foo"]
                }
            }
        ]
    }
}


def send_raw(msg):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect(("localhost", Sender.PORT))
    wfile = sock.makefile('wb', Sender.WRITE_BUF_SIZE)
    rfile = sock.makefile('rb', Sender.READ_BUF_SIZE)

    wfile.write(msg.encode())
    wfile.flush()
    sock.shutdown(socket.SHUT_WR)

    msg = read_message(rfile)

    sock.close()

    ensure_ok_message(msg)


class TestMessagingService(object):
    @classmethod
    def setup_class(cls):
        event = Event()
        cls.service = MessageService(CONFIG)
        cls.service.notify_start = lambda: event.set()
        cls.service_thread = Thread(target=cls.service.on_service_start)
        cls.service_thread.start()
        event.wait()

    @classmethod
    def teardown_class(cls):
        cls.service.on_service_stop()
        cls.service_thread.join()

    def test_connect_disconnect(self):
        with pytest.raises(IOError):
            send_raw('')

    def test_bad_structure(self):
        with pytest.raises(InvalidMessageStructure):
            send_raw("sdkhsds-ss!3l")

    def test_required_fields_missing(self):
        with pytest.raises(RequiredFieldsMissing):
            send_raw("HDR enqueue\nMSG blah\n\n")

    def test_bad_operation(self):
        with pytest.raises(BadOperation):
            send_raw('MSG {"a": "b"}\nOP bad-operation\nQ a.b.c\n\n')

    def test_bad_json(self):
        with pytest.raises(SchemaValidationFailed):
            send_raw('MSG {a": "b"}\nOP enqueue\nQ a.b.c\n\n')

    def test_enqueue_without_queue_header(self):
        with pytest.raises(RequiredFieldsMissing):
            send_raw('MSG {"a": "b"}\nOP enqueue\n\n')

    def test_enqueue_without_task(self):
        s = Sender("a.b.c")
        s.start()
        with pytest.raises(RequiredFieldsMissing):
            s.send(None)

    def test_enqueue_to_unknown_queue(self):
        s = Sender("unknown.queue")
        s.start()
        with pytest.raises(QueueNotFound):
            s.send(Task({"a": "b"}))

    def test_dequeue_from_unknown_queue(self):
        r = Receiver("unknown.queue")
        r.start()
        with pytest.raises(QueueNotFound):
            r.receive()

    def test_enqueue_with_bad_schema(self):
        s = Sender("a.b.c")
        s.start()
        with pytest.raises(SchemaValidationFailed):
            s.send(Task({"foo": [1, 2]}))

    def test_simple_enqueue_dequeue(self):
        msgs = []

        s = Sender("a.b.c")
        r = Receiver("a.b.c")
        s.start()
        r.start()
        r.on_message = lambda msg: msgs.append(msg) or r.stop()

        s.send(Task({"foo": "bar"}))
        thread = Thread(target=r.run)
        thread.start()

        thread.join()

        assert msgs == [{"foo": "bar"}]

    def test_multiple_enqueue_dequeue(self):
        obj = {"foo": "bar"}
{'fraction_non_alphanumeric': 0.131216633885923, 'fraction_numerical': 0.0009834223096375386, 'mean_word_length': 2.1033129904097647, 'pattern_counts': {'":': 37, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, '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': '353083', 'n_tokens_mistral': 2310, 'n_tokens_neox': 2239, 'n_words': 389}