Git Product home page Git Product logo

pywsitest's Introduction

Grid Smarter Cities

Build Status License: MIT PyPI

pywsitest

PYthon WebSocket Integration TESTing framework

A python API to assist with automated websocket integration testing

Installation

pip install pywsitest

Package contents

WSTest is the main test running class in pywsitest. It currently has the following methods:

  • with_parameter: add a query parameter to the connection
  • with_header: add a header to the connection
  • with_response: add an expected response to the test runner
  • with_message: add a message for the test runner to send on connection
  • with_request: attach a rest api request to the instance of this class
  • with_response_timeout: set the timeout in seconds for the test runner to wait for a response from the websocket
  • with_message_timeout: set the timeout in seconds for the test runner to wait while trying to send a message to the websocket
  • with_request_timeout: set the timeout in seconds for the rest request attached to the instance of this class
  • with_test_timeout: set the timeout in seconds for the test runner to run for
  • with_received_response_logging: enable logging of received responses on response timeout error
  • run: asyncronously run the test runner, sending all messages and listening for responses
  • is_complete: check whether all expected responses have been received and messages have been sent

WSResponse is a class to represent an expected response from the websocket

  • with_attribute: add an attribute to check an incoming response against
  • with_trigger: add a message to trigger when a response matching this instance has been received
  • is_match: check whether a received response matches the attributes of this instance

WSMessage is a class to represent a message to send to the websocket

  • with_attribute: add an attribute to the message to be sent to the websocket host
  • with_delay: add a delay to the message to be sent to the websocket host

RestRequest is a class to represent a request to send to rest api

  • with_header: add a header to the request to be sent to the rest api
  • with_body: add a body to the request to be sent to the rest api
  • with_delay: add a delay to the request to be sent to the rest api

Examples

Response testing

Testing a response with a body is received on connection to a websocket host:

from pywsitest import WSTest, WSResponse

ws_test = (
    WSTest("wss://example.com")
    .with_response(
        WSResponse()
        .with_attribute("body")
    )
)

await ws_test.run()

assert ws_test.is_complete()

Testing that a response with the following more complicated body is received on connection to a websocket host:

{
    "body": {
        "attribute": "value"
    }
}
from pywsitest import WSTest, WSResponse

ws_test = (
    WSTest("wss://example.com")
    .with_response(
        WSResponse()
        .with_attribute("body")
        .with_attribute("body/attribute", "value")
    )
)

await ws_test.run()

assert ws_test.is_complete()

Testing that a response with the following body with a list is received on connection to a websocket host:

{
    "body": [
        {"colour": "red"},
        {"colour": "green"},
        {"colour": "blue"}
    ]
}
from pywsitest import WSTest, WSResponse

ws_test = (
    WSTest("wss://example.com")
    .with_response(
        WSResponse()
        .with_attribute("body/0/colour", "red")
        .with_attribute("body/1/colour", "green")
        .with_attribute("body/2/colour", "blue")
    )
)

await ws_test.run()

assert ws_test.is_complete()

Testing that a response with the following body with a list containing the colour green somewhere is received on connection to a websocket host:

{
    "body": [
        {"colour": "red"},
        {"colour": "green"},
        {"colour": "blue"}
    ]
}
from pywsitest import WSTest, WSResponse

ws_test = (
    WSTest("wss://example.com")
    .with_response(
        WSResponse()
        .with_attribute("body//colour", "green")
    )
)

await ws_test.run()

assert ws_test.is_complete()

Message sending

Sending a message on connection to a websocket host:

from pywsitest import WSTest, WSMessage

ws_test = (
    WSTest("wss://example.com")
    .with_message(
        WSMessage()
        .with_attribute("body", "Hello, world!")
    )
)

await ws_test.run()

assert ws_test.is_complete()

Triggering a message to be sent with extracted data when the following response is received:

{
    "body": {
        "message": "Hello, world!"
    }
}
from pywsitest import WSTest, WSResponse, WSMessage

ws_test = (
    WSTest("wss://example.com")
    .with_response(
        WSResponse()
        .with_attribute("body/message")
        .with_trigger(
            WSMessage()
            .with_attribute("body", "${body/message}")
        )
    )
)

await ws_test.run()

assert ws_test.is_complete()

Triggering a message to be sent with the first colour extracted from a list when the following response is received:

{
    "body": [
        {"colour": "red"},
        {"colour": "green"},
        {"colour": "blue"}
    ]
}
from pywsitest import WSTest, WSResponse, WSMessage

ws_test = (
    WSTest("wss://example.com")
    .with_response(
        WSResponse()
        .with_attribute("body/0/colour")
        .with_trigger(
            WSMessage()
            .with_attribute("body", "${body/0/colour}")
        )
    )
)

await ws_test.run()

assert ws_test.is_complete()

Using rest requests

Attaching simple rest get request and sending it:

rest_request = (
    RestRequest("https://example.com", "GET")
    .with_body({"some_key": some_value})
)

ws_test = (
    WSTest("wss://example.com")
    .with_request(rest_request)
)

await ws_test.run()

for response in ws_tester.received_request_responses:
    print(response.status_code)
    print(response.json())

assert ws_test.is_complete()

Error handling

Force a test to fail is execution takes more than 30 seconds (default 60 seconds)

