Tornado, pytest & jsonschema Jira api async testing

jira_utils.py

import json

from tornado.httpclient import AsyncHTTPClient


class JiraClient(object):
    def __init__(self, jira_server, jira_user, jira_password):
        self.__jira_server = jira_server
        self.__jira_user = jira_user
        self.__jira_password = jira_password
        self.__client = AsyncHTTPClient()

    async def __jira_get(self, url):
        _fetch_url = self.__jira_server + url
        return await self.__client.fetch(_fetch_url, auth_username=self.__jira_user, auth_password=self.__jira_password)

    async def get_boards(self):
        url = '/rest/agile/1.0/board'
        response = await self.__jira_get(url)
        boards = json.loads(response.body.decode())
        return boards

    async def get_sprints(self, board_id):
        url = '/rest/agile/1.0/board/%s/sprint' % board_id
        response = await self.__jira_get(url)
        sprints = json.loads(response.body.decode())
        return sprints

    async def get_issues_by_sprint(self, sprint_id, max_results=1000):
        url = '/rest/agile/1.0/sprint/%s/issue?expand=changelog&maxResults=%s' % (sprint_id, max_results)
        response = await self.__jira_get(url)
        sprints = json.loads(response.body.decode())
        return sprints

test_jira_utils.py

from random import sample

import jsonschema
import pytest

from project.application.jira_utils import JiraClient
from project.config.settings import jira_server, jira_user, jira_password

# Docs for json schema validation:
# https://pypi.python.org/pypi/jsonschema
# http://json-schema.org/latest/json-schema-validation.html

schema_boards = {
    'type': 'object',
    'properties': {
        'values': {
            'type': 'array',
            'items': {
                'type': 'object',
                'properties': {
                    'name': {'type': 'string'},
                    'id': {'type': 'number'},
                    'type': {'enum': ['scrum', 'kanban']}
                },
                'required': ['name', 'id', 'type']
            }
        }
    },
    'required': ['values']
}

schema_sprints = {
    'type': 'object',
    'properties': {
        'values': {
            'type': 'array',
            'items': {
                'type': 'object',
                'properties': {
                    'name': {'type': 'string'},
                    'id': {'type': 'number'},
                    'startDate': {'format': 'date-time'},
                    'endDate': {'format': 'date-time'},
                    'state': {'enum': ['future', 'closed', 'active']}
                },
                'required': ['name', 'id', 'state']
            }
        }
    },
    'requred': ['values']
}

schema_issues = {
    'type': 'object',
    'properties': {
        'issues': {
            'type': 'array',
            'items': {
                'type': 'object',
                'properties': {
                    'key': {'type': 'string'},
                    'fields': {
                        'type': 'object',
                        'properties': {
                            'summary': {'type': 'string'},
                            'description': {'type': ['string', 'null']}
                        },
                        'required': ['summary']
                    }
                },
                'required': ['key', 'fields']
            }
        }
    },
    'required': ['issues']
}


@pytest.mark.gen_test(timeout=15)
async def test_jira_client():
    jira_client = JiraClient(jira_server, jira_user, jira_password)
    boards = await jira_client.get_boards()
    assert jsonschema.validate(boards, schema_boards) is None

    # Спринты доступны только для scrum бордов
    scrum_boards = [b for b in boards['values'] if b['type'] == 'scrum']

    for board in sample(scrum_boards, min(5, len(scrum_boards))):
        sprints = await jira_client.get_sprints(board['id'])
        assert jsonschema.validate(sprints, schema_sprints) is None

        test_sprints = sprints['values']
        for sprint in sample(test_sprints, min(3, len(test_sprints))):
            issues = await jira_client.get_issues_by_sprint(sprint['id'], 2)
            assert jsonschema.validate(issues, schema_issues) is None