Pytest function testing multi-processing task queue service

Question

I have a task queue processing service that I'm trying to run pytest function testing on. When running it in 'production', I start this from the command line, e.g. python main.py.

I can't figure out how to start this task service from pytest to do function testing on it. How do I start up the service inside pytest so that I can then add a job to it and see if the job gets processed and added to the database when completed?

def main():
    store = "jobs"
    worker_id = 1
    # Process tasks
    task_processing[store] = multiprocessing.Process(
            target=process_tasks, args=(store, worker_id)
        )
    nanopub_processing[store].start()


if __name__ == "__main__":
    main()

Arrgghhh!!! I imported main and then ran main() and skipped a level - maybe that will teach me to name things better. Please post that as the answer and I'll accept it. — William
– William, Commented Aug 13, 2019 at 11:37

hoefling · Accepted Answer · 2019-08-13 12:25:57Z

1

Just make sure you access the main function correctly:

from main import main

def test_main():
    main()
    ...

answered Aug 13, 2019 at 12:25

hoefling

67.8k15 gold badges182 silver badges228 bronze badges

Sign up to request clarification or add additional context in comments.

Collectives™ on Stack Overflow

Pytest function testing multi-processing task queue service

1 Answer 1

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Related