Coverage for node / src / stigmem_node / plugins / manifest.py: 90%

1"""Plugin manifest validation.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Callable 

6from typing import Any 

7 

8from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator 

9 

10from .capabilities import CAPABILITY_ALLOWLIST 

11from .hooks import KNOWN_HOOKS 

12 

13 

14class PluginManifest(BaseModel): 

15 """Minimum manifest accepted by PR 4-INF.1 manual registration.""" 

16 

17 name: str = Field(pattern=r"^[a-z][a-z0-9-]{2,63}$") 

18 version: str = Field( 

19 pattern=r"^\d+\.\d+\.\d+(?:(?:a|b|rc)\d+|[-+][a-zA-Z0-9.-]+)?$" 

20 ) 

21 requires_stigmem: str = ">=0.9.0a9" 

22 capabilities: frozenset[str] = frozenset() 

23 async_safe: bool = True 

24 hooks: dict[str, Callable[..., Any]] = Field(default_factory=dict) 

25 routes: tuple[Any, ...] = () 

26 health_check: Callable[..., Any] | None = None 

27 config_schema: type[BaseModel] | None = None 

28 depends_on: frozenset[str] = frozenset() 

29 

30 model_config = ConfigDict(arbitrary_types_allowed=True, frozen=True) 

31 

32 @field_validator("capabilities") 

33 @classmethod 

34 def _validate_capabilities(cls, capabilities: frozenset[str]) -> frozenset[str]: 

35 unknown = sorted(set(capabilities) - CAPABILITY_ALLOWLIST) 

36 if unknown: 

37 raise ValueError(f"unknown plugin capabilities: {', '.join(unknown)}") 

38 return capabilities 

39 

40 @field_validator("hooks") 

41 @classmethod 

42 def _validate_hook_names( 

43 cls, hooks: dict[str, Callable[..., Any]] 

44 ) -> dict[str, Callable[..., Any]]: 

45 unknown = sorted(set(hooks) - KNOWN_HOOKS) 

46 if unknown: 

47 raise ValueError(f"unknown plugin hooks: {', '.join(unknown)}") 

48 for name, handler in hooks.items(): 

49 if not callable(handler): 49 ↛ 50line 49 didn't jump to line 50 because the condition on line 49 was never true

50 raise ValueError(f"handler for hook {name!r} is not callable") 

51 return hooks 

52 

53 @field_validator("health_check") 

54 @classmethod 

55 def _validate_health_check( 

56 cls, health_check: Callable[..., Any] | None 

57 ) -> Callable[..., Any] | None: 

58 if health_check is not None and not callable(health_check): 58 ↛ 59line 58 didn't jump to line 59 because the condition on line 58 was never true

59 raise ValueError("health_check must be callable") 

60 return health_check 

61 

62 @model_validator(mode="after") 

63 def _validate_self_dependency(self) -> PluginManifest: 

64 if self.name in self.depends_on: 64 ↛ 65line 64 didn't jump to line 65 because the condition on line 64 was never true

65 raise ValueError("plugin cannot depend on itself") 

66 return self