1

使用 Python3.7，我有两个 py 脚本。server_execute.py 是接受一个 post 请求，一旦被接受，就会触发 server_scripts.py 异步启动测试脚本执行。

server_execute.py 脚本将接受一个发布请求，但我无法让它触发 server_scripts.py 异步函数来执行测试。我是发布请求和异步编程的新手......非常感谢任何帮助，谢谢！

server_execute.py

#!/usr/bin/python3
import server_scripts
from aiohttp import web
import json

global e2e


async def handle(request):
    response_obj = {'status': 'success'}
    return web.Response(text=json.dumps(response_obj), status=200)

#in python command prompt enter the following for the post request:
# import requests
# response = requests.post("http://localhost:8080/e2e?name=222")
# response.json()


async def e2e(request):
    try:
        e2e = request.query['name']
        print("Running e2e tests:", e2e)
        server_scripts.runPyTest222("test")
        print("Parallel script execution completed.")
        response_obj = {'status': 'success', 'message': 'e2e successfully run'}
        return web.Response(text=json.dumps(response_obj), status=200)
    except Exception as e:
        response_obj = {'status': 'failed', 'message': str(e)}
        return web.Response(text=json.dumps(response_obj), status=500)


app = web.Application()
app.router.add_get('/', handle)
app.router.add_post('/e2e', e2e)

web.run_app(app)

server_scripts.py

#!/usr/bin/python3
import asyncio
import time
import datetime
from colorama import Fore
from colorama import Style


async def run(cmd):
        proc = await asyncio.create_subprocess_shell(
            cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE)

        stdout, stderr = await proc.communicate()

        print(f'[{cmd!r} exited with {proc.returncode}]')
        if stdout:
            if proc.returncode == 0:
                print(f'[stdout]\n{Fore.GREEN}{stdout.decode()}{Style.RESET_ALL}!')
            else:
                print(f'[stdout]\n{Fore.RED}{stdout.decode()}{Style.RESET_ALL}!')
        if stderr:
            print(f'[stderr]\n{Fore.RED}{stderr.decode()}{Style.RESET_ALL}!')


async def runPyTest222(filename):
    print('started at this time:', time.strftime('%X'))
    await asyncio.gather(
        run('python3 -m pytest test_222_rlv_overview_005.py'),
        run('python3 -m pytest test_222_ltv_overview_002.py'),
    )
    print('finished at', time.strftime('%X'))


if __name__ == '__main__':
    a = datetime.datetime.now()
    asyncio.run(runPyTest222("test"))
    b = datetime.datetime.now()
    print(b - a)
4

0 回答 0