ClickHouse/tests/ci/ci_settings.py
2024-08-03 21:19:51 +02:00

237 lines
8.8 KiB
Python

import re
from dataclasses import dataclass, asdict
from typing import Optional, List, Dict, Any, Iterable
from ci_config import CI
from git_helper import Runner as GitRunner, GIT_PREFIX
from pr_info import PRInfo
# pylint: disable=too-many-return-statements
@dataclass
class CiSettings:
# job will be included in the run if any keyword from the list matches job name
include_keywords: Optional[List[str]] = None
# job will be excluded in the run if any keyword from the list matches job name
exclude_keywords: Optional[List[str]] = None
# list of specified preconfigured ci sets to run
ci_sets: Optional[List[str]] = None
# list of specified jobs to run
ci_jobs: Optional[List[str]] = None
# batches to run for all multi-batch jobs
job_batches: Optional[List[int]] = None
do_not_test: bool = False
no_ci_cache: bool = False
upload_all: bool = False
no_merge_commit: bool = False
woolen_wolfdog: bool = False
def as_dict(self) -> Dict[str, Any]:
return asdict(self)
@staticmethod
def create_from_run_config(run_config: Dict[str, Any]) -> "CiSettings":
return CiSettings(**run_config["ci_settings"])
@staticmethod
def create_from_pr_message(
debug_message: Optional[str], update_from_api: bool
) -> "CiSettings":
"""
Creates CiSettings instance based on tags found in PR body and/or commit message
@commit_message - may be provided directly for debugging purposes, otherwise it will be retrieved from git.
"""
res = CiSettings()
pr_info = PRInfo()
if (
not pr_info.is_pr and not debug_message
): # if commit_message is provided it's test/debug scenario - do not return
# CI options can be configured in PRs only
# if debug_message is provided - it's a test
return res
message = debug_message or GitRunner(set_cwd_to_git_root=True).run(
f"{GIT_PREFIX} log {pr_info.sha} --format=%B -n 1"
)
# CI setting example we need to match with re:
# - [x] <!---ci_exclude_tsan|msan|ubsan|coverage--> Exclude: All with TSAN, MSAN, UBSAN, Coverage
pattern = r"(#|- \[x\] +<!---)([|\w]+)"
matches = [match[-1] for match in re.findall(pattern, message)]
print(f"CI tags from commit message: [{matches}]")
if not debug_message: # to be skipped if debug/test
pr_info = PRInfo(
pr_event_from_api=update_from_api
) # Fetch updated PR body from GH API
matches_pr = [match[-1] for match in re.findall(pattern, pr_info.body)]
print(f"CI tags from PR body: [{matches_pr}]")
matches = list(set(matches + matches_pr))
if "do not test" in pr_info.labels:
# do_not_test could be set in GH labels
res.do_not_test = True
for match in matches:
if match.startswith("job_"):
if not res.ci_jobs:
res.ci_jobs = []
res.ci_jobs.append(match.removeprefix("job_"))
elif match.startswith("ci_set_") and match in CI.Tags:
if not res.ci_sets:
res.ci_sets = []
res.ci_sets.append(match)
elif match.startswith("ci_include_"):
if not res.include_keywords:
res.include_keywords = []
res.include_keywords.append(
CI.Utils.normalize_string(match.removeprefix("ci_include_"))
)
elif match.startswith("ci_exclude_"):
if not res.exclude_keywords:
res.exclude_keywords = []
keywords = match.removeprefix("ci_exclude_").split("|")
res.exclude_keywords += [
CI.Utils.normalize_string(keyword) for keyword in keywords
]
elif match == CI.Tags.NO_CI_CACHE:
res.no_ci_cache = True
print("NOTE: CI Cache will be disabled")
elif match == CI.Tags.UPLOAD_ALL_ARTIFACTS:
res.upload_all = True
print("NOTE: All binary artifacts will be uploaded")
elif match == CI.Tags.DO_NOT_TEST_LABEL:
res.do_not_test = True
elif match == CI.Tags.NO_MERGE_COMMIT:
res.no_merge_commit = True
print("NOTE: Merge Commit will be disabled")
elif match == CI.Tags.WOOLEN_WOLFDOG_LABEL:
res.woolen_wolfdog = True
print("NOTE: Woolen Wolfdog mode enabled")
elif match.startswith("batch_"):
batches = []
try:
batches = [
int(batch) for batch in match.removeprefix("batch_").split("_")
]
except Exception:
print(f"ERROR: failed to parse commit tag [{match}] - skip")
if batches:
if not res.job_batches:
res.job_batches = []
res.job_batches += batches
res.job_batches = list(set(res.job_batches))
else:
print(
f"WARNING: Invalid tag in commit message or PR body [{match}] - skip"
)
return res
def _check_if_selected(
self,
job: str,
job_config: CI.JobConfig,
is_release: bool,
is_pr: bool,
is_mq: bool,
labels: Iterable[str],
) -> bool: # type: ignore #too-many-return-statements
if self.do_not_test:
label_config = CI.get_tag_config(CI.Tags.DO_NOT_TEST_LABEL)
assert label_config, f"Unknown tag [{CI.Tags.DO_NOT_TEST_LABEL}]"
if job in label_config.run_jobs:
print(
f"Job [{job}] present in CI set [{CI.Tags.DO_NOT_TEST_LABEL}] - pass"
)
return True
return False
if job_config.run_by_label:
if job_config.run_by_label in labels and is_pr:
print(
f"Job [{job}] selected by GH label [{job_config.run_by_label}] - pass"
)
return True
else:
return False
# do not exclude builds
if self.exclude_keywords and not CI.is_build_job(job):
for keyword in self.exclude_keywords:
if keyword in CI.Utils.normalize_string(job):
print(f"Job [{job}] matches Exclude keyword [{keyword}] - deny")
return False
to_deny = False
if self.include_keywords:
# do not exclude builds
if job == CI.JobNames.STYLE_CHECK or CI.is_build_job(job):
# never exclude Style Check by include keywords
return True
for keyword in self.include_keywords:
if keyword in CI.Utils.normalize_string(job):
print(f"Job [{job}] matches Include keyword [{keyword}] - pass")
return True
to_deny = True
if self.ci_sets:
for tag in self.ci_sets:
label_config = CI.get_tag_config(tag)
assert label_config, f"Unknown tag [{tag}]"
if job in label_config.run_jobs:
print(f"Job [{job}] present in CI set [{tag}] - pass")
return True
to_deny = True
if self.ci_jobs:
if job in self.ci_jobs:
print(f"Job [{job}] set by CI #job_ tags [{self.ci_jobs}] - pass")
return True
to_deny = True
if job_config.release_only and not is_release:
return False
elif job_config.pr_only and not is_pr and not is_mq:
return False
return not to_deny
def apply(
self,
job_configs: Dict[str, CI.JobConfig],
is_release: bool,
is_pr: bool,
is_mq: bool,
labels: Iterable[str],
) -> Dict[str, CI.JobConfig]:
"""
Apply CI settings from pr body
"""
res = {}
for job, job_config in job_configs.items():
if self._check_if_selected(
job,
job_config,
is_release=is_release,
is_pr=is_pr,
is_mq=is_mq,
labels=labels,
):
res[job] = job_config
add_parents = []
for job in list(res):
parent_jobs = CI.get_job_parents(job)
for parent_job in parent_jobs:
if parent_job not in res:
add_parents.append(parent_job)
print(f"Job [{job}] requires [{parent_job}] - add")
for job in add_parents:
res[job] = job_configs[job]
return res