diff --git a/backend/scripts/test_executions_rerunner.py b/backend/scripts/test_executions_rerunner.py deleted file mode 100644 index b20643d8..00000000 --- a/backend/scripts/test_executions_rerunner.py +++ /dev/null @@ -1,65 +0,0 @@ -""" -This script will be used by jenkins to rerun test executions as requested -by test observer. Please note that jenkins will download this file then run it. -Therefore, this script can't import from the rest of the codebase and shouldn't be -renamed. Dependencies used by this script must be installed on jenkins. -""" - -import logging -import re -from os import environ - -import requests -from requests.auth import HTTPBasicAuth - -reruns_link = "https://test-observer.canonical.com/v1/test-executions/reruns" - - -class Main: - def __init__(self, jenkins_api_token: str | None = None): - self.jenkins_auth = HTTPBasicAuth( - "admin", jenkins_api_token or environ["JENKINS_API_TOKEN"] - ) - - def run(self): - self._load_rerun_requests() - self._submit_rerun_requests() - - def _load_rerun_requests(self) -> None: - response = requests.get(reruns_link) - self.rerun_requests = response.json() - logging.info(f"Received the following rerun requests:\n{self.rerun_requests}") - - def _submit_rerun_requests(self) -> None: - for rerun_request in self.rerun_requests: - self._submit_rerun(rerun_request) - - def _submit_rerun(self, rerun_request: dict) -> None: - base_job_link = self._extract_base_job_link_from_ci_link( - rerun_request["ci_link"] - ) - if base_job_link: - family = rerun_request["family"] - if family == "deb": - self._submit_deb_rerun(base_job_link) - elif family == "snap": - self._submit_snap_rerun(base_job_link) - else: - logging.error(f"Invalid family name {family}") - - def _extract_base_job_link_from_ci_link(self, ci_link: str) -> str | None: - matching = re.match(r"(.+/)\d+/", ci_link) - if matching: - return matching.group(1) - return None - - def _submit_snap_rerun(self, base_job_link: str) -> None: - rerun_link = f"{base_job_link}/build" - logging.info(f"POST {rerun_link}") - requests.post(rerun_link, auth=self.jenkins_auth) - - def _submit_deb_rerun(self, base_job_link: str) -> None: - rerun_link = f"{base_job_link}/buildWithParameters" - data = {"TESTPLAN": "full"} - logging.info(f"POST {rerun_link} {data}") - requests.post(rerun_link, auth=self.jenkins_auth, json=data) diff --git a/backend/tests/scripts/test_test_executions_rerunner.py b/backend/tests/scripts/test_test_executions_rerunner.py deleted file mode 100644 index 59ffa233..00000000 --- a/backend/tests/scripts/test_test_executions_rerunner.py +++ /dev/null @@ -1,66 +0,0 @@ -import base64 - -from requests_mock import Mocker - -from scripts.test_executions_rerunner import Main, reruns_link -from test_observer.data_access.models_enums import FamilyName - -jenkins_request_headers = { - "Authorization": f"Basic {base64.b64encode(b'admin:token').decode()}" -} - - -def test_does_nothing_when_no_reruns_requested(requests_mock: Mocker): - requests_mock.get(reruns_link, json=[]) - - execute() - - assert requests_mock.called_once - - -def test_correctly_submits_deb_rerun_request(requests_mock: Mocker): - job_link = "http://10.102.156.15:8080/job/fake-job/" - - ci_link = f"{job_link}1/" - requests_mock.get( - reruns_link, - json=[{"test_execution_id": 1, "ci_link": ci_link, "family": FamilyName.DEB}], - ) - - def is_data_valid(request) -> bool: # noqa: ANN001 - return request.json() == {"TESTPLAN": "full"} - - rerun_link = f"{job_link}/buildWithParameters" - matcher = requests_mock.post( - rerun_link, - request_headers=jenkins_request_headers, - additional_matcher=is_data_valid, - ) - - execute() - - assert matcher.called_once - - -def test_correctly_submits_snap_rerun_request(requests_mock: Mocker): - job_link = "http://10.102.156.15:8080/job/fake-job/" - - ci_link = f"{job_link}1/" - requests_mock.get( - reruns_link, - json=[{"test_execution_id": 1, "ci_link": ci_link, "family": FamilyName.SNAP}], - ) - - rerun_link = f"{job_link}/build" - matcher = requests_mock.post( - rerun_link, - request_headers=jenkins_request_headers, - ) - - execute() - - assert matcher.called_once - - -def execute(): - Main(jenkins_api_token="token").run()