Merge remote-tracking branch 'origin/master' into Ticket8402_Jenkins_job_to_check_the_status_of_Instrument_config_dir

Author: rerpha

.env.example

@@ -3,8 +3,8 @@ UPSTREAM_BRANCH_CONFIG=epics
 
 WORKSPACE=temporary_workspace
 
-SSH_CREDENTIALS_USR=***REMOVED***
-SSH_CREDENTIALS_PSW=***REMOVED***
+SSH_CREDENTIALS_USR=username
+SSH_CREDENTIALS_PSW=password
 
 USE_TEST_INSTRUMENT_LIST=false
 TEST_INSTRUMENT_LIST=NDXSCIDEMO

Jenkinsfile_epics_dir

@@ -19,6 +19,7 @@ pipeline {
         DEBUG_MODE = "${DEBUG_MODE}"
         REPO_DIR = "C:\\Instrument\\Apps\\EPICS\\"
         UPSTREAM_BRANCH_CONFIG = "epics"
+        SHOW_UNCOMMITTED_CHANGES_MESSAGES="false"
     }
 
     stages {

hotfix_checker.py

@@ -1,4 +1,5 @@
 """Creates a RepoChecker object and calls the check_instruments method to check for changes in the instruments repository."""
+
 import os
 
 from dotenv import find_dotenv, load_dotenv
@@ -14,12 +15,14 @@
 if __name__ == "__main__":
     if os.environ["DEBUG_MODE"] == "true":
         print("INFO: Running in debug mode")
-        print(f"DEBUG: REPO_DIR: {os.environ['REPO_DIR']}")
-        print(f"DEBUG: UPSTREAM_BRANCH: {os.environ['UPSTREAM_BRANCH_CONFIG']}")
-        print(f"DEBUG: ARTEFACT_DIR: {os.environ['WORKSPACE']}")
-        print(f"DEBUG: USE_TEST_INSTRUMENT_LIST: {os.environ['USE_TEST_INSTRUMENT_LIST']}")
-        print(f"DEBUG: TEST_INSTRUMENT_LIST: {os.environ['TEST_INSTRUMENT_LIST']}")
-        print(f"DEBUG: DEBUG_MODE: {os.environ['DEBUG_MODE']}")
+        print(f"INFO: REPO_DIR: {os.environ['REPO_DIR']}")
+        print(f"INFO: UPSTREAM_BRANCH: {os.environ['UPSTREAM_BRANCH_CONFIG']}")
+        print(f"INFO: ARTEFACT_DIR: {os.environ['WORKSPACE']}")
+        print(
+            f"INFO: USE_TEST_INSTRUMENT_LIST: {os.environ['USE_TEST_INSTRUMENT_LIST']}"
+        )
+        print(f"INFO: TEST_INSTRUMENT_LIST: {os.environ['TEST_INSTRUMENT_LIST']}")
+        print(f"INFO: DEBUG_MODE: {os.environ['DEBUG_MODE']}")
 
     repo_checker = RepoChecker()
     repo_checker.check_instruments()

utils/communication_utils/channel_access.py

@@ -1,4 +1,5 @@
 """Module containing utility methods for interacting with a PV."""
+
 import binascii
 import json
 import zlib
@@ -8,6 +9,7 @@
 from enum import (
     Enum,
 )
+from typing import Dict, Any
 
 from genie_python.channel_access_exceptions import (
     ReadAccessException,
@@ -34,7 +36,7 @@ def __init__(
     def get_value(
         self,
         pv,
-    ):
+    ) ->  str | dict | None:
         """Gets the value of the PV. Returns None if PV is unavailable.
         :return: The PV value as a string, or None if there was an error.
         """
@@ -55,7 +57,7 @@ def get_value(
     @staticmethod
     def _dehex_and_decompress(
         data,
-    ):
+    ) -> bytes:
         """Converts the raw data from a PV to a decompressed string.
         :param data: The raw data from the PV. It is a string of numbers representing the bytes of the raw data of the
         PV.
@@ -65,7 +67,7 @@ def _dehex_and_decompress(
 
     def get_inst_list(
         self,
-    ):
+    ) -> Dict | Any:
         """Gets a list with all instruments running on IBEX from CS:INSTLIST.
         :return: a list of strings of instrument names.
         """
@@ -74,80 +76,6 @@ def get_inst_list(
             {} if pv_value is None else json.loads(self._dehex_and_decompress(pv_value))
         )
 
-    def get_interesting_pvs(
-        self,
-    ):
-        """Returns all PVs with high or medium interest status from the corresponding instrument PV. The
-        instrument for which it returns the list depends on the prefix assigned to this class, which needs to be in the
-        format IN:NAME_OF_INSTRUMENT.
-        :return: A python set with the names of all the PVs with a high or medium interest status.
-        """
-        all_interesting_pvs = (
-            self._get_pvs_by_interesting_level(PvInterestingLevel.HIGH)
-            + self._get_pvs_by_interesting_level(PvInterestingLevel.MEDIUM)
-            + self._get_pvs_by_interesting_level(PvInterestingLevel.LOW)
-            + self._get_pvs_by_interesting_level(PvInterestingLevel.FACILITY)
-        )
-        interesting_pvs = {pv for pv in all_interesting_pvs}
-
-        return interesting_pvs
-
-    def _get_pvs_by_interesting_level(
-        self,
-        interesting_level,
-    ):
-        """Returns the list of all PVs with the specified interesting level from the corresponding instrument PV. The
-        instrument for which it returns the list depends on the prefix assigned to this class, which needs to be in the
-        format IN:NAME_OF_INSTRUMENT.
-        :param interesting_level An enum type representing the interesting level of a PV.
-        :return: A python list with the names of all the PVs with the specified interesting level.
-        """
-        pv_value = self.get_value(
-            "CS:BLOCKSERVER:PVS:INTEREST:" + interesting_level.value
-        )
-        if pv_value is None:
-            return []
-        else:
-            interesting_pvs = json.loads(self._dehex_and_decompress(pv_value))
-            pv_names = [pv[0] for pv in interesting_pvs]
-
-            return pv_names
-
-    def get_valid_iocs(
-        self,
-    ):
-        """Gets the names of all valid IOCS from the PV of IOCs of the instrument.
-        :return: a list of strings representing IOC names.
-        """
-        pv_value = self.get_value("CS:BLOCKSERVER:IOCS")
-        return (
-            None
-            if pv_value is None
-            else json.loads(self._dehex_and_decompress(pv_value)).keys()
-        )
-
-    def get_protected_iocs(
-        self,
-    ):
-        """Gets the names of all protected IOCS from the PV of IOCs of the instrument. Protected IOCs are IOCs that a user
-        is not allowed to stop.
-        :return: a list of strings representing IOC names.
-        """
-        pv_value = self.get_value("CS:BLOCKSERVER:IOCS_NOT_TO_STOP")
-        return (
-            None
-            if pv_value is None
-            else json.loads(self._dehex_and_decompress(pv_value))
-        )
-
-    def get_version_string(
-        self,
-    ):
-        """Gets the version of IBEX server running.
-        :return: a compressed and hex encoded string representing the version.
-        """
-        return self.get_value("CS:VERSION:SVN:REV")
-
 
 class PvInterestingLevel(Enum):
     """Enumerated type representing the possible interesting levels a PV can have."""

utils/communication_utils/ssh_access.py

@@ -1,4 +1,6 @@
 """This module provides utilities for SSH access."""
+from typing import Dict
+
 import paramiko
 
 SSH_PORT = 22
@@ -8,12 +10,12 @@ class SSHAccessUtils(object):
     """Class containing utility methods for SSH access."""
 
     @staticmethod
-    def run_ssh_commandd(
-        host : str,
-        username : str,
-        password : str,
-        command : str,
-    ):
+    def run_ssh_command(
+        host: str,
+        username: str,
+        password: str,
+        command: str,
+    ) -> Dict[str, bool | str]:
         """Run a command on a remote host using SSH.
 
         Args:

utils/hotfix_utils/InstrumentChecker.py

@@ -1,6 +1,7 @@
 """A module for checking the status of an instrument in relation to it's repo."""
+
 import os
-from typing import Union
+from typing import List, Tuple, Union
 
 from ..communication_utils.ssh_access import (
     SSHAccessUtils,
@@ -30,6 +31,7 @@ def __init__(self, hostname: str) -> None:
         self._commits_upstream_not_on_local_enum_messages = None
 
         self._uncommitted_changes_enum = None
+        self._uncommitted_changes_messages = None
 
     @property
     def hostname(self) -> str:
@@ -41,7 +43,7 @@ def hostname(self) -> str:
         """
         return self._hostname
 
-    def check_for_uncommitted_changes(self) -> CHECK:
+    def check_for_uncommitted_changes(self) -> Tuple[CHECK, List[any]]:
         """Check if there are any uncommitted changes on the instrument via SSH.
 
         Args:
@@ -52,7 +54,7 @@ def check_for_uncommitted_changes(self) -> CHECK:
 
         """
         command = f"cd {self.repo_dir} && git status --porcelain"
-        ssh_process = SSHAccessUtils.run_ssh_commandd(
+        ssh_process = SSHAccessUtils.run_ssh_command(
             self.hostname,
             os.environ["SSH_CREDENTIALS_USR"],
             os.environ["SSH_CREDENTIALS_PSW"],
@@ -62,20 +64,20 @@ def check_for_uncommitted_changes(self) -> CHECK:
         if os.environ["DEBUG_MODE"] == "true":
             print(f"DEBUG: Running command {command}")
 
-        # if os.environ["DEBUG_MODE"] == "true":
-        #     print(f"DEBUG: {ssh_process}")
-
         if ssh_process["success"]:
             status = ssh_process["output"]
 
             JenkinsUtils.save_git_status(self.hostname, status, os.environ["WORKSPACE"])
 
-            if status.strip() != "":
-                return CHECK.TRUE
+            status_stripped = status.strip()
+            if status_stripped != "" and os.environ["SHOW_UNCOMMITTED_CHANGES_MESSAGES"] == "true":
+                return CHECK.TRUE, status_stripped.split("\n")
+            elif status_stripped != "":
+                return CHECK.TRUE, []
             else:
-                return CHECK.FALSE
+                return CHECK.FALSE, []
         else:
-            return CHECK.UNDETERMINABLE
+            return CHECK.UNDETERMINABLE, []
 
     def get_parent_epics_branch(
         self,
@@ -91,7 +93,7 @@ def get_parent_epics_branch(
 
         """
         command = f"cd {self.repo_dir} && git log"
-        ssh_process = SSHAccessUtils.run_ssh_commandd(
+        ssh_process = SSHAccessUtils.run_ssh_command(
             hostname,
             os.environ["SSH_CREDENTIALS_USR"],
             os.environ["SSH_CREDENTIALS_PSW"],
@@ -131,9 +133,9 @@ def git_branch_comparer(
         else:
             branch_details = ""
 
-        # fetch latest changes from the remote
+        # Fetch latest changes from the remote, NOT PULL
         fetch_command = f"cd {self.repo_dir} && git fetch origin"
-        ssh_process_fetch = SSHAccessUtils.run_ssh_commandd(
+        ssh_process_fetch = SSHAccessUtils.run_ssh_command(
             hostname,
             os.environ["SSH_CREDENTIALS_USR"],
             os.environ["SSH_CREDENTIALS_PSW"],
@@ -143,29 +145,24 @@ def git_branch_comparer(
         if os.environ["DEBUG_MODE"] == "true":
             print(f"DEBUG: Running command {fetch_command}")
 
-        # if os.environ["DEBUG_MODE"] == "true":
-        #     print(f"DEBUG: {ssh_process_fetch}")
-
         if not ssh_process_fetch["success"]:
             return (
                 CHECK.UNDETERMINABLE,
                 None,
             )
 
         command = f'cd {self.repo_dir} && git log --format="%h %s" {branch_details}'
+
         if os.environ["DEBUG_MODE"] == "true":
             print(f"DEBUG: Running command {command}")
 
-        ssh_process = SSHAccessUtils.run_ssh_commandd(
+        ssh_process = SSHAccessUtils.run_ssh_command(
             hostname,
             os.environ["SSH_CREDENTIALS_USR"],
             os.environ["SSH_CREDENTIALS_PSW"],
             command,
         )
 
-        # if os.environ["DEBUG_MODE"] == "true":
-        #    print(f"DEBUG: {ssh_process}")
-
         if ssh_process["success"]:
             output = ssh_process["output"]
             commit_dict = self.split_git_log(output, prefix)
@@ -210,8 +207,7 @@ def split_git_log(self, git_log: str, prefix: str) -> dict:
             if prefix is None or message.startswith(prefix):
                 commit_dict[hash] = message
         return commit_dict
-
-    # TODO: Improve this function to allow selecting of what checks to run and combine the uncommited one to this function
+    
     def check_instrument(self) -> dict:
         """Check if there are any hotfixes or uncommitted changes on AN instrument.
 
@@ -222,9 +218,8 @@ def check_instrument(self) -> dict:
         # Examples of how to use the git_branch_comparer function decided to not be used in this iteration of the check
         # Check if any hotfixes run on the instrument with the prefix "Hotfix:"
         # hotfix_commits_enum, hotfix_commits_messages = git_branch_comparer(
-        #     hostname, prefix="Hotfix:")
+        #     hostname, local_branch, upstream_branch, prefix="Hotfix:")
 
-        # Check if any unpushed commits run on the instrument
         upstream_branch = None
         if os.environ["UPSTREAM_BRANCH_CONFIG"] == "hostname":
             upstream_branch = "origin/" + self.hostname
@@ -239,7 +234,7 @@ def check_instrument(self) -> dict:
             # if the UPSTREAM_BRANCH_CONFIG is not set to any of the above,  set it to the value of the environment variable assuming user wants custom branch
             upstream_branch = os.environ["UPSTREAM_BRANCH_CONFIG"]
 
-        # Check if any upstream commits are not on the instrument, default to the parent origin branch, either main or galil-old
+        # Check if any commits on upstream that are not on the local branch
         (
             self.commits_upstream_not_on_local_enum,
             self.commits_upstream_not_on_local_messages,
@@ -250,30 +245,19 @@ def check_instrument(self) -> dict:
             prefix=None,
         )
 
-        # print(self.commits_upstream_not_on_local_enum, self.commits_upstream_not_on_local_messages)
-
+        # Check if any commits on local branch that are not on the upstream
         (
             self.commits_local_not_on_upstream_enum,
             self.commits_local_not_on_upstream_messages,
         ) = self.git_branch_comparer(
             self.hostname,
             changes_on="HEAD",
-            # subtracted_against="origin/" + self.hostname,
-            subtracted_against=upstream_branch,  # for inst scripts repo
+            subtracted_against=upstream_branch,
             prefix=None,
         )
 
-        # Check if any uncommitted changes run on the instrument
-        self.uncommitted_changes_enum = self.check_for_uncommitted_changes()
-
-        # # return the result of the checks
-        # instrument_status = {
-        #     "commits_not_pushed_messages": commits_local_not_on_upstream_messages,
-        #     "commits_not_pushed": commits_local_not_on_upstream_enum,
-        #     "uncommitted_changes": uncommitted_changes_enum,
-        # }
-
-        # return instrument_status
+        # Check if any uncommitted changes are on the instrument
+        self.uncommitted_changes_enum, self.uncommitted_changes_messages = self.check_for_uncommitted_changes()
 
     def as_string(self) -> str:
         """Return the Instrument object as a string.