kernelci-api/tests/e2e_tests/test_node_handler.py at 7df261e714fa5e1b0cf9d1f94ff3920c121255e5 · nuclearcat/kernelci-api · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
# SPDX-License-Identifier: LGPL-2.1-or-later
#
# Copyright (C) 2023 Collabora Limited
# Author: Jeny Sadadia <jeny.sadadia@collabora.com>


"""Test functions for KernelCI API node handler"""

import json

import pytest

from .conftest import node_model_fields, paginated_response_keys


async def create_node(test_async_client, node):
    """
    Test Case : Test KernelCI API POST '/node' endpoint
    Expected Result :
        HTTP Response Code 200 OK
        JSON with created Node object attributes
    """
    response = await test_async_client.post(
        "node",
        headers={
            "Accept": "application/json",
            "Authorization": f"Bearer {pytest.BEARER_TOKEN}",  # pylint: disable=no-member
        },
        data=json.dumps(node),
    )
    assert response.status_code == 200
    assert response.json().keys() == node_model_fields
    return response


async def get_node_by_id(test_async_client, node_id):
    """
    Test Case : Test KernelCI API GET /node/{node_id} endpoint
    Expected Result :
        HTTP Response Code 200 OK
        JSON with Node object attributes
    """
    response = await test_async_client.get(
        f"node/{node_id}",
        headers={
            "Accept": "application/json",
            "Authorization": f"Bearer {pytest.BEARER_TOKEN}",  # pylint: disable=no-member
        },
    )
    assert response.status_code == 200
    assert response.json().keys() == node_model_fields
    return response


async def get_node_by_attribute(test_async_client, params):
    """
    Test Case : Test KernelCI API GET /nodes matching query parameters
    Expected Result :
        HTTP Response Code 200 OK
        Returns dictionary with matching Node objects, total number of nodes
        returned along with limit and offset values
    """
    response = await test_async_client.get(
        "nodes",
        params=params,
        headers={
            "Accept": "application/json",
            "Authorization": f"Bearer {pytest.BEARER_TOKEN}",  # pylint: disable=no-member
        },
    )
    assert response.status_code == 200
    assert response.json().keys() == paginated_response_keys
    assert response.json()["total"] >= 0
    return response


async def update_node(test_async_client, node):
    """
    Test Case : Test KernelCI API PUT /node/{node_id} endpoint
    Expected Result :
        HTTP Response Code 200 OK
        JSON with updated Node object
    """
    response = await test_async_client.put(
        f"node/{node['id']}",
        headers={
            "Accept": "application/json",
            "Authorization": f"Bearer {pytest.BEARER_TOKEN}",  # pylint: disable=no-member
        },
        data=json.dumps(node),
    )
    assert response.status_code == 200
    assert response.json().keys() == node_model_fields


async def patch_node(test_async_client, node_id, patch_data):
    """
    Test Case : Test KernelCI API PATCH /node/{node_id} endpoint
    Expected Result :
        HTTP Response Code 200 OK
        JSON with updated Node object
    """
    response = await test_async_client.patch(
        f"node/{node_id}",
        headers={
            "Accept": "application/json",
            "Authorization": f"Bearer {pytest.BEARER_TOKEN}",  # pylint: disable=no-member
        },
        data=json.dumps(patch_data),
    )
    assert response.status_code == 200
    assert response.json().keys() == node_model_fields
    return response