DataOps-platform/tests/integration/test_dataflow_create_saga_neo4j.py

from __future__ import annotations

import os

import pytest
from neo4j import GraphDatabase

from app import create_app
from app.core.common.identifiers import new_governance_uid
from app.core.data_flow.dataflows import DataFlowService


def test_real_neo4j_uid_constraint_merge_replay_and_conflict():
    uri = os.environ.get("DATA_RULE_NEO4J_ACCEPTANCE_URI")
    password = os.environ.get("DATA_RULE_NEO4J_ACCEPTANCE_PASSWORD")
    if not uri or not password:
        pytest.skip("real Neo4j acceptance connection is not configured")
    user = os.environ.get("DATA_RULE_NEO4J_ACCEPTANCE_USER", "neo4j")
    app = create_app()
    app.config.update(
        TESTING=True,
        NEO4J_URI=uri,
        NEO4J_USER=user,
        NEO4J_PASSWORD=password,
        NEO4J_ENCRYPTED=False,
    )
    uid = new_governance_uid()
    node = {
        "uid": uid,
        "name_zh": f"Saga验收-{uid}",
        "name_en": f"saga-{uid}",
        "script_type": "governed",
        "script_requirement": '{"dataflow_spec":{"schema_version":"2.0"}}',
        "script_path": "",
    }
    driver = GraphDatabase.driver(uri, auth=(user, password), encrypted=False)
    try:
        with app.app_context():
            first_id, first = DataFlowService._merge_governed_dataflow(node)
            second_id, second = DataFlowService._merge_governed_dataflow(node)
            assert first_id == second_id
            assert first == second
            with pytest.raises(ValueError, match="dataflow_uid_conflict"):
                DataFlowService._merge_governed_dataflow(
                    {**node, "name_zh": f"篡改-{uid}"}
                )
        with driver.session() as session:
            count = session.run(
                "MATCH (n:DataFlow {uid: $uid}) RETURN count(n) AS count",
                {"uid": uid},
            ).single()["count"]
            constraints = [
                record["name"]
                for record in session.run(
                    "SHOW CONSTRAINTS YIELD name WHERE name = 'data_flow_uid' "
                    "RETURN name"
                )
            ]
        assert count == 1
        assert constraints == ["data_flow_uid"]
    finally:
        with driver.session() as session:
            session.run("MATCH (n:DataFlow {uid: $uid}) DETACH DELETE n", {"uid": uid})
        driver.close()