IFC-727 calculate diff tree in proposed change pipeline (#4546)

* remove test that belongs in sdk

* calculate diff in pc pipeline

* update integration test

* point to SDK commit of partner PR

* fix unit test, generate docs

* clean up NodeDiff imports following SDK changes

* force full import for docs generation

* update python_sdk to latest stable commit

* try to catch EntityNotFoundError

Author: ajtmccarty

backend/infrahub/core/validators/determiner.py

@@ -1,7 +1,7 @@
 from collections import defaultdict
 from typing import Union
 
-from infrahub_sdk.client import NodeDiff
+from infrahub_sdk.diff import NodeDiff
 
 from infrahub.core.constants import RelationshipKind, SchemaPathType
 from infrahub.core.constants.schema import UpdateSupport

backend/infrahub/database/__init__.py

@@ -417,7 +417,10 @@ async def wrapper(*args, **kwargs):
             for attempt in range(1, config.SETTINGS.database.retry_limit + 1):
                 try:
                     return await func(*args, **kwargs)
-                except TransientError as exc:
+                except (TransientError, ClientError) as exc:
+                    if isinstance(exc, ClientError):
+                        if exc.code != "Neo.ClientError.Statement.EntityNotFound":
+                            raise exc
                     retry_time: float = random.randrange(100, 500) / 1000
                     log.info(
                         f"Retrying database transaction, attempt {attempt}/{config.SETTINGS.database.retry_limit}",

backend/infrahub/message_bus/operations/requests/proposed_change.py

@@ -134,6 +134,12 @@ async def pipeline(message: messages.RequestProposedChangePipeline, service: Inf
 
         await _gather_repository_repository_diffs(repositories=repositories)
 
+        destination_branch = await registry.get_branch(db=service.database, branch=message.destination_branch)
+        source_branch = await registry.get_branch(db=service.database, branch=message.source_branch)
+        component_registry = get_component_registry()
+        async with service.database.start_transaction() as dbt:
+            diff_coordinator = await component_registry.get_component(DiffCoordinator, db=dbt, branch=source_branch)
+            await diff_coordinator.update_branch_diff(base_branch=destination_branch, diff_branch=source_branch)
         diff_summary = await service.client.get_diff_summary(branch=message.source_branch)
         branch_diff = ProposedChangeBranchDiff(diff_summary=diff_summary, repositories=repositories)
         await _populate_subscribers(branch_diff=branch_diff, service=service, branch=message.source_branch)

backend/infrahub/message_bus/types.py

@@ -3,7 +3,7 @@
 import re
 from enum import Enum
 
-from infrahub_sdk.client import NodeDiff  # noqa: TCH002
+from infrahub_sdk.diff import NodeDiff  # noqa: TCH002
 from pydantic import BaseModel, Field
 
 from infrahub.core.constants import InfrahubKind, RepositoryInternalStatus

backend/tests/integration/message_bus/operations/request/test_proposed_change.py

@@ -136,6 +136,7 @@ async def test_run_pipeline_validate_requested_jobs(
     services.service._client = client
     services.service.log = fake_log
     services.service.message_bus = bus_pre_data_changes
+    services.service._database = db
     services.prepare(service=services.service)
     await pipeline(message=message, service=services.service)
 

backend/tests/unit/core/constraint_validators/test_determiner.py

@@ -1,5 +1,5 @@
 import pytest
-from infrahub_sdk.client import NodeDiff
+from infrahub_sdk.diff import NodeDiff
 
 from infrahub.core import registry
 from infrahub.core.branch import Branch

backend/tests/unit/graphql/test_graphql_query.py

@@ -8,7 +8,6 @@
 from infrahub.core import registry
 from infrahub.core.branch import Branch
 from infrahub.core.constants import BranchSupportType, InfrahubKind
-from infrahub.core.initialization import create_branch
 from infrahub.core.manager import NodeManager
 from infrahub.core.node import Node
 from infrahub.core.schema import NodeSchema
@@ -687,251 +686,6 @@ async def test_display_label_nested_query(
     assert DeepDiff(result.data["TestPerson"]["edges"][0]["node"], expected_result, ignore_order=True).to_dict() == {}
 
 
-def _check_diff_for_branch_and_id(all_dicts: list[dict], branch_name: str, id: str, things_to_check: dict) -> dict:
-    this_dict = None
-    for one_dict in all_dicts:
-        if one_dict["branch"] == branch_name and one_dict["id"] == id:
-            this_dict = one_dict
-            break
-    if not this_dict:
-        raise ValueError(f"No diff for branch={branch_name} and id={id}")
-    for key, value in things_to_check.items():
-        assert this_dict.get(key) == value
-    return this_dict
-
-
-async def test_query_diffsummary(db: InfrahubDatabase, default_branch: Branch, car_person_schema: SchemaBranch):
-    car = registry.schema.get(name="TestCar")
-    person = registry.schema.get(name="TestPerson")
-
-    p1_main = await Node.init(db=db, schema=person)
-    await p1_main.new(db=db, name="John", height=180)
-    await p1_main.save(db=db)
-    p2_main = await Node.init(db=db, schema=person)
-    await p2_main.new(db=db, name="Jane", height=170)
-    await p2_main.save(db=db)
-
-    c1_main = await Node.init(db=db, schema=car)
-    await c1_main.new(db=db, name="volt", nbr_seats=4, is_electric=True, owner=p1_main)
-    await c1_main.save(db=db)
-    c2_main = await Node.init(db=db, schema=car)
-    await c2_main.new(db=db, name="bolt", nbr_seats=4, is_electric=True, owner=p1_main)
-    await c2_main.save(db=db)
-    c3_main = await Node.init(db=db, schema=car)
-    await c3_main.new(db=db, name="nolt", nbr_seats=4, is_electric=True, owner=p2_main)
-    await c3_main.save(db=db)
-
-    branch2 = await create_branch(branch_name="branch2", db=db)
-    await c1_main.delete(db=db)
-    p1_branch2 = await NodeManager.get_one_by_id_or_default_filter(
-        id=p1_main.id, db=db, kind="TestPerson", branch=branch2
-    )
-    p1_branch2.name.value = "Jonathan"
-    await p1_branch2.save(db=db)
-    p2_main.name.value = "Jeanette"
-    await p2_main.save(db=db)
-    c2_main.name.value = "bolting"
-    await c2_main.save(db=db)
-    c3_branch2 = await NodeManager.get_one_by_id_or_default_filter(id=c3_main.id, db=db, kind="TestCar", branch=branch2)
-    await c3_branch2.owner.update(data=p1_branch2.id, db=db)
-    await c3_branch2.save(db=db)
-
-    query = """
-        query {
-            DiffSummary {
-                branch
-                id
-                kind
-                action
-                display_label
-                elements {
-                    element_type
-                    name
-                    action
-                    summary {
-                        added
-                        updated
-                        removed
-                    }
-                    ... on DiffSummaryElementRelationshipMany {
-                        peers {
-                            action
-                            summary {
-                                added
-                                updated
-                                removed
-                            }
-                        }
-                    }
-                }
-            }
-        }
-    """
-    gql_params = prepare_graphql_params(db=db, include_mutation=False, include_subscription=False, branch=branch2)
-    result = await graphql(
-        schema=gql_params.schema,
-        source=query,
-        context_value=gql_params.context,
-        root_value=None,
-        variable_values={},
-    )
-
-    assert result.errors is None
-    assert result.data
-    diff_summary = result.data["DiffSummary"]
-
-    assert len(diff_summary) == 5
-
-    _check_diff_for_branch_and_id(
-        all_dicts=diff_summary,
-        branch_name="main",
-        id=c1_main.id,
-        things_to_check={
-            "branch": "main",
-            "id": c1_main.id,
-            "kind": "TestCar",
-            "action": "REMOVED",
-            "display_label": "",
-        },
-    )
-    _check_diff_for_branch_and_id(
-        all_dicts=diff_summary,
-        branch_name="main",
-        id=c2_main.id,
-        things_to_check={
-            "branch": "main",
-            "id": c2_main.id,
-            "kind": "TestCar",
-            "action": "UPDATED",
-            "display_label": "bolting #444444",
-            "elements": [
-                {
-                    "element_type": "ATTRIBUTE",
-                    "name": "name",
-                    "action": "UPDATED",
-                    "summary": {"added": 0, "updated": 1, "removed": 0},
-                }
-            ],
-        },
-    )
-    c3_branch2_diff = _check_diff_for_branch_and_id(
-        all_dicts=diff_summary,
-        branch_name="branch2",
-        id=c3_branch2.id,
-        things_to_check={
-            "branch": "branch2",
-            "id": c3_branch2.id,
-            "kind": "TestCar",
-            "action": "UPDATED",
-            "display_label": "nolt #444444",
-        },
-    )
-    c3_branch2_diff_elements = c3_branch2_diff["elements"]
-    assert len(c3_branch2_diff_elements) == 1
-    assert c3_branch2_diff_elements[0]["element_type"] == "RELATIONSHIP_ONE"
-    assert c3_branch2_diff_elements[0]["name"] == "owner"
-    assert c3_branch2_diff_elements[0]["action"] == "UPDATED"
-    p2_main_diff = _check_diff_for_branch_and_id(
-        all_dicts=diff_summary,
-        branch_name="main",
-        id=p2_main.id,
-        things_to_check={
-            "branch": "main",
-            "id": p2_main.id,
-            "kind": "TestPerson",
-            "action": "UPDATED",
-            "display_label": "Jeanette",
-        },
-    )
-    p2_main_diff_elements = p2_main_diff["elements"]
-    assert len(p2_main_diff_elements) == 1
-    assert p2_main_diff_elements[0]["element_type"] == "ATTRIBUTE"
-    assert p2_main_diff_elements[0]["name"] == "name"
-    assert p2_main_diff_elements[0]["action"] == "UPDATED"
-    assert p2_main_diff_elements[0]["summary"] == {"added": 0, "updated": 1, "removed": 0}
-    p1_branch2_diff = _check_diff_for_branch_and_id(
-        all_dicts=diff_summary,
-        branch_name="branch2",
-        id=p1_branch2.id,
-        things_to_check={
-            "branch": "branch2",
-            "id": p1_branch2.id,
-            "kind": "TestPerson",
-            "action": "UPDATED",
-            "display_label": "Jonathan",
-        },
-    )
-    p1_branch2_diff_elements = p1_branch2_diff["elements"]
-    assert len(p1_branch2_diff_elements) == 2
-    p1_branch2_diff_elements_map = {elem["name"]: elem for elem in p1_branch2_diff_elements}
-    assert {"cars", "name"} == set(p1_branch2_diff_elements_map.keys())
-    name_element = p1_branch2_diff_elements_map["name"]
-    assert name_element["name"] == "name"
-    assert name_element["element_type"] == "ATTRIBUTE"
-    assert name_element["action"] == "UPDATED"
-    cars_element = p1_branch2_diff_elements_map["cars"]
-    assert cars_element["name"] == "cars"
-    assert cars_element["element_type"] == "RELATIONSHIP_MANY"
-    assert cars_element["action"] == "ADDED"
-    assert len(cars_element["peers"]) == 1
-    assert cars_element["peers"][0]["action"] == "ADDED"
-
-
-async def test_diffsummary_on_default_branch(
-    db: InfrahubDatabase, default_branch: Branch, car_person_schema: SchemaBranch
-):
-    person = registry.schema.get(name="TestPerson")
-
-    before_create = Timestamp()
-    p1 = await Node.init(db=db, schema=person)
-    await p1.new(db=db, name="John", height=180)
-    await p1.save(db=db)
-    p2 = await Node.init(db=db, schema=person)
-    await p2.new(db=db, name="Jane", height=170)
-    await p2.save(db=db)
-
-    query = """
-        query DiffSummaries($time_from: String) {
-            DiffSummary(time_from: $time_from) {
-                branch
-                id
-                kind
-                action
-            }
-        }
-    """
-    gql_params = prepare_graphql_params(
-        db=db, include_mutation=False, include_subscription=False, branch=default_branch
-    )
-    result = await graphql(
-        schema=gql_params.schema,
-        source=query,
-        context_value=gql_params.context,
-        root_value=None,
-        variable_values={},
-    )
-    assert result.errors
-    assert len(result.errors) == 1
-    assert result.errors[0].message == "time_from is required on default branch"
-
-    gql_params = prepare_graphql_params(
-        db=db, include_mutation=False, include_subscription=False, branch=default_branch
-    )
-    result = await graphql(
-        schema=gql_params.schema,
-        source=query,
-        context_value=gql_params.context,
-        root_value=None,
-        variable_values={"time_from": before_create.to_string()},
-    )
-    assert result.errors is None
-    assert result.data
-    summaries = result.data["DiffSummary"]
-    assert len(summaries) == 2
-    assert {"branch": default_branch.name, "kind": "TestPerson", "id": p1.get_id(), "action": "ADDED"} in summaries
-    assert {"branch": default_branch.name, "kind": "TestPerson", "id": p2.get_id(), "action": "ADDED"} in summaries
-
-
 async def test_query_typename(db: InfrahubDatabase, default_branch: Branch, car_person_schema: SchemaBranch):
     car = registry.schema.get(name="TestCar")
     person = registry.schema.get(name="TestPerson")

backend/tests/unit/message_bus/operations/requests/test_proposed_change.py

@@ -288,7 +288,7 @@ async def test_schema_integrity(
     branch2_schema.set(name="TestPerson", schema=person_schema)
 
     # Ignore creation of Task Report response
-    httpx_mock.add_response(method="POST", url="http://mock/graphql", json={"data": {}})
+    httpx_mock.add_response(method="POST", url="http://mock/graphql/main", json={"data": {}})
     await proposed_change.schema_integrity(message=schema_integrity_01, service=service_all)
 
     checks = await registry.manager.query(db=db, schema=InfrahubKind.SCHEMACHECK)

docs/docs/infrahubctl/infrahubctl-protocols.mdx

@@ -10,7 +10,9 @@ $ infrahubctl protocols [OPTIONS]
 
 **Options**:
 
+* `--schemas PATH`: List of schemas or directory to load.
 * `--branch TEXT`: Branch of schema to export Python protocols for.
+* `--sync / --no-sync`: Generate for sync or async.  [default: no-sync]
 * `--config-file TEXT`: [env var: INFRAHUBCTL_CONFIG; default: infrahubctl.toml]
 * `--out TEXT`: Path to a file to save the result.  [default: schema_protocols.py]
 * `--install-completion`: Install completion for the current shell.