init 3.0: add httprunner application

httprunner/app/main.py

@@ -0,0 +1,22 @@
+from fastapi import FastAPI
+
+from httprunner import __version__
+from .routers import deps, debugtalk, testcase
+
+app = FastAPI()
+
+
+@app.get("/hrun/version")
+async def get_hrun_version():
+    return {
+        "code": 0,
+        "message": "success",
+        "result": {
+            "HttpRunner": __version__
+        }
+    }
+
+
+app.include_router(deps.router)
+app.include_router(debugtalk.router)
+app.include_router(testcase.router)

httprunner/app/routers/debugtalk.py

@@ -0,0 +1,46 @@
+import contextlib
+import logging
+import sys
+from io import StringIO
+
+from fastapi import APIRouter
+from starlette.requests import Request
+
+router = APIRouter()
+
+
+@contextlib.contextmanager
+def stdout_io(stdout=None):
+    old = sys.stdout
+    if stdout is None:
+        stdout = StringIO()
+    sys.stdout = stdout
+    yield stdout
+    sys.stdout = old
+
+
+@router.post("/hrun/debug/debugtalk", tags=["debugtalk"])
+async def debug_python(request: Request):
+    body = await request.body()
+
+    if request.headers.get('content-transfer-encoding') == "base64":
+        # TODO: decode base64
+        pass
+
+    resp = {
+        "code": 0,
+        "message": "success",
+        "result": ""
+    }
+    try:
+        with stdout_io() as s:
+            exec(body, globals())
+            output = s.getvalue()
+            resp["result"] = output
+    except Exception as ex:
+        resp["code"] = 1
+        resp["message"] = "fail"
+        resp["result"] = str(ex)
+        logging.error(resp)
+
+    return resp

httprunner/app/routers/deps.py

@@ -0,0 +1,42 @@
+import logging
+import subprocess
+from typing import List
+
+import pkg_resources
+from fastapi import APIRouter
+
+router = APIRouter()
+
+
+@router.get("/hrun/deps", tags=["deps"])
+async def get_installed_dependenies():
+    resp = {
+        "code": 0,
+        "message": "success",
+        "result": {}
+    }
+    for p in pkg_resources.working_set:
+        resp["result"][p.project_name] = p.version
+
+    return resp
+
+
+@router.post("/hrun/deps", tags=["deps"])
+async def install_dependenies(deps: List[str]):
+    resp = {
+        "code": 0,
+        "message": "success",
+        "result": {}
+    }
+    for dep in deps:
+        try:
+            p = subprocess.run(["pip", "install", dep])
+            assert p.returncode == 0
+            resp["result"][dep] = True
+        except (AssertionError, subprocess.SubprocessError):
+            resp["result"][dep] = False
+            resp["code"] = 1
+            resp["message"] = "fail"
+            logging.error(f"failed to install dependency: {dep}")
+
+    return resp

httprunner/app/routers/testcase.py

@@ -0,0 +1,63 @@
+from fastapi import APIRouter
+
+from httprunner.api import HttpRunner
+
+router = APIRouter()
+runner = HttpRunner()
+
+
+@router.get("/hrun/debug/api", tags=["testcase"])
+async def debug_single_api():
+    pass
+
+
+@router.get("/hrun/debug/testcase", tags=["testcase"])
+async def debug_single_testcase():
+    resp = {
+        "code": 0,
+        "message": "success",
+        "result": {}
+    }
+    testcases = [
+        {
+            "config": {
+                'name': "post data",
+                'variables': {
+                    "var1": "abc",
+                    "var2": "def"
+                },
+                "export": ["status_code", "req_data"]
+            },
+            "teststeps": [
+                {
+                    "name": "post data",
+                    "request": {
+                        "url": "http://httpbin.org/post",
+                        "method": "POST",
+                        "headers": {
+                            "User-Agent": "python-requests/2.18.4",
+                            "Content-Type": "application/json"
+                        },
+                        "data": "$var1"
+                    },
+                    "extract": {
+                        "status_code": "status_code",
+                        "req_data": "content.data"
+                    },
+                    "validate": [
+                        {"eq": ["status_code", 201]}
+                    ]
+                }
+            ]
+        }
+    ]
+    tests_mapping = {
+        "testcases": testcases
+    }
+    summary = runner.run_tests(tests_mapping)
+    if not summary["success"]:
+        resp["code"] = 1
+        resp["message"] = "fail"
+
+    resp["result"] = summary
+    return resp