DataOps-platform/tests/integration/test_dataflow_create_saga_neo4j.py

from __future__ import annotations

import os
from concurrent.futures import ThreadPoolExecutor
from threading import Barrier

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_concurrent_different_uids_same_name_is_closed():
    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,
    )
    name = f"同名并发生产线-{new_governance_uid()}"
    uids = [new_governance_uid(), new_governance_uid()]
    start = Barrier(2)
    driver = GraphDatabase.driver(
        uri, auth=(user, password), encrypted=False
    )

    def create(uid):
        start.wait()
        try:
            with app.app_context():
                DataFlowService._merge_governed_dataflow(
                    {
                        "uid": uid,
                        "name_zh": name,
                        "name_en": f"concurrent_{uid.replace('-', '')}",
                        "script_type": "governed",
                        "script_requirement": "{}",
                        "script_path": "",
                    }
                )
            return "created"
        except ValueError as exc:
            return str(exc)

    try:
        with driver.session() as session:
            session.run(
                "MATCH (n:DataFlow {name_zh: $name}) DETACH DELETE n",
                {"name": name},
            ).consume()
        with ThreadPoolExecutor(max_workers=2) as pool:
            outcomes = list(pool.map(create, uids))
        assert sorted(outcomes) == ["created", "dataflow_uid_conflict"]
        with driver.session() as session:
            count = session.run(
                "MATCH (n:DataFlow {name_zh: $name}) "
                "RETURN count(n) AS count",
                {"name": name},
            ).single()["count"]
            constraints = session.run(
                "SHOW CONSTRAINTS YIELD name "
                "WHERE name IN ['data_flow_uid', 'data_flow_name_zh'] "
                "RETURN collect(name) AS names"
            ).single()["names"]
        assert count == 1
        assert set(constraints) == {"data_flow_uid", "data_flow_name_zh"}
    finally:
        with driver.session() as session:
            session.run(
                "MATCH (n:DataFlow {name_zh: $name}) DETACH DELETE n",
                {"name": name},
            ).consume()
        driver.close()


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 IN ['data_flow_uid', 'data_flow_name_zh'] "
                    "RETURN name"
                )
            ]
        assert count == 1
        assert set(constraints) == {"data_flow_uid", "data_flow_name_zh"}
    finally:
        with driver.session() as session:
            session.run("MATCH (n:DataFlow {uid: $uid}) DETACH DELETE n", {"uid": uid})
        driver.close()