how to put a method in to thread and use it when performing the test

Question

I have this part of code which is doing psubscribe to redis. I want to run this part of code in a thread an working in the background while the other part of code will check some notifications from this below.

def psubscribe(context, param1, param2, param3):
    context.test_config = load_config()
    RedisConnector(context.test_config["redis_host"],
                   context.test_config["redis_db_index"])
    redis_notification_subscriber_connector = RedisConnector(context.test_config["notification__redis_host"],
                                                             int(param3),
                                                             int(context.test_config[
                                                                     "notification_redis_port"]))
    context.redis_connectors = redis_notification_connector.psubscribe_to_redis_event(param1,
                                                                                                 timeout_seconds=int(
                                                                                                     param2)

)

what I have done till now: but its not running :(

context.t = threading.Thread(target=psubscribe, args=['param1', 'param2', 'param3'])
    context.t.start()

Answer 1

It is actually working. I think you didn't need actually to pass context variable to your psubscribe function. Here is an example:

1. Start http server that listens on port 8000 as a background thread
2. Send http requests to it and validate response

Feature scenario:

  Scenario: Run background process and validate responses
    Given Start background process
    Then Validate outputs

background_steps.py file:

import threading
import logging

from behave import *
from features.steps.utils import run_server
import requests


@given("Start background process")
def step_impl(context):
    context.t = threading.Thread(target=run_server, args=[8000])
    context.t.daemon = True
    context.t.start()


@then("Validate outputs")
def step_impl(context):
    response = requests.get('http://127.0.0.1:8000')
    assert response.status_code == 501

utils.py file

from http.server import HTTPServer, BaseHTTPRequestHandler


def run_server(port, server_class=HTTPServer, handler_class=BaseHTTPRequestHandler):
    server_address = ('', port)
    httpd = server_class(server_address, handler_class)
    httpd.serve_forever()