Git Product home page Git Product logo

spintest's Introduction

Spintest

Functional scenario interpreter.

Spintest is a library that facilitates the integration and functional test of APIs. It takes as parameters a list of URLs and a list of tasks (also called scenarios) that will be executed against the specified URLs.

Each task represents an API call and provides some options in order to validate or react to the response. By default the task is a success if the HTTP code returned is 2XX (200 to 299 included), but it is possible to specify the error code or the body that are expected. It is also possible to provide a list of rollback tasks (or task references) that are executed should the task fail.
The response of the API calls can be stored in order to be used in a future task. The task scenarios can also be run concurrently in each URL.

Installation

The package can be installed using PIP.

$ pip install spintest

URLs and tasks definition

The url list is a list of endpoints. A route added here will not be evaluated because the route definition is set on the task.

[
    "https://foo.com",
    "https://bar.com"
]

The task definition is a little bit more detailed.
A scenario is a list of tasks possibly dependent to each other.

A single task follows the following schema :

{
    "method": str,
    Optional("route", default="/"): str,
    Optional("name"): str,
    Optional("body"): dict,
    Optional(
        "headers",
        default={"Accept": "application/json", "Content-Type": "application/json"},
    ): dict,
    Optional("output"): str,
    Optional("expected"): {
        Optional("code"): int,
        Optional("body"): Or(dict, str),
        Optional("expected_match", default="strict"): Or("partial", "strict"),
    },
    Optional("fail_on"): [{
        Optional("code"): int,
        Optional("body"): Or(dict, str),
        Optional("expected_match", default="strict"): Or("partial", "strict"),
    }],
    Optional("retry", default=0): int,
    Optional("delay", default=1): int,
    Optional("ignore", default=False): bool,
    Optional("rollback"): [Or(str, dict)],
}
  • method is the HTTP method of the request (GET, POST, DELETE, ...). Only a valid HTTP method is accepted.
  • route (optional) is the route to test on the endpoint. It will be appended of the current URL (default is "/")
  • name (optional) is the name of the task. Mandatory if you want to use that task in a rollback.
  • body (optional) is a request body.
  • header (optional) is a dictionary of headers. Default is JSON application headers. For Oauth endpoint it is not necessary to add the appropriate header with the token (if the token is specified).
  • output (optional) Variable definition where Spintest puts the result of the call. This result can be used later in another task using Jinja syntax.
  • expected (optional) is an expected HTTP response code or response body.
    • code (optional) is the expected HTTP code.
    • body (optional) is an expected response body. You can put a value to null if you don't want to check the value of a key but you will have to set all keys. It also checks nested list and dictionary unless you put "null" instead.
    • expected_match is an option to check partially the keys present on your response body. By default it is set to strict.
  • fail_on (optional) is a list of error HTTP response code or response body. Once one of these error occurs, the test fails without retries.
    • code (optional) is the expected HTTP code.
    • body (optional) is an expected response body. You can put a value to null if you don't want to check the value of a key but you will have to set all keys. It also checks nested list and dictionary unless you put "null" instead.
    • expected_match is an option to check partially the keys present on your response body. By default it is set to strict.
  • retry (optional) is the number of retries if it fails (default is 0).
  • delay (optional) is the time in second to wait between retries (default is 1).
  • ignore (optional) is to allow to continue the scenario in case of error of the task.
  • rollback (optional) is a list of task names or tasks that are triggered should the task fail.

Usage

A first example with a single route.

from spintest import spintest

urls = ["https://test.com"]
tasks = [
    {
        "method": "GET",
        "route": "test",
    }
]

result = spintest(urls, tasks)
assert True is result

This test will perform a GET call into https://test.com/test and expect a return code between 200 and 299 included.

Here is another example with an interaction between two routes :

from spintest import spintest

urls = ["https://test.com"]
tasks = [
    {
        "method": "POST",
        "route": "test",
        "output": "test_output",
        "body": {"name": "Disk1", "size": 20},
    },
    {
        "method": "DELETE",
        "route": "volumes/{{ test_output['id'] }}",
        "expected": {"code": 204},
    }
]

result = spintest(urls, tasks)
assert True is result

As seen here, the first task has a key output. This way it is possible to store the output of this first task into a test_output variables and be able to use it in following tasks in Jinja templating language. Moreover, the second task has a key expected. The specific return code 204 is expected.

