From d3b941134e078e2dfc790956ce9f6e1c56677f24 Mon Sep 17 00:00:00 2001 From: pyr0ball Date: Wed, 25 Feb 2026 08:13:14 -0800 Subject: [PATCH] feat: integration base class + auto-discovery registry --- scripts/integrations/__init__.py | 44 +++++++++++ scripts/integrations/base.py | 77 +++++++++++++++++++ tests/test_integrations.py | 128 +++++++++++++++++++++++++++++++ 3 files changed, 249 insertions(+) create mode 100644 scripts/integrations/__init__.py create mode 100644 scripts/integrations/base.py create mode 100644 tests/test_integrations.py diff --git a/scripts/integrations/__init__.py b/scripts/integrations/__init__.py new file mode 100644 index 0000000..48df066 --- /dev/null +++ b/scripts/integrations/__init__.py @@ -0,0 +1,44 @@ +"""Integration registry — auto-discovers all IntegrationBase subclasses. + +Import this module to get REGISTRY: {name: IntegrationClass}. +Integration modules are imported here; only successfully imported ones +appear in the registry. +""" +from __future__ import annotations +from scripts.integrations.base import IntegrationBase + +# Import all integration modules to register their subclasses. +# Wrapped in try/except so missing modules don't break the registry. +_INTEGRATION_MODULES = [ + "scripts.integrations.notion", + "scripts.integrations.google_drive", + "scripts.integrations.google_sheets", + "scripts.integrations.airtable", + "scripts.integrations.dropbox", + "scripts.integrations.onedrive", + "scripts.integrations.mega", + "scripts.integrations.nextcloud", + "scripts.integrations.google_calendar", + "scripts.integrations.apple_calendar", + "scripts.integrations.slack", + "scripts.integrations.discord", + "scripts.integrations.home_assistant", +] + +for _mod in _INTEGRATION_MODULES: + try: + __import__(_mod) + except ImportError: + pass # module not yet implemented or missing optional dependency + + +def _build_registry() -> dict[str, type[IntegrationBase]]: + """Collect all IntegrationBase subclasses that have a name attribute.""" + registry: dict[str, type[IntegrationBase]] = {} + for cls in IntegrationBase.__subclasses__(): + if hasattr(cls, "name") and cls.name: + registry[cls.name] = cls + return registry + + +REGISTRY: dict[str, type[IntegrationBase]] = _build_registry() diff --git a/scripts/integrations/base.py b/scripts/integrations/base.py new file mode 100644 index 0000000..56f19d0 --- /dev/null +++ b/scripts/integrations/base.py @@ -0,0 +1,77 @@ +"""Base class for all Peregrine integrations.""" +from __future__ import annotations +from abc import ABC, abstractmethod +from pathlib import Path +import yaml + + +class IntegrationBase(ABC): + """All integrations inherit from this class. + + Subclasses must declare class-level attributes: + name : str — machine key, matches yaml filename (e.g. "notion") + label : str — display name (e.g. "Notion") + tier : str — minimum tier required: "free" | "paid" | "premium" + """ + + name: str + label: str + tier: str + + @abstractmethod + def fields(self) -> list[dict]: + """Return form field definitions for the wizard connection card. + + Each dict must contain: + key : str — yaml key for config + label : str — display label + type : str — "text" | "password" | "url" | "checkbox" + placeholder : str — hint text + required : bool — whether the field must be non-empty to connect + help : str — help tooltip text + """ + + @abstractmethod + def connect(self, config: dict) -> bool: + """Store config in memory, return True if required fields are present. + + Does not verify credentials — call test() for that. + """ + + @abstractmethod + def test(self) -> bool: + """Verify the stored credentials actually work. Returns True on success.""" + + def sync(self, jobs: list[dict]) -> int: + """Push jobs to the external service. Returns count synced. + + Override in subclasses that support job syncing (e.g. Notion, Airtable). + Default implementation is a no-op returning 0. + """ + return 0 + + @classmethod + def config_path(cls, config_dir: Path) -> Path: + """Return the path where this integration's config yaml is stored.""" + return config_dir / "integrations" / f"{cls.name}.yaml" + + @classmethod + def is_configured(cls, config_dir: Path) -> bool: + """Return True if a config file exists for this integration.""" + return cls.config_path(config_dir).exists() + + def save_config(self, config: dict, config_dir: Path) -> None: + """Write config to config/integrations/.yaml. + + Only call this after test() returns True. + """ + path = self.config_path(config_dir) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(yaml.dump(config, default_flow_style=False, allow_unicode=True)) + + def load_config(self, config_dir: Path) -> dict: + """Load and return this integration's config yaml, or {} if not configured.""" + path = self.config_path(config_dir) + if not path.exists(): + return {} + return yaml.safe_load(path.read_text()) or {} diff --git a/tests/test_integrations.py b/tests/test_integrations.py new file mode 100644 index 0000000..a858792 --- /dev/null +++ b/tests/test_integrations.py @@ -0,0 +1,128 @@ +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).parent.parent)) + + +def test_base_class_is_importable(): + from scripts.integrations.base import IntegrationBase + assert IntegrationBase is not None + + +def test_base_class_is_abstract(): + from scripts.integrations.base import IntegrationBase + import inspect + assert inspect.isabstract(IntegrationBase) + + +def test_registry_is_importable(): + from scripts.integrations import REGISTRY + assert isinstance(REGISTRY, dict) + + +def test_registry_returns_integration_base_subclasses(): + """Any entries in the registry must be IntegrationBase subclasses.""" + from scripts.integrations import REGISTRY + from scripts.integrations.base import IntegrationBase + for name, cls in REGISTRY.items(): + assert issubclass(cls, IntegrationBase), f"{name} is not an IntegrationBase subclass" + + +def test_base_class_has_required_class_attributes(): + """Subclasses must define name, label, tier at the class level.""" + from scripts.integrations.base import IntegrationBase + + class ConcreteIntegration(IntegrationBase): + name = "test" + label = "Test Integration" + tier = "free" + + def fields(self): return [] + def connect(self, config): return True + def test(self): return True + + instance = ConcreteIntegration() + assert instance.name == "test" + assert instance.label == "Test Integration" + assert instance.tier == "free" + + +def test_fields_returns_list_of_dicts(): + """fields() must return a list of dicts with key, label, type.""" + from scripts.integrations.base import IntegrationBase + + class TestIntegration(IntegrationBase): + name = "test2" + label = "Test 2" + tier = "free" + + def fields(self): + return [{"key": "token", "label": "API Token", "type": "password", + "placeholder": "abc", "required": True, "help": ""}] + + def connect(self, config): return bool(config.get("token")) + def test(self): return True + + inst = TestIntegration() + result = inst.fields() + assert isinstance(result, list) + assert len(result) == 1 + assert result[0]["key"] == "token" + + +def test_save_and_load_config(tmp_path): + """save_config writes yaml; load_config reads it back.""" + from scripts.integrations.base import IntegrationBase + import yaml + + class TestIntegration(IntegrationBase): + name = "savetest" + label = "Save Test" + tier = "free" + def fields(self): return [] + def connect(self, config): return True + def test(self): return True + + inst = TestIntegration() + config = {"token": "abc123", "database_id": "xyz"} + inst.save_config(config, tmp_path) + + saved_file = tmp_path / "integrations" / "savetest.yaml" + assert saved_file.exists() + + loaded = inst.load_config(tmp_path) + assert loaded["token"] == "abc123" + assert loaded["database_id"] == "xyz" + + +def test_is_configured(tmp_path): + from scripts.integrations.base import IntegrationBase + + class TestIntegration(IntegrationBase): + name = "cfgtest" + label = "Cfg Test" + tier = "free" + def fields(self): return [] + def connect(self, config): return True + def test(self): return True + + assert TestIntegration.is_configured(tmp_path) is False + # Create the file + (tmp_path / "integrations").mkdir(parents=True) + (tmp_path / "integrations" / "cfgtest.yaml").write_text("token: x\n") + assert TestIntegration.is_configured(tmp_path) is True + + +def test_sync_default_returns_zero(): + from scripts.integrations.base import IntegrationBase + + class TestIntegration(IntegrationBase): + name = "synctest" + label = "Sync Test" + tier = "free" + def fields(self): return [] + def connect(self, config): return True + def test(self): return True + + inst = TestIntegration() + assert inst.sync([]) == 0 + assert inst.sync([{"id": 1}]) == 0