ws_test = (
    WSTest("wss://example.com")
    .with_test_timeout(30)
    .with_response(
        WSResponse()
        .with_attribute("body")
    )
)

await ws_test.run()

assert ws_test.is_complete()

Force a test to fail if no response is received for 15 seconds (default 10 seconds)

  • Any responses that haven't been sent will be output along with the WSTimeoutError
  • Received responses can be output too by calling with_received_response_logging on the WSTest instance
ws_test = (
    WSTest("wss://example.com")
    .with_response_timeout(15)
    .with_received_response_logging()
    .with_response(
        WSResponse()
        .with_attribute("body")
    )
)

await ws_test.run()

assert ws_test.is_complete()

Force a test to fail is a message takes longer than 15 seconds to send (default 10 seconds)

  • The message that the test runner failed to send will be output along with the WSTimeoutError
ws_test = (
    WSTest("wss://example.com")
    .with_message_timeout(15)
    .with_message(
        WSMessage()
        .with_attribute("body", "Hello, world!")
    )
)

await ws_test.run()

assert ws_test.is_complete()

Documentation

Users can get the docstring help by running:

from pywsitest import WSTest
help(WSTest.with_response)

Links

pywsitest's People

Contributors

danpudwelluk avatar eulogio-gutierrez avatar jamsidedown avatar made-by-jonny avatar maxoyd avatar miriamt94 avatar

Stargazers

 avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar

Watchers

 avatar

pywsitest's Issues

Assert that an element in a list contains a property

There currently isn't a way to assert that an object with a certain property is present in a list/array of objects.

I'd like to be able to access an object in a list/array with or without an index, and assert that properties are present within that object

An alternative or future feature could be to allow a function that returns True/False on the response object.

Assuming the input json

{
    "body": [
        {"colour": "red"},
        {"colour": "green"},
        {"colour": "blue"}
    ]
}

It could be asserted that the first object has the colour red with:

WSResponse().with_attribute("body[0]colour", "red")

It could be asserted that any object has the colour blue with:

WSResponse().with_attribute("body[]colour", "blue")
# or
WSResponse().with_attribute("body[?]colour", "blue")
# or with any other special character

Add configurable sleep to triggers

Would be nice to be able to configure sleeps in triggered messages, to help with recreating user interactions

(
    WSTest("wss://example.com")
    .with_response(
        WSResponse()
        .with_attribute("body")
        .with_trigger(
            WSMessage()
            .with_delay(5)
            .with_attribute("message", "Hello")
        )
    )
)

pywsitest doesn't work with websocket servers without ssl certificates

wss:// works, ws:// doesn't

Set up a simple websocket server locally with python:

import asyncio
import websockets

async def handler(websocket, path):
    await websocket.send("Hello")

start_server = websockets.serve(handler, "localhost", 7890)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

Use the WSTest class to connect to the websocket server:

import asyncio
from pywsitest import WSTest

ws_test = WSTest("ws://localhost:7890")

asyncio.get_event_loop().run_until_complete(ws_test.run())

Produces:

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/base_events.py", line 584, in run_until_complete
    return future.result()
  File "/Users/robanderson/Documents/code/python/temp/env/lib/python3.7/site-packages/pywsitest/ws_test.py", line 169, in run
    websocket = await websockets.connect(self._get_connection_string(), ssl=ssl.SSLContext())
  File "/Users/robanderson/Documents/code/python/temp/env/lib/python3.7/site-packages/websockets/client.py", line 421, in __init__
    "connect() received a ssl argument for a ws:// URI, "
ValueError: connect() received a ssl argument for a ws:// URI, use a wss:// URI to enable TLS

Handle incoming messages that aren't json

Would be nice to handle and check for responses that aren't json based

E.g. asserting that the websocket tester receives a ping message

(
    WSResponse()
    .with_value("Ping")
)

Modify utils.get_resolved_values to handle list responses

Method signature currently states that it takes a dict as the response, and while it may work for lists with a fixed/known index, the path.lstrip("/") breaks lists where the index is not pre-defined.

I'd like for get_resolved_values to be able to handle lists too, as lists are valid json roots, and to be able to handle cases where the index isn't known.

I would expect the following test to pass

def test_resolve_root_level_list(self):
    ws_response = WSResponse().with_attribute("//colour", "blue")

    test_data = [
        {"colour": "red"},
        {"colour": "green"},
        {"colour": "blue"}
    ]

    self.assertTrue(ws_response.is_match(test_data))

Add support for REST requests to be triggered

Is your feature request related to a problem? Please describe.
Currently all functionality is expected to be performed via websockets, however there are cases when a rest call may trigger a websocket response that needs to be tested.

Describe the solution you'd like
I want to be able to trigger rest requests to be sent when websocket responses are received. I would also like to be able to add triggers to the response from a rest request.

Describe alternatives you've considered
An alternative would be to have separate rest requests being made while a websocket test is running asynchronously.

Additional context
No additional context.

Add session support

Is your feature request related to a problem? Please describe.
cuz session save cookie

Describe the solution you'd like
session/cookie support

Add authorization to ws connecting

Hi

I need authorization by token in headers. Please add it.
Now I override method 'run' to adding 'extra_headers' to 'websockets.connect(...'

Thanks =)

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.