From 7e690a907e6ff824822e06f8b17711169e02ce99 Mon Sep 17 00:00:00 2001 From: Benjamin Labrecque Date: Wed, 22 Jul 2026 12:04:55 +0200 Subject: [PATCH] refactor: add paths, services ctx, and extract models to own files --- cli/pyproject.toml | 8 +- cli/src/config/paths.py | 31 +-- cli/src/core/models.py | 307 --------------------------- cli/src/core/service_creator.py | 188 +++------------- cli/src/models/__init__.py | 20 ++ cli/src/models/context.py | 54 +++++ cli/src/models/iocs_overview.py | 29 +++ cli/src/models/master_hla_names.py | 23 ++ cli/src/models/master_ioc_subs.py | 45 ++++ cli/src/models/service.py | 40 ++++ cli/src/models/service_manager_ui.py | 136 ++++++++++++ cli/src/models/service_registry.py | 46 ++++ cli/src/service.py | 41 +--- cli/tests/test_paths.py | 8 - cli/tests/test_service.py | 6 - cli/tests/test_service_creator.py | 16 ++ 16 files changed, 474 insertions(+), 524 deletions(-) delete mode 100644 cli/src/core/models.py create mode 100644 cli/src/models/__init__.py create mode 100644 cli/src/models/context.py create mode 100644 cli/src/models/iocs_overview.py create mode 100644 cli/src/models/master_hla_names.py create mode 100644 cli/src/models/master_ioc_subs.py create mode 100644 cli/src/models/service.py create mode 100644 cli/src/models/service_manager_ui.py create mode 100644 cli/src/models/service_registry.py delete mode 100644 cli/tests/test_paths.py delete mode 100644 cli/tests/test_service.py create mode 100644 cli/tests/test_service_creator.py diff --git a/cli/pyproject.toml b/cli/pyproject.toml index 00d3ceb..f2f10f6 100644 --- a/cli/pyproject.toml +++ b/cli/pyproject.toml @@ -54,8 +54,12 @@ include = [ line-length = 100 [tool.ruff.lint] -# use isort to sort imports -extend-select = ["I"] +extend-select = [ + # use isort to sort imports + "I", + # remove unused imports + "F401", +] [tool.ruff.lint.isort] known-first-party = ["agebd"] diff --git a/cli/src/config/paths.py b/cli/src/config/paths.py index d3050d3..4e232d9 100644 --- a/cli/src/config/paths.py +++ b/cli/src/config/paths.py @@ -1,14 +1,13 @@ from pathlib import Path -from pydantic import DirectoryPath, FilePath -from pydantic_settings import BaseSettings +from pydantic import BaseModel, DirectoryPath, FilePath from core.utils import get_git_root REPO_ROOT = get_git_root(__file__) -class Paths(BaseSettings): +class Paths(BaseModel): # fmt: off repo_root: DirectoryPath = REPO_ROOT @@ -19,17 +18,25 @@ class Paths(BaseSettings): service_manager_ui_file: FilePath = repo_root / "qt" / "A_BD_ServiceManager.ui" # fmt: on + def get_files_to_stage(self, service_dir_name: str) -> list[str]: + """ + Returns all required relative file paths for git staging. + """ + service_current = self.repo_root / "services" / service_dir_name / "current" + + staged_paths = [ + service_current, + self.service_registry_file, + self.master_hla_names_file, + self.master_ioc_subs_file, + self.iocs_overview_filename, + self.service_manager_ui_file, + ] + + return [self.relative(p) for p in staged_paths] + def relative(self, path: Path) -> str: """ Convert any absolute path into a string path relative to repo_root. """ return str(path.relative_to(self.repo_root)) - - def get_service_dir(self, service_dir_name: str) -> Path: - """ - Helper to get a service path dynamically. - """ - return self.repo_root / "services" / service_dir_name / "current" - - -paths = Paths() diff --git a/cli/src/core/models.py b/cli/src/core/models.py deleted file mode 100644 index 6006982..0000000 --- a/cli/src/core/models.py +++ /dev/null @@ -1,307 +0,0 @@ -import re -import xml.etree.ElementTree as ET -from pathlib import Path -from typing import Any, ClassVar - -import yaml -from pydantic import BaseModel, ConfigDict, field_validator - -from config.paths import paths -from core.enums import ServiceStatus - - -class Service(BaseModel): - name: str - ioc_port: int - status: ServiceStatus - - @property - def name_upper(self): - return self.name.upper() - - @property - def name_lower(self): - return self.name.lower() - - @property - def name_lower_underscores(self): - return self.name.replace("-", "_").lower() - - @property - def dir_name(self): - return self.name.lower() - - def to_dict(self): - return {"name": self.name_lower, "ioc_port": self.ioc_port, "status": str(self.status)} - - class Config: - use_enum_values = True - - @field_validator("name") - @classmethod - def validate_name_chars(cls, value: str) -> str: - if not re.match(r"^[a-z0-9-]+$", value): - raise ValueError("name can only contain lowercase letters, numbers, and dashes") - return value - - -class ServiceRegistry(BaseModel): - config_path: ClassVar[Path] = paths.service_registry_file - - next_available_ioc_port: int - services: list[dict] - - @classmethod - def read_from_config(cls) -> "ServiceRegistry": - with open(cls.config_path) as f: - d = yaml.safe_load(f) - - registry = ServiceRegistry.model_validate(d) - return registry - - def get_service(self, name: str) -> Service: - for svc_data in self.services: - service = Service(**svc_data) - if name == service.name_lower: - return service - - raise ValueError(f"Could not find service with name: {name}") - - def get_services(self) -> list[Service]: - return [Service(**svc_data) for svc_data in self.services] - - def add_service(self, name: str) -> Service: - svc = Service( - name=name, - ioc_port=self.next_available_ioc_port, - status=ServiceStatus.ACTIVE, - ) - self.services.append(svc.to_dict()) - self.next_available_ioc_port += 1 - - return svc - - def write_to_config(self) -> None: - with open(self.config_path, "w") as f: - yaml.safe_dump(self.model_dump(), f, sort_keys=False) - - -class MasterHLANames(BaseModel): - config_path: ClassVar[Path] = paths.master_hla_names_file - - hla_apps: list[str] - - @classmethod - def read_from_config(cls) -> "MasterHLANames": - with open(cls.config_path) as f: - d = yaml.safe_load(f) - - hla_names = MasterHLANames.model_validate(d) - return hla_names - - def add_hla_name(self, name: str): - self.hla_apps.append(name) - - def write_to_config(self) -> None: - with open(self.config_path, "w") as f: - yaml.safe_dump(self.model_dump(), f, indent=4) - - -class IocsOverview(BaseModel): - config_path: ClassVar[Path] = paths.iocs_overview_filename - - content: str - - @classmethod - def read_from_file(cls) -> "IocsOverview": - with open(cls.config_path, "r", encoding="utf-8") as f: - s = f.read() - - return IocsOverview(content=s) - - def add_ioc(self, service: Service, description: str): - new_row = f"| AGEBD-CPCL-{service.name_upper} | {description} |\n" - - # Ensuring it starts on a clean new line if the file didn't end with one - if not self.content.endswith("\n"): - self.content += "\n" - - self.content += new_row - - def write_to_file(self): - with open(self.config_path, "w") as f: - f.write(self.content) - - -class MasterIocSubs(BaseModel): - config_path: ClassVar[Path] = paths.master_ioc_subs_file - - content: str - - @classmethod - def read_from_file(cls) -> "MasterIocSubs": - with open(cls.config_path) as f: - file_content = f.read() - - return MasterIocSubs(content=file_content) - - def update_content( - self, - service_name: str, - starton: str = "0", # TODO: ok defaults? - autooff: str = "0", # TODO: ok defaults? - ): - """ - Parses the configuration text and appends a new row before the closing bracket. - """ - quoted_service = f'"{service_name}"' - new_row = f' {{ "{{{{ agebd_env }}}}", {quoted_service:<22} , "{starton}", "{autooff}" }}' - - # Use regex to find the last closing curly brace of the configuration block - # This targets the line that contains only a lone closing brace, optional spaces, and maybe a comma/semicolon. - pattern = r"(\s*\n\s*\}(?:;|,)?\s*$)" - - if re.search(pattern, self.content, re.MULTILINE): - # Insert our new row right before that closing brace line - self.content = re.sub( - pattern, f"\n{new_row}\\1", self.content, count=1, flags=re.MULTILINE - ) - else: - raise ValueError( - "Could not locate the closing structure format '}' inside the configuration." - ) - - def write_to_file(self) -> None: - with open(self.config_path, "w") as f: - f.write(self.content) - - -# TODO: maybe separate models into their own files -class ServiceManagerUI(BaseModel): - model_config = ConfigDict(arbitrary_types_allowed=True) - config_path: ClassVar[Path] = paths.service_manager_ui_file - - tree: Any - - @classmethod - def read_from_file(cls) -> "ServiceManagerUI": - tree = ET.parse(cls.config_path) - return ServiceManagerUI(tree=tree) - - def update_xml(self, service: Service, ui_name: str, ui_filename: str) -> None: - num_macro_elements = self._update_macro( - service=service, ui_name=ui_name, ui_filename=ui_filename - ) - self._update_heights(num_macro_elements=num_macro_elements) - - def _update_macro(self, service: Service, ui_name: str, ui_filename: str) -> int: - new_macro_element = _ServiceManagerUIMacroElement( - service_name_upper=service.name_upper, ui_name=ui_name, ui_filename=ui_filename - ) - - num_macro_elements = None - root = self.tree.getroot() - for widget in root.iter("widget"): - if widget.get("class") == "caInclude" and widget.get("name") == "cainclude": - for prop in widget.findall("property"): - # Update the macro data string - if prop.get("name") == "macro": - string_elem = prop.find("string") - if string_elem is not None and string_elem.text is not None: - string_elem.text = ( - string_elem.text + ";" + new_macro_element.to_string() - ) - # Dynamically update the count match so caInclude renders all rows - if prop.get("name") == "numberOfItems": - number_elem = prop.find("number") - assert number_elem is not None and number_elem.text is not None - num_macro_elements = int(number_elem.text) - number_elem.text = str(num_macro_elements + 1) - - assert num_macro_elements is not None, "could not find number of macro elements in xml" - return num_macro_elements - - def _update_heights(self, num_macro_elements: int) -> None: - """ - There are 3 heights to be adjusted: - - 1. Main window - 2. Tab widget - 3. Widget inside the tab widget - """ - root = self.tree.getroot() - - ca_include_height = self._get_height_cainclude() - height_to_add = int( - (ca_include_height / num_macro_elements * (num_macro_elements + 1)) - ca_include_height - ) - - # 1. Main window - mainwindow = root.find(".//widget[@class='QMainWindow']") - if mainwindow is not None: - rect = mainwindow.find("./property[@name='geometry']/rect") - assert rect is not None - height = rect.find("height") - assert height is not None and height.text is not None - height.text = str(int(height.text) + height_to_add) - - for widget in root.iter("widget"): - widget_name = widget.get("name") - - # 2. Tab widget - if widget_name == "tabWidget": - rect = widget.find("./property[@name='geometry']/rect") - assert rect is not None - height = rect.find("height") - assert height is not None and height.text is not None - height.text = str(int(height.text) + height_to_add) - - # 3. Widget inside the tab widget - elif widget_name == "cainclude": - rect = widget.find("./property[@name='geometry']/rect") - assert rect is not None - height = rect.find("height") - assert height is not None and height.text is not None - height.text = str(int(height.text) + height_to_add) - - def _get_height_cainclude(self) -> int: - """ - Get the height of the area that includes all the services - """ - root = self.tree.getroot() - cainclude_height = None - - for widget in root.iter("widget"): - widget_name = widget.get("name") - - # Target the caInclude Widget of Required Services - if widget_name == "cainclude": - rect = widget.find("./property[@name='geometry']/rect") - assert rect is not None - height = rect.find("height") - assert height is not None and height.text is not None - cainclude_height = int(height.text) - - assert cainclude_height, "could not get height of cainclude widget" - return cainclude_height - - def write_to_file(self) -> None: - self.tree.write(self.config_path, encoding="utf-8", xml_declaration=True) - - -class _ServiceManagerUIMacroElement(BaseModel): - service_name_upper: str - ui_name: str - panel: int = 1 - svc_exist: int = 1 - ui_filename: str - - def to_string(self) -> str: - return ( - f"IOC=AGEBD-CPCL-{self.service_name_upper}," - f"service={self.service_name_upper}," - f"name={self.ui_name}," - f"panel={self.panel}," - f"svc_exist={self.svc_exist}," - f"panelcmd=A_BD_{self.ui_filename}.ui" - ) diff --git a/cli/src/core/service_creator.py b/cli/src/core/service_creator.py index b48be6b..9a719b8 100644 --- a/cli/src/core/service_creator.py +++ b/cli/src/core/service_creator.py @@ -6,7 +6,6 @@ from pathlib import Path import copier import git -from config.paths import paths from core.exceptions import UVError from core.git import ( assert_clean_repo, @@ -14,15 +13,11 @@ from core.git import ( git_push_changes, switch_branch, ) -from core.models import ( - IocsOverview, - MasterHLANames, - MasterIocSubs, - Service, - ServiceManagerUI, - ServiceRegistry, -) from core.utils import get_git_root +from models import ( + Service, +) +from models.context import ServicesContext logger = logging.getLogger(__name__) @@ -32,34 +27,29 @@ SERVICE_DEST_DIR = REPO_ROOT / "services" class ServiceCreator: - def __init__(self, user: str, ioc_description: str, ui_name: str, ui_filename: str) -> None: - self.user = user - self.ioc_description = ioc_description - self.ui_name = ui_name - self.ui_filename = ui_filename + def __init__(self, ctx: ServicesContext) -> None: + self.ctx = ctx - def add_service(self, name: str): + def add_service( + self, + name: str, + user: str, + ioc_description: str, + ui_name: str, + ui_filename: str, + ): repo = git.Repo(REPO_ROOT) assert_clean_repo(repo) original_branch = repo.active_branch.name - # Keep old configs to revert in case of Exception - registry_old = ServiceRegistry.read_from_config() - registry = ServiceRegistry.read_from_config() - service = registry.add_service(name) - hla_names_old = MasterHLANames.read_from_config() - hla_names = MasterHLANames.read_from_config() - hla_names.add_hla_name(service.name_upper) - iocs_overview_old = IocsOverview.read_from_file() - iocs_overview = IocsOverview.read_from_file() - iocs_overview.add_ioc(service, description=self.ioc_description) - master_ioc_subs_old = MasterIocSubs.read_from_file() - master_ioc_subs = MasterIocSubs.read_from_file() - master_ioc_subs.update_content(service_name=service.name_upper) - service_manager_old = ServiceManagerUI.read_from_file() - service_manager = ServiceManagerUI.read_from_file() - service_manager.update_xml( - service=service, ui_name=self.ui_name, ui_filename=self.ui_filename + backup = self.ctx.snapshot() + + service = self.ctx.service_registry.add_service(name) + self.ctx.master_hla_names.add_hla_name(service.name_upper) + self.ctx.iocs_overview.add_ioc(service, description=ioc_description) + self.ctx.master_ioc_subs.update_content(service_name=service.name_upper) + self.ctx.service_manager_ui.update_xml( + service=service, ui_name=ui_name, ui_filename=ui_filename ) branch_name = f"feature/add-service-{service.dir_name}" @@ -69,23 +59,15 @@ class ServiceCreator: try: self._render_new_service_templates( service=service, - user=self.user, + user=user, + ui_filename=ui_filename, ) self._generate_uv_lock(service=service) - registry.write_to_config() - hla_names.write_to_config() - iocs_overview.write_to_file() - master_ioc_subs.write_to_file() - service_manager.write_to_file() - files_to_stage = [ - paths.relative(paths.get_service_dir(service.dir_name)), - paths.relative(paths.service_registry_file), - paths.relative(paths.master_hla_names_file), - paths.relative(paths.master_ioc_subs_file), - paths.relative(paths.iocs_overview_filename), - paths.relative(paths.service_manager_ui_file), - ] + self.ctx.write_all() + + files_to_stage = self.ctx.paths.get_files_to_stage(service_dir_name=service.dir_name) + # TODO: can't push if branch already exists in remote git_push_changes( repo=repo, @@ -96,119 +78,11 @@ class ServiceCreator: switch_branch(repo, original_branch) except Exception as e: shutil.rmtree(Path(SERVICE_DEST_DIR / service.dir_name), ignore_errors=True) - registry_old.write_to_config() - hla_names_old.write_to_config() - iocs_overview_old.write_to_file() - master_ioc_subs_old.write_to_file() - service_manager_old.write_to_file() + backup.write_all() switch_branch(repo, original_branch) raise e - # TODO: remove - def migrate_service(self, name: str): - repo = git.Repo(REPO_ROOT) - assert_clean_repo(repo) - original_branch = repo.active_branch.name - - # Keep old configs to revert in case of Exception - registry_old = ServiceRegistry.read_from_config() - registry = ServiceRegistry.read_from_config() - service = registry.add_service(name) - iocs_overview_old = IocsOverview.read_from_file() - iocs_overview = IocsOverview.read_from_file() - iocs_overview.add_ioc(service, description=self.ioc_description) - - branch_name = f"feature/add-new-service-all" - # delete_local_branch(repo, branch_name) - switch_branch(repo, branch_name) - - def copy_files_from_current_service(): - # ui - # CURRENT_FILENAME = ( - # REPO_ROOT / ".." / "sls_bd_qt" / f"A_BD_{self.ui_filename}.ui" - # ).resolve() - # DEST_FILENAME = str( - # REPO_ROOT - # / "services" - # / service.dir_name - # / "current" - # / "qt" - # / f"A_BD_{self.ui_filename}.ui" - # ) - # shutil.copyfile(CURRENT_FILENAME, DEST_FILENAME) - - # ioc subs - CURRENT_FILENAME = ( - REPO_ROOT - / ".." - / "A_BD" - / f"AGEBD-CPCL-{service.name_upper}" - / f"AGEBD-CPCL-{service.name_upper}_main.subs" - ).resolve() - DEST_FILENAME = str( - REPO_ROOT - / "services" - / service.dir_name - / "current" - / "ioc" - / f"AGEBD-CPCL-{service.name_upper}_main.subs" - ) - shutil.copyfile(CURRENT_FILENAME, DEST_FILENAME) - - # ioc template - # CURRENT_FILENAME = ( - # REPO_ROOT - # / ".." - # / "A_BD" - # / f"AGEBD-CPCL-{service.name_upper}" - # / f"{service.name_upper}.template" - # ).resolve() - # DEST_FILENAME = str( - # REPO_ROOT - # / "services" - # / service.dir_name - # / "current" - # / "ioc" - # / f"{service.name_upper}.template" - # ) - shutil.copyfile(CURRENT_FILENAME, DEST_FILENAME) - - try: - self._render_new_service_templates( - service=service, - user=self.user, - ) - self._generate_uv_lock(service=service) - registry_old.write_to_config() - iocs_overview.write_to_file() - - copy_files_from_current_service() - - files_to_stage = [ - f"services/{service.dir_name}/current", - # "config/services_registry.yml", - "docs/ioc/user/ioc_overview.md", - ] - # TODO: can't push if branch already exists in remote - git_push_changes( - repo=repo, - branch_name=branch_name, - files_to_stage=files_to_stage, - commit_msg=f"feature: add new service {service.name_lower}", - ) - switch_branch(repo, original_branch) - except Exception as e: - shutil.rmtree(Path(SERVICE_DEST_DIR / service.dir_name), ignore_errors=True) - registry_old.write_to_config() - iocs_overview_old.write_to_file() - switch_branch(repo, original_branch) - raise e - - def _render_new_service_templates( - self, - service: Service, - user: str, - ): + def _render_new_service_templates(self, service: Service, user: str, ui_filename: str): """ Render template files for a new service """ @@ -220,7 +94,7 @@ class ServiceCreator: "service_name_lower": service.name_lower, "service_name_lower_underscores": service.name_lower_underscores, "user": user, - "ui_filename": self.ui_filename, + "ui_filename": ui_filename, }, ) diff --git a/cli/src/models/__init__.py b/cli/src/models/__init__.py new file mode 100644 index 0000000..7dcfbc3 --- /dev/null +++ b/cli/src/models/__init__.py @@ -0,0 +1,20 @@ +# Order matters to avoid circular imports +# ruff: isort: off +from .service import Service +from .iocs_overview import IocsOverview +from .master_hla_names import MasterHLANames +from .master_ioc_subs import MasterIocSubs +from .service_manager_ui import ServiceManagerUI +from .service_registry import ServiceRegistry +from .context import ServicesContext +# ruff: isort: on + +__all__ = [ + "ServicesContext", + "IocsOverview", + "MasterHLANames", + "MasterIocSubs", + "Service", + "ServiceManagerUI", + "ServiceRegistry", +] diff --git a/cli/src/models/context.py b/cli/src/models/context.py new file mode 100644 index 0000000..b375058 --- /dev/null +++ b/cli/src/models/context.py @@ -0,0 +1,54 @@ +from typing import Any, Dict + +from pydantic import BaseModel + +from config.paths import Paths +from models import IocsOverview, MasterHLANames, MasterIocSubs, ServiceManagerUI, ServiceRegistry + + +class ServicesContext(BaseModel): + paths: Paths + + service_registry: ServiceRegistry + master_hla_names: MasterHLANames + iocs_overview: IocsOverview + master_ioc_subs: MasterIocSubs + service_manager_ui: ServiceManagerUI + + @classmethod + def load_from_disk(cls, paths: Paths) -> "ServicesContext": + return cls( + paths=paths, + service_registry=ServiceRegistry.read_from_file(paths.service_registry_file), + master_hla_names=MasterHLANames.read_from_file(paths.master_hla_names_file), + iocs_overview=IocsOverview.read_from_file(paths.iocs_overview_filename), + master_ioc_subs=MasterIocSubs.read_from_file(paths.master_ioc_subs_file), + service_manager_ui=ServiceManagerUI.read_from_file(paths.service_manager_ui_file), + ) + + def snapshot(self) -> "ServicesContext": + """Creates an in-memory backup of all configs for rollback.""" + return ServicesContext( + paths=self.paths, + service_registry=self.service_registry.model_copy(), + master_hla_names=self.master_hla_names.model_copy(), + iocs_overview=self.iocs_overview.model_copy(), + master_ioc_subs=self.master_ioc_subs.model_copy(), + service_manager_ui=self.service_manager_ui.model_copy(), + ) + + def restore_snapshot(self, snapshot: Dict[str, Any]) -> None: + """Restores state from a backup dictionary.""" + self.service_registry = snapshot["service_registry"] + self.master_hla_names = snapshot["master_hla_names"] + self.iocs_overview = snapshot["iocs_overview"] + self.master_ioc_subs = snapshot["master_ioc_subs"] + self.service_manager_ui = snapshot["service_manager_ui"] + + def write_all(self) -> None: + """Persists all modified objects to disk.""" + self.service_registry.write_to_file(self.paths.service_registry_file) + self.master_hla_names.write_to_file(self.paths.master_hla_names_file) + self.iocs_overview.write_to_file(self.paths.iocs_overview_filename) + self.master_ioc_subs.write_to_file(self.paths.master_ioc_subs_file) + self.service_manager_ui.write_to_file(self.paths.service_manager_ui_file) diff --git a/cli/src/models/iocs_overview.py b/cli/src/models/iocs_overview.py new file mode 100644 index 0000000..07779ae --- /dev/null +++ b/cli/src/models/iocs_overview.py @@ -0,0 +1,29 @@ +from pathlib import Path + +from pydantic import BaseModel + +from models import Service + + +class IocsOverview(BaseModel): + content: str + + @classmethod + def read_from_file(cls, path: Path) -> "IocsOverview": + with open(path, "r", encoding="utf-8") as f: + s = f.read() + + return IocsOverview(content=s) + + def add_ioc(self, service: Service, description: str): + new_row = f"| AGEBD-CPCL-{service.name_upper} | {description} |\n" + + # Ensuring it starts on a clean new line if the file didn't end with one + if not self.content.endswith("\n"): + self.content += "\n" + + self.content += new_row + + def write_to_file(self, path: Path): + with open(path, "w") as f: + f.write(self.content) diff --git a/cli/src/models/master_hla_names.py b/cli/src/models/master_hla_names.py new file mode 100644 index 0000000..ec8993b --- /dev/null +++ b/cli/src/models/master_hla_names.py @@ -0,0 +1,23 @@ +from pathlib import Path + +import yaml +from pydantic import BaseModel + + +class MasterHLANames(BaseModel): + hla_apps: list[str] + + @classmethod + def read_from_file(cls, path: Path) -> "MasterHLANames": + with open(path) as f: + d = yaml.safe_load(f) + + hla_names = MasterHLANames.model_validate(d) + return hla_names + + def add_hla_name(self, name: str): + self.hla_apps.append(name) + + def write_to_file(self, path: Path) -> None: + with open(path, "w") as f: + yaml.safe_dump(self.model_dump(), f, indent=4) diff --git a/cli/src/models/master_ioc_subs.py b/cli/src/models/master_ioc_subs.py new file mode 100644 index 0000000..b8435e9 --- /dev/null +++ b/cli/src/models/master_ioc_subs.py @@ -0,0 +1,45 @@ +import re +from pathlib import Path + +from pydantic import BaseModel + + +class MasterIocSubs(BaseModel): + content: str + + @classmethod + def read_from_file(cls, path: Path) -> "MasterIocSubs": + with open(path) as f: + file_content = f.read() + + return MasterIocSubs(content=file_content) + + def update_content( + self, + service_name: str, + starton: str = "0", # TODO: ok defaults? + autooff: str = "0", # TODO: ok defaults? + ): + """ + Parses the configuration text and appends a new row before the closing bracket. + """ + quoted_service = f'"{service_name}"' + new_row = f' {{ "{{{{ agebd_env }}}}", {quoted_service:<22} , "{starton}", "{autooff}" }}' + + # Use regex to find the last closing curly brace of the configuration block + # This targets the line that contains only a lone closing brace, optional spaces, and maybe a comma/semicolon. + pattern = r"(\s*\n\s*\}(?:;|,)?\s*$)" + + if re.search(pattern, self.content, re.MULTILINE): + # Insert our new row right before that closing brace line + self.content = re.sub( + pattern, f"\n{new_row}\\1", self.content, count=1, flags=re.MULTILINE + ) + else: + raise ValueError( + "Could not locate the closing structure format '}' inside the configuration." + ) + + def write_to_file(self, path: Path) -> None: + with open(path, "w") as f: + f.write(self.content) diff --git a/cli/src/models/service.py b/cli/src/models/service.py new file mode 100644 index 0000000..bee5eb5 --- /dev/null +++ b/cli/src/models/service.py @@ -0,0 +1,40 @@ +import re + +from pydantic import BaseModel, field_validator + +from core.enums import ServiceStatus + + +class Service(BaseModel): + name: str + ioc_port: int + status: ServiceStatus + + @property + def name_upper(self): + return self.name.upper() + + @property + def name_lower(self): + return self.name.lower() + + @property + def name_lower_underscores(self): + return self.name.replace("-", "_").lower() + + @property + def dir_name(self): + return self.name.lower() + + def to_dict(self): + return {"name": self.name_lower, "ioc_port": self.ioc_port, "status": str(self.status)} + + class Config: + use_enum_values = True + + @field_validator("name") + @classmethod + def validate_name_chars(cls, value: str) -> str: + if not re.match(r"^[a-z0-9-]+$", value): + raise ValueError("name can only contain lowercase letters, numbers, and dashes") + return value diff --git a/cli/src/models/service_manager_ui.py b/cli/src/models/service_manager_ui.py new file mode 100644 index 0000000..a99d023 --- /dev/null +++ b/cli/src/models/service_manager_ui.py @@ -0,0 +1,136 @@ +import xml.etree.ElementTree as ET +from pathlib import Path +from typing import Any + +from pydantic import BaseModel, ConfigDict + +from models import Service + + +class ServiceManagerUI(BaseModel): + model_config = ConfigDict(arbitrary_types_allowed=True) + + tree: Any + + @classmethod + def read_from_file(cls, path: Path) -> "ServiceManagerUI": + tree = ET.parse(path) + return ServiceManagerUI(tree=tree) + + def update_xml(self, service: Service, ui_name: str, ui_filename: str) -> None: + num_macro_elements = self._update_macro( + service=service, ui_name=ui_name, ui_filename=ui_filename + ) + self._update_heights(num_macro_elements=num_macro_elements) + + def _update_macro(self, service: Service, ui_name: str, ui_filename: str) -> int: + new_macro_element = _ServiceManagerUIMacroElement( + service_name_upper=service.name_upper, ui_name=ui_name, ui_filename=ui_filename + ) + + num_macro_elements = None + root = self.tree.getroot() + for widget in root.iter("widget"): + if widget.get("class") == "caInclude" and widget.get("name") == "cainclude": + for prop in widget.findall("property"): + # Update the macro data string + if prop.get("name") == "macro": + string_elem = prop.find("string") + if string_elem is not None and string_elem.text is not None: + string_elem.text = ( + string_elem.text + ";" + new_macro_element.to_string() + ) + # Dynamically update the count match so caInclude renders all rows + if prop.get("name") == "numberOfItems": + number_elem = prop.find("number") + assert number_elem is not None and number_elem.text is not None + num_macro_elements = int(number_elem.text) + number_elem.text = str(num_macro_elements + 1) + + assert num_macro_elements is not None, "could not find number of macro elements in xml" + return num_macro_elements + + def _update_heights(self, num_macro_elements: int) -> None: + """ + There are 3 heights to be adjusted: + + 1. Main window + 2. Tab widget + 3. Widget inside the tab widget + """ + root = self.tree.getroot() + + ca_include_height = self._get_height_cainclude() + height_to_add = int( + (ca_include_height / num_macro_elements * (num_macro_elements + 1)) - ca_include_height + ) + + # 1. Main window + mainwindow = root.find(".//widget[@class='QMainWindow']") + if mainwindow is not None: + rect = mainwindow.find("./property[@name='geometry']/rect") + assert rect is not None + height = rect.find("height") + assert height is not None and height.text is not None + height.text = str(int(height.text) + height_to_add) + + for widget in root.iter("widget"): + widget_name = widget.get("name") + + # 2. Tab widget + if widget_name == "tabWidget": + rect = widget.find("./property[@name='geometry']/rect") + assert rect is not None + height = rect.find("height") + assert height is not None and height.text is not None + height.text = str(int(height.text) + height_to_add) + + # 3. Widget inside the tab widget + elif widget_name == "cainclude": + rect = widget.find("./property[@name='geometry']/rect") + assert rect is not None + height = rect.find("height") + assert height is not None and height.text is not None + height.text = str(int(height.text) + height_to_add) + + def _get_height_cainclude(self) -> int: + """ + Get the height of the area that includes all the services + """ + root = self.tree.getroot() + cainclude_height = None + + for widget in root.iter("widget"): + widget_name = widget.get("name") + + # Target the caInclude Widget of Required Services + if widget_name == "cainclude": + rect = widget.find("./property[@name='geometry']/rect") + assert rect is not None + height = rect.find("height") + assert height is not None and height.text is not None + cainclude_height = int(height.text) + + assert cainclude_height, "could not get height of cainclude widget" + return cainclude_height + + def write_to_file(self, path: Path) -> None: + self.tree.write(path, encoding="utf-8", xml_declaration=True) + + +class _ServiceManagerUIMacroElement(BaseModel): + service_name_upper: str + ui_name: str + panel: int = 1 + svc_exist: int = 1 + ui_filename: str + + def to_string(self) -> str: + return ( + f"IOC=AGEBD-CPCL-{self.service_name_upper}," + f"service={self.service_name_upper}," + f"name={self.ui_name}," + f"panel={self.panel}," + f"svc_exist={self.svc_exist}," + f"panelcmd=A_BD_{self.ui_filename}.ui" + ) diff --git a/cli/src/models/service_registry.py b/cli/src/models/service_registry.py new file mode 100644 index 0000000..fc52547 --- /dev/null +++ b/cli/src/models/service_registry.py @@ -0,0 +1,46 @@ +from pathlib import Path + +import yaml +from pydantic import BaseModel + +from core.enums import ServiceStatus +from models.service import Service + + +class ServiceRegistry(BaseModel): + next_available_ioc_port: int + services: list[dict] + + @classmethod + def read_from_file(cls, path: Path) -> "ServiceRegistry": + with open(path) as f: + d = yaml.safe_load(f) + + registry = ServiceRegistry.model_validate(d) + return registry + + def get_service(self, name: str) -> Service: + for svc_data in self.services: + service = Service(**svc_data) + if name == service.name_lower: + return service + + raise ValueError(f"Could not find service with name: {name}") + + def get_services(self) -> list[Service]: + return [Service(**svc_data) for svc_data in self.services] + + def add_service(self, name: str) -> Service: + svc = Service( + name=name, + ioc_port=self.next_available_ioc_port, + status=ServiceStatus.ACTIVE, + ) + self.services.append(svc.to_dict()) + self.next_available_ioc_port += 1 + + return svc + + def write_to_file(self, path: Path) -> None: + with open(path, "w") as f: + yaml.safe_dump(self.model_dump(), f, sort_keys=False) diff --git a/cli/src/service.py b/cli/src/service.py index c2032d8..6aa4a30 100644 --- a/cli/src/service.py +++ b/cli/src/service.py @@ -1,8 +1,10 @@ import typer -from core.models import ServiceRegistry +from config.paths import Paths from core.service_creator import ServiceCreator from core.utils import get_git_root, init_logging +from models import ServiceRegistry +from models.context import ServicesContext service = typer.Typer(no_args_is_help=True) @@ -40,46 +42,21 @@ def add( ), ): init_logging() - service_creator = ServiceCreator( + ctx = ServicesContext.load_from_disk(Paths()) + service_creator = ServiceCreator(ctx=ctx) + service_creator.add_service( + name=name, user=user, # TODO: read from 'whoami'? ioc_description=ioc_description, ui_name=ui_name, # TODO: make this standard? ui_filename=ui_filename, # TODO: make this standard? ) - service_creator.add_service(name=name) - - -# TODO: remove -@service.command(no_args_is_help=True) -def migrate( - name: str = typer.Option(..., "--name", "-n"), - user: str = typer.Option(..., "--user", "-u"), - ioc_description: str = typer.Option( - ..., - "--ioc-description", - "-d", - help="See /docs/user/ioc/ioc_overview.md for examples", - ), - ui_filename: str = typer.Option( - ..., - "--ui-filename", - "-f", - help="The in 'A_BD_.ui'; See /qt/A_BD_ServiceManager.ui for examples", - ), -): - init_logging() - service_creator = ServiceCreator( - user=user, # TODO: read from 'whoami'? - ioc_description=ioc_description, - ui_name="", - ui_filename=ui_filename, # TODO: make this standard? - ) - service_creator.migrate_service(name=name) @service.command() def list(): - registry = ServiceRegistry.read_from_config() + paths = Paths() + registry = ServiceRegistry.read_from_file(paths.service_registry_file) services = registry.get_services() for service in services: print(service) diff --git a/cli/tests/test_paths.py b/cli/tests/test_paths.py deleted file mode 100644 index 1269557..0000000 --- a/cli/tests/test_paths.py +++ /dev/null @@ -1,8 +0,0 @@ -from config.paths import paths - - -def test_paths(): - """ - Will fail when paths is imported if a path is invalid - """ - assert True diff --git a/cli/tests/test_service.py b/cli/tests/test_service.py deleted file mode 100644 index ee03a8e..0000000 --- a/cli/tests/test_service.py +++ /dev/null @@ -1,6 +0,0 @@ -from core.models import IocsOverview -from core.service_creator import ServiceCreator - - -def test_add_new_service(): - pass diff --git a/cli/tests/test_service_creator.py b/cli/tests/test_service_creator.py new file mode 100644 index 0000000..88341aa --- /dev/null +++ b/cli/tests/test_service_creator.py @@ -0,0 +1,16 @@ +from config.paths import Paths +from core.service_creator import ServiceCreator +from models.context import ServicesContext + + +def test_add_new_service(): + paths = Paths() + ctx = ServicesContext.load_from_disk(paths) + creator = ServiceCreator(ctx=ctx) + # creator.add_service( + # name="test-service", + # user="test-user", + # ioc_description="test ioc description", + # ui_name="Test UI Name", + # ui_filename="TestUiName", + # )