Finally here is a last example that shows how to run tasks in parallel.

from spintest import spintest

urls = ["https://foo.com", "https://bar.com"]
tasks = [
    {
        "method": "GET",
        "route": "test",
        "expected": {
            "body": {"result": None},
            "expected_match": "partial",
        }
    }
]

result = spintest(urls, tasks, parallel=True)
assert True is result

Here two URLS are provided and the option parallel wad added in the spintest function.
Without this option, the scenario will be executed iteratively on every URLS.

But with this option, the each task of the scenario will be executed concurrently for every URLS.

One last word on the expected option. Here we want to validate that a certain key (result) is present from the output. We don't mind about the value of this key so we just set it to None. The option expected_match set to partial indicates that we don't want to a task failure if there is more key in the API response than expected.

Token management

Oauth token can be automatically included into the task headers.

  • Tokens can be directly hard coded
urls = ["http://test.com"]
tasks = []

spintest(urls, tasks, token= 'ABC')
  • A method that generates a token can be given instead of a token
urls = ["http://test.com"]
tasks = []

spintest(urls, tasks, token=create_token)

Rollback actions

A list of rollback tasks that are executed in case of a task failure can be specified.

from spintest import spintest

urls = ["https://test.com"]
tasks = [
    {
        "method": "POST",
        "route": "test",
        "rollback": [
            {
                "method": "DELETE",
                "route": "test,
            }
        ]
    }
]

spintest(urls, tasks)

The name of a task can be specified in order to prevent rewriting them

from spintest import spintest

urls = ["https://test.com"]
tasks = [
    {
        "method": "POST",
        "route": "test",
        "rollback": [
            "test_delete"
        ]
    },
    {
        "name": "test_delete",
        "method": "DELETE",
        "route": "test",
    }
]

spintest(urls, tasks)

Run the tasks one by one

It is also possible to further control the flow of the task execution to perform additional actions between tasks ( clean up / additional settings / ... )

import asyncio

from spintest import TaskManager


urls = ["http://test.com"]
tasks = [{"method": "GET", "route": "/test"}]
token = "90b7aa25-870a-4dda-a1fc-b57cf0fbf278"

loop = asyncio.get_event_loop()

manager = TaskManager(urls, tasks, token=token)
result = loop.run_until_complete(manager.next())

assert "SUCCESS" == result["status"]

The next() method throws a StopAsyncIteration if there are no tasks left to execute.

Note: The method next() can be used in parallel mode. In this case the method returns a list with the result of the task against each URLs.

Type convertion

Task template evaluation always returns a string, but sometimes the target API expects a non-string value.
It is possible to convert it a the corresponding type if needed

Spintest provides a set of json value converters that provide such functionality.

  • Int -> Converts value to a int
  • List -> Converts value to a list
  • Float -> Converts value to a float
  • Bool -> Converts value to a bool
from spintest import spintest

urls = ["http://test.com"]
tasks = [
    {
        "method": "GET",
        "route": "persons",
        "output": "value",
        # Returns
        # {
        #     "age": 20,
        #     "height": 1.85,
        #     "alive": True,
        # }
    },
    {
        "method": "POST",
        "route": "persons",
        "body": {
            # int convertion
            "age_str": "{{ value.person['age'] }}", # {"age_str": "20"},
            "age": Int("{{ value.person['age'] }}"), # {"age": 20},

            # float convertion
            "height_str": "{{ value.person['height'] }}", # {"height_str": "1.85"},
            "height": Float("{{ value.person['height'] }}"), # {"height": 1.85},

            # bool convertion
            "alive_str": "{{ value.person['alive'] }}", # {"alive_str": "True"},
            "alive": Bool("{{ value.person['alive'] }}"), # {"alive": true},
        }
    }
]

spintest(urls, tasks, token=token_write)

Generate report

Since the version 0.3.0 of spintest, generating reports of test execution is possible.

The report will contain all information that were written here, and on each tasks the return payload and the execution time are attached.
At the end of the report the total execution time of the tests is indicated.

To use this functionality use this piece of code :

from spintest import spintest

urls = ["https://test.com"]
tasks = [
    {
        "method": "GET",
        "route": "test",
    }
]

result = spintest(urls, tasks, generate_report="report_name")
assert True is result

