Python - 如何将单独的模块（不是函数）作为单独的进程运行？

Question

tl,dr：如何以编程方式将 python 模块（非函数）作为与不同 python 模块的单独进程执行？

在我的开发笔记本电脑上，我有一个包含瓶服务器的“服务器”模块。在这个模块中，name==main 子句启动了 Bottle 服务器。

@bt_app.post("/")
def server_post():
    << Generate response to 'http://server.com/' >>

if __name__ == '__main__':
    serve(bt_app, port=localhost:8080)

我还有一个包含 pytests 的“test_server”模块。在这个模块中，name==main 子句运行 pytest 并显示结果。

def test_something():
        _rtn = some_server_function()
        assert _rtn == desired

if __name__ == '__main__':
    _rtn = pytest.main([__file__])
    print("Pytest returned: ", _rtn)

目前，我手动运行服务器模块（在 localhost 上启动 Web 服务器），然后手动启动 pytest 模块，该模块向正在运行的服务器模块发出 html 请求并检查响应。

有时我忘记启动服务器模块。没什么大不了的，但很烦人。所以我想知道我是否可以以编程方式将服务器模块作为与 pytest 模块分开的进程启动（就像我现在手动做的那样），所以我不会忘记手动启动它。

谢谢

Answer 1

这是我的测试用例目录树：

test
├── server.py
└── test_server.py

server.py用flask启动一个web服务器。

from flask import Flask                                                                                                                                                
app = Flask(__name__)

@app.route('/')
def hello_world():
    return 'Hello, World!'


if __name__ == '__main__':
    app.run()

test_server.py 请求测试。

import sys                                                                                                                                                             
import requests
import subprocess
import time

p = None  # server process

def start_server():
    global p
    sys.path.append('/tmp/test')
    # here you may want to do some check. 
    # whether the server is already started, then pass this fucntion
    kwargs = {}  # here u can pass other args needed
    p = subprocess.Popen(['python','server.py'], **kwargs)

def test_function():
    response = requests.get('http://localhost:5000/')
    print('This is response body: ', response.text)

if __name__ == '__main__':
    start_server()
    time.sleep(3)  # waiting server started
    test_function()
    p.kill()

然后就可以python test_server启动服务器，做测试用例了。

PS：Popen() 需要 python3.5+。如果是旧版本，请改用run

Answer 2

import logging
import threading
import time

def thread_function(name):
    logging.info("Thread %s: starting", name)
    time.sleep(2)
    logging.info("Thread %s: finishing", name)

if __name__ == "__main__":
    format = "%(asctime)s: %(message)s"
    logging.basicConfig(format=format, level=logging.INFO,
                        datefmt="%H:%M:%S")

    threads = list()
    for index in range(3):
        logging.info("Main    : create and start thread %d.", index)
        x = threading.Thread(target=thread_function, args=(index,))
        threads.append(x)
        x.start()

    for index, thread in enumerate(threads):
        logging.info("Main    : before joining thread %d.", index)
        thread.join()
        logging.info("Main    : thread %d done", index)

使用线程，您可以一次运行多个进程！

Answer 3

Wim 基本上回答了这个问题。我查看了子流程模块。在阅读它时，我偶然发现了 os.system 函数。

简而言之，子进程是一个用于运行程序的高度灵活和功能强大的程序。而 os.system 则要简单得多，功能要少得多。

只运行一个python模块很简单，所以我选择了os.system。

import os
server_path = "python -m ../src/server.py"
os.system(server_path)

维姆，谢谢你的指点。如果这是一个完整的答案，我会赞成的。重做一个完整的答案，我会这样做。

Answer 4

异步救援。

import gevent
from gevent import monkey, spawn
monkey.patch_all()
from gevent.pywsgi import WSGIServer

@bt_app.post("/")
def server_post():
    << Generate response to 'http://server.com/' >>

def test_something():
    _rtn = some_server_function()
    assert _rtn == desired
    print("Pytest returned: ",_rtn)
    sleep(0)

if __name__ == '__main__':
    spawn(test_something) #runs async
    server = WSGIServer(("0.0.0.0", 8080, bt_app)
    server.serve_forever()