[#4882] Adjustment and refactoring after review

akvo/rsr/tests/usecases/test_fix_inconsistent_results.py

@@ -49,7 +49,7 @@ def test_initial_states(self):
         self.assertEqual(self.project.indicators.get(result=result2).parent_indicator, self.false_lead_indicator)
         self.assertEqual(self.project.periods.get(indicator__result=result2).parent_period, self.false_lead_period)
 
-    def test_not_touching_correct_parents(self):
+    def test_leave_correct_parents_untouched(self):
         command.fix_inconsistent_results(self.project.results.all())
 
         result1 = self.project.results.get(title='Result #1')
@@ -86,7 +86,7 @@ def test_initial_states(self):
         self.assertEqual(self.project.indicators.get(result=result).parent_indicator, self.false_lead_indicator)
         self.assertEqual(self.project.periods.get(indicator__result=result).parent_period, self.false_lead_period)
 
-    def test_fix_false_parents(self):
+    def test_delete_false_parents(self):
         command.fix_inconsistent_results(self.project.results.all())
 
         result = self.project.results.get(title='Result #1')
@@ -123,7 +123,7 @@ def test_initial_states(self):
         self.assertEqual(self.project.indicators.get(result=result).parent_indicator, self.false_lead_indicator)
         self.assertEqual(self.project.periods.get(indicator__result=result).parent_period, self.false_lead_period)
 
-    def test_fix_false_parents(self):
+    def test_delete_false_parents(self):
         command.fix_inconsistent_results(self.project.results.all())
 
         result = self.project.results.get(title='Result #1')

akvo/rsr/usecases/change_project_parent.py

@@ -4,22 +4,23 @@
 # See more details in the license.txt file located at the root folder of the Akvo RSR module.
 # For additional details on the GNU license please see < http://www.gnu.org/licenses/agpl.html >.
 
+from typing import Dict, Optional
 from django.db import transaction
 from akvo.rsr.models import Project, RelatedProject
 from akvo.rsr.usecases.utils import (
-    RF_MODELS_CONFIG, get_direct_lineage_hierarchy_ids, make_tree_from_list, make_target_parent_map
+    RF_MODELS_CONFIG, get_direct_lineage_hierarchy_ids, make_trees_from_list, make_source_to_target_map
 )
 
 
-def get_rf_change_candidates(project: Project, new_parent: Project):
+def get_rf_change_candidates(project: Project, new_parent: Project) -> Dict[str, Dict[int, Optional[int]]]:
     project_ids = get_direct_lineage_hierarchy_ids(project, new_parent)
     candidates = {}
     for key, config in RF_MODELS_CONFIG.items():
         model, parent_attr, project_relation, _ = config
         filter_arg = {f"{project_relation}__in": project_ids}
         items = model.objects.filter(**filter_arg).values('id', parent_attr, project_relation)
-        items_tree = make_tree_from_list(items, parent_attr)
-        candidates[key] = make_target_parent_map(items_tree, project_relation, project.id, new_parent.id)
+        items_tree = make_trees_from_list(items, parent_attr)
+        candidates[key] = make_source_to_target_map(items_tree, project_relation, project.pk, new_parent.pk)
     return candidates
 
 
@@ -36,7 +37,7 @@ def change_parent(project: Project, new_parent: Project, reimport=False, verbosi
     """
 
     old_parent = project.parents_all().first()
-    if old_parent.id == new_parent.id:
+    if old_parent is None or old_parent.pk == new_parent.pk:
         if verbosity > 0:
             print("New parent same as current parent")
         return
@@ -49,7 +50,7 @@ def change_parent(project: Project, new_parent: Project, reimport=False, verbosi
                 print(f"Change {key} parent of {item_id} to {target_id}")
             model.objects.filter(id__in=[item_id]).update(**{f"{parent_attr}_id": target_id})
     if verbosity > 0:
-        print(f"Change project {project.title} (ID:{project.id}) parent to {new_parent.title} (ID:{new_parent.id})")
+        print(f"Change project {project.title} (ID:{project.pk}) parent to {new_parent.title} (ID:{new_parent.pk})")
     RelatedProject.objects.filter(
         project=old_parent, related_project=project, relation='2'
     ).update(project=new_parent)

akvo/rsr/usecases/fix_inconsistent_results.py

@@ -8,7 +8,7 @@
 from django.db import transaction
 from akvo.rsr.models import Result, Project
 from akvo.rsr.usecases.utils import (
-    RF_MODELS_CONFIG, get_direct_lineage_hierarchy_ids, make_tree_from_list, filter_trees, make_target_parent_map
+    RF_MODELS_CONFIG, get_direct_lineage_hierarchy_ids, make_trees_from_list, filter_trees, make_source_to_target_map
 )
 
 
@@ -19,26 +19,39 @@ def fix_inconsistent_results(results: Iterable[Result]):
 
 @transaction.atomic
 def fix_inconsistent_result(result: Result):
+    """Fix inconsisten result between parent result project and parent project.
+
+    This function fixes the parent of a result that have inconsistency between the project
+    of the parent result and the parent project of the result. It also fixes the parent of
+    all the indicators and indicator periods under the result. It resolve the correct parents
+    by traversing from the false parent result project up to the project hierarchy to find
+    the nearest common ancestor than creates a binary lineage tree connecting the false
+    parent result project to to correct result's parent project using the nearest common
+    ancestor as root. Then, it uses the lineage tree path to resolve the correct parent
+    of the result, indicators, and periods respectively.
+
+    """
+
     if result.parent_result is None:
         return
 
     parent_project = result.project.parents_all().first()
     parent_result = result.parent_result
-    parent_result_project = parent_result.project
 
-    if parent_project == parent_result_project:
+    if parent_project == parent_result.project:
         return
 
-    change_candidates = get_change_candidates(parent_result, parent_result_project, parent_project)
+    change_candidates = get_change_candidates(parent_result, parent_project)
     for key, candidates in change_candidates.items():
         model, parent_attr, project_relation, _ = RF_MODELS_CONFIG[key]
-        for orig_parent_id, target_parent_id in candidates.items():
-            filter_args = {parent_attr: orig_parent_id, project_relation: result.project}
+        for source_parent_id, target_parent_id in candidates.items():
+            filter_args = {parent_attr: source_parent_id, project_relation: result.project}
             model.objects.filter(**filter_args).update(**{f"{parent_attr}_id": target_parent_id})
 
 
-def get_change_candidates(result: Result, result_project: Project, target_project: Optional[Project]) -> Dict[str, Dict[int, int]]:
-    project_ids = get_direct_lineage_hierarchy_ids(result_project, target_project)
+def get_change_candidates(result: Result, target_project: Optional[Project]) -> Dict[str, Dict[int, Optional[int]]]:
+    source_project = result.project
+    project_ids = get_direct_lineage_hierarchy_ids(source_project, target_project)
     candidates = {}
     for key, config in RF_MODELS_CONFIG.items():
         if key not in ['results', 'indicators', 'periods']:
@@ -47,10 +60,10 @@ def get_change_candidates(result: Result, result_project: Project, target_projec
         filter_args = {f"{project_relation}__in": project_ids}
         select_attrs = [a for a in ['id', parent_attr, project_relation, result_relation] if a is not None]
         items = model.objects.filter(**filter_args).values(*select_attrs)
-        filter_func = (lambda node: node.item[result_relation] == result.pk) \
+        is_member_of_result = (lambda node: node.item[result_relation] == result.pk) \
             if result_relation is not None \
             else (lambda node: node.item['id'] == result.pk)
-        item_trees = make_tree_from_list(items, parent_attr)
-        filtered_item_trees = filter_trees(item_trees, filter_func)
-        candidates[key] = make_target_parent_map(filtered_item_trees, project_relation, result_project.pk, target_project.pk if target_project else None)
+        item_trees = make_trees_from_list(items, parent_attr)
+        filtered_item_trees = filter_trees(item_trees, is_member_of_result)
+        candidates[key] = make_source_to_target_map(filtered_item_trees, project_relation, source_project.pk, target_project.pk if target_project else None)
     return candidates

akvo/rsr/usecases/utils.py

@@ -7,7 +7,7 @@
 from akvo.rsr.models import Project, RelatedProject, Result, Indicator, IndicatorPeriod, IndicatorDimensionName, IndicatorDimensionValue, DefaultPeriod
 from dataclasses import dataclass, field
 from django.db.models import Q
-from typing import List, Any, Callable, Set, Optional, Iterable, Dict
+from typing import List, Callable, Set, Optional, Iterable, Dict, TypedDict, Tuple
 
 RF_MODELS_CONFIG = {
     # key: (Model, parent_attribute, project_relation, result_relation)
@@ -47,6 +47,45 @@ def get_first_common_item(left: List[int], right: List[int]) -> Optional[int]:
 
 
 def get_direct_lineage_hierarchy_ids(lhs_project: Optional[Project], rhs_project: Optional[Project]) -> Set[Optional[int]]:
+    """Resolve the direct lineage path from lhs_project node to rhs_project node in a projects hierarchy tree
+
+    Given a projects hierarchy
+
+                 A
+                / \
+               B   C
+                 / | \
+                D  E  F
+                     / \
+                    G   H
+
+    The lineage path from D to G
+
+                   C
+                 /   \
+                D     F
+                     /
+                    G
+
+    With C as the common ancestor, the ancestor lineage above C is omitted as it is unnecessary
+
+    get_direct_lineage_hierarchy_ids(D, G) -> [D.id, C.id, F.id, G.id]
+
+
+    If any projects of the function arguments are empty, the function will resolve the lineage
+    path of the non-empty project up to the root project
+
+    The lineage path from D to G
+
+                 A
+                  \
+                   C
+                 /
+                D
+
+    get_direct_lineage_hierarchy_ids(D, None) -> [D.id, C.id, A.id]
+
+    """
     lhs_lineage = get_project_lineage_ids(lhs_project) if lhs_project else []
     rhs_lineage = get_project_lineage_ids(rhs_project) if rhs_project else []
     # Only hierarchy up to the nearest common ancestor are needed to link between the project and the new parent.
@@ -57,13 +96,17 @@ def get_direct_lineage_hierarchy_ids(lhs_project: Optional[Project], rhs_project
     return set(lhs_lineage).symmetric_difference(set(rhs_lineage)) | {common_ancestor}
 
 
+class DictWithId(TypedDict):
+    id: int
+
+
 @dataclass(frozen=True)
 class TreeNode:
-    item: Any
+    item: DictWithId
     children: List['TreeNode'] = field(default_factory=list)
 
 
-def make_tree_from_list(items, parent_attr) -> List[TreeNode]:
+def make_trees_from_list(items: Iterable[DictWithId], parent_attr: str) -> List[TreeNode]:
     tree = []
     ids = [it['id'] for it in items]
     lookup = {item['id']: TreeNode(item=item) for item in items}
@@ -80,46 +123,76 @@ def make_tree_from_list(items, parent_attr) -> List[TreeNode]:
 
 
 def filter_trees(trees: List[TreeNode], predicate: Callable[[TreeNode], bool]) -> List[TreeNode]:
-    return [tree for tree in trees if is_tree_contains(tree, predicate)]
+    return [tree for tree in trees if does_tree_contain(tree, predicate)]
 
 
-def is_tree_contains(tree: TreeNode, predicate: Callable[[TreeNode], bool]) -> bool:
+def does_tree_contain(tree: TreeNode, predicate: Callable[[TreeNode], bool]) -> bool:
     if predicate(tree):
         return True
     for child in tree.children:
-        if is_tree_contains(child, predicate):
+        if does_tree_contain(child, predicate):
             return True
     return False
 
 
-def get_leaf_item(node: TreeNode):
-    return get_leaf_item(node.children[0]) if len(node.children) else node.item
+def get_leaf_item_of_single_lineage_path_tree(node: TreeNode):
+    if len(node.children) > 1:
+        raise ValueError('Expect node to be a TreeNode with single child')
+    return get_leaf_item_of_single_lineage_path_tree(node.children[0]) if len(node.children) else node.item
 
 
-def make_target_parent_map(trees: Iterable[TreeNode], project_attr, original_project_id, target_project_id) -> Dict[int, int]:
+def make_source_to_target_map(trees: Iterable[TreeNode], project_attr: str, source_project_id: int, target_project_id: Optional[int]) -> Dict[int, int]:
+    """Resolve node item ids of source and target projects of each tree
+    """
     target_map = {}
     for node in trees:
-        if len(node.children) == 0:
-            target_map[node.item['id']] = None
-            continue
-        first_leaf = get_leaf_item(node.children[0])
-        if len(node.children) == 1 \
-                and node.item[project_attr] == original_project_id \
-                and first_leaf[project_attr] == target_project_id:
-            target_map[node.item['id']] = first_leaf['id']
-            continue
-        if len(node.children) == 1 \
-                and first_leaf[project_attr] == original_project_id:
-            target_map[first_leaf['id']] = None
-            continue
-        second_leaf = get_leaf_item(node.children[1])
-        if len(node.children) == 2 \
-                and first_leaf[project_attr] == original_project_id:
-            target_map[first_leaf['id']] = second_leaf['id'] if second_leaf[project_attr] == target_project_id else None
-            continue
-        if len(node.children) == 2 \
-                and second_leaf[project_attr] == original_project_id:
-            target_map[second_leaf['id']] = first_leaf['id'] if first_leaf[project_attr] == target_project_id else None
-            continue
-        print('Ignoring ambiguous lineage tree node', node)
+        source_id, target_id = get_source_to_target_pair(node, project_attr, source_project_id, target_project_id)
+        if source_id is None and target_id is None:
+            print('Ignoring ambiguous lineage tree node', node)
+        else:
+            target_map[source_id] = target_id
     return target_map
+
+
+def get_source_to_target_pair(node: TreeNode, project_attr: str, source_project_id: int, target_project_id: Optional[int]) -> Tuple[Optional[int], Optional[int]]:
+    """Resolve node item ids of source and target projects of a direct lineage path tree
+
+    Expected node structure is that the root node (as the common ancestor) only have two children (two path)
+    and each of the children only have single lineage path to either the source or the target project
+
+                 A
+                / \
+               B   C
+              /     \
+             D       E
+                      \
+                       F
+
+    Descendants nodes with multiple children will be considered ambiguous and will not be resolved
+
+    """
+    if len(node.children) == 0 and node.item[project_attr] == source_project_id:
+        return node.item['id'], None
+
+    first_leaf = get_leaf_item_of_single_lineage_path_tree(node.children[0])
+
+    if len(node.children) == 1 \
+            and node.item[project_attr] == source_project_id \
+            and first_leaf[project_attr] == target_project_id:
+        return node.item['id'], first_leaf['id']
+
+    if len(node.children) == 1 \
+            and first_leaf[project_attr] == source_project_id:
+        return first_leaf['id'], None
+
+    second_leaf = get_leaf_item_of_single_lineage_path_tree(node.children[1])
+
+    if len(node.children) == 2 \
+            and first_leaf[project_attr] == source_project_id:
+        return first_leaf['id'], second_leaf['id'] if second_leaf[project_attr] == target_project_id else None
+
+    if len(node.children) == 2 \
+            and second_leaf[project_attr] == source_project_id:
+        return second_leaf['id'], first_leaf['id'] if first_leaf[project_attr] == target_project_id else None
+
+    return None, None