diff --git a/src/infra_controller/discovery.py b/src/infra_controller/discovery.py index ce5baa4..9958c2d 100644 --- a/src/infra_controller/discovery.py +++ b/src/infra_controller/discovery.py @@ -5,6 +5,7 @@ from datetime import datetime import logging from pathlib import Path import subprocess +import re import tomllib import yaml @@ -21,6 +22,50 @@ class InfraMetadata: requires: dict[str, object] metadata: dict[str, object] = field(default_factory=dict) + _PROJECT_RE = re.compile(r"^[A-Za-z0-9._-]+$") + + @classmethod + def _validate_project(cls, project: object, path: Path) -> str: + if not isinstance(project, str): + raise ValueError(f"Invalid 'project' (expected string) in {path}") + if not project.strip(): + raise ValueError(f"Invalid 'project' (empty) in {path}") + if cls._PROJECT_RE.fullmatch(project.strip()) is None: + raise ValueError( + f"Invalid 'project' (must match {cls._PROJECT_RE.pattern}) in {path}" + ) + return project.strip() + + @classmethod + def _validate_requires(cls, requires: object, path: Path) -> dict[str, object]: + if requires is None: + requires = {} + if not isinstance(requires, dict): + raise ValueError(f"Invalid 'requires' (expected mapping) in {path}") + + services = requires.get("services") + if services is None: + return dict(requires) + + if isinstance(services, str): + if not services.strip(): + raise ValueError(f"Invalid 'requires.services' (empty string) in {path}") + return dict(requires) + + if isinstance(services, list): + if not services: + raise ValueError(f"Invalid 'requires.services' (empty list) in {path}") + for idx, item in enumerate(services): + if not isinstance(item, str) or not item.strip(): + raise ValueError( + f"Invalid 'requires.services[{idx}]' (expected non-empty string) in {path}" + ) + return dict(requires) + + raise ValueError( + f"Invalid 'requires.services' (expected string or list of strings) in {path}" + ) + @classmethod def from_file(cls, path: Path) -> "InfraMetadata": if path.suffix == ".toml": @@ -32,8 +77,14 @@ class InfraMetadata: else: raise ValueError(f"Unsupported format: {path}") - project = data.get("project") or path.parent.name - requires = data.get("requires") or {} + if not isinstance(data, dict): + raise ValueError(f"Invalid config (expected mapping at top level) in {path}") + + project_val = data.get("project") + if project_val is None: + project_val = path.parent.name + project = cls._validate_project(project_val, path) + requires = cls._validate_requires(data.get("requires"), path) metadata = {k: v for k, v in data.items() if k not in {"project", "requires"}} return cls(project=str(project), requires=dict(requires), metadata=metadata) diff --git a/tests/test_infra_metadata.py b/tests/test_infra_metadata.py new file mode 100644 index 0000000..03103c9 --- /dev/null +++ b/tests/test_infra_metadata.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest + +from infra_controller.discovery import InfraMetadata + + +def _write(path: Path, content: str) -> Path: + path.write_text(content, encoding="utf-8") + return path + + +def test_infra_metadata_toml_valid(tmp_path: Path) -> None: + p = _write( + tmp_path / ".infra.toml", + """ +project = "my-app" + +[requires] +services = ["grafana", "prometheus"] +""".lstrip(), + ) + + md = InfraMetadata.from_file(p) + assert md.project == "my-app" + assert md.requires["services"] == ["grafana", "prometheus"] + + +def test_infra_metadata_yaml_valid_uses_dir_name_when_project_missing(tmp_path: Path) -> None: + proj_dir = tmp_path / "some-app" + proj_dir.mkdir() + p = _write( + proj_dir / ".infra.yml", + """ +requires: + services: grafana +""".lstrip(), + ) + + md = InfraMetadata.from_file(p) + assert md.project == "some-app" + assert md.requires["services"] == "grafana" + + +def test_infra_metadata_invalid_top_level(tmp_path: Path) -> None: + p = _write(tmp_path / ".infra.yml", "- not-a-mapping\n") + + with pytest.raises(ValueError, match=r"expected mapping"): + InfraMetadata.from_file(p) + + +def test_infra_metadata_invalid_project(tmp_path: Path) -> None: + p = _write( + tmp_path / ".infra.toml", + """ +project = "bad name" +[requires] +services = "grafana" +""".lstrip(), + ) + + with pytest.raises(ValueError, match=r"Invalid 'project'"): + InfraMetadata.from_file(p) + + +def test_infra_metadata_invalid_requires_type(tmp_path: Path) -> None: + p = _write( + tmp_path / ".infra.yml", + """ +project: app +requires: not-a-mapping +""".lstrip(), + ) + + with pytest.raises(ValueError, match=r"Invalid 'requires'"): + InfraMetadata.from_file(p) + + +@pytest.mark.parametrize( + "services", + [ + "[]", + "[\"\"]", + "[1]", + ], +) +def test_infra_metadata_invalid_requires_services_list(tmp_path: Path, services: str) -> None: + p = _write( + tmp_path / ".infra.toml", + ( + "project = \"app\"\n\n" + "[requires]\n" + f"services = {services}\n" + ), + ) + + with pytest.raises(ValueError, match=r"requires\.services"): + InfraMetadata.from_file(p) + + +def test_infra_metadata_invalid_requires_services_type(tmp_path: Path) -> None: + p = _write( + tmp_path / ".infra.yml", + """ +project: app +requires: + services: 123 +""".lstrip(), + ) + + with pytest.raises(ValueError, match=r"requires\.services"): + InfraMetadata.from_file(p)