Validate infra metadata configs

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)
 

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)