<commit_before>import random
import socket
from copy import deepcopy
from threading import Thread, Event, Semaphore

import pytest
from weavelib.exceptions import ObjectNotFound, ObjectAlreadyExists
from weavelib.exceptions import SchemaValidationFailed, ProtocolError
from weavelib.exceptions import BadOperation, InternalError, ObjectClosed
from weavelib.exceptions import AuthenticationFailed
from weavelib.messaging import Sender, Receiver, Creator, read_message
from weavelib.messaging import ensure_ok_message

from weaveserver.services.messaging import MessageService


CONFIG = {
    "redis_config": {
        "USE_FAKE_REDIS": True
    },
    "apps": {
        "auth1": {
            "appid": "blah",
            "type": "SYSTEM"
        },
        "auth2": {
            "appid": "plugin",
            "package": "com.plugin"
        },
        "auth3": {
            "appid": "blah1",
            "type": "SYSTEM"
        },
        "auth4": {
            "appid": "plugin2",
            "package": "com.plugin2"
        }
    }
}

TEST_QUEUES = [
    {
        "queue_name": "a.b.c",
        "request_schema": {
            "type": "object",
            "properties": {
                "foo": {
                    "type": "string"
                }
            },
            "required": ["foo"]
        }
    },
    {
        "queue_name": "a.sticky",
        "queue_type": "sticky",
        "request_schema": {
            "type": "object",
            "properties": {
                "foo": {
                    "type": "string"
                }
            },
            "required": ["foo"]
        }
    },
    {
        "queue_name": "x.keyedsticky",
        "queue_type": "keyedsticky",
        "request_schema": {"type": "object"}
    }
]


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)


def make_receiver(count, obj, sem, r):
    def on_message(msg, headers):
        obj.append(msg)
        sem.release()
        nonlocal count
        count -= 1
        if not count:
            r.stop()
    return on_message


class TestMessagingService(object):
    @classmethod
    def setup_class(cls):
        event = Event()
        cls.service = MessageService(None, CONFIG)
        cls.service.notify_start = lambda: event.set()
        cls.service_thread = Thread(target=cls.service.on_service_start)
        cls.service_thread.start()
        event.wait()
        creator = Creator()
        creator.start()
        for queue_info in TEST_QUEUES:
            creator.create(queue_info, headers={"AUTH": "auth1"})

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

    def test_connect_disconnect(self):
        with pytest.raises(IOError):<commit_msg>Test MessageServer, not the service.<commit_after>import random
import socket
from copy import deepcopy
from threading import Thread, Event, Semaphore

import pytest
from weavelib.exceptions import ObjectNotFound, ObjectAlreadyExists
from weavelib.exceptions import SchemaValidationFailed, ProtocolError
from weavelib.exceptions import BadOperation, InternalError, ObjectClosed
from weavelib.exceptions import AuthenticationFailed
from weavelib.messaging import Sender, Receiver, Creator, read_message
from weavelib.messaging import ensure_ok_message

from weaveserver.services.core.server import MessageServer


CONFIG = {
    "apps": {
        "auth1": {
            "appid": "blah",
            "type": "SYSTEM"
        },
        "auth2": {
            "appid": "plugin",
            "package": "com.plugin"
        },
    }
}

TEST_QUEUES = [
    {
        "queue_name": "a.b.c",
        "request_schema": {
            "type": "object",
            "properties": {
                "foo": {
                    "type": "string"
                }
            },
            "required": ["foo"]
        }
    },
    {
        "queue_name": "a.sticky",
        "queue_type": "sticky",
        "request_schema": {
            "type": "object",
            "properties": {
                "foo": {
                    "type": "string"
                }
            },
            "required": ["foo"]
        }
    },
    {
        "queue_name": "x.keyedsticky",
        "queue_type": "keyedsticky",
        "request_schema": {"type": "object"}
    }
]


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)


def make_receiver(count, obj, sem, r):
    def on_message(msg, headers):
        obj.append(msg)
        sem.release()
        nonlocal count
        count -= 1
        if not count:
            r.stop()
    return on_message


class TestMessageServer(object):
    @classmethod
    def setup_class(cls):
        event = Event()
        cls.server = MessageServer(11023, CONFIG["apps"], lambda: None)
        cls.server_thread = Thread(target=cls.server.run)
        cls.server_thread.start()
        event.wait()
        creator = Creator()
        creator.start()
        for queue_info in TEST_QUEUES:
            creator.create(queue_info, headers={"AUTH": "auth1"})

    @classmethod
    def teardown_class(cls):
        cls.server.shutdown()
        cls.server_thread.join()

    def test_connect_disconnect(self):
        with pytest.raises(IOError):
{'fraction_non_alphanumeric': 0.12226555875869086, 'fraction_numerical': 0.003052399525182296, 'mean_word_length': 2.1009463722397475, 'pattern_counts': {'":': 62, '<': 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': '5266334', 'n_tokens_mistral': 1822, 'n_tokens_neox': 1774, 'n_words': 357}