Page MenuHomePhorge
Files F33101518

ci_file.py
No OneTemporary
Actions

Size
5 KB
Referenced Files
None
Subscribers
None

ci_file.py
View Options

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
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('-')
class CIJob:
def __init__(self, job_name, job_stage, job_struct, defaults):
self.name = job_name
self.stage = job_stage
self.struct_raw = job_struct
self.defaults_raw = defaults
self.image = job_struct.get('image', defaults.get('image'))
self.before_script = normalize_script(job_struct.get('before_script', defaults.get('before_script')))
self.script = normalize_script(job_struct.get('script', defaults.get('script')))
self.after_script = normalize_script(job_struct.get('after_script', defaults.get('after_script')))
self.artifacts = job_struct.get('artifacts', defaults.get('artifacts', {}))
self.dependencies = job_struct.get('dependencies')
def get_predefined_ci_variables(self):
return {
'CI': 'true',
'CI_JOB_NAME': self.name,
'CI_JOB_NAME_SLUG': ci_slugify(self.name),
'CI_JOB_STAGE': self.stage,
}
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,
'defaults_raw': self.defaults_raw,
}
@classmethod
def from_prop(cls, prop):
return cls(
prop['name'],
prop['stage'],
prop['struct_raw'],
prop['defaults_raw']
)
OLD_TOPLEVEL_DEFAULTS = ['image', 'services', 'cache', 'before_script', 'after_script']
DEFAULT_STAGES = ['.pre', 'build', 'test', 'deploy', '.post']
DEFAULT_JOB_STAGE = 'test'
class CIValidationError(Exception):
pass
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]
for job_name, job_struct in f.items():
# 'pages' is a special job
if job_name != 'pages' and job_name in toplevel_entries():
continue
# jobs starting with . will only be used for base jobs of other jobs,
# they themselves are not run
if job_name.startswith('.'):
continue
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, defaults)
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

File Metadata

Mime Type
text/x-python
Expires
Tue, Jan 20, 12:26 PM (1 d, 9 h)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
973535
Default Alt Text
ci_file.py (5 KB)

Event Timeline