• Mocking a function in a process started with multiprocessing.Process

    From Norman Robins@norman.robins59@gmail.com to comp.lang.python on Tue Aug 27 06:48:09 2024
    From Newsgroup: comp.lang.python

    Is it possible to mock out a function in a process started with multiprocessing.Process? AI tells me it is, but I have not gotten it to
    work.

    The use case is that I have a server I want to write tests for. I want the tests to start up the server, but I need to mock out the JWT based auth functions in the server.

    If it is possible, perhaps I am doing something wrong.

    The function I want to mock out is in foo.bar.baz.validate_jwt`.

    The server imports it as from .baz import validate_jwt. I have tried to
    mock it many ways, here is the latest attempt:

    @pytest.fixture(scope="module")
    def mock_get_public_key():
    return PUBLIC_KEY

    @patch('foo.bar.baz.validate_jwt')
    def test_fn(mock_validate_jwt, mock_get_public_key):

    I also tried `@patch('foo.bar.main.validate_jwt')`

    But no matter what I tried the real functions get called.

    Here is an example:

    In directory foo/bar file baz.py

    def get_public_key(token: str) -> str:
    print("in get_public_key)

    def validate_jwt(token: str) -> dict
    public_key = get_public_key(token)
    print("in validate_jwt)

    In directory foo/bar file main.py:

    from .baz import validate_jwt

    def server(token):
    valid = validate_jwt(token)

    In my test code, which is in test/test_main.py (test and foo are at the
    same level)

    @pytest.fixture(scope="module")def mock_get_public_key():
    return PUBLIC_KEY
    def run_server(port):
    from foo.bar.main import start
    start(is_prod=False, port=port)
    @pytest.fixture(scope="module")def mock_validate_jwt(token):
    return {"user": "test_user"}
    @pytest.fixture(scope="module")def start_server():
    port = 8002
    server_process = multiprocessing.Process(target=run_server, args=(port,))
    server_process.start()

    max_retries = 10
    for i in range(max_retries):
    try:
    response = requests.get(f"http://127.0.0.1:{port}/echo?message=test")
    if response.status_code == 200:
    break
    except requests.ConnectionError:
    time.sleep(1)
    else:
    pytest.fail("Server did not start in time") @pytest.mark.usefixtures("start_server")@patch('dispatcher.main.validate_jwt')def
    test_function(mock_validate_jwt, mock_get_public_key):
    ,
    .
    .





    When test_function runs the real validate_jwt runs, not my mock.
    --- Synchronet 3.20a-Linux NewsLink 1.114