datastore/ci/praktika/info.py

import json
import os
import traceback
import urllib
from pathlib import Path
from typing import Optional

from .settings import Settings

class Info:

    def __init__(self):
        from ._environment import _Environment

        self.env = _Environment.get()
        self.workflow = None

    @property
    def sha(self):
        return self.env.SHA

    @property
    def pr_number(self):
        return self.env.PR_NUMBER

    @property
    def linked_pr_number(self):
        """
        PR associated with the merge commit for Push or Merge Queue workflow
        :return: PR number or 0 if not applicable or not found
        """
        return self.env.LINKED_PR_NUMBER

    def set_parent_pr_number(self, pr_number):
        self.env.JOB_KV_DATA["parent_pr_number"] = pr_number
        self.env.dump()
        return self

    @property
    def workflow_name(self):
        return self.env.WORKFLOW_NAME

    @property
    def event_time(self):
        return self.env.EVENT_TIME

    @property
    def job_config(self):
        return self.env.JOB_CONFIG

    @property
    def job_name(self):
        return self.env.JOB_NAME

    @property
    def pr_body(self):
        return self.env.PR_BODY

    @property
    def pr_title(self):
        return self.env.PR_TITLE

    @property
    def updated_at(self):
        return self.env.EVENT_TIME

    @property
    def pr_url(self):
        return self.env.CHANGE_URL

    @property
    def commit_url(self):
        return self.env.COMMIT_URL

    @property
    def change_url(self):
        return self.pr_url if self.pr_number else self.commit_url

    @property
    def git_branch(self):
        return self.env.BRANCH

    @property
    def base_branch(self):
        return self.env.BASE_BRANCH

    @property
    def git_sha(self):
        return self.env.SHA

    @property
    def repo_name(self):
        return self.env.REPOSITORY

    @property
    def repo_owner(self):
        return os.getenv("GITHUB_REPOSITORY_OWNER", "")

    @property
    def fork_name(self):
        return self.env.FORK_NAME

    @property
    def commit_message(self):
        return self.env.COMMIT_MESSAGE

    @property
    def user_name(self):
        return self.env.USER_LOGIN

    @property
    def commit_authors(self):
        return self.env.COMMIT_AUTHORS or []

    @property
    def run_url(self):
        return self.env.RUN_URL

    @property
    def run_id(self):
        return self.env.RUN_ID

    @property
    def pr_labels(self):
        return self.env.PR_LABELS

    @property
    def instance_type(self):
        return self.env.INSTANCE_TYPE

    @property
    def is_merge_queue_event(self):
        return self.env.EVENT_TYPE == "merge_group"

    @property
    def is_push_event(self):
        return self.env.EVENT_TYPE == "push"

    @property
    def is_dispatch_event(self):
        return self.env.EVENT_TYPE == "dispatch"

    @property
    def instance_lifecycle(self):
        return self.env.INSTANCE_LIFE_CYCLE

    @property
    def instance_id(self):
        return self.env.INSTANCE_ID

    @property
    def is_local_run(self):
        return self.env.LOCAL_RUN

    # TODO: Consider defining secrets outside of workflow as it project data in most of the cases
    def get_secret(self, name):
        from .mangle import _get_workflows

        if not self.workflow:
            self.workflow = _get_workflows(self.env.WORKFLOW_NAME)[0]
        return self.workflow.get_secret(name)

    def get_job_url(self):
        if not self.env.WORKFLOW_JOB_DATA:
            return ""
        return f"{self.env.RUN_URL}/job/{self.env.WORKFLOW_JOB_DATA['check_run_id']}"

    def get_job_report_url(self, latest=False):
        url = self.get_report_url(latest=latest)
        return url + f"&name_1={urllib.parse.quote(self.env.JOB_NAME, safe='')}"

    def get_report_url(self, latest=False):
        sha = self.env.SHA
        if latest:
            sha = "latest"
        return self.get_specific_report_url(
            pr_number=self.env.PR_NUMBER, branch=self.env.BRANCH, sha=sha
        )

    def dump(self):
        self.env.dump()

    def get_specific_report_url(
        self, pr_number, branch, sha, job_name="", workflow_name=""
    ):
        from .settings import Settings

        if pr_number:
            ref_param = f"PR={pr_number}"
        else:
            assert branch
            ref_param = f"REF={branch}"
        path = Settings.S3_REPORT_BUCKET
        for bucket, endpoint in Settings.S3_BUCKET_TO_HTTP_ENDPOINT.items():
            if bucket in path:
                path = path.replace(bucket, endpoint)
                break
        workflow_name = workflow_name or self.env.WORKFLOW_NAME
        res = f"https://{path}/{Path(Settings.HTML_PAGE_FILE).name}?{ref_param}&sha={sha}&name_0={urllib.parse.quote(workflow_name, safe='')}"
        if job_name:
            res += f"&name_1={urllib.parse.quote(job_name, safe='')}"
        return res

    @staticmethod
    def get_specific_report_url_static(pr_number, branch, sha, job_name, workflow_name):
        from .settings import Settings

        if pr_number:
            ref_param = f"PR={pr_number}"
        else:
            assert branch
            ref_param = f"REF={branch}"
        path = Settings.S3_REPORT_BUCKET
        for bucket, endpoint in Settings.S3_BUCKET_TO_HTTP_ENDPOINT.items():
            if bucket in path:
                path = path.replace(bucket, endpoint)
                break
        res = f"https://{path}/{Path(Settings.HTML_PAGE_FILE).name}?{ref_param}&sha={sha}&name_0={urllib.parse.quote(workflow_name, safe='')}"
        if job_name:
            res += f"&name_1={urllib.parse.quote(job_name, safe='')}"
        return res

    @staticmethod
    def get_workflow_input_value(input_name) -> Optional[str]:
        from .settings import _Settings

        try:
            with open(_Settings.WORKFLOW_INPUTS_FILE, "r", encoding="utf8") as f:
                input_obj = json.load(f)
                return input_obj[input_name]
        except Exception as e:
            print(f"ERROR: Exception, while reading workflow input [{e}]")
        return None

    @staticmethod
    def set_workflow_inputs(inputs: dict) -> None:
        """Persist workflow_dispatch inputs for jobs to read via
        `get_workflow_input_value`.

        Mirrors the heredoc the YAML generator emits in CI; used by the
        praktika `--workflow-input` CLI flag for local job runs.
        """
        from .settings import _Settings

        os.makedirs(_Settings.TEMP_DIR, exist_ok=True)
        with open(_Settings.WORKFLOW_INPUTS_FILE, "w", encoding="utf8") as f:
            json.dump(inputs, f)

    def set_pr_labels(self, labels, reset=False):
        self.env.set_pr_labels(labels, reset=reset)

    def add_pr_label(self, label):
        self.env.add_pr_label(label)

    def remove_pr_label(self, label):
        self.env.remove_pr_label(label)

    def store_kv_data(self, key, value):
        print(f"Store workflow kv data: key [{key}], value [{value}]")
        self.env.JOB_KV_DATA[key] = value
        self.env.dump()

    def get_kv_data(self, key=None):
        kv_data = self.env.JOB_KV_DATA
        if key:
            return kv_data.get(key, None)
        return kv_data

    def get_changed_files(self):
        return self.get_kv_data().get("changed_files", None)

    def store_traceback(self):
        self.env.TRACEBACKS.append(traceback.format_exc())
        self.env.dump()

    def add_workflow_warning(self, message):
        """
        Add a warning visible on both the job report page and the workflow
        report page.

        The message is stored as ``{"message": str, "from": str}`` in both the
        current job's ``Result.ext["warnings"]`` and the workflow-level
        ``Result.ext["warnings"]``.  If the same message is posted by multiple
        jobs, the report page groups them into a single entry at render time.

        Unlike ``Result.add_warning``, which only affects the specific result
        it is called on, this method ensures the message appears at both levels.
        """
        self.env.add_workflow_warning(message)

    def add_workflow_error(self, message):
        """
        Add an error visible on both the job and workflow report pages.

        See ``add_workflow_warning`` for propagation semantics.
        """
        self.env.add_workflow_error(message)

    def add_workflow_note(self, message):
        """
        Add a note visible on both the job and workflow report pages.

        See ``add_workflow_warning`` for propagation semantics.
        """
        self.env.add_workflow_note(message)

    def is_workflow_ok(self):
        """
        Experimental function
        :return:
        """
        from .result import Result

        result = Result.from_fs(self.env.WORKFLOW_NAME)
        for subresult in result.results:
            if subresult.name == Settings.FINISH_WORKFLOW_JOB_NAME:
                continue
            if not subresult.is_ok():
                print(f"Job [{subresult.name}] is not ok, status [{subresult.status}]")
                return False
        return True

    def docker_tag(self, image_name):
        if self.env.WORKFLOW_CONFIG:
            digest_dockers = self.env.WORKFLOW_CONFIG.get("digest_dockers", None)
            if digest_dockers:
                return digest_dockers.get(image_name, None)
        return None