【问题标题】：How to unit-test code depending on a complex graph database interaction如何根据复杂的图形数据库交互对代码进行单元测试
【发布时间】：2021-05-31 10:54:43
【问题描述】：

我正在开发一个基于 Neo4j 的 Python 应用程序，我面临着编写与 Neo4j 数据库隔离的单元测试的挑战，因此我可以运行我的单元测试而不依赖于启动一个成熟的数据库。

Neo4j Java SDK 包含一个 "In-Memory / Impermanent database"

很遗憾，Python driver 没有。

想法

模拟

我考虑过模拟数据库，但是你 shouldn't mock what you don't own。尤其是第三方库。

因为我希望我的单元测试可靠并遵循良好的设计，所以我不想走这条路。

端口和适配器

从软件架构的角度来看，集成像 Neo4j 这样的第三方库应该遵循端口和适配器模式。这将为我带来以下好处

将我的业务逻辑与图形逻辑解耦
具有明确定义的合同（端口），适配器必须实施
有一个假 Neo4j 适配器实现，将传递给我的业务逻辑以进行单元测试

这是有希望的，但我对要编写的代码量感到不知所措。

带有官方 Neo4j 驱动程序的示例代码

driver = GraphDatabase.driver("bolt://localhost:7687")
with driver.session() as session:
    query = """MATCH (u:USER)
             WHERE u.name = $name
             RETURN u"""
    cursor = session.run(query, parameters={'name', 'Alfred'})
    for result in cursor:
        acc_type = result['u']['account_type']
        return acc_type

“端口和适配器”实现示例

定义 API（端口）

from abc import ABC, abstractmethod

class AbstractGraphService(ABC):
    def __init__(url: str):
        pass

    @abstractmethod
    def __enter__():
        """returns a session"""
        pass

    @abstractmethod
    def __exit__(*args):
        """close session object"""
        pass


class AbstractSession(ABC):

    @abstractmethod
    def run(self, query, parameters):
        """execute a cypher query"""
        pass

➡️ 我怎样才能写一个假的，内存中的实现？我是否也应该编写一个 Fake Cypher 执行引擎？

➡️ 使用上面演示的 API，我还没有将 Cypher 查询与业务逻辑代码解耦

➡️ 是否有更好的方法来对我的代码进行单元测试？

感谢您的建议

【问题讨论】：

标签： python unit-testing neo4j architecture cypher

【解决方案1】：

您可以使用单独的 Python 函数来连接数据库、查询它并处理返回。对于后者，Python pandas 包具有很好的功能。使用它，其中 Q 是您的密码查询，Neo4jServer 可能是远程或本地的，如 Neo4Server ="bolt://{IP or localhost}:7687"。如果您使用的是 Neo4j 4.x，则需要指定具体的数据库。

from neo4j import GraphDatabase
from pandas import DataFrame

def CypherToPandas(Q, database):
    driver=GraphDatabase.driver(Neo4Server, auth=(Neo4UserName,Neo4Pswd),database=database ) 
    with driver.session(database=database) as cyphersession:
        rslt = cyphersession.run(Q)
        df = DataFrame(rslt.data())
        cyphersession.close()
        driver.close()
        return df

我有多种功能，可以重新用于多种用途。 Neo4j 代码与业务逻辑是分开的。然后，您可以运行单元测试，并在测试后将它们无缝地整合到您的解决方案中。

【讨论】：