GithubBackup/httprunner
1
0
Fork 0
mirror of https://github.com/httprunner/httprunner.git synced 2026-05-13 17:19:54 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into master

Browse Source
This commit is contained in:
debugtalk
2022-04-08 09:38:37 +08:00
committed by GitHub
parent a0d9bcaba6 5e40e3692d
commit de2faaa652
50 changed files with 1258 additions and 999 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.github/workflows/hrp-release.yml vendored
View File

@@ -4,6 +4,9 @@ on:
release:
types: [created]
env:
DISABLE_GA: "true"
jobs:
releases-matrix:
name: Release hrp cli binaries

3
.github/workflows/hrp-scaffold.yml vendored
View File

@@ -6,6 +6,9 @@ on:
- master
pull_request:
env:
DISABLE_GA: "true"
jobs:
scaffold-with-python-plugin:
strategy:

3
.github/workflows/smoketest.yml vendored
View File

@@ -6,6 +6,9 @@ on:
- master
pull_request:
env:
DISABLE_GA: "true"
jobs:
smoke-test:

3
.github/workflows/unittest.yml vendored
View File

@@ -6,6 +6,9 @@ on:
- master
pull_request:
env:
DISABLE_GA: "true"
jobs:
py-httprunner:
runs-on: ${{ matrix.os }}

5
docs/CHANGELOG.md
View File

@@ -3,16 +3,17 @@
## v4.0.0-alpha
- refactor: merge [hrp] into httprunner v4, which will include golang and python dual engine
- refactor: redesign `IStep` to make step extensible to support implementing new protocols and test types
- feat: disable GA events report by setting environment `DISABLE_GA=true`
**go version**
- feat: add `--profile` flag for har2case to support overwrite headers/cookies with specified yaml/json profile file
- feat: support run testcases in specified folder path, including testcases in sub folders
- feat: support HTTP/2 protocol
- change: integrate [sentry sdk][sentry sdk] for panic reporting and analysis
- change: lock funplugin version when creating scaffold project
- fix: call referenced api/testcase with relative path
- refactor: redesign `IStep` to make step extensible to support implementing new protocols and test types
- feat: support HTTP/2 protocol
**python version**

2
examples/httpbin/basic_test.py
View File

@@ -1,4 +1,4 @@
# NOTE: Generated By HttpRunner v3.1.7
# NOTE: Generated By HttpRunner v4.0.0-alpha
# FROM: basic.yml

2
examples/httpbin/hooks_test.py
View File

@@ -1,4 +1,4 @@
# NOTE: Generated By HttpRunner v3.1.7
# NOTE: Generated By HttpRunner v4.0.0-alpha
# FROM: hooks.yml

2
examples/httpbin/load_image_test.py
View File

@@ -1,4 +1,4 @@
# NOTE: Generated By HttpRunner v3.1.7
# NOTE: Generated By HttpRunner v4.0.0-alpha
# FROM: load_image.yml

2
examples/httpbin/upload_test.py
View File

@@ -1,4 +1,4 @@
# NOTE: Generated By HttpRunner v3.1.7
# NOTE: Generated By HttpRunner v4.0.0-alpha
# FROM: upload.yml

2
examples/httpbin/validate_test.py
View File

@@ -1,4 +1,4 @@
# NOTE: Generated By HttpRunner v3.1.7
# NOTE: Generated By HttpRunner v4.0.0-alpha
# FROM: validate.yml

18
examples/postman_echo/conftest.py
View File

