<commit_before># -*- coding: utf-8 -*-
'''
Test fixtures for the napalm-logs profiles.
'''
from __future__ import absolute_import

# Import python std lib
import os
import json
import time
import socket
import logging
from multiprocessing import Process

# Import third party lib
import zmq
import pytest

# Import napalm-logs pkgs
import napalm_logs.config
from napalm_logs.base import NapalmLogs

log = logging.getLogger(__name__)

NL_BASE = None
NL_PROC = None
TEST_SKT = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
TEST_CLIENT = None

NAPALM_LOGS_TEST_LOG_LEVEL = os.getenv('NAPALM_LOGS_TEST_LOG_LEVEL', default='warning')
NAPALM_LOGS_TEST_ADDR = os.getenv('NAPALM_LOGS_TEST_ADDR', default='0.0.0.0')
NAPALM_LOGS_TEST_PORT = os.getenv('NAPALM_LOGS_TEST_PORT', default=17191)
NAPALM_LOGS_TEST_PUB_ADDR = os.getenv('NAPALM_LOGS_TEST_PUB_ADDR', default='0.0.0.0')
NAPALM_LOGS_TEST_PUB_PORT = os.getenv('NAPALM_LOGS_TEST_PUB_PORT', default=17193)

logging_level = napalm_logs.config.LOGGING_LEVEL.get(NAPALM_LOGS_TEST_LOG_LEVEL.lower())
logging.basicConfig(level=logging_level, format=napalm_logs.config.LOG_FORMAT)


def startup_proc():
    '''
    Startup the napalm-logs process.
    '''
    global NL_BASE
    global NL_PROC
    log.debug('Starting up the napalm-logs process')
    NL_BASE = NapalmLogs(disable_security=True,
                         address=NAPALM_LOGS_TEST_ADDR,
                         port=NAPALM_LOGS_TEST_PORT,
                         publisher=[{'zmq': {}}],
                         listener=[{'udp': {}}],
                         publish_address=NAPALM_LOGS_TEST_PUB_ADDR,
                         publish_port=NAPALM_LOGS_TEST_PUB_PORT,
                         log_level=NAPALM_LOGS_TEST_LOG_LEVEL)
    NL_PROC = Process(target=NL_BASE.start_engine)
    NL_PROC.start()


# Startup the napalm-logs process
startup_proc()


def startup_local_client():
    '''
    Startup a local ZMQ client to receive the published messages.
    '''
    time.sleep(2)
    global TEST_CLIENT
    context = zmq.Context()
    TEST_CLIENT = context.socket(zmq.SUB)
    TEST_CLIENT.connect('tcp://{addr}:{port}'.format(
        addr=NAPALM_LOGS_TEST_PUB_ADDR,
        port=NAPALM_LOGS_TEST_PUB_PORT)<commit_msg>Send RAW and UNKNOWN message types, so it helps debugging<commit_after># -*- coding: utf-8 -*-
'''
Test fixtures for the napalm-logs profiles.
'''
from __future__ import absolute_import

# Import python std lib
import os
import json
import time
import socket
import logging
from multiprocessing import Process

# Import third party lib
import zmq
import pytest

# Import napalm-logs pkgs
import napalm_logs.config
from napalm_logs.base import NapalmLogs

log = logging.getLogger(__name__)

NL_BASE = None
NL_PROC = None
TEST_SKT = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
TEST_CLIENT = None

NAPALM_LOGS_TEST_LOG_LEVEL = os.getenv('NAPALM_LOGS_TEST_LOG_LEVEL', default='warning')
NAPALM_LOGS_TEST_ADDR = os.getenv('NAPALM_LOGS_TEST_ADDR', default='0.0.0.0')
NAPALM_LOGS_TEST_PORT = os.getenv('NAPALM_LOGS_TEST_PORT', default=17191)
NAPALM_LOGS_TEST_PUB_ADDR = os.getenv('NAPALM_LOGS_TEST_PUB_ADDR', default='0.0.0.0')
NAPALM_LOGS_TEST_PUB_PORT = os.getenv('NAPALM_LOGS_TEST_PUB_PORT', default=17193)

logging_level = napalm_logs.config.LOGGING_LEVEL.get(NAPALM_LOGS_TEST_LOG_LEVEL.lower())
logging.basicConfig(level=logging_level, format=napalm_logs.config.LOG_FORMAT)


def startup_proc():
    '''
    Startup the napalm-logs process.
    '''
    global NL_BASE
    global NL_PROC
    log.debug('Starting up the napalm-logs process')
    NL_BASE = NapalmLogs(disable_security=True,
                         address=NAPALM_LOGS_TEST_ADDR,
                         port=NAPALM_LOGS_TEST_PORT,
                         publisher=[{'zmq': {'send_raw': True, 'send_unknown': True}}],
                         listener=[{'udp': {}}],
                         publish_address=NAPALM_LOGS_TEST_PUB_ADDR,
                         publish_port=NAPALM_LOGS_TEST_PUB_PORT,
                         log_level=NAPALM_LOGS_TEST_LOG_LEVEL)
    NL_PROC = Process(target=NL_BASE.start_engine)
    NL_PROC.start()


# Startup the napalm-logs process
startup_proc()


def startup_local_client():
    '''
    Startup a local ZMQ client to receive the published messages.
    '''
    time.sleep(2)
    global TEST_CLIENT
    context = zmq.Context()
    TEST_CLIENT = context.socket(zmq.SUB)
    TEST_CLIENT.connect('tcp://{addr}:{port}'.format(
        addr=NAPALM_LOGS_TEST_PUB_ADDR,
        port=NAPALM_LOGS_TEST_PUB_PORT)
{'fraction_non_alphanumeric': 0.0960352422907489, 'fraction_numerical': 0.00881057268722467, 'mean_word_length': 3.667009249743063, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, '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': '2057528', 'n_tokens_mistral': 1665, 'n_tokens_neox': 1566, 'n_words': 276}