Index: suite_leahi/shared/scripts/configuration/getrejectiontext.py =================================================================== diff -u --- suite_leahi/shared/scripts/configuration/getrejectiontext.py (revision 0) +++ suite_leahi/shared/scripts/configuration/getrejectiontext.py (revision 6d83d7cbea9b040ee4717d7b90fdc44bf618bdcf) @@ -0,0 +1,152 @@ + +# -*- coding: utf-8 -*- +""" +ScopedRejectionRepository for Ubuntu/Linux paths. +Reads INI-style conf with numbered sections like: + +[number0] +Title = sample + +[number1] +Title = example +""" + +import os +import re +import configparser + + +def _normalize_path(p): + """Expand env vars and ~, then return absolute path.""" + if p is None: + return None + p = os.path.expanduser(os.path.expandvars(p)) + return os.path.abspath(p) + + +class ScopedRejectionRepository(object): + def __init__(self, path, section_regex=r"^number(\d+)$"): + """ + :param path: Absolute path to Rejections.conf (outside suite allowed; REQUIRED) + Example (Ubuntu): + "/opt/configs/payments/Rejections.conf" + "/etc/payments/Rejections.conf" + "/home/ubuntu/configs/Rejections.conf" + :param section_regex: Regex to identify numbered sections, captures index. + Default matches [number0], [number1], ... + """ + if path is None: + raise IOError("Explicit 'path' to Rejections.conf is required since the file is not in the suite.") + + self.path = _normalize_path(path) + + # Validate the path early and clearly + if os.path.isdir(self.path): + raise IsADirectoryError("Path points to a directory, not a file: {}".format(self.path)) + if not os.path.isfile(self.path): + raise IOError("Rejections.conf not found at: {}".format(self.path)) + + self.section_regex = re.compile(section_regex) + self._config = None + self._sections_in_order = [] # e.g., ["number0", "number1", ...] + self.reload() + + def reload(self): + """Load or reload the conf file into memory.""" + # Defensive re-validation + if os.path.isdir(self.path): + raise IsADirectoryError("Path points to a directory, not a file: {}".format(self.path)) + if not os.path.isfile(self.path): + raise IOError("Rejections.conf missing or not a file: {}".format(self.path)) + + config = configparser.ConfigParser() + # Preserve original key case (configparser lowercases by default) + config.optionxform = str + with open(self.path, "r", encoding="utf-8") as f: + config.read_file(f) + + # Determine ordered numbered sections by extracted integer index + numeric_sections = [] + for section in config.sections(): + m = self.section_regex.match(section) + if m: + idx = int(m.group(1)) + numeric_sections.append((idx, section)) + + # Sort by numeric index for stable ordering + numeric_sections.sort(key=lambda t: t[0]) + self._sections_in_order = [name for _, name in numeric_sections] + self._config = config + + # Optional: warn if gaps in numbering + self._warn_for_gaps(numeric_sections) + + def _warn_for_gaps(self, numeric_sections): + """Log gaps or non-sequential indexes using Squish logger (if available).""" + try: + from squish import test + except Exception: + test = None + + if not numeric_sections: + if test: + test.warning("No numbered sections found matching regex.") + return + + indexes = [idx for idx, _ in numeric_sections] + expected = list(range(indexes[0], indexes[-1] + 1)) + if indexes != expected and test: + test.warning("Non-sequential section indexes found: {} (expected: {})" + .format(indexes, expected)) + + # -------------------------- + # Public API + # -------------------------- + def list_sections(self): + """Return ordered section names, e.g., ['number0', 'number1', ...].""" + return list(self._sections_in_order) + + def size(self): + """Number of numbered sections.""" + return len(self._sections_in_order) + + def has_section(self, section_name): + return self._config is not None and self._config.has_section(section_name) + + def get(self, section_name, key= "'title'", default=None): + """ + Fetch a value by (section, key). + Example: get('number0', 'Title') -> 'sample' + """ + if self._config is None or not self._config.has_section(section_name): + return default if default is not None else "".format(section_name) + sect = self._config[section_name] + return sect.get(key, default if default is not None else "".format(section_name, key)) + + def get_by_index(self, index, key, default=None): + """ + Fetch value by numeric index (e.g., 0 for [number0]) and key. + """ + try: + section_name = self._sections_in_order[index] + except IndexError: + return default if default is not None else "".format(index) + return self.get(section_name, key, default=default) + + def get_section_dict(self, section_name): + """ + Return a dict of all key/value pairs in a section. + """ + if not self.has_section(section_name): + return {} + return dict(self._config[section_name]) + + def get_all_titles(self, key_name="Title"): + """ + Return a list of titles from all numbered sections, in order. + """ + titles = [] + for section_name in self._sections_in_order: + val = self._config[section_name].get(key_name, None) + titles.append(val) + return titles Index: suite_leahi/tst_ultrafiltration/test.py =================================================================== diff -u -rbc991ff3f0c12142cc4d10c52021f3bc01f4fbeb -r6d83d7cbea9b040ee4717d7b90fdc44bf618bdcf --- suite_leahi/tst_ultrafiltration/test.py (.../test.py) (revision bc991ff3f0c12142cc4d10c52021f3bc01f4fbeb) +++ suite_leahi/tst_ultrafiltration/test.py (.../test.py) (revision 6d83d7cbea9b040ee4717d7b90fdc44bf618bdcf) @@ -23,9 +23,12 @@ from leahi_dialin.common.td_defs import TDOpModes,TDTreatmentStates from leahi_dialin.common import msg_ids from leahi_dialin.utils import conversions +from configuration.getrejectiontext import ScopedRejectionRepository + td =TD_Messaging() + MAX_UF_VOLUME = 10.00 MAX_UF_RATE = 10.00 MAX_VOLUME_DELIVERED = 8.00 @@ -43,9 +46,9 @@ UF_VOLUME_GOAL = "UF Volume Goal" NEW_UF_VOLUME = "New UF Volume" NEW_UF_RATE = "New UF Rate" -REJECT_TEXT = "[1] Request is not allowed in the current operating mode" + MAX_NEW_UF_VOLUME = 8.0 MAX_NEW_UF_RATE = 2.0 @@ -93,6 +96,10 @@ def main(): utils.tstStart(__file__) + + conf_path = "/home/denali/Public/luis/config/configurations/Alarms/Rejections.conf" + repo = ScopedRejectionRepository(path=conf_path) + startApplication(config.AUT_NAME) td.td_operation_mode(TDOpModes.MODE_STAN.value) @@ -187,7 +194,8 @@ rejectionReason = 1) reject_text = waitForObject(names.o_notificationBar_NotificationBarSmall).text - test.compare(reject_text,REJECT_TEXT,"Text should be ->" +str(REJECT_TEXT)) + REJECT_TEXT = repo.get("1", "Title") + test.compare(reject_text,"[1] "+REJECT_TEXT ,"Text should be ->" +str(REJECT_TEXT)) test.endSection() utils.tstDone()