codecov-cli/codecov_cli/services/report/__init__.py at 01b3e73908e599679135a6a955fbd3e120708069 · zetaab/codecov-cli · 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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
import json
import logging
import time

from codecov_cli.helpers import request
from codecov_cli.helpers.config import CODECOV_API_URL
from codecov_cli.helpers.encoder import encode_slug
from codecov_cli.helpers.request import (
    get_token_header_or_fail,
    log_warnings_and_errors_if_any,
    request_result,
    send_post_request,
)

logger = logging.getLogger("codecovcli")
MAX_NUMBER_TRIES = 3


def create_report_logic(
    commit_sha: str,
    code: str,
    slug: str,
    service: str,
    token: str,
    enterprise_url: str,
    fail_on_error: bool = False,
):
    encoded_slug = encode_slug(slug)
    sending_result = send_create_report_request(
        commit_sha, code, service, token, encoded_slug, enterprise_url
    )
    log_warnings_and_errors_if_any(sending_result, "Report creating", fail_on_error)
    return sending_result


def send_create_report_request(
    commit_sha, code, service, token, encoded_slug, enterprise_url
):
    data = {"code": code}
    headers = get_token_header_or_fail(token)
    upload_url = enterprise_url or CODECOV_API_URL
    url = f"{upload_url}/upload/{service}/{encoded_slug}/commits/{commit_sha}/reports"
    return send_post_request(url=url, headers=headers, data=data)


def create_report_results_logic(
    commit_sha: str,
    code: str,
    slug: str,
    service: str,
    token: str,
    enterprise_url: str,
    fail_on_error: bool = False,
):
    encoded_slug = encode_slug(slug)
    sending_result = send_reports_result_request(
        commit_sha=commit_sha,
        report_code=code,
        encoded_slug=encoded_slug,
        service=service,
        token=token,
        enterprise_url=enterprise_url,
    )

    log_warnings_and_errors_if_any(
        sending_result, "Report results creating", fail_on_error
    )
    return sending_result


def send_reports_result_request(
    commit_sha, report_code, encoded_slug, service, token, enterprise_url
):
    headers = get_token_header_or_fail(token)
    upload_url = enterprise_url or CODECOV_API_URL
    url = f"{upload_url}/upload/{service}/{encoded_slug}/commits/{commit_sha}/reports/{report_code}/results"
    return send_post_request(url=url, headers=headers)


def send_reports_result_get_request(
    commit_sha,
    report_code,
    encoded_slug,
    service,
    token,
    enterprise_url,
    fail_on_error=False,
):
    headers = get_token_header_or_fail(token)
    upload_url = enterprise_url or CODECOV_API_URL
    url = f"{upload_url}/upload/{service}/{encoded_slug}/commits/{commit_sha}/reports/{report_code}/results"
    number_tries = 0
    while number_tries < MAX_NUMBER_TRIES:
        resp = request.get(url=url, headers=headers)
        response_obj = request_result(resp)
        response_content = json.loads(response_obj.text)

        # if response_status is 400 and higher
        if response_obj.error:
            log_warnings_and_errors_if_any(
                response_obj, "Getting report results", fail_on_error
            )
            return response_obj

        state = response_content.get("state").lower()
        if state == "error":
            logger.error(
                "An error occured while processing the report. Please try again later.",
                extra=dict(
                    extra_log_attributes=dict(
                        response_status_code=response_obj.status_code,
                        state=response_content.get("state"),
                        result=response_content.get("result"),
                    )
                ),
            )
            return response_obj
        elif state == "pending":
            logger.info("Report with the given code is still being processed.")
        elif state == "completed":
            logger.info(
                "Finished processing report results",
                extra=dict(
                    extra_log_attributes=dict(
                        state=response_content.get("result")["state"],
                        message=response_content.get("result")["message"],
                    )
                ),
            )
            return response_obj
        else:
            logger.error("Please try again later.")
            return response_obj
        time.sleep(5)
        number_tries += 1
    return response_obj