fix: recursive references across multiple schemas

Author: simosho

jsf/parser.py

@@ -124,7 +124,9 @@ def __parse_primitive(self, name: str, path: str, schema: Dict[str, Any]) -> Pri
             }
         )
 
-    def __parse_object(self, name: str, path: str, schema: Dict[str, Any]) -> Object:
+    def __parse_object(
+        self, name: str, path: str, schema: Dict[str, Any], root: Optional[AllTypes] = None
+    ) -> Object:
         _, is_nullable = self.__is_field_nullable(schema)
         model = Object.from_dict(
             {
@@ -136,21 +138,25 @@ def __parse_object(self, name: str, path: str, schema: Dict[str, Any]) -> Object
                 **schema,
             }
         )
-        self.root = model if not self.root else self.root
+        root = model if root is None else root
         props = []
         for _name, definition in schema.get("properties", {}).items():
-            props.append(self.__parse_definition(_name, path=f"{path}/{_name}", schema=definition))
+            props.append(
+                self.__parse_definition(_name, path=f"{path}/{_name}", schema=definition, root=root)
+            )
         model.properties = props
         pattern_props = []
         for _name, definition in schema.get("patternProperties", {}).items():
             pattern_props.append(
-                self.__parse_definition(_name, path=f"{path}/{_name}", schema=definition)
+                self.__parse_definition(_name, path=f"{path}/{_name}", schema=definition, root=root)
             )
         model.patternProperties = pattern_props
 
         return model
 
-    def __parse_array(self, name: str, path: str, schema: Dict[str, Any]) -> Array:
+    def __parse_array(
+        self, name: str, path: str, schema: Dict[str, Any], root: Optional[AllTypes] = None
+    ) -> Array:
         _, is_nullable = self.__is_field_nullable(schema)
         arr = Array.from_dict(
             {
@@ -162,11 +168,13 @@ def __parse_array(self, name: str, path: str, schema: Dict[str, Any]) -> Array:
                 **schema,
             }
         )
-        self.root = arr if not self.root else self.root
-        arr.items = self.__parse_definition(name, f"{path}/items", schema["items"])
+        root = arr if root is None else root
+        arr.items = self.__parse_definition(name, f"{path}/items", schema["items"], root=root)
         return arr
 
-    def __parse_tuple(self, name: str, path: str, schema: Dict[str, Any]) -> JSFTuple:
+    def __parse_tuple(
+        self, name: str, path: str, schema: Dict[str, Any], root: Optional[AllTypes] = None
+    ) -> JSFTuple:
         _, is_nullable = self.__is_field_nullable(schema)
         arr = JSFTuple.from_dict(
             {
@@ -178,10 +186,12 @@ def __parse_tuple(self, name: str, path: str, schema: Dict[str, Any]) -> JSFTupl
                 **schema,
             }
         )
-        self.root = arr if not self.root else self.root
+        root = arr if root is None else root
         arr.items = []
         for i, item in enumerate(schema["items"]):
-            arr.items.append(self.__parse_definition(name, path=f"{path}/{name}[{i}]", schema=item))
+            arr.items.append(
+                self.__parse_definition(name, path=f"{path}/{name}[{i}]", schema=item, root=root)
+            )
         return arr
 
     def __is_field_nullable(self, schema: Dict[str, Any]) -> Tuple[str, bool]:
@@ -196,46 +206,55 @@ def __is_field_nullable(self, schema: Dict[str, Any]) -> Tuple[str, bool]:
                 return random.choice(item_type_deep_copy), False
         return item_type, False
 
-    def __parse_anyOf(self, name: str, path: str, schema: Dict[str, Any]) -> AnyOf:
+    def __parse_anyOf(
+        self, name: str, path: str, schema: Dict[str, Any], root: Optional[AllTypes] = None
+    ) -> AnyOf:
         model = AnyOf(name=name, path=path, max_recursive_depth=self.max_recursive_depth, **schema)
-        self.root = model if not self.root else self.root
+        root = model if root is None else root
         schemas = []
         for d in schema["anyOf"]:
-            schemas.append(self.__parse_definition(name, path, d))
+            schemas.append(self.__parse_definition(name, path, d, root=root))
         model.schemas = schemas
         return model
 
-    def __parse_allOf(self, name: str, path: str, schema: Dict[str, Any]) -> AllOf:
+    def __parse_allOf(
+        self, name: str, path: str, schema: Dict[str, Any], root: Optional[AllTypes] = None
+    ) -> AllOf:
         combined_schema = dict(ChainMap(*schema["allOf"]))
         model = AllOf(name=name, path=path, max_recursive_depth=self.max_recursive_depth, **schema)
-        self.root = model if not self.root else self.root
-        model.combined_schema = self.__parse_definition(name, path, combined_schema)
+        root = model if root is None else root
+        model.combined_schema = self.__parse_definition(name, path, combined_schema, root=root)
         return model
 
-    def __parse_oneOf(self, name: str, path: str, schema: Dict[str, Any]) -> OneOf:
+    def __parse_oneOf(
+        self, name: str, path: str, schema: Dict[str, Any], root: Optional[AllTypes] = None
+    ) -> OneOf:
         model = OneOf(name=name, path=path, max_recursive_depth=self.max_recursive_depth, **schema)
-        self.root = model if not self.root else self.root
+        root = model if root is None else root
         schemas = []
         for d in schema["oneOf"]:
-            schemas.append(self.__parse_definition(name, path, d))
+            schemas.append(self.__parse_definition(name, path, d, root=root))
         model.schemas = schemas
         return model
 
