feat: python in jobs

  - local: yaml/yamllint.yml

  - local: project-automation/pipeline-scheduler.yml

  - local: project-automation/pages-hugo.yml

  - local: project-automation/gitlab/py-in-ci.yml

  - local: docs/prettier.yml

pages:

#!/usr/bin/env sh

TARGET="${1:=${PWD}}"

prettier --write --ignore-unknown --prose-wrap always ${TARGET} || \

docker run -u $(id -u):$(id -g) -v "${TARGET}:${TARGET}" registry.gitlab.com/just-ci/images/prettier:latest prettier --write --ignore-unknown --prose-wrap always ${TARGET}

#!/usr/bin/env sh

set -e

cd "$(dirname "$0")/../.."

if ! command -v docker >/dev/null && ! command -v prettier >/dev/null; then

    echo "[!] You need either 'prettier' or 'docker' on PATH to run this script."

    exit 1

fi

for FILE in $(find . -type f ! -path "*/tests/*" -name "*.py"); do

    NAME_ONLY="${FILE%.*}"

    echo "[*] Fixing ${NAME_ONLY}..."

    JOB_FILE="${NAME_ONLY}.yml"

    if ! test -f "${JOB_FILE}"; then

        echo "[!] ${JOB_FILE} does not exist."

        continue

    fi

    black "${FILE}"

    isort --profile=black "${FILE}"

    echo 'python - << EOF' > /tmp/script.yaml

    cat "${FILE}" >> /tmp/script.yaml

    echo 'EOF' >> /tmp/script.yaml

    SCRIPT="$(cat /tmp/script.yaml)" yq -i '(..|select(has("script")).["script"].[0]) |= strenv(SCRIPT)' "${JOB_FILE}"

    docs/prettier.sh "${JOB_FILE}"

done

---

gitlab:py-in-ci-check:

  stage: test

  image:

    name: docker.io/alpine/git:latest

    entrypoint: [""]

  script:

    - apk add -q yq

    - |

      for FILE in $(find ${CI_PROJECT_DIR} -type f ! -path "*/tests/*" -name "*.py"); do

        NAME_ONLY="${FILE%.*}"

        echo "[*] Checking ${NAME_ONLY}..."

        JOB_FILE="${NAME_ONLY}.yml"

        if ! test -f "${JOB_FILE}"; then

          echo "[!] ${JOB_FILE} does not exist."

          ERROR=1

          continue

        fi

        echo 'python - << EOF' > /tmp/script.yaml

        cat "${FILE}" >> /tmp/script.yaml

        echo 'EOF' >> /tmp/script.yaml

        SCRIPT="$(cat /tmp/script.yaml)" yq -i '(..|select(has("script")).["script"].[0]) |= strenv(SCRIPT)' "${JOB_FILE}"

        if ! git --no-pager -c color.diff=always diff --exit-code "${JOB_FILE}"; then

          echo "[!] ${JOB_FILE} does not match script."

          ERROR=1

        else

          echo "Looks good!"

        fi

      done

    - |

      if [ "${ERROR}" != "" ]; then

        echo "[!] One or more jobs don't match their script, see above."

        echo "To fix, run: project-automation/gitlab/py-in-ci.sh"

        exit 1

      fi

  needs: []

import os

import subprocess

import sys

from typing import Dict, List

install_cmd = f"{sys.executable} -m pip install -q python-gitlab"

subprocess.run(install_cmd.split(), check=True)

import gitlab

from gitlab.base import RESTObject

from gitlab.exceptions import GitlabAuthenticationError

no_error = True

data = {

    "ref": os.getenv("CI_DEFAULT_BRANCH"),

    "description": os.getenv("SCHEDULE_PIPELINE_DESCRIPTION", "Default"),

    "cron": os.getenv("SCHEDULE_PIPELINE_CRON", "0 1 * * *"),

    "cron_timezone": os.getenv("TZ", "UTC"),

}

project_ids = os.getenv(

    "SCHEDULE_PIPELINE_PROJECT_IDS", os.getenv("CI_PROJECT_ID")

).split(" ")

print(f"Project IDs: {' '.join(project_ids)}")

print(f"Description: {data['description']}")

print(f"Cron: {data['cron']}")

print(f"Timezone: {data['cron_timezone']}")

user_variables = list()

if variables := os.getenv("SCHEDULE_PIPELINE_VARIABLES"):

    for variable in variables.splitlines():

        print(f"User variable: {variable}")

        key, value = variable.split("=", 1)

        user_variables.append({"key": key, "value": value, "variable_type": "env_var"})

gl = gitlab.Gitlab(private_token=os.getenv("GL_TOKEN"))

try:

    gl.auth()

except GitlabAuthenticationError:

    print("[!] Could not authenticate. Is your GL_TOKEN correct?")

    sys.exit(1)

def check_variables(current_variables: List[Dict[str, str]]) -> bool:

    new_variables = [

        {

            "key": "AUTO_SCHEDULED",

            "value": "true",

            "variable_type": "env_var",

        }

    ] + user_variables

    return new_variables == current_variables

def check_ownership(schedule: RESTObject, project: RESTObject) -> bool:

    if os.getenv("DISABLE_SCHEDULE_OWNERSHIP_WARNING"):

        print("[-] Not checking ownership.")

        return True

    GL_TOKEN_user = gl.users.list(username=gl.user.username)[0]

    if schedule.attributes["owner"]["id"] == GL_TOKEN_user.id:

        print(

            "[!] The schedule is owned by the GL_TOKEN user named "

            f"'{GL_TOKEN_user.name}' with username '{GL_TOKEN_user.username}', "

            "so you probably will not receive any warnings of pipeline failures."

        )

        print(

            f"Go here and take ownership: {project.attributes['web_url']}/-/pipeline_schedules"

        )

        print(

            "Or disable this warning by setting a variable called DISABLE_SCHEDULE_OWNERSHIP_WARNING to 'true'."

        )

        return False

    return True

def process_schedule(

    schedule: RESTObject, already_exists: bool, project: RESTObject

) -> bool:

    global no_error

    schedule = project.pipelineschedules.get(schedule.get_id())

    attrs = schedule.attributes

    variable_values = [key["key"] for key in attrs["variables"]]

    attrs_data = {

        "ref": attrs["ref"],

        "description": attrs["description"],

        "cron": attrs["cron"],

        "cron_timezone": attrs["cron_timezone"],

    }

    if (

        attrs["description"] == data["description"]

        and "AUTO_SCHEDULED" in variable_values

    ):

        if already_exists:

            print("[-] Duplicate schedule found. Deleting.")

            schedule.delete()

        if attrs_data == data and check_variables(attrs["variables"]):

            print("[+] Existing matching schedule found. Skipping.")

            no_error = check_ownership(schedule, project)

            already_exists = True

        else:

            print(

                f"[-] Outdated schedule found. Deleting before creating new schedule."

            )

            schedule.delete()

    return already_exists

for project_id in project_ids:

    already_exists = False

    project = gl.projects.get(project_id)

    current_schedules = project.pipelineschedules.list()

    for schedule in current_schedules:

        already_exists = process_schedule(schedule, already_exists, project)

    if not already_exists:

        print("[+] Creating new schedule")

        schedule = project.pipelineschedules.create(data)

        schedule.variables.create({"key": "AUTO_SCHEDULED", "value": "true"})

        for variable in user_variables:

            schedule.variables.create(

                {"key": variable["key"], "value": variable["value"]}

            )

        no_error = check_ownership(schedule, project)

if not no_error:

    sys.exit(1)