diff --git a/core/src/agentforge_core/registry/policy_store.py b/core/src/agentforge_core/registry/policy_store.py new file mode 100644 index 0000000..e81e03e --- /dev/null +++ b/core/src/agentforge_core/registry/policy_store.py @@ -0,0 +1,64 @@ +"""PolicyStore basado en sistema de ficheros (YAML por versión + index.yaml).""" + +from __future__ import annotations + +from datetime import datetime +from pathlib import Path +from typing import Any + +import yaml + +from agentforge_core.domain.policy import PolicyDefinition, PolicyVersionMeta + + +class FileSystemPolicyStore: + """Lee políticas de policies//{index.yaml,versions/vN.yaml}.""" + + def __init__(self, root: Path) -> None: + self._root = Path(root) + + def list_policies(self) -> list[PolicyDefinition]: + if not self._root.exists(): + return [] + result: list[PolicyDefinition] = [] + for d in sorted(self._root.iterdir()): + if d.is_dir() and (d / "index.yaml").exists(): + result.append(self.get_policy(d.name)) + return result + + def get_policy(self, name: str, version: str | None = None) -> PolicyDefinition: + index = self._load_index(name) + v = version or index["active_version"] + path = self._root / name / "versions" / f"{v}.yaml" + if not path.exists(): + raise FileNotFoundError(f"policy version not found: {name}@{v}") + with path.open(encoding="utf-8") as f: + data = yaml.safe_load(f) + return PolicyDefinition.model_validate(data) + + def list_versions(self, name: str) -> list[PolicyVersionMeta]: + index = self._load_index(name) + return [ + PolicyVersionMeta( + id=v["id"], + hash=v["hash"], + author=v["author"], + message=v["message"], + created_at=self._parse_dt(v["created_at"]), + ) + for v in index["versions"] + ] + + def _load_index(self, name: str) -> dict[str, Any]: + index_path = self._root / name / "index.yaml" + if not index_path.exists(): + raise FileNotFoundError(f"policy not found: {name}") + with index_path.open(encoding="utf-8") as f: + data: dict[str, Any] = yaml.safe_load(f) + return data + + @staticmethod + def _parse_dt(value: str | datetime) -> datetime: + if isinstance(value, datetime): + return value + return datetime.fromisoformat(value.replace("Z", "+00:00")) diff --git a/pyproject.toml b/pyproject.toml index 2a4ffc3..ce048e7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,7 +29,7 @@ disallow_untyped_defs = true plugins = ["pydantic.mypy"] [[tool.mypy.overrides]] -module = ["guardrails.*", "presidio_analyzer.*", "nemoguardrails.*", "langgraph.*"] +module = ["guardrails.*", "presidio_analyzer.*", "nemoguardrails.*", "langgraph.*", "yaml"] ignore_missing_imports = true [tool.pytest.ini_options] diff --git a/tests/fixtures/policies/test_policy/index.yaml b/tests/fixtures/policies/test_policy/index.yaml new file mode 100644 index 0000000..6c8330a --- /dev/null +++ b/tests/fixtures/policies/test_policy/index.yaml @@ -0,0 +1,8 @@ +name: test_policy +versions: + - id: v1 + hash: deadbeef + author: pytest + message: fixture inicial + created_at: 2026-01-01T00:00:00Z +active_version: v1 diff --git a/tests/fixtures/policies/test_policy/versions/v1.yaml b/tests/fixtures/policies/test_policy/versions/v1.yaml new file mode 100644 index 0000000..005a9ed --- /dev/null +++ b/tests/fixtures/policies/test_policy/versions/v1.yaml @@ -0,0 +1,13 @@ +name: test_policy +version: v1 +description: Política de prueba +input_validators: + - type: detect_pii + config: + entities: [PERSON, EMAIL] + severity_on_match: block +output_validators: + - type: schema_match + config: + severity_on_mismatch: block +on_validator_error: fail_closed diff --git a/tests/unit/test_policy_store.py b/tests/unit/test_policy_store.py new file mode 100644 index 0000000..1bbbdf4 --- /dev/null +++ b/tests/unit/test_policy_store.py @@ -0,0 +1,36 @@ +"""Tests del PolicyStore.""" + +from pathlib import Path + +import pytest + +from agentforge_core.registry.policy_store import FileSystemPolicyStore + +FIXTURES = Path(__file__).parent.parent / "fixtures" / "policies" + + +def test_list_policies() -> None: + store = FileSystemPolicyStore(FIXTURES) + names = [p.name for p in store.list_policies()] + assert "test_policy" in names + + +def test_get_policy_devuelve_version_activa() -> None: + store = FileSystemPolicyStore(FIXTURES) + p = store.get_policy("test_policy") + assert p.version == "v1" + assert len(p.input_validators) == 1 + assert p.input_validators[0].type == "detect_pii" + + +def test_get_policy_inexistente_lanza_error() -> None: + store = FileSystemPolicyStore(FIXTURES) + with pytest.raises(FileNotFoundError): + store.get_policy("inexistente") + + +def test_list_versions() -> None: + store = FileSystemPolicyStore(FIXTURES) + metas = store.list_versions("test_policy") + assert metas[0].id == "v1" + assert metas[0].author == "pytest"