tekton-compile-pipeline.py

# -*- coding: utf-8 -*-
# Author: Chmouel Boudjnah <[email protected]>
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import copy
import glob

import yaml

try:
    from yaml import CLoader as Loader, CDumper as Dumper
except ImportError:
    from yaml import Loader, Dumper

TEKTON_TYPE = ("pipeline", "pipelinerun", "task", "taskrun")


def parse(yamlfiles):
    yaml_documents = {}
    for yamlfile in yamlfiles:
        for document in yaml.load_all(open(yamlfile), Loader=Loader):
            if 'apiVersion' not in document or 'kind' not in document:
                print("Skipping not a kubernetes file")
                continue

            taskname = document['metadata']['name'].lower()
            kind = document['kind'].lower()

            if kind not in TEKTON_TYPE:
                print("Skipping not a tekton file")
                continue

            yaml_documents.setdefault(kind, {})
            yaml_documents[kind][taskname] = document

    if 'pipelinerun' not in yaml_documents:
        raise Exception("We need at least a PipelineRun")

    for pipeline in yaml_documents['pipeline']:
        mpipe = copy.deepcopy(yaml_documents['pipeline'][pipeline])
        for task in mpipe['spec']['tasks']:
            if 'taskRef' in task:
                reftask = task['taskRef']['name']
                if reftask not in yaml_documents['task']:
                    raise Exception(
                        f"Pipeline: {pipeline} reference a Task: {reftask} not in repository"
                    )

                del task['taskRef']
                task['taskSpec'] = yaml_documents['task'][reftask]['spec']

        yaml_documents['pipeline'][pipeline] = copy.deepcopy(mpipe)

    for pr in yaml_documents['pipelinerun']:
        mpr = copy.deepcopy(yaml_documents['pipelinerun'][pr])
        if 'pipelineRef' in mpr['spec']:
            refpipeline = mpr['spec']['pipelineRef']['name']
            if refpipeline not in yaml_documents['pipeline']:
                raise Exception(
                    f"PR: {pr} reference a Pipeline: {refpipeline} not in repository"
                )
            del mpr['spec']['pipelineRef']
            mpr['spec']['pipelineSpec'] = yaml_documents['pipeline'][
                refpipeline]['spec']
        print(yaml.dump(mpr, Dumper=Dumper, default_flow_style=False))


if __name__ == '__main__':
    parse(glob.glob("*.yaml"))