-    def __parse_named_definition(self, path: str, def_name: str) -> AllTypes:
+    def __parse_named_definition(self, path: str, def_name: str, root) -> AllTypes:
         schema = self.root_schema
         parsed_definition = None
         for def_tag in ("definitions", "$defs"):
             if path.startswith(f"#/{def_tag}/{def_name}"):
-                self.root.is_recursive = True
-                return self.root
-            elif definition := schema.get(def_tag, {}).get(def_name):
+                root.is_recursive = True
+                return root
+            definition = schema.get(def_tag, {}).get(def_name)
+            if definition is not None:
                 parsed_definition = self.__parse_definition(
-                    def_name, path=f"{path}/#/{def_tag}/{def_name}", schema=definition
+                    def_name, path=f"{path}/#/{def_tag}/{def_name}", schema=definition, root=root
                 )
                 self.definitions[f"#/{def_tag}/{def_name}"] = parsed_definition
         return parsed_definition
 
-    def __parse_definition(self, name: str, path: str, schema: Dict[str, Any]) -> AllTypes:
+    def __parse_definition(
+        self, name: str, path: str, schema: Dict[str, Any], root: Optional[AllTypes] = None
+    ) -> AllTypes:
         self.base_state["__all_json_paths__"].append(path)
         item_type, is_nullable = self.__is_field_nullable(schema)
         if "const" in schema:
@@ -259,20 +278,20 @@ def __parse_definition(self, name: str, path: str, schema: Dict[str, Any]) -> Al
             )
         elif "type" in schema:
             if item_type == "object" and "properties" in schema:
-                return self.__parse_object(name, path, schema)
+                return self.__parse_object(name, path, schema, root)
             elif item_type == "object" and "anyOf" in schema:
-                return self.__parse_anyOf(name, path, schema)
+                return self.__parse_anyOf(name, path, schema, root)
             elif item_type == "object" and "allOf" in schema:
-                return self.__parse_allOf(name, path, schema)
+                return self.__parse_allOf(name, path, schema, root)
             elif item_type == "object" and "oneOf" in schema:
-                return self.__parse_oneOf(name, path, schema)
+                return self.__parse_oneOf(name, path, schema, root)
             elif item_type == "array":
                 if (schema.get("contains") is not None) or isinstance(schema.get("items"), dict):
-                    return self.__parse_array(name, path, schema)
+                    return self.__parse_array(name, path, schema, root)
                 if isinstance(schema.get("items"), list) and all(
                     isinstance(x, dict) for x in schema.get("items", [])
                 ):
-                    return self.__parse_tuple(name, path, schema)
+                    return self.__parse_tuple(name, path, schema, root)
             else:
                 return self.__parse_primitive(name, path, schema)
         elif "$ref" in schema:
@@ -283,20 +302,23 @@ def __parse_definition(self, name: str, path: str, schema: Dict[str, Any]) -> Al
                 else:
                     # parse referenced definition
                     ref_name = frag.split("/")[-1]
-                    cls = self.__parse_named_definition(path, ref_name)
+                    cls = self.__parse_named_definition(path, ref_name, root)
             else:
                 with s_open(ext, "r") as f:
                     external_jsf = JSF(json.load(f))
                 cls = deepcopy(external_jsf.definitions.get(f"#{frag}"))
-            cls.name = name
-            cls.path = path
+            if path != "#" and cls == root:
+                cls.name = name
+            elif path != "#":
+                cls.name = name
+                cls.path = path
             return cls
         elif "anyOf" in schema:
-            return self.__parse_anyOf(name, path, schema)
+            return self.__parse_anyOf(name, path, schema, root)
         elif "allOf" in schema:
-            return self.__parse_allOf(name, path, schema)
+            return self.__parse_allOf(name, path, schema, root)
         elif "oneOf" in schema:
-            return self.__parse_oneOf(name, path, schema)
+            return self.__parse_oneOf(name, path, schema, root)
         else:
             raise ValueError(f"Cannot parse schema {repr(schema)}")  # pragma: no cover
 

jsf/tests/data/anyof_recursive.json

@@ -0,0 +1,45 @@
+{
+    "$ref": "#/definitions/tree",
+    "definitions": {
+        "tree": {
+            "anyOf": [
+                {
+                    "$ref": "#/definitions/node"
+                },
+                {
+                    "type": "string"
+                }
+            ]
+        },
+        "node": {
+            "type": "object",
+            "allOf": [
+                {
+                    "type": "object",
+                    "oneOf": [
+                        {
+                            "type": "object",
+                            "properties": {
+                                "value": {
+                                    "$ref": "#/definitions/tree"
+                                }
+                            },
+                            "required": [
+                                "value"
+                            ]
+                        },
+                        {
+                            "type": "object",
+                            "properties": {
+                                "value": {
+                                    "type": "string"
+                                }
+                            },
+                            "required": ["value"]
+                        }
+                    ]
+                }
+            ]
+        }
+    }
+}

jsf/tests/data/complex_recursive.json

@@ -488,14 +488,13 @@ def test_fake_oneof_recursive(TestData):
             assert isinstance(item, int) or isinstance(item, list)
 
 
-def test_fake_anyof_recursive(TestData):
-    with open(TestData / "anyof_recursive.json") as file:
+def test_fake_complex_recursive(TestData):
+    with open(TestData / "complex_recursive.json") as file:
         schema = json.load(file)
     p = JSF(schema, max_recursive_depth=2)
 
     fake_data = [p.generate() for _ in range(10)]
     for d in fake_data:
-        for item in d:
-            assert isinstance(item, str) or isinstance(item, dict)
-            if isinstance(item, dict):
-                assert "value" in item
+        assert isinstance(d, str) or isinstance(d, dict)
+        if isinstance(d, dict):
+            assert "value" in d