conftest.py

"""
This file will contain the fixtures that are commonly needed by all other test files
for example: setting up the database, input_queue, output_queue, etc..
"""
import pytest
import os, sys, inspect
from multiprocessing import Queue
from unittest.mock import patch
from slips_files.core.database.database_manager import DBManager
from slips_files.core.output import Output
from slips_files.core.flows.zeek import Conn


# add parent dir to path for imports to work
current_dir = os.path.dirname(
    os.path.abspath(inspect.getfile(inspect.currentframe()))
)
parent_dir = os.path.dirname(current_dir)
sys.path.insert(0, parent_dir)



@pytest.fixture
def mock_rdb():
    # Create a mock version of the database object
    with patch('slips_files.core.database.database_manager.DBManager') as mock:
        yield mock.return_value

def do_nothing(*arg):
    """Used to override the print function because using the self.print causes broken pipes"""
    pass

@pytest.fixture
def input_queue():
    """This input_queue will be passed to all module constructors that need it"""
    input_queue = Queue()
    input_queue.put = do_nothing
    return input_queue


@pytest.fixture
def profiler_queue():
    """This profiler_queue will be passed to all module constructors that need it"""
    profiler_queue = Queue()
    profiler_queue.put = do_nothing
    return profiler_queue


@pytest.fixture
def database():
    db = DBManager(Output(), 'output/', 6379)
    db.print = do_nothing
    return db

@pytest.fixture
def flow():
    """returns a dummy flow for testing"""
    return Conn(
        '1601998398.945854',
        '1234',
        '192.168.1.1',
        '8.8.8.8',
        5,
        'TCP',
        'dhcp',
        80,88,
        20,20,
        20,20,
        '','',
        'Established',''
    )

# Define a fixture to run before each test
@pytest.fixture(autouse=True)
def setup_teardown_before_each_test(request):
    # Code to run before each test
    print(f"\nSetting up for test: {request.node.name}")
    #
    # # Code to run after each test
    # yield
    #
    # print(f"Tearing down after test: {request.node.name}")
    # # This is where you can perform any teardown actions needed for each test
    # ...