Fix: resolve relative $ref correctly when fetching schemas from HTTP URLs (#2600)

* Fix: Improve base URL context handling and URL validation

* Fix: Enhance HTTP URL validation and improve reference resolution logic

Author: koxudaxi

src/datamodel_code_generator/reference.py

@@ -477,9 +477,16 @@ def current_base_path_context(self, base_path: Path | None) -> Generator[None, N
             yield
 
     @contextmanager
-    def base_url_context(self, base_url: str) -> Generator[None, None, None]:
-        """Temporarily set the base URL within a context."""
-        if self._base_url:
+    def base_url_context(self, base_url: str | None) -> Generator[None, None, None]:
+        """Temporarily set the base URL within a context.
+
+        Only sets the base_url if:
+        - The new value is actually a URL (http:// or https://)
+        - OR _base_url was already set (switching between URLs)
+        This preserves backward compatibility for local file parsing where
+        this method was previously a no-op.
+        """
+        if self._base_url or (base_url and is_url(base_url)):
             with context_variable(self.set_base_url, self.base_url, base_url):
                 yield
         else:
@@ -523,7 +530,7 @@ def add_id(self, id_: str, path: Sequence[str]) -> None:
         """Register an identifier mapping to a resolved reference path."""
         self.ids["/".join(self.current_root)][id_] = self.resolve_ref(path)
 
-    def resolve_ref(self, path: Sequence[str] | str) -> str:  # noqa: PLR0911, PLR0912
+    def resolve_ref(self, path: Sequence[str] | str) -> str:  # noqa: PLR0911, PLR0912, PLR0914
         """Resolve a reference path to its canonical form."""
         joined_path = path if isinstance(path, str) else self.join_path(path)
         if joined_path == "#":
@@ -548,18 +555,23 @@ def resolve_ref(self, path: Sequence[str] | str) -> str:  # noqa: PLR0911, PLR09
         else:
             if "#" not in joined_path:
                 joined_path += "#"
-            elif joined_path[0] == "#":
+            elif joined_path[0] == "#" and not self.base_url:
                 joined_path = f"{'/'.join(self.current_root)}{joined_path}"
 
             file_path, fragment = joined_path.split("#", 1)
             ref = f"{file_path}#{fragment}"
-            if self.root_id_base_path and not (is_url(joined_path) or Path(self._base_path, file_path).is_file()):
+            if (
+                self.root_id_base_path
+                and not self.base_url
+                and not (is_url(joined_path) or Path(self._base_path, file_path).is_file())
+            ):
                 ref = f"{self.root_id_base_path}/{ref}"
 
         if self.base_url:
             from .http import join_url  # noqa: PLC0415
 
-            joined_url = join_url(self.base_url, ref)
+            effective_base = self.root_id or self.base_url
+            joined_url = join_url(effective_base, ref)
             if "#" in joined_url:
                 return joined_url
             return f"{joined_url}#"
@@ -846,5 +858,5 @@ def snake_to_upper_camel(word: str, delimiter: str = "_") -> str:
 
 
 def is_url(ref: str) -> bool:
-    """Check if a reference string is a URL."""
+    """Check if a reference string is an HTTP(S) URL."""
     return ref.startswith(("https://", "http://"))

tests/test_reference.py

@@ -6,7 +6,7 @@
 
 import pytest
 
-from datamodel_code_generator.reference import ModelResolver, get_relative_path
+from datamodel_code_generator.reference import ModelResolver, get_relative_path, is_url
 
 
 @pytest.mark.parametrize(
@@ -70,3 +70,140 @@ def test_model_resolver_add_ref_unevaluated() -> None:
     model_resolver = ModelResolver()
     reference = model_resolver.add_ref("meta/unevaluated")
     assert reference.original_name == "unevaluated"
+
+
+def test_base_url_context_sets_url_when_base_url_already_set() -> None:
+    """When _base_url is already set, base_url_context should switch to new URL."""
+    resolver = ModelResolver(base_url="https://example.com/original.json")
+    assert resolver.base_url == "https://example.com/original.json"
+
+    with resolver.base_url_context("https://example.com/new.json"):
+        assert resolver.base_url == "https://example.com/new.json"
+
+    # Should restore original
+    assert resolver.base_url == "https://example.com/original.json"
+
+
+def test_base_url_context_sets_url_when_new_value_is_url() -> None:
+    """When _base_url is None but new value is a URL, should set base_url."""
+    resolver = ModelResolver()
+    assert resolver.base_url is None
+
+    with resolver.base_url_context("https://example.com/schema.json"):
+        assert resolver.base_url == "https://example.com/schema.json"
+
+    # Should restore to None
+    assert resolver.base_url is None
+
+
+def test_base_url_context_noop_when_new_value_is_not_url() -> None:
+    """When _base_url is None and new value is not a URL, should do nothing."""
+    resolver = ModelResolver()
+    assert resolver.base_url is None
+
+    with resolver.base_url_context("../relative/path.json"):
+        # Should remain None because the value is not a URL
+        assert resolver.base_url is None
+
+    assert resolver.base_url is None
+
+
+def test_base_url_context_nested() -> None:
+    """Nested base_url_context should properly restore values."""
+    resolver = ModelResolver(base_url="https://example.com/level0.json")
+
+    with resolver.base_url_context("https://example.com/level1.json"):
+        assert resolver.base_url == "https://example.com/level1.json"
+
+        with resolver.base_url_context("https://example.com/level2.json"):
+            assert resolver.base_url == "https://example.com/level2.json"
+
+        assert resolver.base_url == "https://example.com/level1.json"
+
+    assert resolver.base_url == "https://example.com/level0.json"
+
+
+def test_resolve_ref_with_base_url_does_not_prepend_root_id_base_path() -> None:
+    """When base_url is set, root_id_base_path should not be prepended to refs."""
+    resolver = ModelResolver(base_url="https://example.com/schemas/main.json")
+    resolver.set_root_id("https://example.com/schemas/main.json")
+
+    # Resolve a relative ref
+    result = resolver.resolve_ref("../other/schema.json")
+
+    # Should resolve via join_url, not prepend root_id_base_path
+    assert result == "https://example.com/other/schema.json#"
+    # Should NOT be like "https://example.com/schemas/../other/schema.json#"
+
+
+def test_resolve_ref_with_base_url_nested_relative_refs() -> None:
+    """Nested relative refs should resolve correctly when base_url is set."""
+    resolver = ModelResolver(base_url="https://example.com/a/b/c/main.json")
+
+    # Resolve a deeply nested relative ref
+    result = resolver.resolve_ref("../../other/schema.json")
+
+    assert result == "https://example.com/a/other/schema.json#"
+
+
+def test_resolve_ref_with_base_url_context_switch() -> None:
+    """Relative refs should resolve correctly after base_url context switch."""
+    resolver = ModelResolver(base_url="https://example.com/schemas/person.json")
+
+    # Switch context to a different file
+    with resolver.base_url_context("https://example.com/schemas/definitions/pet.json"):
+        # Resolve a relative ref from the new context
+        result = resolver.resolve_ref("../common/types.json")
+
+        assert result == "https://example.com/schemas/common/types.json#"
+
+
+def test_resolve_ref_local_fragment_with_base_url() -> None:
+    """Local fragment refs should resolve to full URL when base_url is set."""
+    resolver = ModelResolver(base_url="https://example.com/schemas/main.json")
+
+    result = resolver.resolve_ref("#/definitions/Foo")
+
+    # When base_url is set, local fragments are resolved to full URL
+    assert result == "https://example.com/schemas/main.json#/definitions/Foo"
+
+
+@pytest.mark.parametrize(
+    ("ref", "expected"),
+    [
+        # HTTP/HTTPS URLs (only supported schemes)
+        ("https://example.com/schema.json", True),
+        ("http://example.com/schema.json", True),
+        ("https://example.com/path/to/schema.json", True),
+        # file:// URLs - NOT recognized (fetcher only supports HTTP)
+        ("file:///home/user/schema.json", False),
+        ("file:/home/user/schema.json", False),
+        # Other URL schemes - NOT recognized
+        ("ftp://example.com/schema.json", False),
+        # Relative paths (not URLs)
+        ("../relative/path.json", False),
+        ("relative/path.json", False),
+        # Local fragments (not URLs)
+        ("#/definitions/Foo", False),
+        ("#", False),
+        # Absolute paths (not URLs)
+        ("/absolute/path.json", False),
+        # Windows paths (not URLs)
+        ("c:/windows/path.json", False),
+        ("d:/path/to/file.json", False),
+    ],
+)
+def test_is_url(ref: str, expected: bool) -> None:
+    """Test is_url correctly identifies HTTP(S) URLs only."""
+    assert is_url(ref) == expected
+
+
+def test_resolve_ref_with_root_id_differs_from_base_url() -> None:
+    """When $id differs from fetch URL, refs should resolve against $id."""
+    # Scenario: Schema fetched from CDN but has canonical $id
+    resolver = ModelResolver(base_url="https://cdn.example.com/latest/schema.json")
+    resolver.set_root_id("https://example.com/v1/schema.json")
+
+    result = resolver.resolve_ref("../common/types.json")
+
+    assert result == "https://example.com/common/types.json#"