@@ -12,13 +12,13 @@ from httprunner.utils import get_platform, ExtendJSONEncoder
@pytest.fixture(scope="session", autouse=True)
def session_fixture(request):
"""setup and teardown each task"""
logger.info(f"start running testcases ...")
logger.info("start running testcases ...")
start_at = time.time()
yield
logger.info(f"task finished, generate task summary for --save-tests")
logger.info("task finished, generate task summary for --save-tests")
summary = {
"success": True,
@@ -36,24 +36,27 @@ def session_fixture(request):
summary["success"] &= testcase_summary.success
summary["stat"]["testcases"]["total"] += 1
summary["stat"]["teststeps"]["total"] += len(testcase_summary.step_datas)
summary["stat"]["teststeps"]["total"] += len(testcase_summary.step_results)
if testcase_summary.success:
summary["stat"]["testcases"]["success"] += 1
summary["stat"]["teststeps"]["successes"] += len(
testcase_summary.step_datas
testcase_summary.step_results
)
else:
summary["stat"]["testcases"]["fail"] += 1
summary["stat"]["teststeps"]["successes"] += (
len(testcase_summary.step_datas) - 1
len(testcase_summary.step_results) - 1
)
summary["stat"]["teststeps"]["failures"] += 1
testcase_summary_json = testcase_summary.dict()
testcase_summary_json["records"] = testcase_summary_json.pop("step_datas")
testcase_summary_json["records"] = testcase_summary_json.pop("step_results")
summary["details"].append(testcase_summary_json)
summary_path = r"/Users/debugtalk/MyProjects/HttpRunner-dev/httprunner/examples/postman_echo/logs/request_methods/hardcode.summary.json"
summary_path = os.path.join(
os.getcwd(),
"examples/postman_echo/logs/request_methods/hardcode.summary.json"
)
summary_dir = os.path.dirname(summary_path)
os.makedirs(summary_dir, exist_ok=True)
@@ -61,4 +64,3 @@ def session_fixture(request):
json.dump(summary, f, indent=4, ensure_ascii=False, cls=ExtendJSONEncoder)
logger.info(f"generated task summary: {summary_path}")

2
examples/postman_echo/request_methods/conftest.py
View File

@@ -23,7 +23,7 @@ def session_fixture(request):
yield
logger.debug(f"teardown task fixture")
logger.debug("teardown task fixture")
# teardown task
# TODO: upload task summary

1
examples/postman_echo/request_methods/demo_testsuite_yml/request_with_functions_test.py
View File

@@ -20,7 +20,6 @@ class TestCaseRequestWithFunctions(HttpRunner):
.base_url("https://postman-echo.com")
.verify(False)
.export(*["foo3"])
.locust_weight(2)
)
teststeps = [

1
examples/postman_echo/request_methods/demo_testsuite_yml/request_with_testcase_reference_test.py
View File

@@ -29,7 +29,6 @@ class TestCaseRequestWithTestcaseReference(HttpRunner):
)
.base_url("https://postman-echo.com")
.verify(False)
.locust_weight(3)
)
teststeps = [

1
examples/postman_echo/request_methods/request_with_functions_test.py
View File

@@ -20,7 +20,6 @@ class TestCaseRequestWithFunctions(HttpRunner):
.base_url("https://postman-echo.com")
.verify(False)
.export(*["foo3"])
.locust_weight(2)
)
teststeps = [

15
docs/dev.md → hrp/README.md
View File

@@ -1,3 +1,4 @@
# 代码阅读指南（golang 部分）
## 核心数据结构
@@ -21,14 +22,14 @@ type IStep interface {
}
```
我们只需遵循 `IStep` 的接口定义，即可实现各种类型的测试步骤类型。当前已支持的步骤类型包括：
我们只需遵循 `IStep` 的接口定义，即可实现各种类型的测试步骤类型。当前 hrp 已支持的步骤类型包括：
- request：发起单次 HTTP 请求
- api：引用执行其它 API 文件
- testcase：引用执行其它测试用例文件
- thinktime：思考时间，按照配置的逻辑进行等待
- transaction：事务机制，用于压测
- rendezvous：集合点机制，用于压测
- [request](step_request.go)：发起单次 HTTP 请求
- [api](step_api.go)：引用执行其它 API 文件
- [testcase](step_testcase.go)：引用执行其它测试用例文件
- [thinktime](step_thinktime.go)：思考时间，按照配置的逻辑进行等待
- [transaction](step_transaction.go)：事务机制，用于压测
- [rendezvous](step_rendezvous.go)：集合点机制，用于压测
基于该机制，我们可以扩展支持新的协议类型，例如 HTTP2/WebSocket/RPC 等；同时也可以支持新的测试类型，例如 UI 自动化。甚至我们还可以在一个测试用例中混合调用多种不同的 Step 类型，例如实现 HTTP/RPC/UI 混合场景。

7
hrp/boomer.go
View File

@@ -1,6 +1,7 @@
package hrp
import (
"os"
"sync"
"time"
@@ -42,14 +43,16 @@ func (b *HRPBoomer) Run(testcases ...ITestCase) {
// load all testcases
testCases, err := loadTestCases(testcases...)
if err != nil {
panic(err)
log.Error().Err(err).Msg("failed to load testcases")
os.Exit(1)
}
for _, testcase := range testCases {
cfg := testcase.Config
err = initParameterIterator(cfg, "boomer")
if err != nil {
panic(err)
log.Error().Err(err).Msg("failed to init parameter iterator")
os.Exit(1)
}
rendezvousList := initRendezvous(testcase, int64(b.GetSpawnCount()))
task := b.convertBoomerTask(testcase, rendezvousList)

4
hrp/internal/builtin/utils.go
View File

@@ -227,14 +227,14 @@ func loadFromCSV(path string) []map[string]interface{} {
file, err := readFile(path)
if err != nil {
log.Error().Err(err).Msg("read csv file failed")
panic(err)
os.Exit(1)
}
r := csv.NewReader(strings.NewReader(string(file)))
content, err := r.ReadAll()
if err != nil {
log.Error().Err(err).Msg("parse csv file failed")
panic(err)
os.Exit(1)
}
var result []map[string]interface{}
for i := 1; i < len(content); i++ {

2
hrp/internal/sdk/client_test.go
View File

@@ -10,7 +10,7 @@ func TestSendEvents(t *testing.T) {
Action: "SendEvents",
Value: 123,
}
err := gaClient.SendEvent(event)
err := SendEvent(event)
if err != nil {
t.Fatal(err)
}

5
hrp/internal/sdk/init.go
View File

@@ -2,6 +2,7 @@ package sdk
import (
"fmt"
"os"
"github.com/denisbrodbeck/machineid"
"github.com/getsentry/sentry-go"
@@ -46,5 +47,9 @@ func init() {
}
func SendEvent(e IEvent) error {
if os.Getenv("DISABLE_GA") == "true" {
// do not send GA events in CI environment
return nil
}
return gaClient.SendEvent(e)
}

20
hrp/parser.go
View File

@@ -4,6 +4,7 @@ import (
builtinJSON "encoding/json"
"fmt"
"net/url"
"path"
"reflect"
"regexp"
"strings"
@@ -26,18 +27,29 @@ type Parser struct {
}
func buildURL(baseURL, stepURL string) string {
uConfig, err := url.Parse(baseURL)
uStep, err := url.Parse(stepURL)
if err != nil {
log.Error().Str("baseURL", baseURL).Err(err).Msg("[buildURL] parse baseURL failed")
log.Error().Str("stepURL", stepURL).Err(err).Msg("[buildURL] parse url failed")
return ""
}
uStep, err := uConfig.Parse(stepURL)
// step url is absolute url
if uStep.Host != "" {
return stepURL
}
// step url is relative, based on base url
uConfig, err := url.Parse(baseURL)
if err != nil {
log.Error().Str("stepURL", stepURL).Err(err).Msg("[buildURL] parse stepURL failed")
log.Error().Str("baseURL", baseURL).Err(err).Msg("[buildURL] parse url failed")
return ""
}
// merge url
uStep.Scheme = uConfig.Scheme
uStep.Host = uConfig.Host
uStep.Path = path.Join(uConfig.Path, uStep.Path)
// base url missed
return uStep.String()
}

35
hrp/parser_test.go
View File

@@ -11,25 +11,44 @@ import (
func TestBuildURL(t *testing.T) {
var url string
url = buildURL("https://postman-echo.com", "/get")
if url != "https://postman-echo.com/get" {
t.Fatalf("buildURL error, %s != 'https://postman-echo.com/get'", url)
if !assert.Equal(t, url, "https://postman-echo.com/get") {
t.Fail()
}
url = buildURL("https://postman-echo.com", "get")
if !assert.Equal(t, url, "https://postman-echo.com/get") {
t.Fail()
}
url = buildURL("https://postman-echo.com/", "/get")
if !assert.Equal(t, url, "https://postman-echo.com/get") {
t.Fail()
}
url = buildURL("https://postman-echo.com/abc/", "/get?a=1&b=2")
if url != "https://postman-echo.com/get?a=1&b=2" {
t.Fatalf("buildURL error, %s != 'https://postman-echo.com/get'", url)
if !assert.Equal(t, url, "https://postman-echo.com/abc/get?a=1&b=2") {
t.Fail()
}
url = buildURL("https://postman-echo.com/abc", "get?a=1&b=2")
if !assert.Equal(t, url, "https://postman-echo.com/abc/get?a=1&b=2") {
t.Fail()
}
// omit query string in base url
url = buildURL("https://postman-echo.com/abc?x=6&y=9", "/get?a=1&b=2")
if !assert.Equal(t, url, "https://postman-echo.com/abc/get?a=1&b=2") {
t.Fail()
}
url = buildURL("", "https://postman-echo.com/get")
if url != "https://postman-echo.com/get" {
t.Fatalf("buildURL error, %s != 'https://postman-echo.com/get'", url)
if !assert.Equal(t, url, "https://postman-echo.com/get") {
t.Fail()
}
// notice: step request url > config base url
url = buildURL("https://postman-echo.com", "https://httpbin.org/get")
if url != "https://httpbin.org/get" {
t.Fatalf("buildURL error, %s != 'https://httpbin.org/get'", url)
if !assert.Equal(t, url, "https://httpbin.org/get") {
t.Fail()
}
}

14
hrp/response.go
View File

@@ -175,13 +175,13 @@ func (v *responseObject) Validate(iValidators []interface{}, variablesMapping ma
Msgf("validate %s", checkItem)
if !result {
v.t.Fail()
return errors.New(fmt.Sprintf(
"do assertion failed, checkExpr: %v, assertMethod: %v, checkValue: %v, expectValue: %v",
validator.Check,
assertMethod,
checkValue,
expectValue,
))
log.Error().
Str("checkExpr", validator.Check).
Str("assertMethod", assertMethod).
Interface("checkValue", checkValue).
Interface("expectValue", expectValue).
Msg("assert failed")
return errors.New("step validation failed")
}
}
return nil

6
hrp/runner_test.go
View File

@@ -16,7 +16,8 @@ func buildHashicorpGoPlugin() {
cmd := exec.Command("go", "build",
"-o", templatesDir+"debugtalk.bin", templatesDir+"plugin/debugtalk.go")
if err := cmd.Run(); err != nil {
panic(err)
log.Error().Err(err).Msg("build hashicorp go plugin failed")
os.Exit(1)
}
}
@@ -30,7 +31,8 @@ func buildHashicorpPyPlugin() {
pluginFile := templatesDir + "debugtalk.py"
err := scaffold.CopyFile("templates/plugin/debugtalk.py", pluginFile)
if err != nil {
panic(err)
log.Error().Err(err).Msg("build hashicorp python plugin failed")
os.Exit(1)
}
}

34
hrp/session.go
View File

@@ -69,24 +69,40 @@ func (r *SessionRunner) Start() error {
r.startTime = time.Now()
// run step in sequential order
for _, step := range r.testCase.TestSteps {
_, err := step.Run(r)
log.Info().Str("step", step.Name()).
Str("type", string(step.Type())).Msg("run step start")
stepResult, err := step.Run(r)
if err != nil && r.hrpRunner.failfast {
log.Error().
Str("step", stepResult.Name).
Str("type", string(stepResult.StepType)).
Bool("success", false).
Msg("run step end")
return errors.Wrap(err, "abort running due to failfast setting")
}
// update extracted variables
for k, v := range stepResult.ExportVars {
r.sessionVariables[k] = v
}
// update testcase summary
r.updateSummary(stepResult)
log.Info().
Str("step", stepResult.Name).
Str("type", string(stepResult.StepType)).
Bool("success", stepResult.Success).
Interface("exportVars", stepResult.ExportVars).
Msg("run step end")
}
log.Info().Str("testcase", config.Name).Msg("run testcase end")
return nil
}
func (r *SessionRunner) UpdateSession(vars map[string]interface{}) {
for k, v := range vars {
r.sessionVariables[k] = v
}
}
// UpdateSummary appends step result to summary
func (r *SessionRunner) UpdateSummary(stepResult *StepResult) {
// updateSummary appends step result to summary
func (r *SessionRunner) updateSummary(stepResult *StepResult) {
r.summary.Records = append(r.summary.Records, stepResult)
r.summary.Stat.Total += 1
if stepResult.Success {

4
hrp/step_api.go
View File

@@ -3,8 +3,6 @@ package hrp
import (
"fmt"
"github.com/rs/zerolog/log"
"github.com/httprunner/httprunner/hrp/internal/builtin"
)
@@ -93,8 +91,6 @@ func (s *StepAPIWithOptionalArgs) Struct() *TStep {
}
func (s *StepAPIWithOptionalArgs) Run(r *SessionRunner) (*StepResult, error) {
log.Info().Str("api", s.step.Name).Msg("run referenced api")
// extend request with referenced API
api, _ := s.step.API.(*API)
extendWithAPI(s.step, api)

24
hrp/step_request.go
View File

@@ -250,8 +250,6 @@ func (r *requestBuilder) prepareBody(stepVariables map[string]interface{}) error
}
func runStepRequest(r *SessionRunner, step *TStep) (stepResult *StepResult, err error) {
log.Info().Str("step", step.Name).Msg("run step start")
stepResult = &StepResult{
Name: step.Name,
StepType: stepTypeRequest,
@@ -262,20 +260,16 @@ func runStepRequest(r *SessionRunner, step *TStep) (stepResult *StepResult, err
defer func() {
// update testcase summary
if err != nil {
log.Error().Err(err).Msg("run request step failed")
stepResult.Attachment = err.Error()
} else {
// update extracted variables
r.UpdateSession(stepResult.ExportVars)
log.Info().
Str("step", step.Name).
Bool("success", stepResult.Success).
Interface("exportVars", stepResult.ExportVars).
Msg("run step end")
}
r.UpdateSummary(stepResult)
}()
// override step variables
stepVariables, err := r.MergeStepVariables(step.Variables)
if err != nil {
return
}
sessionData := newSessionData()
parser := r.GetParser()
config := r.GetConfig()
@@ -283,12 +277,6 @@ func runStepRequest(r *SessionRunner, step *TStep) (stepResult *StepResult, err
rb := newRequestBuilder(parser, config, step.Request)
rb.req.Method = string(step.Request.Method)
// override step variables
stepVariables, err := r.MergeStepVariables(step.Variables)
if err != nil {
return
}
err = rb.prepareUrlParams(stepVariables)
if err != nil {
return

37
hrp/step_testcase.go
View File

@@ -44,26 +44,27 @@ func (s *StepTestCaseWithOptionalArgs) Struct() *TStep {
}
func (s *StepTestCaseWithOptionalArgs) Run(r *SessionRunner) (*StepResult, error) {
stepVariables, err := r.MergeStepVariables(s.step.Variables)
if err != nil {
return nil, err
}
s.step.Variables = stepVariables
log.Info().Str("testcase", s.step.Name).Msg("run referenced testcase")
stepResult := &StepResult{
Name: s.step.Name,
StepType: stepTypeTestCase,
Success: false,
}
testcase := s.step.TestCase.(*TestCase)
// copy testcase to avoid data racing
copiedTestCase := &TestCase{}
if err := copier.Copy(copiedTestCase, testcase); err != nil {
log.Error().Err(err).Msg("copy testcase failed")
stepVariables, err := r.MergeStepVariables(s.step.Variables)
if err != nil {
return stepResult, err
}
// copy step to avoid data racing
copiedStep := &TStep{}
if err := copier.Copy(copiedStep, s.step); err != nil {
log.Error().Err(err).Msg("copy step failed")
return stepResult, err
}
copiedStep.Variables = stepVariables
copiedTestCase := copiedStep.TestCase.(*TestCase)
// override testcase config
extendWithTestCase(s.step, copiedTestCase)
@@ -73,16 +74,14 @@ func (s *StepTestCaseWithOptionalArgs) Run(r *SessionRunner) (*StepResult, error
err = sessionRunner.Start()
stepResult.Elapsed = time.Since(start).Milliseconds()
if err != nil {
log.Error().Err(err).Msg("run referenced testcase step failed")
log.Info().Str("step", s.step.Name).Bool("success", false).Msg("run step end")
stepResult.Attachment = err.Error()
r.summary.Success = false
return stepResult, err
}
summary := sessionRunner.GetSummary()
stepResult.Data = summary
stepResult.Data = summary.Records
// export testcase export variables
stepResult.ExportVars = sessionRunner.summary.InOut.ExportVars
stepResult.ExportVars = summary.InOut.ExportVars
stepResult.Success = true
// update extracted variables
@@ -96,12 +95,6 @@ func (s *StepTestCaseWithOptionalArgs) Run(r *SessionRunner) (*StepResult, error
r.summary.Stat.Successes += summary.Stat.Successes
r.summary.Stat.Failures += summary.Stat.Failures
log.Info().
Str("step", s.step.Name).
Bool("success", true).
Interface("exportVars", stepResult.ExportVars).
Msg("run step end")
return stepResult, nil
}

3
hrp/step_thinktime.go
View File

@@ -30,8 +30,7 @@ func (s *StepThinkTime) Struct() *TStep {
func (s *StepThinkTime) Run(r *SessionRunner) (*StepResult, error) {
thinkTime := s.step.ThinkTime
log.Info().Str("name", s.step.Name).
Float64("time", thinkTime.Time).Msg("think time")
log.Info().Float64("time", thinkTime.Time).Msg("think time")
stepResult := &StepResult{
Name: s.step.Name,

115
httprunner/README.md Normal file
View File

@@ -0,0 +1,115 @@
# 代码阅读指南（python 部分）
## 核心数据结构
HttpRunner 以 `TestCase` 为核心，将任意测试场景抽象为有序步骤的集合。
```py
class TestCase(BaseModel):
config: TConfig
teststeps: List[TStep]
```
针对每种测试步骤，统一继承自 `IStep`，并要求必须至少实现如下 4 个方法；步骤内容统一在 `run` 方法中进行实现。
```py
class IStep(object):
def name(self) -> str:
raise NotImplementedError
def type(self) -> str:
raise NotImplementedError
def struct(self) -> TStep:
raise NotImplementedError
def run(self, runner) -> StepData:
# runner: HttpRunner
raise NotImplementedError
```
我们只需遵循 `IStep` 的接口定义，即可实现各种类型的测试步骤类型。当前 python 版本已支持的步骤类型包括：
- [request](step_request.py)：发起单次 HTTP 请求
- [testcase](step_testcase.py)：引用执行其它测试用例文件
基于该机制，我们可以扩展支持新的协议类型，例如 HTTP2/WebSocket/RPC 等；同时也可以支持新的测试类型，例如 UI 自动化。甚至我们还可以在一个测试用例中混合调用多种不同的 Step 类型，例如实现 HTTP/RPC/UI 混合场景。
## 用例编写
## 运行主流程
### 整体控制器 pytest
不同于 golang 版本，python 版本的控制逻辑都基于 `pytest` 的用例发现和执行机制。
- 如果是运行 JSON/YAML 格式的用例，hrp 会将用例转换为 pytest 支持的用例格式
- 如果是要自行编写 pytest 测试用例，需要遵循 HttpRunner 的格式要求
### pytest 用例格式要求
所有测试用例要求都继承自 `HttpRunner`，然后
结构如下所示：
```py
class TestCaseRequestWithFunctions(HttpRunner):
config = (
Config("request methods testcase with functions")
)
teststeps = [
Step(
RunRequest("get with params")...
),
Step(
RunRequest("post raw text")...
),
Step(
RunRequest("post form data")...
),
]
```
完整案例可参考：
- [request_with_functions_test.py](../examples/postman_echo/request_methods/request_with_functions_test.py)：用例中包含了 requests 的情况
- [request_with_testcase_reference_test.py](../examples/postman_echo/request_methods/request_with_testcase_reference_test.py)：用例中包含了引用其它测试用例的情况
### 用例执行器 SessionRunner
测试用例的具体执行都由 `SessionRunner` 完成，每个 TestCase 对应一个实例，在该实例中除了包含测试用例自身内容外，还会包含测试过程的 session 数据和最终测试结果 summary。
```py
class SessionRunner(object):
config: Config
teststeps: List[object] # list of Step
...
```
重点关注一个方法：
- test_start：该方法将被 pytest 发现，作为启动执行入口，依次执行所有测试步骤
```go
def test_start(self, param: Dict = None) -> "SessionRunner":
"""main entrance, discovered by pytest"""
self.__start_at = time.time()
try:
# run step in sequential order
for step in self.teststeps:
self.__run_step(step)
finally:
logger.info(f"generate testcase log: {self.__log_path}")
self.__duration = time.time() - self.__start_at
```
在主流程中，SessionRunner 并不需要关注 step 的具体类型，统一都是调用 `step.run(self)`，具体实现逻辑都在对应 step 的 `run` 方法中。
```py
def run(self, runner: HttpRunner) -> StepData:
return self.__step.run(runner)
```

6
httprunner/__init__.py
View File

@@ -1,10 +1,12 @@
__version__ = "4.0.0-alpha"
__description__ = "One-stop solution for HTTP(S) testing."
# import firstly for monkey patch if needed
from httprunner.config import Config
from httprunner.parser import parse_parameters as Parameters
from httprunner.runner import HttpRunner
from httprunner.testcase import Config, Step, RunRequest, RunTestCase
from httprunner.step import Step
from httprunner.step_request import RunRequest
from httprunner.step_testcase import RunTestCase
__all__ = [
"__version__",

4
httprunner/cli.py
View File

@@ -63,7 +63,7 @@ def main():
)
subparsers = parser.add_subparsers(help="sub-command help")
sub_parser_run = init_parser_run(subparsers)
init_parser_run(subparsers)
sub_parser_make = init_make_parser(subparsers)
if len(sys.argv) == 1:
@@ -93,7 +93,7 @@ def main():
sys.exit(0)
extra_args = []
if len(sys.argv) >= 2 and sys.argv[1] in ["run", "locusts"]:
if len(sys.argv) >= 2 and sys.argv[1] in ["run"]:
args, extra_args = parser.parse_known_args()
else:
args = parser.parse_args()

14
httprunner/cli_test.py
View File

@@ -5,7 +5,8 @@ import unittest
import pytest
from httprunner.cli import main
from httprunner import loader
from httprunner.cli import main, main_run
class TestCli(unittest.TestCase):
@@ -45,8 +46,17 @@ class TestCli(unittest.TestCase):
try:
os.chdir(os.path.join(cwd, "examples", "postman_echo"))
exit_code = pytest.main(
["-s", "request_methods/request_with_testcase_reference_test.py",]
["-s", "request_methods/request_with_testcase_reference_test.py"]
)
self.assertEqual(exit_code, 0)
finally:
os.chdir(cwd)
def test_run_testcase_with_abnormal_path(self):
loader.project_meta = None
exit_code = main_run(["examples/data/a-b.c/2 3.yml"])
self.assertEqual(exit_code, 0)
self.assertTrue(os.path.exists("examples/data/a_b_c/__init__.py"))
self.assertTrue(os.path.exists("examples/data/debugtalk.py"))
self.assertTrue(os.path.exists("examples/data/a_b_c/T1_test.py"))
self.assertTrue(os.path.exists("examples/data/a_b_c/T2_3_test.py"))

14
httprunner/compat.py
View File

@@ -257,12 +257,12 @@ def ensure_cli_args(args: List) -> List:
"""
# remove deprecated --failfast
if "--failfast" in args:
logger.warning(f"remove deprecated argument: --failfast")
logger.warning("remove deprecated argument: --failfast")
args.pop(args.index("--failfast"))
# convert --report-file to --html
if "--report-file" in args:
logger.warning(f"replace deprecated argument --report-file with --html")
logger.warning("replace deprecated argument --report-file with --html")
index = args.index("--report-file")
args[index] = "--html"
args.append("--self-contained-html")
@@ -270,7 +270,7 @@ def ensure_cli_args(args: List) -> List:
# keep compatibility with --save-tests in v2
if "--save-tests" in args:
logger.warning(
f"generate conftest.py keep compatibility with --save-tests in v2"
"generate conftest.py keep compatibility with --save-tests in v2"
)
args.pop(args.index("--save-tests"))
_generate_conftest_for_summary(args)
@@ -327,21 +327,21 @@ def session_fixture(request):
summary["success"] &= testcase_summary.success
summary["stat"]["testcases"]["total"] += 1
summary["stat"]["teststeps"]["total"] += len(testcase_summary.step_datas)
summary["stat"]["teststeps"]["total"] += len(testcase_summary.step_results)
if testcase_summary.success:
summary["stat"]["testcases"]["success"] += 1
summary["stat"]["teststeps"]["successes"] += len(
testcase_summary.step_datas
testcase_summary.step_results
)
else:
summary["stat"]["testcases"]["fail"] += 1
summary["stat"]["teststeps"]["successes"] += (
len(testcase_summary.step_datas) - 1
len(testcase_summary.step_results) - 1
)
summary["stat"]["teststeps"]["failures"] += 1
testcase_summary_json = testcase_summary.dict()
testcase_summary_json["records"] = testcase_summary_json.pop("step_datas")
testcase_summary_json["records"] = testcase_summary_json.pop("step_results")
summary["details"].append(testcase_summary_json)
summary_path = r"{{SUMMARY_PATH_PLACEHOLDER}}"

71
httprunner/config.py Normal file
View File

@@ -0,0 +1,71 @@
import inspect
from typing import Text
from httprunner.models import TConfig, TConfigThrift
class ConfigThrift(object):
def __init__(self, config: TConfig) -> None:
self.__config = config
self.__config.thrift = TConfigThrift()
def psm(self, psm: Text) -> "ConfigThrift":
self.__config.thrift.psm = psm
return self
def env(self, env: Text) -> "ConfigThrift":
self.__config.thrift.env = env
return self
def cluster(self, cluster: Text) -> "ConfigThrift":
self.__config.thrift.cluster = cluster
return self
def target(self, target: Text) -> "ConfigThrift":
self.__config.thrift.target = target
return self
def struct(self) -> TConfig:
return self.__config
class Config(object):
def __init__(self, name: Text) -> None:
caller_frame = inspect.stack()[1]
self.__config = TConfig(
name=name,
path=caller_frame.filename
)
@property
def name(self) -> Text:
return self.__config.name
@property
def path(self) -> Text:
return self.__config.path
def variables(self, **variables) -> "Config":
self.__config.variables.update(variables)
return self
def base_url(self, base_url: Text) -> "Config":
self.__config.base_url = base_url
return self
def verify(self, verify: bool) -> "Config":
self.__config.verify = verify
return self
def export(self, *export_var_name: Text) -> "Config":
self.__config.export.extend(export_var_name)
self.__config.export = list(set(self.__config.export))
return self
def struct(self) -> TConfig:
return self.__config
def thrift(self) -> ConfigThrift:
return ConfigThrift(self.__config)

7
httprunner/make.py
View File

@@ -198,9 +198,6 @@ def make_config_chain_style(config: Dict) -> Text:
if "export" in config:
config_chain_style += f'.export(*{config["export"]})'
if "weight" in config:
config_chain_style += f'.locust_weight({config["weight"]})'
return config_chain_style
@@ -484,10 +481,6 @@ def make_testsuite(testsuite: Dict):
)
testcase_dict["config"]["variables"].update(testcase_variables)
# override weight
if "weight" in testcase:
testcase_dict["config"]["weight"] = testcase["weight"]
# make testcase
testcase_pytest_path = make_testcase(testcase_dict, testsuite_dir)
pytest_files_run_set.add(testcase_pytest_path)

52
httprunner/models.py
View File

@@ -1,11 +1,8 @@
import os
from enum import Enum
from typing import Any
from typing import Dict, Text, Union, Callable
from typing import List
from typing import Any, Callable, Dict, List, Text, Union
from pydantic import BaseModel, Field
from pydantic import HttpUrl
from pydantic import BaseModel, Field, HttpUrl
Name = Text
Url = Text
@@ -31,6 +28,16 @@ class MethodEnum(Text, Enum):
PATCH = "PATCH"
# configs for thrift rpc
class TConfigThrift(BaseModel):
psm: Text = None
env: Text = None
cluster: Text = None
target: Text = None
include_dirs: List[Text] = None
thrift_client: Any = None
class TConfig(BaseModel):
name: Name
verify: Verify = False
@@ -42,7 +49,8 @@ class TConfig(BaseModel):
# teardown_hooks: Hooks = []
export: Export = []
path: Text = None
weight: int = 1
# configs for other protocols
thrift: TConfigThrift = None
class TRequest(BaseModel):
@@ -153,16 +161,36 @@ class SessionData(BaseModel):
validators: Dict = {}
class StepData(BaseModel):
class StepResult(BaseModel):
"""teststep data, each step maybe corresponding to one request or one testcase"""
name: Text = "" # teststep name
step_type: Text = "" # teststep type, request or testcase
success: bool = False
name: Text = "" # teststep name
data: Union[SessionData, List['StepData']] = None
data: Union[SessionData, List['StepResult']] = None
elapsed: float = 0.0 # teststep elapsed time
content_size: float = 0 # response content size
export_vars: VariablesMapping = {}
attachment: Text = "" # teststep attachment
StepData.update_forward_refs()
StepResult.update_forward_refs()
class IStep(object):
def name(self) -> str:
raise NotImplementedError
def type(self) -> str:
raise NotImplementedError
def struct(self) -> TStep:
raise NotImplementedError
def run(self, runner) -> StepResult:
# runner: HttpRunner
raise NotImplementedError
class TestCaseSummary(BaseModel):
@@ -172,7 +200,7 @@ class TestCaseSummary(BaseModel):
time: TestCaseTime
in_out: TestCaseInOut = {}
log: Text = ""
step_datas: List[StepData] = []
step_results: List[StepResult] = []
class PlatformInfo(BaseModel):

51
httprunner/parser.py
View File

@@ -1,16 +1,15 @@
import ast
import builtins
import re
import os
from typing import Any, Set, Text, Callable, List, Dict, Union
import re
from typing import Any, Callable, Dict, List, Set, Text
from urllib.parse import urljoin, urlparse
from loguru import logger
from sentry_sdk import capture_exception
from httprunner import loader, utils, exceptions
from httprunner.models import VariablesMapping, FunctionsMapping
absolute_http_url_regexp = re.compile(r"^https?://", re.I)
from httprunner import exceptions, loader, utils
from httprunner.models import FunctionsMapping, VariablesMapping
# use $$ to escape $ notation
dolloar_regex_compile = re.compile(r"\$\$")
@@ -37,15 +36,25 @@ def parse_string_value(str_value: Text) -> Any:
return str_value
def build_url(base_url, path):
def build_url(base_url, step_url):
""" prepend url with base_url unless it's already an absolute URL """
if absolute_http_url_regexp.match(path):
return path
elif base_url:
return "{}/{}".format(base_url.rstrip("/"), path.lstrip("/"))
else:
o_step_url = urlparse(step_url)
if o_step_url.netloc != "":
# step url is absolute url
return step_url
# step url is relative, based on base url
o_base_url = urlparse(base_url)
if o_base_url.netloc == "":
# missed base url
raise exceptions.ParamsError("base url missed!")
path = o_base_url.path.rstrip("/") + "/" + o_step_url.path.lstrip("/")
o_step_url = o_step_url._replace(scheme=o_base_url.scheme) \
._replace(netloc=o_base_url.netloc) \
._replace(path=path)
return o_step_url.geturl()
def regex_findall_variables(raw_string: Text) -> List[Text]:
""" extract all variable names from content, which is in format $variable
@@ -572,3 +581,21 @@ def parse_parameters(parameters: Dict,) -> List[Dict]:
parsed_parameters_list.append(parameter_content_list)
return utils.gen_cartesian_product(*parsed_parameters_list)
class Parser(object):
def __init__(self, functions_mapping: FunctionsMapping = None) -> None:
self.functions_mapping = functions_mapping
def parse_string(self, raw_string: Text, variables_mapping: VariablesMapping) -> Any:
return parse_string(raw_string, variables_mapping, self.functions_mapping)
def parse_variables(self, variables_mapping: VariablesMapping) -> VariablesMapping:
return parse_variables_mapping(variables_mapping, self.functions_mapping)
def parse_data(self, raw_data: Any, variables_mapping: VariablesMapping = None) -> Any:
return parse_data(raw_data, variables_mapping, self.functions_mapping)
def get_mapping_function(self, func_name: Text) -> Callable:
return get_mapping_function(func_name, self.functions_mapping)

27
httprunner/parser_test.py
View File

@@ -3,11 +3,36 @@ import time
import unittest
from httprunner import parser
from httprunner.exceptions import VariableNotFound, FunctionNotFound
from httprunner.exceptions import FunctionNotFound, VariableNotFound
from httprunner.loader import load_project_meta
class TestParserBasic(unittest.TestCase):
def test_build_url(self):
url = parser.build_url("https://postman-echo.com", "/get")
self.assertEqual(url, "https://postman-echo.com/get")
url = parser.build_url("https://postman-echo.com", "get")
self.assertEqual(url, "https://postman-echo.com/get")
url = parser.build_url("https://postman-echo.com/", "/get")
self.assertEqual(url, "https://postman-echo.com/get")
url = parser.build_url("https://postman-echo.com/abc/", "/get?a=1&b=2")
self.assertEqual(url, "https://postman-echo.com/abc/get?a=1&b=2")
url = parser.build_url("https://postman-echo.com/abc/", "get?a=1&b=2")
self.assertEqual(url, "https://postman-echo.com/abc/get?a=1&b=2")
# omit query string in base url
url = parser.build_url("https://postman-echo.com/abc?x=6&y=9", "/get?a=1&b=2")
self.assertEqual(url, "https://postman-echo.com/abc/get?a=1&b=2")
url = parser.build_url("", "https://postman-echo.com/get")
self.assertEqual(url, "https://postman-echo.com/get")
# notice: step request url > config base url
url = parser.build_url("https://postman-echo.com", "https://httpbin.org/get")
self.assertEqual(url, "https://httpbin.org/get")
def test_parse_variables_mapping(self):
variables = {"varA": "$varB", "varB": "$varC", "varC": "123", "a": 1, "b": 2}
parsed_variables = parser.parse_variables_mapping(variables)

24
httprunner/response.py
View File

@@ -7,8 +7,8 @@ from loguru import logger
from httprunner import exceptions
from httprunner.exceptions import ValidationFailure, ParamsError
from httprunner.models import VariablesMapping, Validators, FunctionsMapping
from httprunner.parser import parse_data, parse_string_value, get_mapping_function
from httprunner.models import VariablesMapping, Validators
from httprunner.parser import parse_string_value, Parser
def get_uniform_comparator(comparator: Text):
@@ -115,7 +115,7 @@ def uniform_validator(validator):
class ResponseObject(object):
def __init__(self, resp_obj: requests.Response):
def __init__(self, resp_obj: requests.Response, parser: Parser):
""" initialize with a requests.Response object
Args:
@@ -123,6 +123,7 @@ class ResponseObject(object):
"""
self.resp_obj = resp_obj
self.parser = parser
self.validation_results: Dict = {}
def __getattr__(self, key):
@@ -170,7 +171,6 @@ class ResponseObject(object):
def extract(self,
extractors: Dict[Text, Text],
variables_mapping: VariablesMapping = None,
functions_mapping: FunctionsMapping = None,
) -> Dict[Text, Any]:
if not extractors:
return {}
@@ -179,8 +179,8 @@ class ResponseObject(object):
for key, field in extractors.items():
if '$' in field:
# field contains variable or function
field = parse_data(
field, variables_mapping, functions_mapping
field = self.parser.parse_data(
field, variables_mapping
)
field_value = self._search_jmespath(field)
extract_mapping[key] = field_value
@@ -192,11 +192,9 @@ class ResponseObject(object):
self,
validators: Validators,
variables_mapping: VariablesMapping = None,
functions_mapping: FunctionsMapping = None,
):
variables_mapping = variables_mapping or {}
functions_mapping = functions_mapping or {}
self.validation_results = {}
if not validators:
@@ -216,8 +214,8 @@ class ResponseObject(object):
check_item = u_validator["check"]
if "$" in check_item:
# check_item is variable or function
check_item = parse_data(
check_item, variables_mapping, functions_mapping
check_item = self.parser.parse_data(
check_item, variables_mapping
)
check_item = parse_string_value(check_item)
@@ -229,17 +227,17 @@ class ResponseObject(object):
# comparator
assert_method = u_validator["assert"]
assert_func = get_mapping_function(assert_method, functions_mapping)
assert_func = self.parser.get_mapping_function(assert_method)
# expect item
expect_item = u_validator["expect"]
# parse expected value with config/teststep/extracted variables
expect_value = parse_data(expect_item, variables_mapping, functions_mapping)
expect_value = self.parser.parse_data(expect_item, variables_mapping)
# message
message = u_validator["message"]
# parse message with config/teststep/extracted variables
message = parse_data(message, variables_mapping, functions_mapping)
message = self.parser.parse_data(message, variables_mapping)
validate_msg = f"assert {check_item} {assert_method} {expect_value}({type(expect_value).__name__})"

13
httprunner/response_test.py
View File

@@ -2,6 +2,7 @@ import unittest
import requests
from httprunner.parser import Parser
from httprunner.response import ResponseObject
@@ -18,15 +19,16 @@ class TestResponse(unittest.TestCase):
]
},
)
self.resp_obj = ResponseObject(resp)
parser = Parser(functions_mapping={
'get_name': lambda: 'name',
"get_num": lambda x: x
})
self.resp_obj = ResponseObject(resp, parser)
def test_extract(self):
variables_mapping = {
'body': 'body'
}
functions_mapping = {
'get_name': lambda: 'name',
}
extract_mapping = self.resp_obj.extract(
{
"var_1": "body.json.locations[0]",
@@ -35,7 +37,6 @@ class TestResponse(unittest.TestCase):
"var_4": "$body.json.locations[3].${get_name()}",
},
variables_mapping=variables_mapping,
functions_mapping=functions_mapping,
)
self.assertEqual(extract_mapping["var_1"], {"name": "Seattle", "state": "WA"})
self.assertEqual(extract_mapping["var_2"], "Olympia")
@@ -62,9 +63,7 @@ class TestResponse(unittest.TestCase):
def test_validate_functions(self):
variables_mapping = {"index": 1}
functions_mapping = {"get_num": lambda x: x}
self.resp_obj.validate(
[{"eq": ["${get_num(0)}", 0]}, {"eq": ["${get_num($index)}", 1]},],
variables_mapping=variables_mapping,
functions_mapping=functions_mapping,
)

489
httprunner/runner.py
View File

@@ -2,7 +2,7 @@ import os
import time
import uuid
from datetime import datetime
from typing import List, Dict, Text
from typing import Dict, List, Text
try:
import allure
@@ -13,41 +13,29 @@ except ModuleNotFoundError:
from loguru import logger
from httprunner import utils, exceptions
from httprunner.client import HttpSession
from httprunner.exceptions import ValidationFailure, ParamsError
from httprunner.ext.uploader import prepare_upload_step
from httprunner.loader import load_project_meta, load_testcase_file
from httprunner.parser import build_url, parse_data, parse_variables_mapping
from httprunner.response import ResponseObject
from httprunner.testcase import Config, Step
from httprunner.config import Config
from httprunner.exceptions import ParamsError
from httprunner.loader import load_project_meta
from httprunner.models import (ProjectMeta, StepResult, TConfig, TestCaseInOut,
TestCaseSummary, TestCaseTime, VariablesMapping)
from httprunner.parser import Parser
from httprunner.utils import merge_variables
from httprunner.models import (
TConfig,
TStep,
VariablesMapping,
StepData,
TestCaseSummary,
TestCaseTime,
TestCaseInOut,
ProjectMeta,
TestCase,
Hooks,
)
class HttpRunner(object):
class SessionRunner(object):
config: Config
teststeps: List[Step]
teststeps: List[object] # list of Step
parser: Parser = None
session: HttpSession = None
case_id: Text = ""
root_dir: Text = ""
success: bool = False # indicate testcase execution result
__config: TConfig
__teststeps: List[TStep]
__project_meta: ProjectMeta = None
__case_id: Text = ""
__export: List[Text] = []
__step_datas: List[StepData] = []
__session: HttpSession = None
__step_results: List[StepResult] = []
__session_variables: VariablesMapping = {}
# time
__start_at: float = 0
@@ -55,335 +43,62 @@ class HttpRunner(object):
# log
__log_path: Text = ""
def __init_tests__(self):
self.__config = self.config.perform()
self.__teststeps = []
for step in self.teststeps:
self.__teststeps.append(step.perform())
def __init(self):
self.__config = self.config.struct()
self.__session_variables = {}
self.__start_at = 0
self.__duration = 0
@property
def raw_testcase(self) -> TestCase:
if not hasattr(self, "__config"):
self.__init_tests__()
return TestCase(config=self.__config, teststeps=self.__teststeps)
def with_project_meta(self, project_meta: ProjectMeta) -> "HttpRunner":
self.__project_meta = project_meta
return self
def with_session(self, session: HttpSession) -> "HttpRunner":
self.__session = session
return self
def with_case_id(self, case_id: Text) -> "HttpRunner":
self.__case_id = case_id
return self
def with_variables(self, variables: VariablesMapping) -> "HttpRunner":
self.__session_variables = variables
return self
def with_export(self, export: List[Text]) -> "HttpRunner":
self.__export = export
return self
def __call_hooks(self, hooks: Hooks, step_variables: VariablesMapping, hook_msg: Text):
""" call hook actions.
Args:
hooks (list): each hook in hooks list maybe in two format.
format1 (str): only call hook functions.
${func()}
format2 (dict): assignment, the value returned by hook function will be assigned to variable.
{"var": "${func()}"}
step_variables: current step variables to call hook, include two special variables
request: parsed request dict
response: ResponseObject for current response
hook_msg: setup/teardown request/testcase
"""
logger.info(f"call hook actions: {hook_msg}")
if not isinstance(hooks, List):
logger.error(f"Invalid hooks format: {hooks}")
return
for hook in hooks:
if isinstance(hook, Text):
# format 1: ["${func()}"]
logger.debug(f"call hook function: {hook}")
parse_data(hook, step_variables, self.__project_meta.functions)
elif isinstance(hook, Dict) and len(hook) == 1:
# format 2: {"var": "${func()}"}
var_name, hook_content = list(hook.items())[0]
hook_content_eval = parse_data(
hook_content, step_variables, self.__project_meta.functions
)
logger.debug(
f"call hook function: {hook_content}, got value: {hook_content_eval}"
)
logger.debug(f"assign variable: {var_name} = {hook_content_eval}")
step_variables[var_name] = hook_content_eval
else:
logger.error(f"Invalid hook format: {hook}")
def __run_step_request(self, step: TStep) -> StepData:
"""run teststep: request"""
step_data = StepData(name=step.name)
# parse
functions = self.__project_meta.functions
prepare_upload_step(step, functions)
request_dict = step.request.dict()
request_dict.pop("upload", None)
parsed_request_dict = parse_data(
request_dict, step.variables, functions
)
parsed_request_dict["headers"].setdefault(
"HRUN-Request-ID",
f"HRUN-{self.__case_id}-{str(int(time.time() * 1000))[-6:]}",
)
step.variables["request"] = parsed_request_dict
# setup hooks
if step.setup_hooks:
self.__call_hooks(step.setup_hooks, step.variables, "setup request")
# prepare arguments
method = parsed_request_dict.pop("method")
url_path = parsed_request_dict.pop("url")
url = build_url(self.__config.base_url, url_path)
parsed_request_dict["verify"] = self.__config.verify
parsed_request_dict["json"] = parsed_request_dict.pop("req_json", {})
# request
resp = self.__session.request(method, url, **parsed_request_dict)
resp_obj = ResponseObject(resp)
step.variables["response"] = resp_obj
# teardown hooks
if step.teardown_hooks:
self.__call_hooks(step.teardown_hooks, step.variables, "teardown request")
def log_req_resp_details():
err_msg = "\n{} DETAILED REQUEST & RESPONSE {}\n".format("*" * 32, "*" * 32)
# log request
err_msg += "====== request details ======\n"
err_msg += f"url: {url}\n"
err_msg += f"method: {method}\n"
headers = parsed_request_dict.pop("headers", {})
err_msg += f"headers: {headers}\n"
for k, v in parsed_request_dict.items():
v = utils.omit_long_data(v)
err_msg += f"{k}: {repr(v)}\n"
err_msg += "\n"
# log response
err_msg += "====== response details ======\n"
err_msg += f"status_code: {resp.status_code}\n"
err_msg += f"headers: {resp.headers}\n"
err_msg += f"body: {repr(resp.text)}\n"
logger.error(err_msg)
# extract
extractors = step.extract
extract_mapping = resp_obj.extract(extractors, step.variables, functions)
step_data.export_vars = extract_mapping
variables_mapping = step.variables
variables_mapping.update(extract_mapping)
# validate
validators = step.validators
session_success = False
try:
resp_obj.validate(
validators, variables_mapping, functions
)
session_success = True
except ValidationFailure:
session_success = False
log_req_resp_details()
# log testcase duration before raise ValidationFailure
self.__duration = time.time() - self.__start_at
raise
finally:
self.success = session_success
step_data.success = session_success
if hasattr(self.__session, "data"):
# httprunner.client.HttpSession, not locust.clients.HttpSession
# save request & response meta data
self.__session.data.success = session_success
self.__session.data.validators = resp_obj.validation_results
# save step data
step_data.data = self.__session.data
return step_data
def __run_step_testcase(self, step: TStep) -> StepData:
"""run teststep: referenced testcase"""
step_data = StepData(name=step.name)
step_variables = step.variables
step_export = step.export
# setup hooks
if step.setup_hooks:
self.__call_hooks(step.setup_hooks, step_variables, "setup testcase")
if hasattr(step.testcase, "config") and hasattr(step.testcase, "teststeps"):
testcase_cls = step.testcase
case_result = (
testcase_cls()
.with_session(self.__session)
.with_case_id(self.__case_id)
.with_variables(step_variables)
.with_export(step_export)
.run()
)
elif isinstance(step.testcase, Text):
if os.path.isabs(step.testcase):
ref_testcase_path = step.testcase
else:
ref_testcase_path = os.path.join(
self.__project_meta.RootDir, step.testcase
)
case_result = (
HttpRunner()
.with_session(self.__session)
.with_case_id(self.__case_id)
.with_variables(step_variables)
.with_export(step_export)
.run_path(ref_testcase_path)
)
else:
raise exceptions.ParamsError(
f"Invalid teststep referenced testcase: {step.dict()}"
)
# teardown hooks
if step.teardown_hooks:
self.__call_hooks(step.teardown_hooks, step.variables, "teardown testcase")
step_data.data = case_result.get_step_datas() # list of step data
step_data.export_vars = case_result.get_export_variables()
step_data.success = case_result.success
self.success = case_result.success
if step_data.export_vars:
logger.info(f"export variables: {step_data.export_vars}")
return step_data
def __run_step(self, step: TStep) -> Dict:
"""run teststep, teststep maybe a request or referenced testcase"""
logger.info(f"run step begin: {step.name} >>>>>>")
if step.request:
step_data = self.__run_step_request(step)
elif step.testcase:
step_data = self.__run_step_testcase(step)
else:
raise ParamsError(
f"teststep is neither a request nor a referenced testcase: {step.dict()}"
)
self.__step_datas.append(step_data)
logger.info(f"run step end: {step.name} <<<<<<\n")
return step_data.export_vars
def __parse_config(self, config: TConfig):
config.variables.update(self.__session_variables)
config.variables = parse_variables_mapping(
config.variables, self.__project_meta.functions
)
config.name = parse_data(
config.name, config.variables, self.__project_meta.functions
)
config.base_url = parse_data(
config.base_url, config.variables, self.__project_meta.functions
)
def run_testcase(self, testcase: TestCase) -> "HttpRunner":
"""run specified testcase
Examples:
>>> testcase_obj = TestCase(config=TConfig(...), teststeps=[TStep(...)])
>>> HttpRunner().with_project_meta(project_meta).run_testcase(testcase_obj)
"""
self.__config = testcase.config
self.__teststeps = testcase.teststeps
# prepare
self.__project_meta = self.__project_meta or load_project_meta(
self.__config.path
)
self.__parse_config(self.__config)
self.__start_at = time.time()
self.__step_datas: List[StepData] = []
self.__session = self.__session or HttpSession()
# save extracted variables of teststeps
extracted_variables: VariablesMapping = {}
self.case_id = self.case_id or str(uuid.uuid4())
self.root_dir = self.root_dir or self.__project_meta.RootDir
self.__log_path = os.path.join(
self.root_dir, "logs", f"{self.case_id}.run.log"
)
# run teststeps
for step in self.__teststeps:
# override variables
# step variables > extracted variables from previous steps
step.variables = merge_variables(step.variables, extracted_variables)
# step variables > testcase config variables
step.variables = merge_variables(step.variables, self.__config.variables)
self.__step_results.clear()
self.session = self.session or HttpSession()
self.parser = self.parser or Parser(self.__project_meta.functions)
# parse variables
step.variables = parse_variables_mapping(
step.variables, self.__project_meta.functions
)
# run step
if USE_ALLURE:
with allure.step(f"step: {step.name}"):
extract_mapping = self.__run_step(step)
else:
extract_mapping = self.__run_step(step)
# save extracted variables to session variables
extracted_variables.update(extract_mapping)
self.__session_variables.update(extracted_variables)
self.__duration = time.time() - self.__start_at
def with_session(self, session: HttpSession) -> "SessionRunner":
self.session = session
return self
def run_path(self, path: Text) -> "HttpRunner":
if not os.path.isfile(path):
raise exceptions.ParamsError(f"Invalid testcase path: {path}")
def get_config(self) -> TConfig:
return self.__config
testcase_obj = load_testcase_file(path)
return self.run_testcase(testcase_obj)
def with_case_id(self, case_id: Text) -> "SessionRunner":
self.case_id = case_id
return self
def run(self) -> "HttpRunner":
""" run current testcase
def with_variables(self, variables: VariablesMapping) -> "SessionRunner":
self.__session_variables = variables
return self
Examples:
>>> TestCaseRequestWithFunctions().run()
def with_export(self, export: List[Text]) -> "SessionRunner":
self.__export = export
return self
"""
self.__init_tests__()
testcase_obj = TestCase(config=self.__config, teststeps=self.__teststeps)
return self.run_testcase(testcase_obj)
def __parse_config(self, param: Dict = None) -> None:
# parse config variables
self.__config.variables.update(self.__session_variables)
if param:
self.__config.variables.update(param)
self.__config.variables = self.parser.parse_variables(
self.__config.variables
)
def get_step_datas(self) -> List[StepData]:
return self.__step_datas
# parse config name
self.__config.name = self.parser.parse_data(
self.__config.name, self.__config.variables
)
# parse config base url
self.__config.base_url = self.parser.parse_data(
self.__config.base_url, self.__config.variables
)
def get_export_variables(self) -> Dict:
# override testcase export vars with step export
@@ -403,10 +118,17 @@ class HttpRunner(object):
"""get testcase result summary"""
start_at_timestamp = self.__start_at
start_at_iso_format = datetime.utcfromtimestamp(start_at_timestamp).isoformat()
summary_success = True
for step_result in self.__step_results:
if not step_result.success:
summary_success = False
break
return TestCaseSummary(
name=self.__config.name,
success=self.success,
case_id=self.__case_id,
success=summary_success,
case_id=self.case_id,
time=TestCaseTime(
start_at=self.__start_at,
start_at_iso_format=start_at_iso_format,
@@ -417,43 +139,70 @@ class HttpRunner(object):
export_vars=self.get_export_variables(),
),
log=self.__log_path,
step_datas=self.__step_datas,
step_results=self.__step_results,
)
def test_start(self, param: Dict = None) -> "HttpRunner":
def merge_step_variables(self, variables: VariablesMapping) -> VariablesMapping:
# override variables
# step variables > extracted variables from previous steps
variables = merge_variables(variables, self.__session_variables)
# step variables > testcase config variables
variables = merge_variables(variables, self.__config.variables)
# parse variables
return self.parser.parse_variables(variables)
def __run_step(self, step):
"""run teststep, step maybe any kind that implements IStep interface
Args:
step (Step): teststep
"""
logger.info(f"run step begin: {step.name()} >>>>>>")
# run step
if USE_ALLURE:
with allure.step(f"step: {step.name()}"):
step_result: StepResult = step.run(self)
else:
step_result: StepResult = step.run(self)
# save extracted variables to session variables
self.__session_variables.update(step_result.export_vars)
# update testcase summary
self.__step_results.append(step_result)
logger.info(f"run step end: {step.name()} <<<<<<\n")
def test_start(self, param: Dict = None) -> "SessionRunner":
"""main entrance, discovered by pytest"""
self.__init_tests__()
self.__project_meta = self.__project_meta or load_project_meta(
self.__config.path
)
self.__case_id = self.__case_id or str(uuid.uuid4())
self.__log_path = self.__log_path or os.path.join(
self.__project_meta.RootDir, "logs", f"{self.__case_id}.run.log"
)
log_handler = logger.add(self.__log_path, level="DEBUG")
# parse config name
config_variables = self.__config.variables
if param:
config_variables.update(param)
config_variables.update(self.__session_variables)
self.__config.name = parse_data(
self.__config.name, config_variables, self.__project_meta.functions
)
self.__init()
self.__parse_config(param)
if USE_ALLURE:
# update allure report meta
allure.dynamic.title(self.__config.name)
allure.dynamic.description(f"TestCase ID: {self.__case_id}")
allure.dynamic.description(f"TestCase ID: {self.case_id}")
logger.info(
f"Start to run testcase: {self.__config.name}, TestCase ID: {self.__case_id}"
f"Start to run testcase: {self.__config.name}, TestCase ID: {self.case_id}"
)
log_handler = logger.add(self.__log_path, level="DEBUG")
self.__start_at = time.time()
try:
return self.run_testcase(
TestCase(config=self.__config, teststeps=self.__teststeps)
)
# run step in sequential order
for step in self.teststeps:
self.__run_step(step)
finally:
logger.remove(log_handler)
logger.info(f"generate testcase log: {self.__log_path}")
self.__duration = time.time() - self.__start_at
return self
class HttpRunner(SessionRunner):
# split SessionRunner to keep consistant with golang version
pass

40
httprunner/runner_test.py
View File

@@ -1,40 +0,0 @@
import os
import unittest
from httprunner import loader
from httprunner.cli import main_run
from httprunner.runner import HttpRunner
class TestHttpRunner(unittest.TestCase):
def setUp(self):
loader.project_meta = None
self.runner = HttpRunner()
def test_run_testcase_by_path_request_only(self):
self.runner.run_path(
"examples/postman_echo/request_methods/request_with_functions.yml"
)
result = self.runner.get_summary()
self.assertTrue(result.success)
self.assertEqual(result.name, "request methods testcase with functions")
self.assertEqual(result.step_datas[0].name, "get with params")
self.assertEqual(len(result.step_datas), 3)
def test_run_testcase_by_path_ref_testcase(self):
self.runner.run_path(
"examples/postman_echo/request_methods/request_with_testcase_reference.yml"
)
result = self.runner.get_summary()
self.assertTrue(result.success)
self.assertEqual(result.name, "request methods testcase: reference testcase")
self.assertEqual(result.step_datas[0].name, "request with functions")
self.assertEqual(len(result.step_datas), 2)
def test_run_testcase_with_abnormal_path(self):
exit_code = main_run(["examples/data/a-b.c/2 3.yml"])
self.assertEqual(exit_code, 0)
self.assertTrue(os.path.exists("examples/data/a_b_c/__init__.py"))
self.assertTrue(os.path.exists("examples/data/debugtalk.py"))
self.assertTrue(os.path.exists("examples/data/a_b_c/T1_test.py"))
self.assertTrue(os.path.exists("examples/data/a_b_c/T2_3_test.py"))

40
httprunner/step.py Normal file
View File

@@ -0,0 +1,40 @@
from typing import Union
from httprunner.models import StepResult, TRequest, TStep, TestCase
from httprunner.runner import HttpRunner
from httprunner.step_request import RequestWithOptionalArgs, StepRequestExtraction, StepRequestValidation
from httprunner.step_testcase import StepRefCase
class Step(object):
def __init__(
self,
step: Union[
StepRequestValidation,
StepRequestExtraction,
RequestWithOptionalArgs,
StepRefCase,
],
):
self.__step = step
@property
def request(self) -> TRequest:
return self.__step.struct().request
@property
def testcase(self) -> TestCase:
return self.__step.struct().testcase
def struct(self) -> TStep:
return self.__step.struct()
def name(self) -> str:
return self.__step.name()
def type(self) -> str:
return self.__step.type()
def run(self, runner: HttpRunner) -> StepResult:
return self.__step.run(runner)

463
httprunner/step_request.py Normal file
View File

@@ -0,0 +1,463 @@
import time
from typing import Any, Dict, List, Text, Union
from loguru import logger
from httprunner import utils
from httprunner.exceptions import ValidationFailure
from httprunner.ext.uploader import prepare_upload_step
from httprunner.models import (Hooks, IStep, MethodEnum, StepResult, TRequest,
TStep, VariablesMapping)
from httprunner.parser import build_url
from httprunner.response import ResponseObject
from httprunner.runner import HttpRunner
def call_hooks(runner: HttpRunner, hooks: Hooks, step_variables: VariablesMapping, hook_msg: Text):
""" call hook actions.
Args:
hooks (list): each hook in hooks list maybe in two format.
format1 (str): only call hook functions.
${func()}
format2 (dict): assignment, the value returned by hook function will be assigned to variable.
{"var": "${func()}"}
step_variables: current step variables to call hook, include two special variables
request: parsed request dict
response: ResponseObject for current response
hook_msg: setup/teardown request/testcase
"""
logger.info(f"call hook actions: {hook_msg}")
if not isinstance(hooks, List):
logger.error(f"Invalid hooks format: {hooks}")
return
for hook in hooks:
if isinstance(hook, Text):
# format 1: ["${func()}"]
logger.debug(f"call hook function: {hook}")
runner.parser.parse_data(hook, step_variables)
elif isinstance(hook, Dict) and len(hook) == 1:
# format 2: {"var": "${func()}"}
var_name, hook_content = list(hook.items())[0]
hook_content_eval = runner.parser.parse_data(
hook_content, step_variables
)
logger.debug(
f"call hook function: {hook_content}, got value: {hook_content_eval}"
)
logger.debug(f"assign variable: {var_name} = {hook_content_eval}")
step_variables[var_name] = hook_content_eval
else:
logger.error(f"Invalid hook format: {hook}")
def run_step_request(runner: HttpRunner, step: TStep) -> StepResult:
"""run teststep: request"""
step_result = StepResult(
name=step.name,
success=False,
)
start_time = time.time()
step.variables = runner.merge_step_variables(step.variables)
# parse
functions = runner.parser.functions_mapping
prepare_upload_step(step, functions)
request_dict = step.request.dict()
request_dict.pop("upload", None)
parsed_request_dict = runner.parser.parse_data(
request_dict, step.variables
)
parsed_request_dict["headers"].setdefault(
"HRUN-Request-ID",
f"HRUN-{runner.case_id}-{str(int(time.time() * 1000))[-6:]}",
)
step.variables["request"] = parsed_request_dict
# setup hooks
if step.setup_hooks:
call_hooks(runner, step.setup_hooks, step.variables, "setup request")
# prepare arguments
config = runner.get_config()
method = parsed_request_dict.pop("method")
url_path = parsed_request_dict.pop("url")
url = build_url(config.base_url, url_path)
parsed_request_dict["verify"] = config.verify
parsed_request_dict["json"] = parsed_request_dict.pop("req_json", {})
# request
resp = runner.session.request(method, url, **parsed_request_dict)
resp_obj = ResponseObject(resp, runner.parser)
step.variables["response"] = resp_obj
# teardown hooks
if step.teardown_hooks:
call_hooks(runner, step.teardown_hooks, step.variables, "teardown request")
def log_req_resp_details():
err_msg = "\n{} DETAILED REQUEST & RESPONSE {}\n".format("*" * 32, "*" * 32)
# log request
err_msg += "====== request details ======\n"
err_msg += f"url: {url}\n"
err_msg += f"method: {method}\n"
headers = parsed_request_dict.pop("headers", {})
err_msg += f"headers: {headers}\n"
for k, v in parsed_request_dict.items():
v = utils.omit_long_data(v)
err_msg += f"{k}: {repr(v)}\n"
err_msg += "\n"
# log response
err_msg += "====== response details ======\n"
err_msg += f"status_code: {resp.status_code}\n"
err_msg += f"headers: {resp.headers}\n"
err_msg += f"body: {repr(resp.text)}\n"
logger.error(err_msg)
# extract
extractors = step.extract
extract_mapping = resp_obj.extract(extractors, step.variables)
step_result.export_vars = extract_mapping
variables_mapping = step.variables
variables_mapping.update(extract_mapping)
# validate
validators = step.validators
try:
resp_obj.validate(
validators, variables_mapping
)
step_result.success = True
except ValidationFailure:
log_req_resp_details()
raise
finally:
session_data = runner.session.data
session_data.success = step_result.success
session_data.validators = resp_obj.validation_results
# save step data
step_result.data = session_data
step_result.elapsed = time.time() - start_time
return step_result
class StepRequestValidation(IStep):
def __init__(self, step: TStep):
self.__step = step
def assert_equal(
self, jmes_path: Text, expected_value: Any, message: Text = ""
) -> "StepRequestValidation":
self.__step.validators.append(
{"equal": [jmes_path, expected_value, message]}
)
return self
def assert_not_equal(
self, jmes_path: Text, expected_value: Any, message: Text = ""
) -> "StepRequestValidation":
self.__step.validators.append(
{"not_equal": [jmes_path, expected_value, message]}
)
return self
def assert_greater_than(
self, jmes_path: Text, expected_value: Union[int, float], message: Text = ""
) -> "StepRequestValidation":
self.__step.validators.append(
{"greater_than": [jmes_path, expected_value, message]}
)
return self
def assert_less_than(
self, jmes_path: Text, expected_value: Union[int, float], message: Text = ""
) -> "StepRequestValidation":
self.__step.validators.append(
{"less_than": [jmes_path, expected_value, message]}
)
return self
def assert_greater_or_equals(
self, jmes_path: Text, expected_value: Union[int, float], message: Text = ""
) -> "StepRequestValidation":
self.__step.validators.append(
{"greater_or_equals": [jmes_path, expected_value, message]}
)
return self
def assert_less_or_equals(
self, jmes_path: Text, expected_value: Union[int, float], message: Text = ""
) -> "StepRequestValidation":
self.__step.validators.append(
{"less_or_equals": [jmes_path, expected_value, message]}
)
return self
def assert_length_equal(
self, jmes_path: Text, expected_value: int, message: Text = ""
) -> "StepRequestValidation":
self.__step.validators.append(
{"length_equal": [jmes_path, expected_value, message]}
)
return self
def assert_length_greater_than(
self, jmes_path: Text, expected_value: int, message: Text = ""
) -> "StepRequestValidation":
self.__step.validators.append(
{"length_greater_than": [jmes_path, expected_value, message]}
)
return self
def assert_length_less_than(
self, jmes_path: Text, expected_value: int, message: Text = ""
) -> "StepRequestValidation":
self.__step.validators.append(
{"length_less_than": [jmes_path, expected_value, message]}
)
return self
def assert_length_greater_or_equals(
self, jmes_path: Text, expected_value: int, message: Text = ""
) -> "StepRequestValidation":
self.__step.validators.append(
{"length_greater_or_equals": [jmes_path, expected_value, message]}
)
return self
def assert_length_less_or_equals(
self, jmes_path: Text, expected_value: int, message: Text = ""
) -> "StepRequestValidation":
self.__step.validators.append(
{"length_less_or_equals": [jmes_path, expected_value, message]}
)
return self
def assert_string_equals(
self, jmes_path: Text, expected_value: Any, message: Text = ""
) -> "StepRequestValidation":
self.__step.validators.append(
{"string_equals": [jmes_path, expected_value, message]}
)
return self
def assert_startswith(
self, jmes_path: Text, expected_value: Text, message: Text = ""
) -> "StepRequestValidation":
self.__step.validators.append(
{"startswith": [jmes_path, expected_value, message]}
)
return self
def assert_endswith(
self, jmes_path: Text, expected_value: Text, message: Text = ""
) -> "StepRequestValidation":
self.__step.validators.append(
{"endswith": [jmes_path, expected_value, message]}
)
return self
def assert_regex_match(
self, jmes_path: Text, expected_value: Text, message: Text = ""
) -> "StepRequestValidation":
self.__step.validators.append(
{"regex_match": [jmes_path, expected_value, message]}
)
return self
def assert_contains(
self, jmes_path: Text, expected_value: Any, message: Text = ""
) -> "StepRequestValidation":
self.__step.validators.append(
{"contains": [jmes_path, expected_value, message]}
)
return self
def assert_contained_by(
self, jmes_path: Text, expected_value: Any, message: Text = ""
) -> "StepRequestValidation":
self.__step.validators.append(
{"contained_by": [jmes_path, expected_value, message]}
)
return self
def assert_type_match(
self, jmes_path: Text, expected_value: Any, message: Text = ""
) -> "StepRequestValidation":
self.__step.validators.append(
{"type_match": [jmes_path, expected_value, message]}
)
return self
def struct(self) -> TStep:
return self.__step
def name(self) -> Text:
return self.__step.name
def type(self) -> Text:
return f"request-{self.__step.request.method}"
def run(self, runner: HttpRunner):
return run_step_request(runner, self.__step)
class StepRequestExtraction(IStep):
def __init__(self, step: TStep):
self.__step = step
def with_jmespath(self, jmes_path: Text, var_name: Text) -> "StepRequestExtraction":
self.__step.extract[var_name] = jmes_path
return self
# def with_regex(self):
# # TODO: extract response html with regex
# pass
#
# def with_jsonpath(self):
# # TODO: extract response json with jsonpath
# pass
def validate(self) -> StepRequestValidation:
return StepRequestValidation(self.__step)
def struct(self) -> TStep:
return self.__step
def name(self) -> Text:
return self.__step.name
def type(self) -> Text:
return f"request-{self.__step.request.method}"
def run(self, runner: HttpRunner):
return run_step_request(runner, self.__step)
class RequestWithOptionalArgs(IStep):
def __init__(self, step: TStep):
self.__step = step
def with_params(self, **params) -> "RequestWithOptionalArgs":
self.__step.request.params.update(params)
return self
def with_headers(self, **headers) -> "RequestWithOptionalArgs":
self.__step.request.headers.update(headers)
return self
def with_cookies(self, **cookies) -> "RequestWithOptionalArgs":
self.__step.request.cookies.update(cookies)
return self
def with_data(self, data) -> "RequestWithOptionalArgs":
self.__step.request.data = data
return self
def with_json(self, req_json) -> "RequestWithOptionalArgs":
self.__step.request.req_json = req_json
return self
def set_timeout(self, timeout: float) -> "RequestWithOptionalArgs":
self.__step.request.timeout = timeout
return self
def set_verify(self, verify: bool) -> "RequestWithOptionalArgs":
self.__step.request.verify = verify
return self
def set_allow_redirects(self, allow_redirects: bool) -> "RequestWithOptionalArgs":
self.__step.request.allow_redirects = allow_redirects
return self
def upload(self, **file_info) -> "RequestWithOptionalArgs":
self.__step.request.upload.update(file_info)
return self
def teardown_hook(
self, hook: Text, assign_var_name: Text = None
) -> "RequestWithOptionalArgs":
if assign_var_name:
self.__step.teardown_hooks.append({assign_var_name: hook})
else:
self.__step.teardown_hooks.append(hook)
return self
def extract(self) -> StepRequestExtraction:
return StepRequestExtraction(self.__step)
def validate(self) -> StepRequestValidation:
return StepRequestValidation(self.__step)
def struct(self) -> TStep:
return self.__step
def name(self) -> Text:
return self.__step.name
def type(self) -> Text:
return f"request-{self.__step.request.method}"
def run(self, runner: HttpRunner):
return run_step_request(runner, self.__step)
class RunRequest(object):
def __init__(self, name: Text):
self.__step = TStep(name=name)
def with_variables(self, **variables) -> "RunRequest":
self.__step.variables.update(variables)
return self
def setup_hook(self, hook: Text, assign_var_name: Text = None) -> "RunRequest":
if assign_var_name:
self.__step.setup_hooks.append({assign_var_name: hook})
else:
self.__step.setup_hooks.append(hook)
return self
def get(self, url: Text) -> RequestWithOptionalArgs:
self.__step.request = TRequest(method=MethodEnum.GET, url=url)
return RequestWithOptionalArgs(self.__step)
def post(self, url: Text) -> RequestWithOptionalArgs:
self.__step.request = TRequest(method=MethodEnum.POST, url=url)
return RequestWithOptionalArgs(self.__step)
def put(self, url: Text) -> RequestWithOptionalArgs:
self.__step.request = TRequest(method=MethodEnum.PUT, url=url)
return RequestWithOptionalArgs(self.__step)
def head(self, url: Text) -> RequestWithOptionalArgs:
self.__step.request = TRequest(method=MethodEnum.HEAD, url=url)
return RequestWithOptionalArgs(self.__step)
def delete(self, url: Text) -> RequestWithOptionalArgs:
self.__step.request = TRequest(method=MethodEnum.DELETE, url=url)
return RequestWithOptionalArgs(self.__step)
def options(self, url: Text) -> RequestWithOptionalArgs:
self.__step.request = TRequest(method=MethodEnum.OPTIONS, url=url)
return RequestWithOptionalArgs(self.__step)
def patch(self, url: Text) -> RequestWithOptionalArgs:
self.__step.request = TRequest(method=MethodEnum.PATCH, url=url)
return RequestWithOptionalArgs(self.__step)

16
httprunner/step_request_test.py Normal file
View File

@@ -0,0 +1,16 @@
import unittest
from examples.postman_echo.request_methods.request_with_functions_test import TestCaseRequestWithFunctions
class TestRunRequest(unittest.TestCase):
def test_run_request(self):
runner = TestCaseRequestWithFunctions().test_start()
summary = runner.get_summary()
self.assertTrue(summary.success)
self.assertEqual(summary.name, "request methods testcase with functions")
self.assertEqual(len(summary.step_results), 3)
self.assertEqual(summary.step_results[0].name, "get with params")
self.assertEqual(summary.step_results[1].name, "post raw text")
self.assertEqual(summary.step_results[2].name, "post form data")

100
httprunner/step_testcase.py Normal file
View File

@@ -0,0 +1,100 @@
from typing import Callable, Text
from loguru import logger
from httprunner import exceptions
from httprunner.models import IStep, StepResult, TStep, TestCaseSummary
from httprunner.runner import HttpRunner
from httprunner.step_request import call_hooks
def run_step_testcase(runner: HttpRunner, step: TStep) -> StepResult:
"""run teststep: referenced testcase"""
step_result = StepResult(name=step.name)
step_variables = step.variables
step_export = step.export
# setup hooks
if step.setup_hooks:
call_hooks(runner, step.setup_hooks, step_variables, "setup testcase")
# TODO: override testcase with current step name/variables/export
# step.testcase is a referenced testcase, e.g. RequestWithFunctions
ref_case_runner = step.testcase()
ref_case_runner.with_session(runner.session) \
.with_case_id(runner.case_id) \
.with_variables(step_variables) \
.with_export(step_export) \
.test_start()
# teardown hooks
if step.teardown_hooks:
call_hooks(runner, step.teardown_hooks, step.variables, "teardown testcase")
summary: TestCaseSummary = ref_case_runner.get_summary()
step_result.data = summary.step_results # list of step data
step_result.export_vars = summary.in_out.export_vars
step_result.success = summary.success
if step_result.export_vars:
logger.info(f"export variables: {step_result.export_vars}")
return step_result
class StepRefCase(IStep):
def __init__(self, step: TStep):
self.__step = step
def teardown_hook(self, hook: Text, assign_var_name: Text = None) -> "StepRefCase":
if assign_var_name:
self.__step.teardown_hooks.append({assign_var_name: hook})
else:
self.__step.teardown_hooks.append(hook)
return self
def export(self, *var_name: Text) -> "StepRefCase":
self.__step.export.extend(var_name)
return self
def struct(self) -> TStep:
return self.__step
def name(self) -> Text:
return self.__step.name
def type(self) -> Text:
return f"request-{self.__step.request.method}"
def run(self, runner: HttpRunner):
return run_step_testcase(runner, self.__step)
class RunTestCase(object):
def __init__(self, name: Text):
self.__step = TStep(name=name)
def with_variables(self, **variables) -> "RunTestCase":
self.__step.variables.update(variables)
return self
def setup_hook(self, hook: Text, assign_var_name: Text = None) -> "RunTestCase":
if assign_var_name:
self.__step.setup_hooks.append({assign_var_name: hook})
else:
self.__step.setup_hooks.append(hook)
return self
def call(self, testcase: Callable) -> StepRefCase:
if issubclass(testcase, HttpRunner):
# referenced testcase object
self.__step.testcase = testcase
else:
raise exceptions.ParamsError(
f"Invalid teststep referenced testcase: {testcase}"
)
return StepRefCase(self.__step)

23
httprunner/step_testcase_test.py Normal file
View File

@@ -0,0 +1,23 @@
import unittest
from httprunner.runner import HttpRunner
from httprunner.step_testcase import RunTestCase
from examples.postman_echo.request_methods.request_with_functions_test import TestCaseRequestWithFunctions
class TestRunTestCase(unittest.TestCase):
def setUp(self):
self.runner = HttpRunner()
def test_run_testcase_by_path(self):
step_result = RunTestCase("run referenced testcase").call(
TestCaseRequestWithFunctions
).run(self.runner)
self.assertTrue(step_result.success)
self.assertEqual(step_result.name, "run referenced testcase")
self.assertEqual(len(step_result.data), 3)
self.assertEqual(step_result.data[0].name, "get with params")
self.assertEqual(step_result.data[1].name, "post raw text")
self.assertEqual(step_result.data[2].name, "post form data")

415
httprunner/testcase.py
View File

@@ -1,415 +0,0 @@
import inspect
from typing import Text, Any, Union, Callable
from httprunner.models import (
TConfig,
TStep,
TRequest,
MethodEnum,
TestCase,
)
class Config(object):
def __init__(self, name: Text):
self.__name = name
self.__variables = {}
self.__base_url = ""
self.__verify = False
self.__export = []
self.__weight = 1
caller_frame = inspect.stack()[1]
self.__path = caller_frame.filename
@property
def name(self) -> Text:
return self.__name
@property
def path(self) -> Text:
return self.__path
@property
def weight(self) -> int:
return self.__weight
def variables(self, **variables) -> "Config":
self.__variables.update(variables)
return self
def base_url(self, base_url: Text) -> "Config":
self.__base_url = base_url
return self
def verify(self, verify: bool) -> "Config":
self.__verify = verify
return self
def export(self, *export_var_name: Text) -> "Config":
self.__export.extend(export_var_name)
return self
def locust_weight(self, weight: int) -> "Config":
self.__weight = weight
return self
def perform(self) -> TConfig:
return TConfig(
name=self.__name,
base_url=self.__base_url,
verify=self.__verify,
variables=self.__variables,
export=list(set(self.__export)),
path=self.__path,
weight=self.__weight,
)
class StepRequestValidation(object):
def __init__(self, step_context: TStep):
self.__step_context = step_context
def assert_equal(
self, jmes_path: Text, expected_value: Any, message: Text = ""
) -> "StepRequestValidation":
self.__step_context.validators.append(
{"equal": [jmes_path, expected_value, message]}
)
return self
def assert_not_equal(
self, jmes_path: Text, expected_value: Any, message: Text = ""
) -> "StepRequestValidation":
self.__step_context.validators.append(
{"not_equal": [jmes_path, expected_value, message]}
)
return self
def assert_greater_than(
self, jmes_path: Text, expected_value: Union[int, float], message: Text = ""
) -> "StepRequestValidation":
self.__step_context.validators.append(
{"greater_than": [jmes_path, expected_value, message]}
)
return self
def assert_less_than(
self, jmes_path: Text, expected_value: Union[int, float], message: Text = ""
) -> "StepRequestValidation":
self.__step_context.validators.append(
{"less_than": [jmes_path, expected_value, message]}
)
return self
def assert_greater_or_equals(
self, jmes_path: Text, expected_value: Union[int, float], message: Text = ""
) -> "StepRequestValidation":
self.__step_context.validators.append(
{"greater_or_equals": [jmes_path, expected_value, message]}
)
return self
def assert_less_or_equals(
self, jmes_path: Text, expected_value: Union[int, float], message: Text = ""
) -> "StepRequestValidation":
self.__step_context.validators.append(
{"less_or_equals": [jmes_path, expected_value, message]}
)
return self
def assert_length_equal(
self, jmes_path: Text, expected_value: int, message: Text = ""
) -> "StepRequestValidation":
self.__step_context.validators.append(
{"length_equal": [jmes_path, expected_value, message]}
)
return self
def assert_length_greater_than(
self, jmes_path: Text, expected_value: int, message: Text = ""
) -> "StepRequestValidation":
self.__step_context.validators.append(
{"length_greater_than": [jmes_path, expected_value, message]}
)
return self
def assert_length_less_than(
self, jmes_path: Text, expected_value: int, message: Text = ""
) -> "StepRequestValidation":
self.__step_context.validators.append(
{"length_less_than": [jmes_path, expected_value, message]}
)
return self
def assert_length_greater_or_equals(
self, jmes_path: Text, expected_value: int, message: Text = ""
) -> "StepRequestValidation":
self.__step_context.validators.append(
{"length_greater_or_equals": [jmes_path, expected_value, message]}
)
return self
def assert_length_less_or_equals(
self, jmes_path: Text, expected_value: int, message: Text = ""
) -> "StepRequestValidation":
self.__step_context.validators.append(
{"length_less_or_equals": [jmes_path, expected_value, message]}
)
return self
def assert_string_equals(
self, jmes_path: Text, expected_value: Any, message: Text = ""
) -> "StepRequestValidation":
self.__step_context.validators.append(
{"string_equals": [jmes_path, expected_value, message]}
)
return self
def assert_startswith(
self, jmes_path: Text, expected_value: Text, message: Text = ""
) -> "StepRequestValidation":
self.__step_context.validators.append(
{"startswith": [jmes_path, expected_value, message]}
)
return self
def assert_endswith(
self, jmes_path: Text, expected_value: Text, message: Text = ""
) -> "StepRequestValidation":
self.__step_context.validators.append(
{"endswith": [jmes_path, expected_value, message]}
)
return self
def assert_regex_match(
self, jmes_path: Text, expected_value: Text, message: Text = ""
) -> "StepRequestValidation":
self.__step_context.validators.append(
{"regex_match": [jmes_path, expected_value, message]}
)
return self
def assert_contains(
self, jmes_path: Text, expected_value: Any, message: Text = ""
) -> "StepRequestValidation":
self.__step_context.validators.append(
{"contains": [jmes_path, expected_value, message]}
)
return self
def assert_contained_by(
self, jmes_path: Text, expected_value: Any, message: Text = ""
) -> "StepRequestValidation":
self.__step_context.validators.append(
{"contained_by": [jmes_path, expected_value, message]}
)
return self
def assert_type_match(
self, jmes_path: Text, expected_value: Any, message: Text = ""
) -> "StepRequestValidation":
self.__step_context.validators.append(
{"type_match": [jmes_path, expected_value, message]}
)
return self
def perform(self) -> TStep:
return self.__step_context
class StepRequestExtraction(object):
def __init__(self, step_context: TStep):
self.__step_context = step_context
def with_jmespath(self, jmes_path: Text, var_name: Text) -> "StepRequestExtraction":
self.__step_context.extract[var_name] = jmes_path
return self
# def with_regex(self):
# # TODO: extract response html with regex
# pass
#
# def with_jsonpath(self):
# # TODO: extract response json with jsonpath
# pass
def validate(self) -> StepRequestValidation:
return StepRequestValidation(self.__step_context)
def perform(self) -> TStep:
return self.__step_context
class RequestWithOptionalArgs(object):
def __init__(self, step_context: TStep):
self.__step_context = step_context
def with_params(self, **params) -> "RequestWithOptionalArgs":
self.__step_context.request.params.update(params)
return self
def with_headers(self, **headers) -> "RequestWithOptionalArgs":
self.__step_context.request.headers.update(headers)
return self
def with_cookies(self, **cookies) -> "RequestWithOptionalArgs":
self.__step_context.request.cookies.update(cookies)
return self
def with_data(self, data) -> "RequestWithOptionalArgs":
self.__step_context.request.data = data
return self
def with_json(self, req_json) -> "RequestWithOptionalArgs":
self.__step_context.request.req_json = req_json
return self
def set_timeout(self, timeout: float) -> "RequestWithOptionalArgs":
self.__step_context.request.timeout = timeout
return self
def set_verify(self, verify: bool) -> "RequestWithOptionalArgs":
self.__step_context.request.verify = verify
return self
def set_allow_redirects(self, allow_redirects: bool) -> "RequestWithOptionalArgs":
self.__step_context.request.allow_redirects = allow_redirects
return self
def upload(self, **file_info) -> "RequestWithOptionalArgs":
self.__step_context.request.upload.update(file_info)
return self
def teardown_hook(
self, hook: Text, assign_var_name: Text = None
) -> "RequestWithOptionalArgs":
if assign_var_name:
self.__step_context.teardown_hooks.append({assign_var_name: hook})
else:
self.__step_context.teardown_hooks.append(hook)
return self
def extract(self) -> StepRequestExtraction:
return StepRequestExtraction(self.__step_context)
def validate(self) -> StepRequestValidation:
return StepRequestValidation(self.__step_context)
def perform(self) -> TStep:
return self.__step_context
class RunRequest(object):
def __init__(self, name: Text):
self.__step_context = TStep(name=name)
def with_variables(self, **variables) -> "RunRequest":
self.__step_context.variables.update(variables)
return self
def setup_hook(self, hook: Text, assign_var_name: Text = None) -> "RunRequest":
if assign_var_name:
self.__step_context.setup_hooks.append({assign_var_name: hook})
else:
self.__step_context.setup_hooks.append(hook)
return self
def get(self, url: Text) -> RequestWithOptionalArgs:
self.__step_context.request = TRequest(method=MethodEnum.GET, url=url)
return RequestWithOptionalArgs(self.__step_context)
def post(self, url: Text) -> RequestWithOptionalArgs:
self.__step_context.request = TRequest(method=MethodEnum.POST, url=url)
return RequestWithOptionalArgs(self.__step_context)
def put(self, url: Text) -> RequestWithOptionalArgs:
self.__step_context.request = TRequest(method=MethodEnum.PUT, url=url)
return RequestWithOptionalArgs(self.__step_context)
def head(self, url: Text) -> RequestWithOptionalArgs:
self.__step_context.request = TRequest(method=MethodEnum.HEAD, url=url)
return RequestWithOptionalArgs(self.__step_context)
def delete(self, url: Text) -> RequestWithOptionalArgs:
self.__step_context.request = TRequest(method=MethodEnum.DELETE, url=url)
return RequestWithOptionalArgs(self.__step_context)
def options(self, url: Text) -> RequestWithOptionalArgs:
self.__step_context.request = TRequest(method=MethodEnum.OPTIONS, url=url)
return RequestWithOptionalArgs(self.__step_context)
def patch(self, url: Text) -> RequestWithOptionalArgs:
self.__step_context.request = TRequest(method=MethodEnum.PATCH, url=url)
return RequestWithOptionalArgs(self.__step_context)
class StepRefCase(object):
def __init__(self, step_context: TStep):
self.__step_context = step_context
def teardown_hook(self, hook: Text, assign_var_name: Text = None) -> "StepRefCase":
if assign_var_name:
self.__step_context.teardown_hooks.append({assign_var_name: hook})
else:
self.__step_context.teardown_hooks.append(hook)
return self
def export(self, *var_name: Text) -> "StepRefCase":
self.__step_context.export.extend(var_name)
return self
def perform(self) -> TStep:
return self.__step_context
class RunTestCase(object):
def __init__(self, name: Text):
self.__step_context = TStep(name=name)
def with_variables(self, **variables) -> "RunTestCase":
self.__step_context.variables.update(variables)
return self
def setup_hook(self, hook: Text, assign_var_name: Text = None) -> "RunTestCase":
if assign_var_name:
self.__step_context.setup_hooks.append({assign_var_name: hook})
else:
self.__step_context.setup_hooks.append(hook)
return self
def call(self, testcase: Callable) -> StepRefCase:
self.__step_context.testcase = testcase
return StepRefCase(self.__step_context)
def perform(self) -> TStep:
return self.__step_context
class Step(object):
def __init__(
self,
step_context: Union[
StepRequestValidation,
StepRequestExtraction,
RequestWithOptionalArgs,
RunTestCase,
StepRefCase,
],
):
self.__step_context = step_context.perform()
@property
def request(self) -> TRequest:
return self.__step_context.request
@property
def testcase(self) -> TestCase:
return self.__step_context.testcase
def perform(self) -> TStep:
return self.__step_context

13
httprunner/utils.py
View File

@@ -2,6 +2,7 @@ import collections
import copy
import itertools
import json
import os
import os.path
import platform
import uuid
@@ -40,8 +41,13 @@ class GAClient(object):
'cid': uuid.getnode(), # Anonymous Client ID
'ua': f'HttpRunner/{__version__}',
}
# do not send GA events in CI environment
self.__is_ci = os.getenv("DISABLE_GA") == "true"
def track_event(self, category: Text, action: Text, value: int = 0):
if self.__is_ci:
return
data = {
't': 'event', # Event hit type = event
'ec': category, # Required. Event Category.
@@ -51,11 +57,14 @@ class GAClient(object):
}
data.update(self.common_params)
try:
self.http_client.post(self.report_url, data=data)
self.http_client.post(self.report_url, data=data, timeout=5)
except Exception: # ProxyError, SSLError, ConnectionError
pass
def track_user_timing(self, category: Text, variable: Text, duration: int):
if self.__is_ci:
return
data = {
't': 'timing', # Event hit type = timing
'utc': category, # Required. user timing category. e.g. jsonLoader
@@ -65,7 +74,7 @@ class GAClient(object):
}
data.update(self.common_params)
try:
self.http_client.post(self.report_url, data=data)
self.http_client.post(self.report_url, data=data, timeout=5)
except Exception: # ProxyError, SSLError, ConnectionError
pass