fix(python): UriSource — add urlopen timeout + HTTP scheme test coverage

Author: dmealing

server/python/src/metaobjects/loader/sources/uri_source.py

@@ -23,9 +23,15 @@ def _infer_format(uri: str) -> MetaDataFormat:
 class UriSource(MetaDataSource):
     """A URI-backed source. Lazily fetched on ``read()``."""
 
-    def __init__(self, uri: str, format: MetaDataFormat | None = None) -> None:
+    def __init__(
+        self,
+        uri: str,
+        format: MetaDataFormat | None = None,
+        timeout: float = 30.0,
+    ) -> None:
         self._uri = uri
         self._format = format if format is not None else _infer_format(uri)
+        self._timeout = timeout
 
     @property
     def id(self) -> str:
@@ -41,7 +47,9 @@ def read(self) -> str:
             # urlparse splits the leading slashes off the path on file:// URIs.
             return Path(parsed.path).read_text(encoding="utf-8-sig")
         if parsed.scheme in ("http", "https"):
-            with urlopen(self._uri) as resp:  # noqa: S310 — caller-supplied URI
+            # Schemes are explicitly allowlisted (file/http/https) above; arbitrary
+            # URI handlers (ftp, etc.) reject with ValueError before urlopen is called.
+            with urlopen(self._uri, timeout=self._timeout) as resp:  # noqa: S310
                 return resp.read().decode("utf-8")
         raise ValueError(
             f"UriSource: unsupported scheme '{parsed.scheme}' on {self._uri}"

server/python/tests/unit/test_sources.py

@@ -6,6 +6,9 @@
 from __future__ import annotations
 
 from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
 
 from metaobjects.loader.sources import (
     DirectorySource,
@@ -108,3 +111,28 @@ def test_uri_source_format_inferred_from_path(tmp_path: Path) -> None:
     p.write_text("k: v", encoding="utf-8")
     src = UriSource(p.as_uri())
     assert src.format == MetaDataFormat.YAML
+
+
+def test_uri_source_http_scheme_decodes_utf8() -> None:
+    fake_response = MagicMock()
+    fake_response.read.return_value = b'{"metadata.root":{"package":"x","children":[]}}'
+    fake_response.__enter__ = lambda self: self
+    fake_response.__exit__ = lambda self, *a: None
+
+    with patch(
+        "metaobjects.loader.sources.uri_source.urlopen",
+        return_value=fake_response,
+    ) as mock_open:
+        src = UriSource("https://example.com/meta.json")
+        content = src.read()
+
+    assert content == '{"metadata.root":{"package":"x","children":[]}}'
+    # Verify timeout was passed
+    args, kwargs = mock_open.call_args
+    assert kwargs.get("timeout") == 30.0
+
+
+def test_uri_source_rejects_unsupported_scheme() -> None:
+    src = UriSource("ftp://example.com/foo.json", format=MetaDataFormat.JSON)
+    with pytest.raises(ValueError, match="unsupported scheme"):
+        src.read()