diff --git a/tests/codegen/test_transformer.py b/tests/codegen/test_transformer.py index 9d8e5372..9ba69323 100644 --- a/tests/codegen/test_transformer.py +++ b/tests/codegen/test_transformer.py @@ -1,4 +1,5 @@ import pickle +import shutil import tempfile from pathlib import Path from unittest import mock @@ -152,13 +153,63 @@ def test_process_definitions( mock_convert_definitions.assert_called_once_with(fist_def) @mock.patch.object(ResourceTransformer, "process_schema") - def test_process_schemas(self, mock_process_schema) -> None: + @mock.patch.object(ResourceTransformer, "find_included_chameleons") + def test_process_schemas( + self, mock_find_included_chameleons, mock_process_schema + ) -> None: uris = ["http://xsdata/foo.xsd", "http://xsdata/bar.xsd"] + mock_find_included_chameleons.return_value = set() self.transformer.process_schemas(uris) mock_process_schema.assert_has_calls([mock.call(uri) for uri in uris]) + @mock.patch.object(ResourceTransformer, "process_schema") + @mock.patch.object(ResourceTransformer, "find_included_chameleons") + def test_process_schemas_skips_included_chameleons( + self, mock_find_included_chameleons, mock_process_schema + ) -> None: + uris = ["http://xsdata/chameleon.xsd", "http://xsdata/main.xsd"] + mock_find_included_chameleons.return_value = {uris[0]} + + self.transformer.process_schemas(uris) + + mock_process_schema.assert_called_once_with(uris[1]) + + def test_find_included_chameleons(self) -> None: + tmp = Path(tempfile.mkdtemp()) + try: + chameleon = tmp / "chameleon.xsd" + chameleon.write_text( + '\n' + ) + main = tmp / "main.xsd" + main.write_text( + '' + '\n' + "\n" + ) + # standalone chameleon that nobody includes must NOT be skipped + orphan = tmp / "orphan.xsd" + orphan.write_text( + '\n' + ) + + uris = [f.as_uri() for f in (chameleon, main, orphan)] + skip = self.transformer.find_included_chameleons(uris) + + self.assertEqual({chameleon.as_uri()}, skip) + # the content is cached so compilation does not read the file twice + self.assertIn(chameleon.as_uri(), self.transformer.preloaded) + finally: + shutil.rmtree(tmp) + + def test_find_included_chameleons_handles_missing_source(self) -> None: + self.assertEqual( + set(), self.transformer.find_included_chameleons(["file://nonexistent"]) + ) + @mock.patch.object(ClassUtils, "reduce_classes") @mock.patch.object(ElementMapper, "map") @mock.patch.object(TreeParser, "from_bytes") diff --git a/xsdata/codegen/transformer.py b/xsdata/codegen/transformer.py index c070850d..a11e79b5 100644 --- a/xsdata/codegen/transformer.py +++ b/xsdata/codegen/transformer.py @@ -3,11 +3,13 @@ import json import os import pickle +import re import tempfile from collections import defaultdict from collections.abc import Callable from pathlib import Path from typing import NamedTuple +from urllib.parse import urljoin from toposort import CircularDependencyError @@ -191,11 +193,69 @@ def process_definitions(self, uris: list[str]) -> None: def process_schemas(self, uris: list[str]) -> None: """Process a list of xsd resources. + Chameleon schemas (no targetNamespace) that are xs:included by another + schema in the batch are skipped as top-level sources: they are compiled + through the include with the including schema's namespace. Processing + them standalone first would compile their types with no namespace and + break references from the schemas that include them. + Args: uris: A list of xsd URI strings to process """ + skip = self.find_included_chameleons(uris) for uri in uris: - self.process_schema(uri) + if uri not in skip: + self.process_schema(uri) + + def find_included_chameleons(self, uris: list[str]) -> set[str]: + """Return the chameleon schemas that are xs:included by a namespaced one. + + A chameleon schema declares no targetNamespace. When it is included by + a schema that does, it must be compiled through that include so its + types inherit the namespace. Each source is read once; the content is + cached in ``preloaded`` so the subsequent compilation reuses it. + + Args: + uris: A list of xsd URI strings to inspect + + Returns: + The subset of ``uris`` to skip as top-level sources. + """ + has_ns: dict[str, bool] = {} + includes: dict[str, set[str]] = {} + for uri in uris: + try: + data = opener.open(uri).read() # nosec + except OSError: + continue + + self.preloaded[uri] = data + text = data.decode("utf-8", errors="ignore") + header = re.search( + r"<(?:\w+:)?schema\b[^>]*>", text, re.IGNORECASE | re.DOTALL + ) + has_ns[uri] = bool( + header and re.search(r"targetNamespace\s*=", header.group(0)) + ) + includes[uri] = { + urljoin(uri, loc) + for loc in re.findall( + r'<(?:\w+:)?include\b[^>]*schemaLocation="([^"]+)"', + text, + re.IGNORECASE, + ) + } + + included_by_ns: set[str] = set() + for uri, namespaced in has_ns.items(): + if namespaced: + included_by_ns |= includes[uri] + + return { + uri + for uri, namespaced in has_ns.items() + if not namespaced and uri in included_by_ns + } def process_dtds(self, uris: list[str]) -> None: """Process a list of dtd resources.