Fix reference when field and resource name is changed (#1577)

* Fix names of primary and foreign keys when updated

* Add comment

Add comment
Field type check while updating field name

* Fix references when resource and field name is updated

frictionless/steps/field/field_update.py

@@ -71,6 +71,31 @@ def transform_resource(self, resource: Resource):
         if new_name and resource.schema.primary_key:
             resource.schema.primary_key.remove(self.name)
             resource.schema.primary_key.append(new_name)
+            resources = resource.package.resources if resource.package else []
+            # update name in all the resources where it is referenced
+            for package_resource in resources:
+                for index, fk in enumerate(package_resource.schema.foreign_keys):
+                    fields = fk["reference"]["fields"]
+                    if isinstance(fields, list):
+                        if self.name in fk["reference"]["fields"]:
+                            package_resource.schema.foreign_keys[index]["reference"][
+                                "fields"
+                            ].remove(self.name)
+                            package_resource.schema.foreign_keys[index]["reference"][
+                                "fields"
+                            ].append(new_name)
+                    else:
+                        package_resource.schema.foreign_keys[index]["reference"][
+                            "fields"
+                        ] = new_name
+
+                package_resource.schema.foreign_keys = (
+                    package_resource.schema.foreign_keys
+                )
+            if resource.package:
+                resource.package.metadata_descriptor_initial = (
+                    resource.package.to_descriptor()
+                )
 
     # Metadata
 

frictionless/steps/resource/resource_update.py

@@ -51,6 +51,23 @@ def transform_resource(self, resource: Resource):
         options = helpers.create_options(self.descriptor)
         for name, value in options.items():
             setattr(resource, name, value)
+        resources = resource.package.resources if resource.package else []
+        new_name = options.get("name")
+        if new_name and new_name != self.name:
+            # update name in all the resources where it is referenced
+            for package_resource in resources:
+                for index, fk in enumerate(package_resource.schema.foreign_keys):
+                    if fk["reference"]["resource"] == self.name:
+                        package_resource.schema.foreign_keys[index]["reference"][
+                            "resource"
+                        ] = new_name
+                package_resource.schema.foreign_keys = (
+                    package_resource.schema.foreign_keys
+                )
+            if resource.package:
+                resource.package.metadata_descriptor_initial = (
+                    resource.package.to_descriptor()
+                )
 
     # Metadata
 

tests/steps/field/test_field_update.py

@@ -1,4 +1,5 @@
-from frictionless import Pipeline, steps
+from frictionless import Pipeline, steps, transform
+from frictionless.package.package import Package
 from frictionless.resources import TableResource
 from frictionless.schema.schema import Schema
 
@@ -92,3 +93,54 @@ def test_step_field_update_field_name_with_primary_key():
     )
     target = source.transform(pipeline)
     assert target.schema.primary_key == ["pkey"]
+
+
+def test_step_field_update_referenced_as_foreign_key():
+    resource1 = TableResource(name="resource1", path="data/transform.csv")
+    resource2 = TableResource(name="resource2")
+    resource1.schema = Schema.from_descriptor(
+        {
+            "fields": [
+                {"name": "id", "type": "integer"},
+                {"name": "name", "type": "string"},
+                {"name": "population", "type": "integer"},
+            ],
+            "primaryKey": ["id"],
+        }
+    )
+    resource2.schema = Schema.from_descriptor(
+        {
+            "fields": [
+                {"name": "id", "type": "integer"},
+                {"name": "address", "type": "string"},
+                {"name": "country_name", "type": "integer"},
+            ],
+            "primaryKey": ["id"],
+            "foreignKeys": [
+                {
+                    "fields": ["country_name"],
+                    "reference": {"fields": ["id"], "resource": "resource1"},
+                }
+            ],
+        }
+    )
+    package = Package(name="test-package", resources=[resource1, resource2])
+    transform(
+        package,
+        steps=[
+            steps.resource_transform(
+                name="resource1",
+                steps=[steps.field_update(name="id", descriptor={"name": "pkey"})],
+            )
+        ],
+    )
+    assert (
+        package.get_resource("resource1").validate().flatten(["title", "message"]) == []
+    )
+    assert package.get_resource("resource1").schema.primary_key == ["pkey"]
+    assert package.get_resource("resource2").schema.foreign_keys == [
+        {
+            "fields": ["country_name"],
+            "reference": {"fields": ["pkey"], "resource": "resource1"},
+        }
+    ]

tests/steps/resource/test_resource_update.py

@@ -1,4 +1,4 @@
-from frictionless import Package, Pipeline, steps
+from frictionless import Package, Pipeline, Schema, steps, transform
 from frictionless.resources import TableResource
 
 # General
@@ -38,3 +38,56 @@ def test_step_resource_update_standalone_issue_1351():
     )
     target = source.transform(pipeline)
     assert target.title == "New title"
+
+
+def test_step_resource_update_referenced_as_foreign_key():
+    resource1 = TableResource(name="resource1", path="data/transform.csv")
+    resource2 = TableResource(name="resource2")
+    resource1.schema = Schema.from_descriptor(
+        {
+            "fields": [
+                {"name": "id", "type": "integer"},
+                {"name": "name", "type": "string"},
+                {"name": "population", "type": "integer"},
+            ],
+            "primaryKey": ["id"],
+        }
+    )
+    resource2.schema = Schema.from_descriptor(
+        {
+            "fields": [
+                {"name": "id", "type": "integer"},
+                {"name": "address", "type": "string"},
+                {"name": "country_name", "type": "integer"},
+            ],
+            "primaryKey": ["id"],
+            "foreignKeys": [
+                {
+                    "fields": ["country_name"],
+                    "reference": {"fields": ["id"], "resource": "resource1"},
+                }
+            ],
+        }
+    )
+    package = Package(name="test-package", resources=[resource1, resource2])
+    transform(
+        package,
+        steps=[
+            steps.resource_transform(
+                name="resource1",
+                steps=[
+                    steps.resource_update(
+                        name="resource1", descriptor={"name": "first-resource"}
+                    )
+                ],
+            )
+        ],
+    )
+    assert (
+        package.get_resource("first-resource").validate().flatten(["title", "message"])
+        == []
+    )
+    assert (
+        package.get_resource("resource2").schema.foreign_keys[0]["reference"]["resource"]
+        == "first-resource"
+    )