A report with the name "report_name" will be create.
To avoid to creating multiple "report_name", this report will be overwrote on each test execution.

Raise to avoid long test execution

The test no longer retries and fails immediately once one of the "fail_on" definition is met.

from spintest import spintest
urls = ["https://test.com"]

tasks = [
    {
        "method": "GET",
        "route": "test",
        "expected": {
            "body": {"result": "Success"},
            "expected_match": "partial",
        },
        "fail_on": [
            {
                "code": 409,
            },
            {
                "body": {"result": "Failed"},
                "match": "partial",
            },
            {
                "body": {"result": "Error"},
                "match": "partial",
            },
        ],
        "retry": 15,
    }
]

result = spintest(urls, tasks, generate_report="report_name")
assert True is result

spintest's People

Contributors

kyp76 avatar bew avatar bewsg avatar lordpatate avatar hellfire01 avatar qkponton avatar

Stargazers

Matthieu Gouel avatar Quentin Huber avatar Abder avatar Killian avatar Mohamed-Ali ELOUAER avatar

Watchers

Michel Rasschaert avatar Vincent Girard-Reydet avatar ahmed arbaoui avatar James Cloos avatar Patrice Lachance avatar Marouan Bl avatar CARRIERE Etienne avatar Yunshi TAN avatar  avatar Anand Manissery avatar Guillaume Fournier avatar Abdelhamid Henni avatar FBKZ avatar Fabrice VOLKAERT avatar  avatar Vincent Fuchs avatar  avatar Julien Debbia avatar Pierre-Emmanuel Fraisse avatar  avatar  avatar Baptiste Hebert avatar David FIOU avatar Aissa EL OUAFI avatar Amir Jaballah avatar  avatar NanLIANG avatar

spintest's Issues

Token type in spintest method

Hi,

I want to give a token generator to the function:

def spintest( urls: List[str], tasks: List[Dict[str, str]], token: str = None, parallel: bool = False, verify: bool = True, generate_report: Optional[str] = None, ):

Like the documentation says, we can give token string directly or a callable to generate the token, but the method shows that the token type is a string.

When I give a callable it works but there are some warnings due to the token type defined.

Thanks

As a QA I want to parallelize tasks.

Example :
I have an API where on 1 application I could add several features (15).
I do not want to create 15 tests files in order to run the test executions, but on 1 test file I could execute tasks in parallel.

Proposal :
Add new field
Optional("parallelize", default=False): bool,

As a QA I want to create a complete EndtoEnd test with several Endpoint

For the moment, with spintest we can test the same service (which is accessible from API).
If this service has a multiple regions, we can play the same test scenario on different endpoints of of this service (North America, South America, Europe ....).

However it's NOT possible to use multiple APIs from the same scenario (e.g: multiple region of same service, or multiple services altogether).

For example:
For the API1, I create a object OBJ1, and I want to use OBJ1 to create/update existing OBJ2 present in API2.

Spintest task with wrong format does not seems to return to correct error

I have encounter an error will doing some simple test with the library.

I'm using

python==3.9.7
spintest==0.4.2

I'm testing a dummy API having it running or not does not affect the behavior that I'm encountering.

Here is my python file

from spintest import spintest
from spintest.types import Int

urls = ["http://localhost:8080/"]
task = [{
    "method": "POST",
    "route": "rollback/",
    "output": "rollback_doc",
    "body": {"task": "string"},
    "expected": {
        "code": 201,
        "body": {
            "task": "string",
            "id": None
        },
        "expected_match": "partial"
    },
    "ignore": True
}, {
    "method": "GET",
    "route": "rollback/5",
    "expected": {
        "code": 200,
        "body": {
            "task": 'string',
            "id": Int("{{ rollback_doc['id']}}"),
            "status": "RUNNING"
        },
        "expected_match": "partial"
    },
    "fail_on": {
        "expected_match": "partial",
        "body": {"status": "ERROR"}
    },
    "rollback": ["delete_rollback"]
}, {
    "name": "delete_rollback",
    "method": "DELETE",
    "route": "rollback/5",
    "expected": {"code": 204}
}]
result = spintest(urls, task)
assert True is result

