Fix edge cases in steps.field_update when primaryKey and foreignKeys are renamed

data/transform-fk.csv

@@ -0,0 +1,4 @@
+id,address,country_name
+1,a,germany
+2,b,france
+3,c,spain

frictionless/steps/field/field_update.py

@@ -68,34 +68,37 @@ def transform_resource(self, resource: Resource):
             resource.data = table.update(self.name, self.value)  # type: ignore
         elif new_name:
             resource.data = table.rename({self.name: new_name})  # type: ignore
-        if new_name and resource.schema.primary_key:
+        if new_name and self.name in 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:
+        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):
+                if new_name and self.name in fk["fields"]:
+                    package_resource.schema.foreign_keys[index]["fields"].remove(
+                        self.name
+                    )
+                    package_resource.schema.foreign_keys[index]["fields"].append(new_name)
+                fields = fk["reference"]["fields"]
+                if isinstance(fields, list):
+                    if self.name in fk["reference"]["fields"]:
                         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()
-                )
+                        ].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
 

tests/steps/field/test_field_update.py

@@ -89,6 +89,7 @@ def test_step_field_update_field_name_with_primary_key():
     pipeline = Pipeline(
         steps=[
             steps.field_update(name="id", descriptor={"name": "pkey"}),
+            steps.field_update(name="population", descriptor={"name": "pop"}),
         ],
     )
     target = source.transform(pipeline)
@@ -97,7 +98,7 @@ def test_step_field_update_field_name_with_primary_key():
 
 def test_step_field_update_referenced_as_foreign_key():
     resource1 = TableResource(name="resource1", path="data/transform.csv")
-    resource2 = TableResource(name="resource2")
+    resource2 = TableResource(name="resource2", path="data/transform-fk.csv")
     resource1.schema = Schema.from_descriptor(
         {
             "fields": [
@@ -134,13 +135,94 @@ def test_step_field_update_referenced_as_foreign_key():
             )
         ],
     )
+
+    transform(
+        package,
+        steps=[
+            steps.resource_transform(
+                name="resource2",
+                steps=[
+                    steps.field_update(
+                        name="country_name", descriptor={"name": "country"}
+                    )
+                ],
+            )
+        ],
+    )
+
     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"],
+            "fields": ["country"],
             "reference": {"fields": ["pkey"], "resource": "resource1"},
         }
     ]
+
+
+def test_step_field_update_referenced_as_foreign_key_with_same_name():
+    resource1 = TableResource.from_descriptor(
+        {
+            "name": "resource1",
+            "data": [
+                ["id", "name", "population"],
+                [1, "germany", 83],
+                [2, "france", 66],
+                [3, "spain", 47],
+            ],
+            "schema": {
+                "fields": [
+                    {"name": "id", "type": "integer"},
+                    {"name": "name", "type": "string"},
+                    {"name": "population", "type": "integer"},
+                ],
+                "primaryKey": ["id"],
+                "foreignKeys": [
+                    {
+                        "fields": ["name"],
+                        "reference": {"fields": ["name"], "resource": "resource2"},
+                    }
+                ],
+            },
+        }
+    )
+    resource2 = TableResource.from_descriptor(
+        {
+            "name": "resource2",
+            "data": [
+                ["name", "continent"],
+                ["germany", "europe"],
+                ["france", "europe"],
+                ["spain", "europe"],
+            ],
+            "schema": {
+                "fields": [
+                    {"name": "name", "type": "string"},
+                    {"name": "continent", "type": "string"},
+                ],
+                "primaryKey": ["name"],
+            },
+        }
+    )
+    package = Package(name="test-package", resources=[resource1, resource2])
+
+    transform(
+        package,
+        steps=[
+            steps.resource_transform(
+                name="resource1",
+                steps=[
+                    steps.field_update(name="name", descriptor={"name": "country_name"})
+                ],
+            )
+        ],
+    )
+
+    assert package.get_resource("resource1").schema.foreign_keys == [
+        {
+            "fields": ["country_name"],
+            "reference": {"fields": ["name"], "resource": "resource2"},
+        }
+    ]