No OneTemporary
Actions

Size

56 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lilybuild/lilybuild/ci_steps.py b/lilybuild/lilybuild/ci_steps.py
	index 85cfcbb..6d76179 100644
	--- a/lilybuild/lilybuild/ci_steps.py
	+++ b/lilybuild/lilybuild/ci_steps.py
	@@ -1,577 +1,578 @@

	from buildbot.plugins import *
	from buildbot.process import buildstep, logobserver
	from buildbot.interfaces import IRenderable
	from twisted.internet import defer
	from .ci_syntax import ci_file
	from .ci_syntax import rules as ci_rules
	from .helpers import rsync_rules_from_artifacts, get_job_script, normalize_image, normalize_services
	from .phorge import SendCoverageToPhorge
	import re
	import sys
	import json

	SAFETAR_EXEC = '/lilybuild/lilybuild/safetar.py'
	COVERAGE_EXEC = '/lilybuild/lilybuild/coverage.py'

	def on_success(step):
	return step.build.results == util.SUCCESS

	def on_always(_step):
	return True

	def fill_list(*args):
	return list(args)

	class RunCIJobStep(steps.BuildStep):
	# 200 MiB
	artifact_max_size = 200 * 1024 * 1024
	default_image = 'alpine'
	master_job_artifact_dir_pattern = '%(kw:st)s/repos/%(prop:lilybuild_repo_id)s/builds/%(prop:lilybuild_root_build_id)s/jobs/%(kw:job)s/artifacts'
	artifact_file_name = 'artifacts.tar'
	master_job_artifact_file_name_pattern = master_job_artifact_dir_pattern + '/' + artifact_file_name
	reports_file_name = 'reports.tar'
	master_reports_file_name_pattern = master_job_artifact_dir_pattern + '/' + reports_file_name
	master_pages_dir_pattern = '%(kw:st)s/repos/%(prop:lilybuild_repo_id)s/pages'
	phorge_coverage_file_name = 'coverage-phorge.json'

	def __init__(
	self,
	lbc,
	src_relative=None,
	src_dir=None,
	storage_dir=None,
	repo_id=None,
	result_relative=None,
	result_dir=None,
	artifact_stage_relative=None,
	artifact_stage_dir=None,
	job_prop=None,
	artifact_link_base=None,
	**kwargs):
	self.lbc = lbc
	self.src_relative = src_relative
	self.src_dir = src_dir
	self.work_root_dir = kwargs['workdir']
	self.script_dir = 'script'
	self.storage_dir = storage_dir
	self.repo_id = repo_id
	self.artifact_stage_relative = artifact_stage_relative
	self.artifact_stage_dir = artifact_stage_dir
	self.result_relative = result_relative
	self.result_dir = result_dir
	self.artifact_link_base = artifact_link_base
	super().__init__(name='Run step', **kwargs)

	def get_cur_repo_config(self):
	return self.lbc.repos[self.getProperty('lilybuild_repo_id')]

	@defer.inlineCallbacks
	def run(self):
	job_prop = self.getProperty('lilybuild_job_prop')
	job = ci_file.CIJob.from_prop(job_prop)
	job_index = self.getProperty('lilybuild_job_index')
	variables = yield self.get_ci_variables(job)

	should_run = True
	if len(job.rules):
	should_run = False
	+ default_when = job.struct_raw.get('when', 'on_success')
	for r in job.rules:
	try:
	- when = r.get('when', 'on_success')
	- if when == 'never':
	+ when = r.get('when', default_when)
	+ if when == 'never' or when == 'manual':
	should_run_to_set = False
	else:
	should_run_to_set = True

	rule_str = r.get('if')
	if not rule_str:
	# No condition == always true
	# TODO: `changes` rule
	should_run = should_run_to_set
	break

	res = ci_rules.evaluate_rule(ci_rules.parse_rule(rule_str), variables)
	if not res:
	continue
	should_run = should_run_to_set
	break
	except SyntaxError:
	self.addCompleteLog('error', f'Rule "{rule_str}" has syntax errors')
	except:
	pass

	if should_run:
	next_steps = self.job_to_steps(job, job_index, variables)

	self.build.addStepsAfterCurrentStep(next_steps)
	return util.SUCCESS
	else:
	self.addCompleteLog('info', 'Job skipped by a rule')
	return util.SKIPPED

	@defer.inlineCallbacks
	def get_ci_variables(self, job):
	res = {}
	res.update(job.get_predefined_ci_variables())
	res.update(self.getProperty('lilybuild_pipeline_vars'))
	res['CI_JOB_IMAGE'] = job.image or self.default_image
	res['CI_JOB_URL'] = yield self.build.getUrl()
	res['CI_JOB_ID'] = self.build.buildid
	res['CI_PROJECT_DIR'] = '/build'
	try:
	repo = self.lbc.repos[self.getProperty('lilybuild_repo_id')]
	variables = yield repo['variables_getter'](self.build)
	res_vars = {}
	for var in variables:
	value = variables[var]
	if IRenderable.providedBy(value):
	value = yield self.build.render(value)
	res_vars[var] = value
	res.update(res_vars)
	except Exception as e:
	self.addCompleteLog('exception', f'{e}')
	return res

	def get_upload_artifacts_jobs(self, short_name, artifact_type, artifact_name, base_dir, paths, exclude, master_pattern, job_index, doStepIf=on_success, has_pages=False):
	archive_artifact_step = steps.ShellCommand(
	name=f'Archive artifacts: {short_name}',
	command=[
	SAFETAR_EXEC,
	],
	initialStdin=json.dumps({
	'op': 'create',
	'archive_file': artifact_name,
	'base_dir': base_dir,
	'content': paths,
	'items_to_exclude': exclude,
	'compression': 'gz',
	'limit_bytes': self.get_cur_repo_config()['artifact_uncompressed_limit'],
	}),
	workdir=self.work_root_dir,
	doStepIf=doStepIf,
	)
	masterdest = util.Interpolate(
	master_pattern,
	st=self.storage_dir,
	job=job_index,
	doStepIf=doStepIf,
	)
	parent_build_id = self.getProperty('lilybuild_pipeline_vars')['CI_PIPELINE_ID']
	artifact_url = f'{self.artifact_link_base}/plugins/lilybuild_artifacts/builds/{parent_build_id}/jobs/{job_index}/artifacts/{artifact_type}' if self.artifact_link_base else None
	upload_artifact_step = steps.FileUpload(
	workersrc=artifact_name,
	maxsize=self.get_cur_repo_config()['artifact_compressed_limit'],
	name=f'Upload artifacts: {short_name}',
	masterdest=masterdest,
	workdir=self.work_root_dir,
	url=artifact_url,
	doStepIf=doStepIf,
	)
	r = [archive_artifact_step, upload_artifact_step]
	if has_pages:
	r.append(steps.MasterShellCommand(
	command=util.Transform(fill_list,
	sys.executable,
	'-m', 'lilybuild.pages',
	util.Interpolate(
	self.master_pages_dir_pattern,
	st=self.storage_dir,
	),
	masterdest,
	),
	name='Deploy pages',
	logEnviron=False,
	doStepIf=doStepIf,
	))
	return r

	def job_to_steps(self, job, job_index, variables):
	script_name = self.script_dir + '/run.sh'

	source_step = self.lbc.create_source_step()

	script_step = steps.StringDownload(
	get_job_script(variables, job),
	name='Set up script',
	workerdest=script_name,
	workdir=self.work_root_dir,
	doStepIf=on_success,
	)

	chmod_step = steps.ShellCommand(
	name='Make script executable',
	command=['chmod', '+x', script_name],
	workdir=self.work_root_dir,
	doStepIf=on_success,
	)

	artifact_steps = []
	dep_job_indices = self.getProperty('lilybuild_dependency_job_indices')
	if dep_job_indices:
	for i in dep_job_indices:
	# The steps may not run or may not have an artifact even if it runs
	download_job = steps.FileDownload(
	mastersrc=util.Interpolate(
	self.master_job_artifact_file_name_pattern,
	st=self.storage_dir,
	job=i,
	),
	maxsize=self.get_cur_repo_config()['artifact_compressed_limit'],
	name=f'Download artifacts from job #{i}',
	workerdest=self.artifact_file_name,
	workdir=self.work_root_dir,
	doStepIf=on_success,
	haltOnFailure=False,
	flunkOnFailure=False,
	# https://github.com/buildbot/buildbot/issues/3709
	blocksize=256 * 1024,
	)

	unarchive_job = steps.ShellCommand(
	name=f'Unarchive artifacts from job #{i}',
	command=[
	SAFETAR_EXEC,
	],
	initialStdin=json.dumps({
	'op': 'extract',
	'archive_file': self.artifact_file_name,
	'target_dir': self.src_relative,
	}),
	workdir=self.work_root_dir,
	doStepIf=on_success,
	haltOnFailure=False,
	flunkOnFailure=False,
	)

	artifact_steps += [download_job, unarchive_job]

	run_step = steps.ShellCommand(
	name='Run script in container',
	command=[
	'/lilybuild/podman-helper',
	normalize_image(job.image or self.default_image),
	self.src_relative,
	self.script_dir,
	self.result_relative,
	normalize_services(job.services),
	],
	workdir=self.work_root_dir,
	doStepIf=on_success,
	# 2h timeout by default
	# TODO support timeout by each job
	timeout=60 * 60 * 2,
	)

	clean_script_step = steps.ShellCommand(
	name='Clean script dir',
	command=[
	'rm',
	'-rf',
	self.script_dir,
	],
	workdir=self.work_root_dir,
	alwaysRun=True,
	)

	steps_to_run = [source_step, script_step, chmod_step] + artifact_steps + [run_step, clean_script_step]
	if 'paths' in job.artifacts:
	steps_to_run += self.get_upload_artifacts_jobs(
	'files',
	'archive',
	self.artifact_file_name,
	self.result_relative,
	job.artifacts.get('paths', []),
	job.artifacts.get('exclude', []),
	self.master_job_artifact_file_name_pattern,
	job_index,
	has_pages=job.is_pages()
	)
	if job.has_supported_coverage_report():
	steps_to_run += [steps.ShellCommand(
	name='Process reports',
	command=[
	COVERAGE_EXEC,
	],
	initialStdin=json.dumps({
	'source_dir': self.src_relative,
	'result_dir': self.result_relative,
	'untrusted_coverage_file': job.artifacts['reports']['coverage_report']['path'],
	'output_dir': self.artifact_stage_relative,
	}),
	workdir=self.work_root_dir,
	doStepIf=on_always,
	)] + self.get_upload_artifacts_jobs(
	'reports',
	'reports',
	self.reports_file_name,
	self.artifact_stage_relative,
	['*'],
	[],
	self.master_reports_file_name_pattern,
	job_index,
	doStepIf=on_always
	) + [SendCoverageToPhorge(
	self.lbc,
	self.artifact_stage_relative + '/' + self.phorge_coverage_file_name,
	workdir=self.work_root_dir,
	)]

	clean_stage_dir_again_step = steps.ShellCommand(
	name='Clean stage, result and artifact',
	command=[
	'rm',
	'-rf',
	self.result_relative,
	self.artifact_file_name,
	self.artifact_stage_relative,
	self.reports_file_name,
	],
	workdir=self.work_root_dir,
	alwaysRun=True,
	)
	steps_to_run.append(clean_stage_dir_again_step)

	return steps_to_run

	class TriggerMultipleJobsStep(steps.Trigger):
	properties_to_keep = [
	'branch',
	'revision',
	'repository',
	'harbormaster_build_target_phid',
	'harbormaster_variable_buildable.diff',
	'harbormaster_variable_repository.staging.ref',
	'harbormaster_variable_repository.staging.uri',
	'harbormaster_variable_repository.uri',
	'lilybuild_repo',
	'lilybuild_repo_id',
	'lilybuild_pipeline_vars',
	]

	def __init__(self, lbc, jobs_with_data, **kwargs):
	self.lbc = lbc
	self.jobs = jobs_with_data
	super().__init__(schedulerNames=[self.lbc.triggerable_scheduler_name], **kwargs)

	def getSchedulersAndProperties(self):
	ret = []
	common_properties = {
	'lilybuild_root_build_id': self.build.buildid,
	}
	for prop in self.properties_to_keep:
	if self.hasProperty(prop):
	common_properties[prop] = self.getProperty(prop)

	for (job, i, dep_job_indices) in self.jobs:
	properties = common_properties.copy()
	properties['lilybuild_job_prop'] = job.to_prop()
	properties['lilybuild_job_index'] = i
	properties['virtual_builder_name'] = 'lilybuild-job - ' + common_properties['lilybuild_repo'] + ' - ' + job.name
	properties['lilybuild_dependency_job_indices'] = dep_job_indices
	ret.append({
	'sched_name': self.lbc.triggerable_scheduler_name,
	'props_to_set': properties,
	'unimportant': False,
	})

	return ret

	class LatestMixin:
	latest_build_dir_pattern = '%(kw:st)s/repos/%(prop:lilybuild_repo_id)s/latest'
	latest_good_build_dir_pattern = '%(kw:st)s/repos/%(prop:lilybuild_repo_id)s/latest-good'
	latest_name_pattern = '%(kw:st)s/repos/%(prop:lilybuild_repo_id)s/latest/%(kw:name)s'
	latest_good_name_pattern = '%(kw:st)s/repos/%(prop:lilybuild_repo_id)s/latest-good/%(kw:name)s'

	class EnsureLatestDirs(steps.MasterShellCommand, LatestMixin):
	def __init__(self, lbc, storage_dir, **kwargs):
	self.lbc = lbc
	self.storage_dir = storage_dir
	command = util.Transform(
	fill_list,
	'mkdir',
	'-pv',
	util.Interpolate(self.latest_build_dir_pattern, st=self.storage_dir),
	util.Interpolate(self.latest_good_build_dir_pattern, st=self.storage_dir))
	super().__init__(
	name='Ensure latest dirs',
	command=command,
	logEnviron=False,
	doStepIf=on_always,
	**kwargs
	)

	class MarkLatest(steps.MasterShellCommand, LatestMixin):
	def __init__(self, lbc, storage_dir, buildid, ref_name, is_good, **kwargs):
	self.lbc = lbc
	self.storage_dir = storage_dir

	name_pattern = self.latest_good_name_pattern if is_good else self.latest_name_pattern

	command = util.Transform(
	fill_list,
	'ln',
	'-sfvn',
	util.Interpolate('../builds/%(kw:buildid)s', buildid=buildid),
	util.Interpolate(name_pattern, st=self.storage_dir, name=ref_name),
	)
	super().__init__(
	name='Mark build as latest-good' if is_good else 'Mark build as latest',
	command=command,
	logEnviron=False,
	doStepIf=on_success if is_good else on_always,
	**kwargs
	)

	class AnalyzeCIFileCommand(buildstep.ShellMixin, steps.BuildStep):
	ci_def_file = '.gitlab-ci.yml'
	build_target_prop_name = 'harbormaster_build_target_phid'

	def __init__(
	self,
	lbc,
	src_relative=None,
	src_dir=None,
	storage_dir=None,
	repo_id=None,
	result_relative=None,
	result_dir=None,
	artifact_stage_relative=None,
	artifact_stage_dir=None,
	**kwargs):
	kwargs['name'] = 'Analyze CI file'
	kwargs['command'] = ['cat', self.ci_def_file]
	self.lbc = lbc
	self.src_relative = src_relative
	self.src_dir = src_dir
	self.work_root_dir = kwargs['workdir']
	self.script_dir = 'script'
	self.storage_dir = storage_dir
	self.repo_id = repo_id
	self.artifact_stage_relative = artifact_stage_relative
	self.artifact_stage_dir = artifact_stage_dir
	self.result_relative = result_relative
	self.result_dir = result_dir
	kwargs['workdir'] = self.src_dir
	kwargs = self.setupShellMixin(kwargs)
	super().__init__(**kwargs)
	self.observer = logobserver.BufferLogObserver()
	self.addLogObserver('stdio', self.observer)

	def stage_to_step(self, stage_name, stage_jobs, job_name_to_index_map, ci_file):
	jobs_with_data = []
	for job in stage_jobs:
	dep_job_names = [
	jn
	for jn in ci_file.get_jobs_to_pull_artifacts_from(job.name)
	if ci_file.jobs[jn].has_artifacts_archive()
	]
	dep_job_indices = [job_name_to_index_map[jn] for jn in dep_job_names]
	jobs_with_data.append((job, job_name_to_index_map[job.name], dep_job_indices))

	trigger = TriggerMultipleJobsStep(
	name=stage_name,
	lbc=self.lbc,
	jobs_with_data=jobs_with_data,
	waitForFinish=True,
	doStepIf=on_success,
	)

	return trigger

	def get_steps_and_job_map(self, stdout):
	f = ci_file.CIFile(stdout)

	stages = f.get_grouped_jobs()
	jobs = [job for (stage, js) in f.get_grouped_jobs() for job in js]
	job_names = [job.name for job in jobs]
	job_name_to_index_map = {}
	for (i, j) in enumerate(jobs):
	job_name_to_index_map[j.name] = i

	steps = [self.stage_to_step(stage_name, stage_jobs, job_name_to_index_map, f) for (stage_name, stage_jobs) in stages]
	print('steps:', steps)
	return (steps, job_name_to_index_map)

	def get_is_phorge(self):
	return not not self.getProperty(self.build_target_prop_name)

	def get_ref_and_type(self):
	ref_type = 'branch'
	ref = self.getProperty('branch')

	if self.getProperty('category') == 'tag':
	ref_type = 'tag'

	if ref is not None:
	m = re.match(r'^refs/(heads\|tags)/(.+)$', ref)
	if m:
	ref = m.group(2)

	return (ref, ref_type)

	def get_cur_repo_config(self):
	return self.lbc.repos[self.getProperty('lilybuild_repo_id')]

	@defer.inlineCallbacks
	def get_pipeline_ci_vars(self):
	url = yield self.build.getUrl()
	res = {
	'CI_PIPELINE_ID': self.build.buildid,
	'CI_PIPELINE_IID': self.build.buildid,
	'CI_PIPELINE_URL': url,
	'CI_PROJECT_ID': self.getProperty('lilybuild_repo_id'),
	'CI_CONFIG_PATH': self.ci_def_file,
	}
	if not self.get_is_phorge():
	res['CI_COMMIT_SHA'] = self.getProperty('got_revision')
	res['CI_COMMIT_SHORT_SHA'] = res['CI_COMMIT_SHA'][:8]
	(ref, ref_type) = self.get_ref_and_type()
	res['CI_COMMIT_REF_NAME'] = ref
	res['CI_COMMIT_REF_SLUG'] = ci_file.ci_slugify(ref)
	res['CI_COMMIT_REF_PROTECTED'] = 'false'
	if ref_type == 'tag':
	res['CI_COMMIT_TAG'] = ref
	elif ref_type == 'branch':
	res['CI_COMMIT_BRANCH'] = ref
	return res

	@defer.inlineCallbacks
	def run(self):
	# run './build.sh --list-stages' to generate the list of stages
	cmd = yield self.makeRemoteShellCommand()
	yield self.runCommand(cmd)

	# if the command passes extract the list of stages
	result = cmd.results()
	if result == util.SUCCESS:
	pipeline_vars = yield self.get_pipeline_ci_vars()
	self.setProperty('lilybuild_pipeline_vars', pipeline_vars, self.__class__.__name__)
	# create a ShellCommand for each stage and add them to the build
	(steps, job_map) = self.get_steps_and_job_map(self.observer.getStdout())
	self.setProperty('lilybuild_job_map', job_map, self.__class__.__name__)
	self.build.addStepsAfterCurrentStep(steps)

	latest_branch_map = self.get_cur_repo_config()['artifact_latest_branch_map']
	ref, _ref_type = self.get_ref_and_type()
	if ref in latest_branch_map:
	latest_name = latest_branch_map[ref]
	self.build.addStepsAfterLastStep([
	EnsureLatestDirs(lbc=self.lbc, storage_dir=self.storage_dir),
	MarkLatest(
	lbc=self.lbc,
	storage_dir=self.storage_dir,
	buildid=self.build.buildid,
	ref_name=latest_name,
	is_good=True,
	),
	MarkLatest(
	lbc=self.lbc,
	storage_dir=self.storage_dir,
	buildid=self.build.buildid,
	ref_name=latest_name,
	is_good=False,
	),
	])

	return result
	diff --git a/lilybuild/lilybuild/ci_syntax/ci_file.py b/lilybuild/lilybuild/ci_syntax/ci_file.py
	index 6417502..46c0f8a 100644
	--- a/lilybuild/lilybuild/ci_syntax/ci_file.py
	+++ b/lilybuild/lilybuild/ci_syntax/ci_file.py
	@@ -1,241 +1,281 @@

	import yaml
	import jsonschema
	import json
	import os
	import re

	# https://gitlab.com/gitlab-org/gitlab/-/blob/master/app/assets/javascripts/editor/schema/ci.json
	schema_file = os.path.join(os.path.dirname(__file__), 'ci.json')
	schema = None
	extend_limit = 11

	class CIValidationError(Exception):
	pass

	def get_schema():
	global schema
	if schema is not None:
	return schema
	else:
	with open(schema_file, 'r') as f:
	schema = json.loads(f.read())
	return schema

	def toplevel_entries():
	return get_schema()['properties']

	def normalize_script(script):
	if script is None:
	return []
	elif isinstance(script, str):
	return [script]
	else:
	res = []
	for l in script:
	if isinstance(l, str):
	res.append(l)
	else:
	res += l
	return res

	slug_re = re.compile('[^0-9a-z]')
	def ci_slugify(s):
	return re.sub(slug_re, '-', s.lower()[:63]).strip('-')

	def get_job_extend_seq(job_name, all_jobs, depth=0):
	# DFS traversal; child-first, self-last order
	res = []
	if job_name not in all_jobs:
	raise CIValidationError(f'Job "{job_name}" does not exist, but occurs in `extends`.')
	job_struct = all_jobs.get(job_name) or {}
	parents = job_struct.get('extends', [])
	if depth > extend_limit:
	raise CIValidationError(f'`extends` depth is over the limit of {extend_limit}.')
	if isinstance(parents, str):
	parents = [parents]
	for p in parents:
	res += get_job_extend_seq(p, all_jobs, depth + 1)
	res.append(job_name)

	return res

	depth_limit = 20
	def merge_job_deep(res, parent, depth=0):
	res_keys = set(res.keys())
	if depth > depth_limit:
	raise CIValidationError(f'depth is over the limit when merging job.')
	for k in parent:
	if k not in res_keys:
	res[k] = parent[k]
	elif isinstance(parent[k], dict) and isinstance(res[k], dict):
	child = res[k]
	# Necessary to avoid changing anything inside res[k] that was shallow-copied
	res[k] = {}
	merge_job_deep(res[k], child, depth + 1)
	merge_job_deep(res[k], parent[k], depth + 1)
	else:
	# Presenting in both parent and child, and it's not a dict,
	# so the child should take preference.
	pass

	def expand_job(job_name, all_jobs, defaults):
	seq = get_job_extend_seq(job_name, all_jobs)
	res = {}
	for ancestor_name in reversed(seq):
	merge_job_deep(res, all_jobs[ancestor_name])
	merge_job_deep(res, defaults)
	if 'extends' in res:
	del res['extends']
	return res

	+def get_ref_from_ref_cond(ref_cond):
	+ # We don't support project name so ignore everything after @
	+ pos = ref_cond.rfind('@')
	+ if pos != -1:
	+ ref_cond = ref_cond[:pos]
	+ if ref_cond[0] == '/' and ref_cond[-1] == '/':
	+ return '$CI_COMMIT_REF_NAME =~ ' + ref_cond
	+ else:
	+ return '$CI_COMMIT_REF_NAME == "' + ref_cond.replace('\\', '\\\\').replace('"', '\\"') + '"'
	+
	+def get_alt_rules_from_job_struct(job_struct):
	+ # Convert when/only/except to rules
	+ when = None
	+ if job_struct.get('when'):
	+ when = job_struct.get('when')
	+ positive_cond = []
	+ if job_struct.get('only'):
	+ for ref_cond in job_struct.get('only'):
	+ positive_cond.append(get_ref_from_ref_cond(ref_cond))
	+ negative_cond = []
	+ if job_struct.get('except'):
	+ for ref_cond in job_struct.get('except'):
	+ negative_cond.append(get_ref_from_ref_cond(ref_cond))
	+ rule_if = ''
	+ if positive_cond:
	+ rule_if += '(' + ' \|\| '.join(positive_cond) + ')'
	+ if negative_cond:
	+ rule_if += ' && '
	+ if negative_cond:
	+ rule_if += '!(' + ' \|\| '.join(negative_cond) + ')'
	+ rule = {}
	+ if when:
	+ rule['when'] = when
	+ if positive_cond or negative_cond:
	+ rule['if'] = rule_if
	+ if rule:
	+ return [rule]
	+ else:
	+ return []
	+
	class CIJob:
	def __init__(self, job_name, job_stage, job_struct):
	self.name = job_name
	self.stage = job_stage
	self.struct_raw = job_struct
	self.image = job_struct.get('image')
	self.before_script = normalize_script(job_struct.get('before_script'))
	self.script = normalize_script(job_struct.get('script'))
	self.after_script = normalize_script(job_struct.get('after_script'))
	self.artifacts = job_struct.get('artifacts') or {}
	- self.rules = job_struct.get('rules') or []
	+ self.rules = job_struct.get('rules') or get_alt_rules_from_job_struct(job_struct)
	self.dependencies = job_struct.get('dependencies')
	self.services = job_struct.get('services') or []

	def get_predefined_ci_variables(self):
	vs = {
	'CI': 'true',
	'CI_JOB_NAME': self.name,
	'CI_JOB_NAME_SLUG': ci_slugify(self.name),
	'CI_JOB_STAGE': self.stage,
	}
	vs.update(self.struct_raw.get('variables', {}))
	return vs

	def has_artifacts_archive(self):
	return self.artifacts and 'paths' in self.artifacts

	def to_prop(self):
	return {
	'name': self.name,
	'stage': self.stage,
	'struct_raw': self.struct_raw,
	}

	@classmethod
	def from_prop(cls, prop):
	return cls(
	prop['name'],
	prop['stage'],
	prop['struct_raw']
	)

	def is_pages(self):
	return self.name == 'pages' or self.struct_raw.get('pages', False)

	def has_supported_coverage_report(self):
	return (
	'reports' in self.artifacts
	and self.artifacts['reports'].get('coverage_report')
	and self.artifacts['reports']['coverage_report'].get('coverage_format') == 'cobertura'
	and self.artifacts['reports']['coverage_report'].get('path')
	)

	OLD_TOPLEVEL_DEFAULTS = ['image', 'services', 'cache', 'before_script', 'after_script']
	DEFAULT_STAGES = ['.pre', 'build', 'test', 'deploy', '.post']
	DEFAULT_JOB_STAGE = 'test'

	class CIFile:
	'''
	Class for parsing CI file.
	'''

	def __init__(self, file_content):
	'''
	Construct a CI File from its text content.
	'''
	f = yaml.safe_load(file_content)
	if f is None:
	self.stages = []
	self.jobs = {}
	return

	jsonschema.validate(instance=f, schema=get_schema())

	self.stages = f.get('stages', DEFAULT_STAGES)
	self.jobs = {}

	defaults = f.get('default', {})
	if f.get('variables'):
	defaults['variables'] = f['variables']
	for kw in OLD_TOPLEVEL_DEFAULTS:
	if kw not in defaults and kw in f:
	defaults[kw] = f[kw]

	all_jobs = {}
	for job_name, job_struct in f.items():
	# 'pages' is a special job
	if job_name != 'pages' and job_name in toplevel_entries():
	continue
	all_jobs[job_name] = job_struct

	for job_name in all_jobs:
	# jobs starting with . will only be used for base jobs of other jobs,
	# they themselves are not run
	if job_name.startswith('.'):
	continue
	job_struct = expand_job(job_name, all_jobs, defaults)
	job_stage = job_struct.get('stage', DEFAULT_JOB_STAGE)
	if job_stage not in self.stages:
	raise CIValidationError(f'Job "{job_name}": Stage "{job_stage}" is not specified in CI file')
	self.jobs[job_name] = CIJob(job_name, job_stage, job_struct)

	self.validate_logic()

	def validate_logic(self):
	for job_name in self.jobs:
	self.validate_job(job_name)

	def validate_job(self, job_name):
	j = self.jobs[job_name]
	my_stage_order = self.stage_order(job_name)
	if j.dependencies is not None:
	for d in j.dependencies:
	try:
	stage_order = self.stage_order(d)
	except KeyError:
	raise CIValidationError(f'Dependency "{d}" of job "{job_name}" does not exist')
	if stage_order >= my_stage_order:
	raise CIValidationError(f'Dependency "{d}" of job "{job_name}" is not before the job in stage')

	def stage_order(self, jn):
	return self.stages.index(self.jobs[jn].stage)

	def get_grouped_jobs(self):
	groups = {}
	for job_name in self.jobs:
	job = self.jobs[job_name]
	if job.stage not in groups:
	groups[job.stage] = []
	groups[job.stage].append(job)
	res = []
	for stage in self.stages:
	if stage in groups:
	res.append((stage, groups[stage]))
	return res

	def get_jobs_to_pull_artifacts_from(self, job_name):
	'''
	Get a list of names of jobs of which the artifacts will be pulled
	from for the job named `job_name`.
	'''
	dependencies = self.jobs[job_name].dependencies
	if dependencies is None:
	dependencies = []
	cur_job_stage_order = self.stage_order(job_name)
	for jn in self.jobs:
	so = self.stage_order(jn)
	if so < cur_job_stage_order:
	dependencies.append(jn)
	return dependencies
	diff --git a/lilybuild/lilybuild/ci_syntax/rules.py b/lilybuild/lilybuild/ci_syntax/rules.py
	index d50e431..afcc55d 100644
	--- a/lilybuild/lilybuild/ci_syntax/rules.py
	+++ b/lilybuild/lilybuild/ci_syntax/rules.py
	@@ -1,122 +1,132 @@

	import re

	# XXX: Since we are stuck with Python 3.9, we can't disable backtrack yet.
	# https://docs.python.org/3/library/re.html
	patterns = {
	'OP': re.compile(r'\s*(==\|\\|\\|\|&&\|=~\|\!=\|\!~\|\!)'),
	'PAREN_LEFT': re.compile(r'\s*\('),
	'PAREN_RIGHT': re.compile(r'\s*\)'),
	'STR_DOUBLE': re.compile(r'''\s"((?:[^\\"]\|\\.))"'''),
	'STR_SINGLE': re.compile(r"""\s'((?:[^\\']\|\\.))'"""),
	+ 'REGEX': re.compile(r'''\s/((?:[^\\/]\|\\.))/'''),
	'VAR': re.compile(r'\s\$([A-Za-z_][A-Za-z0-9_])'),
	'NULL': re.compile(r'\s*null'),
	'END': re.compile(r'\s*$'),
	}

	ops = {
	'==': (2, lambda a, b: a == b),
	'!=': (2, lambda a, b: a != b),
	- '=~': (2, lambda a, b: False),
	- '!~': (2, lambda a, b: False),
	+ '=~': (2, lambda a, b: not not re.search(b, a)),
	+ '!~': (2, lambda a, b: not re.search(b, a)),
	'!': (1, lambda a: not a),
	'&&': (2, lambda a, b: a and b),
	'\|\|': (2, lambda a, b: a or b),
	}
	TERM_PRECEDENCE = 9
	PAREN_PRECEDENCE = 8

	def is_term(token):
	- return token[0] in ['STR_DOUBLE', 'STR_SINGLE', 'VAR', 'NULL']
	+ return token[0] in ['STR_DOUBLE', 'STR_SINGLE', 'REGEX', 'VAR', 'NULL']

	def get_precedence(token):
	if is_term(token):
	return TERM_PRECEDENCE
	elif token[0] in ['PAREN_LEFT', 'PAREN_RIGHT']:
	return PAREN_PRECEDENCE
	elif token[0] == 'OP' and (token[1][0] in ['==', '!=', '=~', '!~']):
	return 6
	elif token[0] == 'OP' and (token[1][0] in ['!']):
	return 5
	elif token[0] == 'OP' and (token[1][0] in ['&&']):
	return 4
	elif token[0] == 'OP' and (token[1][0] in ['\|\|']):
	return 3
	raise SyntaxError('Unknown token')

	def tokenize_rule(rule_str):
	pos = 0
	tokenized = []
	while not (len(tokenized) and tokenized[-1][0] == 'END'):
	match = None
	for t in patterns:
	regex = patterns[t]
	match = regex.match(rule_str, pos)
	if match:
	tokenized.append((t, match.groups()))
	pos = match.end()
	break
	if not match:
	raise SyntaxError(f'Bad rule "{rule_str}", at pos {pos}')
	return tokenized[:-1]

	def make_tree(tokenized):
	# https://en.wikipedia.org/wiki/Shunting_yard_algorithm
	stack = []
	res = []
	for t in tokenized:
	if is_term(t):
	res.append(t)
	elif t[0] == 'OP':
	if len(stack) and stack[-1][0] == 'OP' and get_precedence(stack[-1]) >= get_precedence(t):
	res.append(stack.pop())
	stack.append(t)
	elif t[0] == 'PAREN_LEFT':
	stack.append(t)
	elif t[0] == 'PAREN_RIGHT':
	while len(stack) and stack[-1][0] != 'PAREN_LEFT':
	res.append(stack.pop())
	if not len(stack):
	raise SyntaxError('Mismatched parentheses')
	stack.pop()
	while len(stack):
	t = stack.pop()
	if t[0] == 'PAREN_LEFT':
	raise SyntaxError('Mismatched parentheses')
	res.append(t)

	stack = []
	for t in res:
	if t[0] == 'OP':
	opname = t[1][0]
	arity = ops[opname][0]
	if len(stack) < arity:
	raise SyntaxError('Missing operands')
	operands = tuple(reversed([stack.pop() for i in range(arity)]))
	stack.append((t[0], (opname,) + operands))
	else:
	stack.append(t)
	if len(stack) != 1:
	raise SyntaxError('Too many operands')
	return stack[0]

	def parse_rule(rule_str):
	return make_tree(tokenize_rule(rule_str))

	backslash_re = re.compile(r'\\(.)')
	def replace_backslash(match):
	c = match.groups()[0]
	return c

	+def replace_backslash_in_regex(match):
	+ c = match.groups()[0]
	+ if c == '/':
	+ return c
	+ # Treat everything except \/ as is
	+ return '\\' + c
	+
	def evaluate_rule(expr, variables):
	if expr[0] == 'VAR':
	return variables.get(expr[1][0])
	elif expr[0] == 'STR_DOUBLE' or expr[0] == 'STR_SINGLE':
	return backslash_re.sub(replace_backslash, expr[1][0])
	+ elif expr[0] == 'REGEX':
	+ return backslash_re.sub(replace_backslash_in_regex, expr[1][0])
	elif expr[0] == 'NULL':
	return None
	elif expr[0] == 'OP':
	opname = expr[1][0]
	operands = [evaluate_rule(o, variables) for o in expr[1][1:]]
	return ops[opname][1](*operands)
	raise SyntaxError(f'Cannot evaluate expression {expr}')
	diff --git a/lilybuild/lilybuild/tests/ci_syntax/ci_file_test.py b/lilybuild/lilybuild/tests/ci_syntax/ci_file_test.py
	index bffd8c6..504aa2f 100644
	--- a/lilybuild/lilybuild/tests/ci_syntax/ci_file_test.py
	+++ b/lilybuild/lilybuild/tests/ci_syntax/ci_file_test.py
	@@ -1,315 +1,333 @@

	import unittest
	import yaml
	from lilybuild.ci_syntax.ci_file import CIFile, CIValidationError, get_job_extend_seq, merge_job_deep
	from lilybuild.tests.resources import get_res

	class CIFileTest(unittest.TestCase):
	def test_empty(self):
	r = CIFile('')
	self.assertEqual(r.stages, [])
	self.assertEqual(r.jobs, {})

	def test_invalid(self):
	with self.assertRaises(yaml.composer.ComposerError) as m:
	r = CIFile('*xxx')

	def test_simple(self):
	r = CIFile(get_res('simple'))
	self.assertEqual(r.stages, ['build'])
	self.assertEqual(list(r.jobs), ['build-a'])
	self.assertEqual(r.jobs['build-a'].image, 'test')
	self.assertEqual(r.jobs['build-a'].before_script, ['ls'])
	self.assertEqual(r.jobs['build-a'].script, ['make', 'make install'])
	self.assertEqual(r.jobs['build-a'].after_script, ['find', 'echo ok'])

	def test_noscript(self):
	r = CIFile(get_res('noscript'))

	self.assertEqual(r.stages, ['build'])
	self.assertEqual(list(r.jobs), ['build-a'])
	self.assertEqual(r.jobs['build-a'].image, 'test')
	self.assertEqual(r.jobs['build-a'].script, [])

	def test_defaults(self):
	r = CIFile(get_res('defaults'))
	self.assertEqual(r.stages, ['build'])
	self.assertEqual(list(r.jobs), ['build-a'])
	self.assertEqual(r.jobs['build-a'].image, 'test')
	self.assertEqual(r.jobs['build-a'].before_script, ['ls'])
	self.assertEqual(r.jobs['build-a'].script, ['make', 'make install'])
	self.assertEqual(r.jobs['build-a'].after_script, ['find', 'echo ok'])

	def test_pages(self):
	r = CIFile(get_res('pages'))
	self.assertEqual(r.stages, ['test'])
	self.assertEqual(list(r.jobs), ['pages'])
	self.assertEqual(r.jobs['pages'].script, ['make docs'])
	self.assertTrue(r.jobs['pages'].is_pages())

	def test_pages_attr(self):
	r = CIFile(get_res('pages_attr'))
	self.assertFalse(r.jobs['not-pages'].is_pages())
	self.assertTrue(r.jobs['is-pages'].is_pages())

	def test_dotjobs(self):
	r = CIFile(get_res('dotjobs'))
	self.assertEqual(list(r.jobs), [])

	def test_no_such_stage(self):
	with self.assertRaises(CIValidationError) as m:
	r = CIFile(get_res('no_such_stage'))

	def test_default_stages(self):
	r = CIFile(get_res('default_stages'))
	self.assertEqual(r.jobs['a'].stage, 'build')
	self.assertEqual(r.jobs['b'].stage, 'test')
	self.assertEqual(r.jobs['c'].stage, 'deploy')
	self.assertEqual(r.jobs['d'].stage, '.pre')
	self.assertEqual(r.jobs['e'].stage, '.post')
	self.assertEqual(r.jobs['f'].stage, 'test')

	def test_group_jobs(self):
	r = CIFile(get_res('group_jobs'))
	groups = r.get_grouped_jobs()
	self.assertEqual(len(groups), 2)
	self.assertEqual(groups[0][0], 'build')
	self.assertEqual(len(groups[0][1]), 2)
	self.assertTrue(next(j for j in groups[0][1] if j.name == 'a'))
	self.assertTrue(next(j for j in groups[0][1] if j.name == 'b'))
	self.assertEqual(groups[1][0], 'test')
	self.assertEqual(len(groups[1][1]), 2)
	self.assertTrue(next(j for j in groups[1][1] if j.name == 'c'))
	self.assertTrue(next(j for j in groups[1][1] if j.name == 'd'))

	def test_dependencies(self):
	r = CIFile(get_res('dependencies'))
	self.assertEqual(r.get_jobs_to_pull_artifacts_from('a'), [])
	self.assertEqual(r.get_jobs_to_pull_artifacts_from('a2'), [])
	self.assertEqual(r.get_jobs_to_pull_artifacts_from('b'), ['a'])
	self.assertEqual(set(r.get_jobs_to_pull_artifacts_from('b2')), set(['a', 'a2']))

	def test_nonexistent_dep(self):
	with self.assertRaises(CIValidationError) as m:
	r = CIFile(get_res('nonexistent_dep'))
	self.assertEqual(str(m.exception), 'Dependency "a3" of job "b" does not exist')

	def test_late_dep(self):
	with self.assertRaises(CIValidationError) as m:
	r = CIFile(get_res('late_dep'))
	self.assertEqual(str(m.exception), 'Dependency "a2" of job "a" is not before the job in stage')

	def test_artifacts(self):
	r = CIFile(get_res('has_artifacts_archive'))
	self.assertEqual(r.jobs['has_archive'].artifacts, { 'paths': ['a'] })
	self.assertEqual(r.jobs['no_archive'].artifacts, {})
	self.assertEqual(r.jobs['no_artifacts'].artifacts, {})

	def test_has_artifacts_archive(self):
	r = CIFile(get_res('has_artifacts_archive'))
	self.assertTrue(r.jobs['has_archive'].has_artifacts_archive())
	self.assertFalse(r.jobs['no_archive'].has_artifacts_archive())
	self.assertFalse(r.jobs['no_artifacts'].has_artifacts_archive())

	def test_artifacts_reports(self):
	r = CIFile(get_res('artifacts_reports'))
	self.assertFalse(r.jobs['has_archive'].has_supported_coverage_report())
	self.assertFalse(r.jobs['empty_reports'].has_supported_coverage_report())
	self.assertTrue(r.jobs['yes_reports'].has_supported_coverage_report())
	self.assertFalse(r.jobs['no_supported_reports'].has_supported_coverage_report())

	def test_extends(self):
	r = CIFile(get_res('extends'))
	self.assertEqual(r.jobs['build-a'].struct_raw, {
	'image': 'alpine',
	'variables': { 'SECRET': 'abcdef', 'CC': 'gcc' },
	'after_script': ['rm -r cache'],
	'stage': 'build',
	'before_script': ['apk add gcc'],
	'script': ['make a', 'make install'],
	})
	self.assertEqual(r.jobs['build-b'].struct_raw, {
	'image': 'alpine',
	'variables': { 'SECRET': 'abcdef', 'CC': 'cc' },
	'after_script': ['rm -r cache'],
	'stage': 'build',
	'before_script': ['apk add gcc'],
	'script': ['make b', 'make install'],
	})
	self.assertEqual(r.jobs['test'].struct_raw, {
	'image': 'tester',
	'variables': { 'SECRET': 'abcdef' },
	'after_script': ['rm -r cache'],
	'stage': 'test',
	'script': ['make test'],
	})
	self.assertEqual(r.jobs['container-amd64'].struct_raw, {
	'image': 'podman',
	'variables': { 'SECRET': 'abcdef', 'DOCKER': 'podman', 'PLATFORM': 'amd64', 'LOGIN': 'defghi' },
	'after_script': ['rm -r cache'],
	'stage': 'deploy',
	'script': ['podman login', 'make docker'],
	'tags': ['podman']
	})
	self.assertEqual(r.jobs['container-aarch64'].struct_raw, {
	'image': 'podman',
	'variables': { 'SECRET': 'abcdef', 'DOCKER': 'podman', 'PLATFORM': 'aarch64', 'LOGIN': 'defghi' },
	'after_script': ['rm -r cache'],
	'stage': 'deploy',
	'script': ['podman login', 'make docker'],
	'tags': ['podman']
	})
	self.assertEqual(r.jobs['appimage-amd64'].struct_raw, {
	'image': 'alpine',
	'variables': { 'SECRET': 'abcdef', 'PLATFORM': 'amd64', 'LOGIN': 'defghi' },
	'after_script': ['rm -r cache'],
	'stage': 'deploy',
	'script': ['make appimage'],
	'tags': ['amd64']
	})
	self.assertEqual(r.jobs['appimage-aarch64'].struct_raw, {
	'image': 'alpine',
	'variables': { 'SECRET': 'abcdef', 'PLATFORM': 'aarch64', 'LOGIN': 'defghi' },
	'after_script': ['rm -r cache'],
	'stage': 'deploy',
	'script': ['make appimage'],
	'tags': ['aarch64']
	})

	def test_variables(self):
	r = CIFile(get_res('variables'))
	self.assertEqual(r.jobs['build-a'].get_predefined_ci_variables()['foo'], 'bar')

	+ def test_when_only_except(self):
	+ r = CIFile(get_res('when-only-except'))
	+ expected_if = r'''($CI_COMMIT_REF_NAME =~ /a/ \|\| $CI_COMMIT_REF_NAME =~ /bb$/ \|\| $CI_COMMIT_REF_NAME == "ccccccc" \|\| $CI_COMMIT_REF_NAME == "ddd" \|\| $CI_COMMIT_REF_NAME == "ff\\\\\"f" \|\| $CI_COMMIT_REF_NAME == "ff\\'f" \|\| $CI_COMMIT_REF_NAME == "ff\\\"") && !($CI_COMMIT_REF_NAME == "bbb" \|\| $CI_COMMIT_REF_NAME =~ /^ba/ \|\| $CI_COMMIT_REF_NAME =~ /ff\w/ \|\| $CI_COMMIT_REF_NAME =~ /@@@/ \|\| $CI_COMMIT_REF_NAME == "@@@")'''
	+ self.assertEqual(r.jobs['a'].rules, [{
	+ 'when': 'manual',
	+ 'if': expected_if
	+ }])
	+ self.assertEqual(r.jobs['b'].rules, [{
	+ 'if': '($CI_COMMIT_REF_NAME == "servant")'
	+ }])
	+ self.assertEqual(r.jobs['c'].rules, [{
	+ 'if': '!($CI_COMMIT_REF_NAME == "servant")'
	+ }])
	+ self.assertEqual(r.jobs['d'].rules, [])
	+ self.assertEqual(r.jobs['e'].rules, [{
	+ 'when': 'always'
	+ }])
	+
	class GetJobExtendSeqTest(unittest.TestCase):
	def test_no_extends(self):
	all_jobs = {
	'nothing': {}
	}
	self.assertEqual(get_job_extend_seq('nothing', all_jobs), ['nothing'])

	def test_multi_extends(self):
	all_jobs = {
	'a': { 'extends': ['b', 'c'] },
	'b': { 'extends': ['d', 'e'] },
	'c': { 'extends': ['f', 'g'] },
	'd': {},
	'e': {},
	'f': {},
	'g': {},
	}
	self.assertEqual(get_job_extend_seq('a', all_jobs), ['d', 'e', 'b', 'f', 'g', 'c', 'a'])

	def test_common_ancestor(self):
	all_jobs = {
	'a': { 'extends': ['b', 'c'] },
	'b': { 'extends': ['d', 'e'] },
	'c': { 'extends': ['e', 'd'] },
	'd': {},
	'e': {},
	'f': {},
	'g': {},
	}
	self.assertEqual(get_job_extend_seq('a', all_jobs), ['d', 'e', 'b', 'e', 'd', 'c', 'a'])

	def test_self_cycle(self):
	all_jobs = {
	'a': { 'extends': ['a'] },
	}
	with self.assertRaises(CIValidationError) as m:
	r = get_job_extend_seq('a', all_jobs)
	self.assertTrue(str(m.exception).startswith('`extends` depth is over the limit of'))

	def test_complex_cycle(self):
	all_jobs = {
	'a': { 'extends': ['b', 'c'] },
	'b': { 'extends': ['d', 'c'] },
	'c': { 'extends': ['d'] },
	'd': { 'extends': ['a'] },
	}
	with self.assertRaises(CIValidationError) as m:
	r = get_job_extend_seq('a', all_jobs)
	self.assertTrue(str(m.exception).startswith('`extends` depth is over the limit of'))

	def test_nonexistent_parent(self):
	all_jobs = {
	'a': { 'extends': ['b', 'c'] },
	'b': { 'extends': ['d', 'c'] },
	'c': { 'extends': ['d'] },
	}
	with self.assertRaises(CIValidationError) as m:
	r = get_job_extend_seq('a', all_jobs)
	self.assertEqual(str(m.exception), 'Job "d" does not exist, but occurs in `extends`.')

	def test_null_parent(self):
	all_jobs = {
	'a': { 'extends': ['b', 'c'] },
	'b': { 'extends': ['d', 'c'] },
	'c': { 'extends': ['d'] },
	'd': None,
	}
	self.assertEqual(get_job_extend_seq('a', all_jobs), ['d', 'd', 'c', 'b', 'd', 'c', 'a'])

	class MergeJobTest(unittest.TestCase):
	def test_simple(self):
	parent = {
	'a': {
	'a/1': {
	'a/1/1': '3',
	'a/1/2': 4,
	},
	'a/2': {},
	},
	'b': 1,
	}
	child = {
	'a': {
	'a/1': {
	'a/1/2': '6',
	'a/1/3': 5,
	},
	'a/2': 'mew',
	},
	'c': 'test',
	}
	merge_job_deep(child, parent)
	self.assertEqual(parent, {
	'a': {
	'a/1': {
	'a/1/1': '3',
	'a/1/2': 4,
	},
	'a/2': {},
	},
	'b': 1,
	})
	self.assertEqual(child, {
	'a': {
	'a/1': {
	'a/1/1': '3',
	'a/1/2': '6',
	'a/1/3': 5,
	},
	'a/2': 'mew',
	},
	'b': 1,
	'c': 'test',
	})

	def test_no_overflow(self):
	a = {
	'1': '2',
	}
	b = {
	'2': '4',
	'a': a,
	}
	a['a'] = b

	res = {}
	merge_job_deep(res, b)
	with self.assertRaises(CIValidationError) as m:
	merge_job_deep(res, a)
	self.assertEqual(str(m.exception), 'depth is over the limit when merging job.')

	if __name__ == '__main__':
	unittest.main()
	diff --git a/lilybuild/lilybuild/tests/ci_syntax/res/when-only-except.yaml b/lilybuild/lilybuild/tests/ci_syntax/res/when-only-except.yaml
	new file mode 100644
	index 0000000..0a1c655
	--- /dev/null
	+++ b/lilybuild/lilybuild/tests/ci_syntax/res/when-only-except.yaml
	@@ -0,0 +1,31 @@
	+
	+a:
	+ when: manual
	+ only:
	+ - /a/
	+ - /bb$/@abcabc/defdef
	+ - ccccccc@abcabc/defdef
	+ - ddd
	+ - 'ff\\"f'
	+ - "ff\\'f"
	+ - 'ff\"'
	+ except:
	+ - bbb
	+ - /^ba/
	+ - /ff\w/@abcabc/defdef
	+ - /@@@/@abcabc/defdef
	+ - '@@@@abcabc/defdef'
	+
	+b:
	+ only:
	+ - servant
	+
	+c:
	+ except:
	+ - servant
	+
	+d:
	+ image: alpine
	+
	+e:
	+ when: always
	diff --git a/lilybuild/lilybuild/tests/ci_syntax/rules_test.py b/lilybuild/lilybuild/tests/ci_syntax/rules_test.py
	index f6d58f1..9cf31bb 100644
	--- a/lilybuild/lilybuild/tests/ci_syntax/rules_test.py
	+++ b/lilybuild/lilybuild/tests/ci_syntax/rules_test.py
	@@ -1,118 +1,143 @@

	import unittest
	from lilybuild.ci_syntax.rules import tokenize_rule, make_tree, parse_rule, evaluate_rule

	class RulesTest(unittest.TestCase):
	def test_tokenize(self):
	self.assertEqual(tokenize_rule("$VAR_FOO == ''"), [
	('VAR', ('VAR_FOO',)),
	('OP', ('==',)),
	('STR_SINGLE', ('',)),
	])
	+ self.assertEqual(tokenize_rule("$VAR_FOO =~ /abc/"), [
	+ ('VAR', ('VAR_FOO',)),
	+ ('OP', ('=~',)),
	+ ('REGEX', ('abc',)),
	+ ])
	+ self.assertEqual(tokenize_rule(r"$VAR_FOO =~ /ab\s\/c/"), [
	+ ('VAR', ('VAR_FOO',)),
	+ ('OP', ('=~',)),
	+ ('REGEX', ('ab\\s\\/c',)),
	+ ])
	self.assertEqual(tokenize_rule(r'$VAR_FOO == "xxx\""'), [
	('VAR', ('VAR_FOO',)),
	('OP', ('==',)),
	('STR_DOUBLE', ('xxx\\"',)),
	])
	self.assertEqual(tokenize_rule(r'!($VAR_FOO == "xxx\"")'), [
	('OP', ('!',)),
	('PAREN_LEFT', ()),
	('VAR', ('VAR_FOO',)),
	('OP', ('==',)),
	('STR_DOUBLE', ('xxx\\"',)),
	('PAREN_RIGHT', ()),
	])
	with self.assertRaises(SyntaxError) as m:
	tokenize_rule(r'$VAR_FOO =/= "foo"')
	with self.assertRaises(SyntaxError) as m:
	tokenize_rule(r'$VAR_FOO > "foo"')
	# bareword
	with self.assertRaises(SyntaxError) as m:
	tokenize_rule(r'$VAR_FOO == foo')
	# unterminated strings
	with self.assertRaises(SyntaxError) as m:
	tokenize_rule(r'$VAR_FOO == "foo\"')
	with self.assertRaises(SyntaxError) as m:
	tokenize_rule(r'$VAR_FOO == "foo')
	with self.assertRaises(SyntaxError) as m:
	tokenize_rule(r'$VAR_FOO == "foo\\\"')
	+ # bareword after regex
	+ with self.assertRaises(SyntaxError) as m:
	+ tokenize_rule(r"$VAR_FOO =~ /ab\s\/c/d")

	def test_make_tree(self):
	self.assertEqual(make_tree(tokenize_rule(r'($FOO)')), ('VAR', ('FOO',)))
	self.assertEqual(make_tree(tokenize_rule(r'($FOO=="foo(") \|\| "xxx"')), (
	'OP', (
	'\|\|',
	('OP', ('==', ('VAR', ('FOO',)), ('STR_DOUBLE', ('foo(',)))),
	('STR_DOUBLE', ('xxx',)),
	)
	))
	self.assertEqual(make_tree(tokenize_rule(r'$FOO=="foo("\|\| "xxx"')), (
	'OP', (
	'\|\|',
	('OP', ('==', ('VAR', ('FOO',)), ('STR_DOUBLE', ('foo(',)))),
	('STR_DOUBLE', ('xxx',)),
	)
	))
	self.assertEqual(make_tree(tokenize_rule(r'$FOO==("foo("\|\| "xxx")')), (
	'OP', (
	'==',
	('VAR', ('FOO',)),
	('OP', ('\|\|', ('STR_DOUBLE', ('foo(',)), ('STR_DOUBLE', ('xxx',)))),
	)
	))
	self.assertEqual(make_tree(tokenize_rule(r'!$FOO==("foo("\|\| "xxx")')), (
	'OP', (
	'!',
	('OP', (
	'==',
	('VAR', ('FOO',)),
	('OP', ('\|\|', ('STR_DOUBLE', ('foo(',)), ('STR_DOUBLE', ('xxx',)))),
	))
	)
	))
	with self.assertRaises(SyntaxError) as m:
	make_tree(tokenize_rule(r'())'))
	self.assertEqual(str(m.exception), 'Mismatched parentheses')
	with self.assertRaises(SyntaxError) as m:
	make_tree(tokenize_rule(r'('))
	self.assertEqual(str(m.exception), 'Mismatched parentheses')
	with self.assertRaises(SyntaxError) as m:
	make_tree(tokenize_rule(r'("("))'))
	self.assertEqual(str(m.exception), 'Mismatched parentheses')

	with self.assertRaises(SyntaxError) as m:
	make_tree(tokenize_rule(r'$FOO== != "0"'))
	self.assertEqual(str(m.exception), 'Missing operands')

	with self.assertRaises(SyntaxError) as m:
	make_tree(tokenize_rule(r'( == "1" && "0" \|\| "0")'))
	self.assertEqual(str(m.exception), 'Missing operands')

	with self.assertRaises(SyntaxError) as m:
	make_tree(tokenize_rule(r'$VAR( $FOO == "0")'))
	self.assertEqual(str(m.exception), 'Too many operands')

	def test_eval(self):
	self.assertEqual(evaluate_rule(parse_rule("$VAR_FOO == ''"), {}), False)
	self.assertEqual(evaluate_rule(parse_rule("$VAR_FOO == null"), {}), True)
	self.assertEqual(evaluate_rule(parse_rule("$VAR_FOO == 'xx'"), {}), False)
	self.assertEqual(evaluate_rule(parse_rule("$VAR_FOO == 'xx'"), { 'VAR_FOO': 'xx' }), True)
	self.assertEqual(
	evaluate_rule(parse_rule("$VAR_FOO == 'xx' && $BAR"), {
	'VAR_FOO': 'xx',
	'BAR': 'yyy',
	}), 'yyy')
	self.assertEqual(
	evaluate_rule(parse_rule("$VAR_FOO == 'xx' && !$BAR"), {
	'VAR_FOO': 'xx',
	'BAR': 'yyy',
	}), False)
	self.assertEqual(
	evaluate_rule(parse_rule("$VAR_FOO == 'xx' && $Z"), {
	'VAR_FOO': 'xx',
	'BAR': 'yyy',
	}), None)
	self.assertEqual(
	evaluate_rule(parse_rule(r"$VAR_FOO == 'x\a\\\'x'"), {
	'VAR_FOO': r"xa\'x",
	}), True)
	+ self.assertEqual(
	+ evaluate_rule(parse_rule(r"$VAR_FOO =~ /ab\s\/c/"), {
	+ 'VAR_FOO': 'XXab /c/def',
	+ }), True)
	+ self.assertEqual(
	+ evaluate_rule(parse_rule(r"$VAR_FOO =~ /ab\s\/c/"), {
	+ 'VAR_FOO': 'ab \\/c',
	+ }), False)
	+ self.assertEqual(
	+ evaluate_rule(parse_rule(r"$VAR_FOO !~ /ab\s\/c/"), {
	+ 'VAR_FOO': 'ab \\/c',
	+ }), True)

File Metadata

Mime Type: text/x-diff
Expires: Fri, Oct 9, 1:00 AM (1 d, 37 m)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1784372
Default Alt Text: (56 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions