128 lines
3.9 KiB
Python
128 lines
3.9 KiB
Python
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
|