add: first commit

gmmcosta15 · gmmcosta15 · commit bf4362563a0b · 2026-04-15T10:05:28.000+01:00
diff --git a/tests/updater/__init__.py b/tests/updater/__init__.py
diff --git a/tests/updater/test_components_unit.py b/tests/updater/test_components_unit.py
@@ -0,0 +1,167 @@
+"""Unit tests for updater.components """
+import logging
+import pytest
+from pathlib import Path
+from unittest.mock import patch, mock_open
+
+from updater.components import (
+    ComponentConfig, ComponentStatus, load_components
+)
+
+BUNDLE_YAML = """
+pool_internal_minutes: 30
+components:
+  - name: klipper
+  - type: git
+  - path: ~/klipper
+  - service: klipper.service
+  - reset_mode: hard
+  - order: 1
+  - name: BlocksScreen
+  - type: git
+  - path: ~/BlocksScreen
+  - service: BlocksScreen.service
+  - reset_mode: hard
+  - order: 99
+"""
+
+OVERRIDE_YAML = """
+components:
+    - name: klipper
+      path: ~/costum_klipper
+    - name: my-plugin
+      type: git
+      path: ~/my-plugin
+      service: my-plugin.service
+      reset_mode: hard
+      order: 5
+"""
+
+def _mock_load(bundled: str, override: str | None = None):
+    import updater.components as mod
+    bundled_path = Path(mod.__file__).parent / "components.yaml"
+
+    def fake_open(path, *a, **kw):
+        if Path(path) == bundled_path:
+            return mock_open(read_data=bundled)()
+        if override is not None:
+            return mock_open(read_data=override)()
+        raise FileNotFoundError(Path)
+    return fake_open
+
+
+class TestLoadComponents:
+    def test_load_return_sorted_by_order(self):
+        with patch("builtins.open", _mock_load(BUNDLE_YAML)), patch("pathlib.Path.exists", return_value=False), patch("pathlib.Path.is_dir", return_value=True):
+                components = load_components()
+        names = [c.name for c in components]
+        assert names.index("klipper") < names.index("BlocksScreen")
+    
+    def test_missing_override_file_silently_ignored(self):
+        with patch("builtins.open", _mock_load(BUNDLE_YAML)), patch("pathlib.Path.exists", return_value=False), patch("pathlib.Path.is_dir", return_value=True):
+            components = load_components()
+        assert len(components) >= 2
+
+class TestYamlMerge:
+    def test_ovvrides_path_by_name(self):
+        with patch("builtins.open", _mock_load(BUNDLE_YAML)), patch("pathlib.Path.exists", return_value=True), patch("pathlib.Path.is_dir", return_value=True):
+            components = load_components()
+        klipper = next(c for c in components if c.name == "klipper")
+        assert str(klipper.path).endswith("costum_klipper")
+        assert klipper.service == "klipper.service"
+
+    def test_appends_new_componentself(self):
+        with patch("builtins.open", _mock_load(BUNDLE_YAML, OVERRIDE_YAML)), patch("pathlib.Path.exists", return_value=True), patch("pathlib.Path.is_dir", return_value=True):
+            components = load_components()
+        assert any(c.name == "my-plugin" for c in components)
+    def test_keeps_bundled_fields_not_in_override(self):
+        with patch("builtins.open", _mock_load(BUNDLE_YAML, OVERRIDE_YAML)), patch("pathlib.Path.exists", return_value=True), patch("pathlib.Path.is_dir", return_value=True):
+            components = load_components()
+        klipper = next(c for c in components if c.name == "klipper")
+        assert klipper.type == "git"
+        assert klipper.reset_mode == "hard"
+
+    def test_syntax_error_falls_back_to_bundled(self, caplog):
+        with patch("builtins.open", _mock_load(BUNDLE_YAML, OVERRIDE_YAML)), patch("pathlib.Path.exists", return_value=True), patch("pathlib.Path.is_dir", return_value=True), \
+            caplog.at_level(logging.ERROR, logger="updater.components"):
+            components = load_components()
+        names = [c.name for c in components]
+        assert "klipper" in names
+        assert "my-plugin" not in names
+
+
+class TestValidation:
+    def test_invalid_component_skipped_with_warning(self, caplog):
+        bad_yaml = """
+components:
+    - name: bad
+      type: git
+      path: /etc/passwd
+      service: bad.service
+      order: 10
+"""
+        with patch("builtins.open", _mock_load(bad_yaml)), \
+            patch("pathlib.Path.exists", return_value=True), \
+            caplog.at_level(logging.WARNING, logger="updater.components"):
+            components = load_components()
+        assert not any(c.name == "bad" for c in components)
+        assert any("bad" in r.message for r in caplog.records)
+
+    def test_path_outside_home_rejected(self):
+        outside_yaml = """
+components:
+    - name: outside
+      type: git
+      path: /etc/outside
+      service: outside.service
+      order: 10
+"""
+        with patch("builtins.open", _mock_load(outside_yaml)), \
+            patch("pathlib.Path.exists", return_value=False):
+            components = load_components()
+        assert not any(c.name == "outside" for c in components)
+
+    def test_service_name_with_slash_rejected(self):
+        bad_yaml = """
+components:
+    - name: bad
+      type: git
+      path: ~/bad
+      service: ../bad.service
+      order: 10
+"""
+        with patch("builtins.open", _mock_load(bad_yaml)), \
+            patch("pathlib.Path.exists", return_value=False):
+            components = load_components()
+        assert not any(c.name == "evil" for c in components)
+
+    def test_service_name_with_shell_metachar_rejected(self):
+        for metachar in [";", "&", "|", "$", "`"]:
+            bad_yaml = f"""
+components:
+- name: bad
+type: git
+path: ~/bad
+service: bad{metachar}cmd.service
+order: 10
+"""
+            with patch("builtins.open", _mock_load(bad_yaml)), \
+                patch("pathlib.Path.exists", return_value=False):
+                components = load_components()
+            assert not any(c.name == "bad" for c in components), f"metachar {metachar!r} not rejected"
+
+    def test_service_name_valid_pattern_accepted(self):
+        good_yaml = """
+components:
+    - name: my-plugin
+      type: git
+      path: ~/my-plugin
+      service: my-plugin.service
+      order: 5
+"""
+        with patch("builtins.open", _mock_load(good_yaml)), \
+            patch("pathlib.Path.exists", return_value=False), \
+            patch("pathlib.Path.is_dir", return_value=True):
+            components = load_components()
+        assert any(c.name == "my-plugin" for c in components)
+ 
diff --git a/updater/__init__.py b/updater/__init__.py
@@ -0,0 +1,4 @@
+from .components import load_components
+from .models import ComponentConfig, ComponentStatus
+
+__all__ = ["ComponentConfig", "ComponentStatus", "load_components"]
diff --git a/updater/components.py b/updater/components.py
@@ -0,0 +1,112 @@
+from __future__ import annotations
+
+import logging
+import re
+from pathlib import Path
+
+import yaml
+
+from .models import ComponentConfig, ComponentStatus
+
+__all__ = ["ComponentConfig", "ComponentStatus", "load_components"]
+
+logger = logging.getLogger(__name__)
+
+_SERVICE_RE = re.compile(r"^[a-zA-Z0-9@:._-]+\.service$")
+_SERVICE_BANNED = set("/\\..;&|$`") | {" ", "\t"}
+
+OVERRIDE_PATH = Path("~/printer_data/config/blockscreen_updater.yaml").expanduser()
+
+
+def _validate_service(name: str) -> bool:
+    if not name:
+        return False
+    if any(ch in _SERVICE_BANNED for ch in name):
+        return False
+    return bool(_SERVICE_RE.match(name))
+
+
+def _validate_component(data: dict) -> ComponentConfig | None:
+    name = data.get("name", "")
+    if not isinstance(name, str) or not name:
+        logger.warning("Component missing name,  skipped")
+        return None
+
+    comp_type = data.get("type", "")
+    if comp_type not in ("git", "apt"):
+        logger.warning("Component %r has invalid type %r — skipped", name, comp_type)
+        return None
+
+    if comp_type == "git":
+        raw_path = data.get("path")
+        if not raw_path:
+            logger.warning("Component %r missing path — skipped", name)
+            return None
+        resolved = Path(str(raw_path)).expanduser().resolve()
+        if not resolved.is_relative_to(Path.home()):
+            logger.warning("Component %r path escapes home dir — skipped", name)
+            return None
+
+        service = data.get("service")
+        if service is not None and not _validate_service(str(service)):
+            logger.warning(
+                "Component %r has invalid service %r, skipped", name, service
+            )
+            return None
+
+        return ComponentConfig(
+            name=name,
+            kind=comp_type,
+            path=resolved,
+            service=str(service) if service else None,
+            reset_mode=data.get("reset_mode", "hard"),
+            order=int(data.get("order", 50)),
+        )
+    return ComponentConfig(name=name, kind="apt", order=int(data.get("order", 50)))
+
+
+def _merge(base: list[dict], override: list[dict]) -> list[dict]:
+    base_by_name = {c["name"]: dict(c) for c in base if "name" in c}
+    for entry in override:
+        entry_name = entry.get("name")
+        if not entry_name:
+            continue
+        if entry_name in base_by_name:
+            for k, v in entry.items():
+                if v is not None:
+                    base_by_name[entry_name][k] = v
+        else:
+            base_by_name[entry_name] = dict(entry)
+    return list(base_by_name.values())
+
+
+def load_components() -> list[ComponentConfig]:
+    """Load and validate component definitions from bundled YAML, merged with user override."""
+    bundled_path = Path(__file__).parent / "components.yaml"
+    with open(bundled_path) as f:
+        bundled_data = yaml.safe_load(f)
+
+    raw_components: list[dict] = bundled_data.get("components", [])
+
+    if OVERRIDE_PATH.exists():
+        try:
+            with open(OVERRIDE_PATH) as f:
+                override_data = yaml.safe_load(f)
+            if isinstance(override_data, dict):
+                raw_components = _merge(
+                    raw_components,
+                    override_data.get("components", []),
+                )
+        except Exception as exc:
+            logger.error(
+                "Failed to load override YAML %s: %s",
+                OVERRIDE_PATH,
+                exc,
+            )
+    configs: list[ComponentConfig] = []
+    for entry in raw_components:
+        cfg = _validate_component(entry)
+        if cfg is not None:
+            configs.append(cfg)
+    configs.sort(key=lambda c: c.order)
+    return configs
diff --git a/updater/components.yaml b/updater/components.yaml
@@ -0,0 +1,33 @@
+pool_interval_minutes: 1440
+
+components:
+  - name: klipper
+    type: git
+    path: ~/klipper
+    service: klipper.service
+    reset_mode: hard
+    order: 1
+
+  - name: moonraker
+    type: git
+    path: ~/moonraker/
+    service: moonraker.service
+    reset_mode: hard
+    order: 2
+
+  - name: RF50-Klipper
+    type: git
+    path: ~/RF50-Klipper
+    reset_mode: hard
+    order: 3
+
+  - name: system
+    type: apt
+    order: 4
+
+  - name: BlocksScreen
+    type: git
+    path: ~/BlocksScreen
+    service: BlocksScreen.service
+    reset_mode: hard
+    order: 99
diff --git a/updater/models.py b/updater/models.py
@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+
+
+@dataclass
+class ComponentConfig:
+    name: str
+    kind: str
+    path: Path | None = None
+    service: str | None = None
+    reset_mode: str = "hard"
+    order: int = 50
+
+
+@dataclass
+class ComponentStatus:
+    name: str
+    commits_behind: int = 0
+    current_hash: str = ""
+    remote_url: str = ""
+    packages_upgradable: int = 0
+    error: str | None = None