My tasks have 3 steps

  • Creating a resource with a POST -> no issue expected (ignore: True so that you don't need to have a running API)
  • Retrieving the resource with a wrong status -> Should trigger the rollback but will fail due to parsing issue
  • Deleting the resource -> Should always delete

When i'm executing the task i'm getting the following output

 ERROR - {
    "name": null,
    "status": "FAILED",
    "timestamp": "Thu Apr 21 10:14:59 2022",
    "duration_sec": 0.03,
    "url": "http://localhost:8080/",
    "route": "rollback/",
    "message": "Request failed.",
    "code": null,
    "body": null,
    "task": {
        "method": "POST",
        "route": "rollback/",
        "output": "rollback_doc",
        "ignore": true,
        "body": {
            "task": "string"
        },
        "expected": {
            "code": 201,
            "expected_match": "partial",
            "body": {
                "task": "string",
                "id": null
            }
        },
        "retry": 0,
        "delay": 1,
        "headers": {
            "Accept": "application/json",
            "Content-Type": "application/json"
        },
        "duration_sec": 0.03
    },
    "ignore": true
}
Traceback (most recent call last):
  File "/lib/python3.9/site-packages/spintest/manager.py", line 193, in run
    results.append(await self._next())
  File "/lib/python3.9/site-packages/spintest/manager.py", line 178, in _next
    return await self.stack.__anext__()
  File "/lib/python3.9/site-packages/spintest/manager.py", line 116, in _executor
    result = await Task(
  File "/lib/python3.9/site-packages/spintest/task.py", line 174, in run
    return self._response(
  File "/lib/python3.9/site-packages/spintest/task.py", line 48, in _response
    log_level.get(status, logger.critical)(json.dumps(result, indent=4))
  File "python/lib/python3.9/json/__init__.py", line 234, in dumps
    return cls(
  File "python/lib/python3.9/json/encoder.py", line 201, in encode
    chunks = list(chunks)
  File "python/lib/python3.9/json/encoder.py", line 431, in _iterencode
    yield from _iterencode_dict(o, _current_indent_level)
  File "python/lib/python3.9/json/encoder.py", line 405, in _iterencode_dict
    yield from chunks
  File "python/lib/python3.9/json/encoder.py", line 405, in _iterencode_dict
    yield from chunks
  File "python/lib/python3.9/json/encoder.py", line 405, in _iterencode_dict
    yield from chunks
  [Previous line repeated 1 more time]
  File "python/lib/python3.9/json/encoder.py", line 438, in _iterencode
    o = _default(o)
  File "python/lib/python3.9/json/encoder.py", line 179, in default
    raise TypeError(f'Object of type {o.__class__.__name__} '
TypeError: Object of type Int is not JSON serializable

On this python file i have noticed that some the 'fail_on' value under task[1]['fail_on'] has the incorrect type.
It should be a list of dict and not directly a dict.
The issue here is that the error raised does not match the issue at end.

By moving the faulty key ('fail_on') to the first task. I'm getting a more coherent error

2022-04-21 10:10:45,564 - ERROR - {
    "name": null,
    "status": "FAILED",
    "timestamp": "Thu Apr 21 10:10:45 2022",
    "duration_sec": null,
    "url": "http://localhost:8080/",
    "route": "rollback/",
    "message": "Task must follow this schema : Schema({'method': <class 'str'>, Optional('route'): <class 'str'>, Optional('name'): <class 'str'>, Optional('body'): <class 'dict'>, Optional('headers'): <class 'dict'>, Optional('output'): <class 'str'>, Optional('expected'): {Optional('code'): <class 'int'>, Optional('body'): Or(<class 'dict'>, <class 'str'>), Optional('expected_match'): Or('partial', 'strict')}, Optional('fail_on'): [{Optional('code'): <class 'int'>, Optional('body'): Or(<class 'dict'>, <class 'str'>), Optional('expected_match'): Or('partial', 'strict')}], Optional('retry'): <class 'int'>, Optional('delay'): <class 'int'>, Optional('ignore'): <class 'bool'>, Optional('rollback'): [Or(<class 'str'>, <class 'dict'>)]}).",
    "code": null,
    "body": null,
    "task": {
        "method": "POST",
        "route": "rollback/",
        "output": "rollback_doc",
        "body": {
            "task": "string"
        },
        "expected": {
            "code": 201,
            "body": {
                "task": "string",
                "id": null
            },
            "expected_match": "partial"
        },
        "fail_on": {
            "expected_match": "partial",
            "body": {
                "status": "ERROR"
            }
        }
    },
    "ignore": false
}
Traceback (most recent call last):
  File "pydevd.py", line 1415, in _exec
    pydev_imports.execfile(file, globals, locals)  # execute the script
  File "_pydev_execfile.py", line 18, in execfile
    exec(compile(contents+"\n", file, 'exec'), glob, loc)
  File "test_spin.py", line 158, in <module>
    assert True is result
AssertionError

From what i can understand.

the parsing of the second task fails here
.smoke/lib/python3.9/site-packages/spintest/task.py:172
validated_task = input_validator(self.task, TASK_SCHEMA)

Which lead to not parsing the task and skipping the casting the pytest.Int.
The task then failed here
.smoke/lib/python3.9/site-packages/spintest/task.py:48
log_level.get(status, logger.critical)(json.dumps(result, indent=4))

because result still have the "id": Int("{{ rollback_doc['id']}}"), unparsed causing the json.dumps to fail


I don't know the full depth of the parsing order but have noticed that in
.smoke/lib/python3.9/site-packages/spintest/validator.py:41

modifiing the

    try:
        return input_schema.validate(input)
    except SchemaError:
        return None

to

    try:
        return input_schema.validate(input)
    except SchemaError:
        raise

Raises the correct error schema.SchemaError: Key 'fail_on' error:{'expected_match': 'partial', 'body': {'status': 'ERROR'}} should be instance of 'list'

I don't know the full side effect of doing this change but doing something like this
or
Parsing the task (to remove cast) might be better to avoid dumping incorrect json.

Allow other autentification methods

The library allows to define a token parameter that is put in headers and that's great.

However, I'm using an other authentication method in headers: {'x-apikey': 'my_apikey'}
Api key is a custom authentication method that also exists in other forms, as : {'x-api-key': 'my_apikey'}

So, I need a way to authenticate with headers in a custom way.

As a workaround, I can use the 'headers' key in each task definition, but the secret value is not hidden from the report, and that's a huge security issue.

So, it would be nice to be able to give spintest a function as a new parameter, to build the authentication headers, returning a dict with custom keys/values.

For exemple:
The current "Authorizatioin" header is moved to this function:

def build_bearer_authentication(token: Union[str, Callable[..., str]]) -> dict:
    return {"Authorization": "Bearer " + (token() if callable(token) else token)}

The spintest function has a new parameter with the bearer function by default, so no breaking change:

def spintest(
    urls: List[str],
    tasks: List[Dict[str, str]],
    token: Union[str, Callable[..., str], None] = None,
    authentication_callback: Callable[..., dict] = build_bearer_authentication,
    parallel: bool = False,
    verify: bool = True,
    generate_report: Optional[str] = None,
):

and in task.py, it becomes:

                if self.output.get("__token__"):
                    token = self.output["__token__"]
                    authentication_headers = authentication_callback(token)
                    self.task["headers"].update(authentication_headers)

I think, this way, the token is hidden in the report regardless the header form as we always use the same token.

Recommend Projects

  • React photo React

    A declarative, efficient, and flexible JavaScript library for building user interfaces.

  • Vue.js photo Vue.js

    ๐Ÿ–– Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.

  • Typescript photo Typescript

    TypeScript is a superset of JavaScript that compiles to clean JavaScript output.

  • TensorFlow photo TensorFlow

    An Open Source Machine Learning Framework for Everyone

  • Django photo Django

    The Web framework for perfectionists with deadlines.

  • D3 photo D3

    Bring data to life with SVG, Canvas and HTML. ๐Ÿ“Š๐Ÿ“ˆ๐ŸŽ‰

Recommend Topics

  • javascript

    JavaScript (JS) is a lightweight interpreted programming language with first-class functions.

  • web

    Some thing interesting about web. New door for the world.

  • server

    A server is a program made to process requests and deliver data to clients.

  • Machine learning

    Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.

  • Game

    Some thing interesting about game, make everyone happy.

Recommend Org

  • Facebook photo Facebook

    We are working to build community through open source technology. NB: members must have two-factor auth.

  • Microsoft photo Microsoft

    Open source projects and samples from Microsoft.

  • Google photo Google

    Google โค๏ธ Open Source for everyone.

  • D3 photo D3

    Data-Driven Documents codes.