Skip to content

Api Client

QaseClient

Class for interacting with tests runs in Qase API.

Source code in pytest_qaseio/api_client.py 
 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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
class QaseClient:
    """Class for interacting with tests runs in Qase API."""

    def __init__(
        self,
        token: str,
        project_code: str,
        retries: int,
    ) -> None:
        """Init client."""
        super().__init__()
        self._logger = logging.getLogger("qase")
        self._logger.addHandler(
            logging.StreamHandler(sys.stderr),
        )
        self._client = ApiClient(
            configuration=qaseio_config.Configuration(
                api_key={
                    "TokenAuth": token,
                },
            ),
        )
        self._project_code: str = project_code
        self._retries = retries

    def api_retry(
        self,
        function: collections.abc.Callable[FuncParams, ReturnValue],
    ) -> collections.abc.Callable[FuncParams, ReturnValue]:
        """Retrying Qase API requests.

        Sometimes Qase.io API closes the connection without response after a
        large number of requests, so we added retries to
        all qase.io API requests.

        """

        @functools.wraps(function)
        def wrapper(
            *args: FuncParams.args,
            **kwargs: FuncParams.kwargs,
        ) -> ReturnValue:
            for retry in tenacity.Retrying(
                stop=tenacity.stop_after_attempt(self._retries),
                wait=tenacity.wait_exponential(),
                reraise=True,
            ):
                with retry:
                    return function(*args, **kwargs)

            # Just hack for mypy "missing return statement" error
            raise ValueError("No raises and no return from Qase API")

        return wrapper

    def get_run(
        self,
        run_id: int,
    ) -> Run:
        """Get test run from Qase."""
        run_response = self.api_retry(RunsApi(self._client).get_run)(
            code=self._project_code,
            id=run_id,
        )
        return typing.cast(Run, run_response.result)

    def create_run(
        self,
        run_data: RunCreate,
    ) -> Run:
        """Create test run in Qase."""
        response = self.api_retry(RunsApi(self._client).create_run)(
            code=self._project_code,
            run_create=run_data,
        )
        created_run = typing.cast(IdResponseAllOfResult, response.result)

        return self.get_run(typing.cast(int, created_run.id))

    def load_cases_ids(
        self,
    ) -> list[int]:
        """Load all cases ids of project."""
        limit = 100
        cases: list[int] = []

        while True:
            response = self.api_retry(CasesApi(self._client).get_cases)(
                code=self._project_code,
                limit=limit,
                offset=len(cases),
            )
            response_cases = getattr(response.result, "entities", [])
            new_cases = [case.id for case in response_cases]
            cases += new_cases
            if not new_cases:
                break
        return cases

    def report_test_results(
        self,
        run: Run,
        report_data: ResultCreate,
    ) -> tuple[str, ResultCreate]:
        """Report test results back to Qase."""
        create_result = self.api_retry(ResultsApi(self._client).create_result)
        result = create_result(
            code=self._project_code,
            id=typing.cast(int, run.id),
            result_create=report_data,
        ).result
        assert result  # noqa: S101
        return typing.cast(str, result.hash), typing.cast(
            ResultCreate,
            report_data.status,
        )

__init__(token, project_code, retries)

Init client.

Source code in pytest_qaseio/api_client.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def __init__(
    self,
    token: str,
    project_code: str,
    retries: int,
) -> None:
    """Init client."""
    super().__init__()
    self._logger = logging.getLogger("qase")
    self._logger.addHandler(
        logging.StreamHandler(sys.stderr),
    )
    self._client = ApiClient(
        configuration=qaseio_config.Configuration(
            api_key={
                "TokenAuth": token,
            },
        ),
    )
    self._project_code: str = project_code
    self._retries = retries

api_retry(function)

Retrying Qase API requests.

Sometimes Qase.io API closes the connection without response after a large number of requests, so we added retries to all qase.io API requests.

Source code in pytest_qaseio/api_client.py 
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
def api_retry(
    self,
    function: collections.abc.Callable[FuncParams, ReturnValue],
) -> collections.abc.Callable[FuncParams, ReturnValue]:
    """Retrying Qase API requests.

    Sometimes Qase.io API closes the connection without response after a
    large number of requests, so we added retries to
    all qase.io API requests.

    """

    @functools.wraps(function)
    def wrapper(
        *args: FuncParams.args,
        **kwargs: FuncParams.kwargs,
    ) -> ReturnValue:
        for retry in tenacity.Retrying(
            stop=tenacity.stop_after_attempt(self._retries),
            wait=tenacity.wait_exponential(),
            reraise=True,
        ):
            with retry:
                return function(*args, **kwargs)

        # Just hack for mypy "missing return statement" error
        raise ValueError("No raises and no return from Qase API")

    return wrapper

create_run(run_data)

Create test run in Qase.

Source code in pytest_qaseio/api_client.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def create_run(
    self,
    run_data: RunCreate,
) -> Run:
    """Create test run in Qase."""
    response = self.api_retry(RunsApi(self._client).create_run)(
        code=self._project_code,
        run_create=run_data,
    )
    created_run = typing.cast(IdResponseAllOfResult, response.result)

    return self.get_run(typing.cast(int, created_run.id))

get_run(run_id)

Get test run from Qase.

Source code in pytest_qaseio/api_client.py 
79
80
81
82
83
84
85
86
87
88
def get_run(
    self,
    run_id: int,
) -> Run:
    """Get test run from Qase."""
    run_response = self.api_retry(RunsApi(self._client).get_run)(
        code=self._project_code,
        id=run_id,
    )
    return typing.cast(Run, run_response.result)

load_cases_ids()

Load all cases ids of project.

Source code in pytest_qaseio/api_client.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def load_cases_ids(
    self,
) -> list[int]:
    """Load all cases ids of project."""
    limit = 100
    cases: list[int] = []

    while True:
        response = self.api_retry(CasesApi(self._client).get_cases)(
            code=self._project_code,
            limit=limit,
            offset=len(cases),
        )
        response_cases = getattr(response.result, "entities", [])
        new_cases = [case.id for case in response_cases]
        cases += new_cases
        if not new_cases:
            break
    return cases

report_test_results(run, report_data)

Report test results back to Qase.

Source code in pytest_qaseio/api_client.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def report_test_results(
    self,
    run: Run,
    report_data: ResultCreate,
) -> tuple[str, ResultCreate]:
    """Report test results back to Qase."""
    create_result = self.api_retry(ResultsApi(self._client).create_result)
    result = create_result(
        code=self._project_code,
        id=typing.cast(int, run.id),
        result_create=report_data,
    ).result
    assert result  # noqa: S101
    return typing.cast(str, result.hash), typing.cast(
        ResultCreate,
        report_data.status,
    )