Loading ci/get_id_of_last_job_occurence.py +10 −32 Original line number Diff line number Diff line Loading @@ -40,15 +40,9 @@ PAGE_SUFFIX = "&page={}" API_URL_TMPL = "https://forge.3gpp.org/rep/api/v4/projects/{}/pipelines" def get_job_ids( branch_name: str, job_name: str, project_id: int, success_only: bool, n: int ) -> list[int]: """ Find id of last <n> jobs with name <branch_name> for prject with id <project_id>. If <success_only> is true, only jobs with status success are considered. """ job_ids = list() # check last 5000 pipelines max def get_job_id(branch_name, job_name, project_id, success_only): job_id = -1 # check last 500 pipelines max for page in range(100): url_pls = API_URL_TMPL.format(project_id) Loading @@ -70,18 +64,15 @@ def get_job_ids( for job in resp_jobs.json(): include_job = not success_only or job["status"] == "success" if include_job and job["name"] == job_name: job_ids.append(job["id"]) if len(job_ids) == n: job_id = job["id"] break if len(job_ids) == n: if job_id >= 0: break if len(job_ids) == n: if job_id >= 0: break return job_ids return job_id if __name__ == "__main__": Loading @@ -89,22 +80,9 @@ if __name__ == "__main__": parser.add_argument("branch_name", help="Name of the branch to search on") parser.add_argument("job_name", help="Name of the job to get the id of") parser.add_argument("project_id", help="ID of project to search in", type=int) parser.add_argument( "--success_only", help="Only include jobs with status 'success'", action="store_true", ) parser.add_argument("--success_only", help="Only include jobs with status 'success'", action="store_true") args = parser.parse_args() n = 1 job_ids = get_job_ids( args.branch_name, args.job_name, args.project_id, args.success_only, n ) try: job_id = job_ids[0] except IndexError: job_id = -1 job_id = get_job_id(args.branch_name, args.job_name, args.project_id, args.success_only) print(job_id) ci/setup_pages.py +27 −28 Original line number Diff line number Diff line Loading @@ -4,29 +4,26 @@ import pathlib import subprocess import sys import shutil from collections import namedtuple from tempfile import TemporaryDirectory from get_id_of_last_job_occurence import get_job_ids from get_id_of_last_job_occurence import get_job_id PROJECT_ID_FLOAT_REPO = 49 PROJECT_ID_BASOP_REPO = 77 Job = namedtuple("Job", ["name", "n"]) JOBS_FLOAT_REPO = [ Job("complexity-stereo-in-stereo-out", 1), Job("complexity-ism-in-binaural-out", 1), Job("complexity-sba-hoa3-in-hoa3-out", 1), Job("complexity-mc-in-7_1_4-out", 1), Job("complexity-masa-in-7_1_4-out", 1), Job("complexity-StereoDmxEVS-stereo-in-mono-out", 1), Job("coverage-test-on-main-scheduled", 1), "complexity-stereo-in-stereo-out", "complexity-ism-in-binaural-out", "complexity-sba-hoa3-in-hoa3-out", "complexity-mc-in-7_1_4-out", "complexity-masa-in-7_1_4-out", "complexity-StereoDmxEVS-stereo-in-mono-out", "coverage-test-on-main-scheduled", ] JOBS_BASOP_REPO = [ "ivas-pytest-mld-long-dec", ] JOBS_BASOP_REPO = { Job("ivas-pytest-mld-long-dec", 1), } JOBS_FOR_PROJECT_ID = { PROJECT_ID_FLOAT_REPO: JOBS_FLOAT_REPO, Loading Loading @@ -87,16 +84,18 @@ def get_artifacts_for_jobs_and_return_num_failed( for job in jobs: print(os.environ["CI_DEFAULT_BRANCH"], job.name, project_id, success_only) # TODO: change back before merginG!! # job_ids = get_job_ids( # os.environ["CI_DEFAULT_BRANCH"], job.name, project_id, success_only, job.n # job_id = get_job_id( # os.environ["CI_DEFAULT_BRANCH"], job.name, project_id, success_only # ) job_ids = get_job_ids( "ci/setup_gitlab_pages_for_ci_result_presentation", job.name, project_id, success_only, job.n job_id = get_job_id( "ci/setup_gitlab_pages_for_ci_result_presentation", job.name, project_id, success_only, ) print(f"{job_ids} - {job.name}") print(f"{job_id} - {job.name}") try: for job_id in job_ids: with TemporaryDirectory() as tmp_dir: curl_for_artifacts(job_id, tmp_dir) Loading Loading
ci/get_id_of_last_job_occurence.py +10 −32 Original line number Diff line number Diff line Loading @@ -40,15 +40,9 @@ PAGE_SUFFIX = "&page={}" API_URL_TMPL = "https://forge.3gpp.org/rep/api/v4/projects/{}/pipelines" def get_job_ids( branch_name: str, job_name: str, project_id: int, success_only: bool, n: int ) -> list[int]: """ Find id of last <n> jobs with name <branch_name> for prject with id <project_id>. If <success_only> is true, only jobs with status success are considered. """ job_ids = list() # check last 5000 pipelines max def get_job_id(branch_name, job_name, project_id, success_only): job_id = -1 # check last 500 pipelines max for page in range(100): url_pls = API_URL_TMPL.format(project_id) Loading @@ -70,18 +64,15 @@ def get_job_ids( for job in resp_jobs.json(): include_job = not success_only or job["status"] == "success" if include_job and job["name"] == job_name: job_ids.append(job["id"]) if len(job_ids) == n: job_id = job["id"] break if len(job_ids) == n: if job_id >= 0: break if len(job_ids) == n: if job_id >= 0: break return job_ids return job_id if __name__ == "__main__": Loading @@ -89,22 +80,9 @@ if __name__ == "__main__": parser.add_argument("branch_name", help="Name of the branch to search on") parser.add_argument("job_name", help="Name of the job to get the id of") parser.add_argument("project_id", help="ID of project to search in", type=int) parser.add_argument( "--success_only", help="Only include jobs with status 'success'", action="store_true", ) parser.add_argument("--success_only", help="Only include jobs with status 'success'", action="store_true") args = parser.parse_args() n = 1 job_ids = get_job_ids( args.branch_name, args.job_name, args.project_id, args.success_only, n ) try: job_id = job_ids[0] except IndexError: job_id = -1 job_id = get_job_id(args.branch_name, args.job_name, args.project_id, args.success_only) print(job_id)
ci/setup_pages.py +27 −28 Original line number Diff line number Diff line Loading @@ -4,29 +4,26 @@ import pathlib import subprocess import sys import shutil from collections import namedtuple from tempfile import TemporaryDirectory from get_id_of_last_job_occurence import get_job_ids from get_id_of_last_job_occurence import get_job_id PROJECT_ID_FLOAT_REPO = 49 PROJECT_ID_BASOP_REPO = 77 Job = namedtuple("Job", ["name", "n"]) JOBS_FLOAT_REPO = [ Job("complexity-stereo-in-stereo-out", 1), Job("complexity-ism-in-binaural-out", 1), Job("complexity-sba-hoa3-in-hoa3-out", 1), Job("complexity-mc-in-7_1_4-out", 1), Job("complexity-masa-in-7_1_4-out", 1), Job("complexity-StereoDmxEVS-stereo-in-mono-out", 1), Job("coverage-test-on-main-scheduled", 1), "complexity-stereo-in-stereo-out", "complexity-ism-in-binaural-out", "complexity-sba-hoa3-in-hoa3-out", "complexity-mc-in-7_1_4-out", "complexity-masa-in-7_1_4-out", "complexity-StereoDmxEVS-stereo-in-mono-out", "coverage-test-on-main-scheduled", ] JOBS_BASOP_REPO = [ "ivas-pytest-mld-long-dec", ] JOBS_BASOP_REPO = { Job("ivas-pytest-mld-long-dec", 1), } JOBS_FOR_PROJECT_ID = { PROJECT_ID_FLOAT_REPO: JOBS_FLOAT_REPO, Loading Loading @@ -87,16 +84,18 @@ def get_artifacts_for_jobs_and_return_num_failed( for job in jobs: print(os.environ["CI_DEFAULT_BRANCH"], job.name, project_id, success_only) # TODO: change back before merginG!! # job_ids = get_job_ids( # os.environ["CI_DEFAULT_BRANCH"], job.name, project_id, success_only, job.n # job_id = get_job_id( # os.environ["CI_DEFAULT_BRANCH"], job.name, project_id, success_only # ) job_ids = get_job_ids( "ci/setup_gitlab_pages_for_ci_result_presentation", job.name, project_id, success_only, job.n job_id = get_job_id( "ci/setup_gitlab_pages_for_ci_result_presentation", job.name, project_id, success_only, ) print(f"{job_ids} - {job.name}") print(f"{job_id} - {job.name}") try: for job_id in job_ids: with TemporaryDirectory() as tmp_dir: curl_for_artifacts(job_id, tmp_dir) Loading