5：高档和低档的TDD
我们已经引入了服务层，以捕获工作应用程序所需的一些额外的编排职责。服务层帮助我们清晰地定义我们的用例以及每个用例的工作流程：我们需要从仓库中获取什么，我们应该做什么预检查和当前状态验证，以及我们在最后保存什么。
但是目前，我们的许多单元测试都在较低级别运行，直接作用于模型。在本章中，我们将讨论将这些测试提升到服务层级别所涉及的权衡，以及一些更通用的测试指南。
我们的测试金字塔看起来如何？
让我们看看这种转向使用服务层及其自身的服务层测试，会对我们的测试金字塔产生什么影响
$ grep -c test_ */*/test_*.py
tests/unit/test_allocate.py:4
tests/unit/test_batches.py:8
tests/unit/test_services.py:3
tests/integration/test_orm.py:6
tests/integration/test_repository.py:2
tests/e2e/test_api.py:2不错！我们有 15 个单元测试，8 个集成测试，以及仅仅 2 个端到端测试。这已经是一个看起来很健康的测试金字塔了。
领域层测试应该移动到服务层吗？
让我们看看如果我们更进一步会发生什么。既然我们可以针对服务层测试我们的软件，我们实际上不再需要领域模型的测试了。相反，我们可以根据服务层重写来自 [chapter_01_domain_model] 的所有领域级别测试
# domain-layer test:
def test_prefers_current_stock_batches_to_shipments():
in_stock_batch = Batch("in-stock-batch", "RETRO-CLOCK", 100, eta=None)
shipment_batch = Batch("shipment-batch", "RETRO-CLOCK", 100, eta=tomorrow)
line = OrderLine("oref", "RETRO-CLOCK", 10)
allocate(line, [in_stock_batch, shipment_batch])
assert in_stock_batch.available_quantity == 90
assert shipment_batch.available_quantity == 100
# service-layer test:
def test_prefers_warehouse_batches_to_shipments():
in_stock_batch = Batch("in-stock-batch", "RETRO-CLOCK", 100, eta=None)
shipment_batch = Batch("shipment-batch", "RETRO-CLOCK", 100, eta=tomorrow)
repo = FakeRepository([in_stock_batch, shipment_batch])
session = FakeSession()
line = OrderLine('oref', "RETRO-CLOCK", 10)
services.allocate(line, repo, session)
assert in_stock_batch.available_quantity == 90
assert shipment_batch.available_quantity == 100我们为什么要这样做？
测试应该帮助我们无所畏惧地更改我们的系统，但我们经常看到团队针对他们的领域模型编写了太多的测试。当他们来更改他们的代码库并发现他们需要更新数十甚至数百个单元测试时，这会导致问题。
如果你停下来思考自动化测试的目的，这就有道理了。我们使用测试来强制系统的某个属性在我们工作时不会改变。我们使用测试来检查 API 是否继续返回 200，数据库会话是否继续提交，以及订单是否仍在被分配。
如果我们不小心更改了其中一种行为，我们的测试将会失败。然而，另一方面是，如果我们想更改我们代码的设计，任何直接依赖于该代码的测试也会失败。
随着我们深入本书，您将看到服务层如何为我们的系统形成一个 API，我们可以通过多种方式驱动它。针对此 API 进行测试减少了我们在重构领域模型时需要更改的代码量。如果我们仅限于仅针对服务层进行测试，我们将不会有任何直接与模型对象上的“私有”方法或属性交互的测试，这使我们更自由地重构它们。
|
提示
|
我们在测试中放入的每一行代码都像一团胶水，将系统固定成特定的形状。我们拥有的低级别测试越多，更改事物就越困难。 |
关于决定编写哪种类型的测试
您可能会问自己，“那么，我应该重写我所有的单元测试吗？针对领域模型编写测试是错误的吗？” 为了回答这些问题，重要的是要理解耦合和设计反馈之间的权衡（请参阅测试频谱）。
[ditaa, apwp_0501] | Low feedback High feedback | | Low barrier to change High barrier to change | | High system coverage Focused coverage | | | | <--------- ----------> | | | | API Tests Service–Layer Tests Domain Tests |
极限编程 (XP) 告诫我们“倾听代码”。当我们编写测试时，我们可能会发现代码难以使用或注意到代码异味。这是一个触发器，促使我们重构并重新考虑我们的设计。
然而，只有当我们与目标代码紧密合作时，我们才能获得那种反馈。HTTP API 的测试没有告诉我们关于对象的细粒度设计的任何信息，因为它位于更高层次的抽象。
另一方面，我们可以重写我们的整个应用程序，只要我们不更改 URL 或请求格式，我们的 HTTP 测试将继续通过。这让我们确信大规模更改（例如更改数据库模式）没有破坏我们的代码。
在频谱的另一端，我们在 [chapter_01_domain_model] 中编写的测试帮助我们充实了我们对所需对象的理解。这些测试引导我们进行设计，使其有意义并以领域语言阅读。当我们的测试以领域语言阅读时，我们感到舒服，因为我们的代码与我们对我们试图解决的问题的直觉相匹配。
因为测试是用领域语言编写的，所以它们充当我们模型的动态文档。新的团队成员可以阅读这些测试，以快速了解系统如何工作以及核心概念如何相互关联。
我们经常通过在这个级别编写测试来“草拟”新行为，以查看代码可能看起来像什么。但是，当我们想要改进代码的设计时，我们将需要替换或删除这些测试，因为它们与特定的 实现 紧密耦合。
高档和低档
大多数时候，当我们添加新功能或修复错误时，我们不需要对领域模型进行大量更改。在这些情况下，我们更喜欢针对服务编写测试，因为耦合更低，覆盖率更高。
例如，在编写 add_stock 函数或 cancel_order 功能时，我们可以通过针对服务层编写测试，更快且耦合更少地工作。
当开始一个新项目或遇到一个特别棘手的问题时，我们将退回到针对领域模型编写测试，以便我们获得更好的反馈和我们意图的可执行文档。
我们使用的隐喻是换挡。当开始旅程时，自行车需要处于低档，以便它可以克服惯性。一旦我们顺利进行，我们可以通过换到高档来更快更高效地行驶；但是如果我们突然遇到陡坡或被迫因危险而减速，我们将再次降到低档，直到我们可以再次加速。
完全解耦服务层测试与领域
在我们的服务层测试中，我们仍然对领域有直接依赖性，因为我们使用领域对象来设置我们的测试数据并调用我们的服务层函数。
为了拥有一个完全与领域解耦的服务层，我们需要重写其 API 以使用原始类型。
我们的服务层当前接受 OrderLine 领域对象
def allocate(line: OrderLine, repo: AbstractRepository, session) -> str:如果它的参数都是原始类型，它会是什么样子？
def allocate(
orderid: str, sku: str, qty: int,
repo: AbstractRepository, session
) -> str:我们也用这些术语重写测试
def test_returns_allocation():
batch = model.Batch("batch1", "COMPLICATED-LAMP", 100, eta=None)
repo = FakeRepository([batch])
result = services.allocate("o1", "COMPLICATED-LAMP", 10, repo, FakeSession())
assert result == "batch1"但是我们的测试仍然依赖于领域，因为我们仍然手动实例化 Batch 对象。因此，如果有一天我们决定大规模重构我们的 Batch 模型的工作方式，我们将不得不更改大量的测试。
缓解：将所有领域依赖项保留在 Fixture 函数中
我们至少可以将其抽象为一个辅助函数或我们测试中的 fixture。这是你可以做到的一种方式，在 FakeRepository 上添加一个工厂函数
class FakeRepository(set):
@staticmethod
def for_batch(ref, sku, qty, eta=None):
return FakeRepository([
model.Batch(ref, sku, qty, eta),
])
...
def test_returns_allocation():
repo = FakeRepository.for_batch("batch1", "COMPLICATED-LAMP", 100, eta=None)
result = services.allocate("o1", "COMPLICATED-LAMP", 10, repo, FakeSession())
assert result == "batch1"至少这将把我们所有测试对领域的依赖项移动到一个地方。
添加缺失的服务
但是，我们可以更进一步。如果我们有一个添加库存的服务，我们可以使用它，并使我们的服务层测试完全根据服务层的官方用例表达，从而消除对领域的所有依赖
def test_add_batch():
repo, session = FakeRepository([]), FakeSession()
services.add_batch("b1", "CRUNCHY-ARMCHAIR", 100, None, repo, session)
assert repo.get("b1") is not None
assert session.committed|
提示
|
一般来说，如果你发现自己需要在你的服务层测试中直接进行领域层的东西，这可能表明你的服务层是不完整的。 |
而实现只有两行
def add_batch(
ref: str, sku: str, qty: int, eta: Optional[date],
repo: AbstractRepository, session,
) -> None:
repo.add(model.Batch(ref, sku, qty, eta))
session.commit()
def allocate(
orderid: str, sku: str, qty: int,
repo: AbstractRepository, session
) -> str:|
注意
|
您是否应该仅仅因为它可以帮助消除测试中的依赖项而编写新服务？可能不是。但在这种情况下，我们几乎肯定有一天无论如何都需要一个 add_batch 服务 。 |
现在，这使我们能够纯粹根据服务本身，仅使用原始类型，并且没有任何模型依赖项来重写所有服务层测试
def test_allocate_returns_allocation():
repo, session = FakeRepository([]), FakeSession()
services.add_batch("batch1", "COMPLICATED-LAMP", 100, None, repo, session)
result = services.allocate("o1", "COMPLICATED-LAMP", 10, repo, session)
assert result == "batch1"
def test_allocate_errors_for_invalid_sku():
repo, session = FakeRepository([]), FakeSession()
services.add_batch("b1", "AREALSKU", 100, None, repo, session)
with pytest.raises(services.InvalidSku, match="Invalid sku NONEXISTENTSKU"):
services.allocate("o1", "NONEXISTENTSKU", 10, repo, FakeSession())这是一个非常好的状态。我们的服务层测试仅依赖于服务层本身，让我们完全自由地重构模型，随心所欲。
将改进延续到 E2E 测试
正如添加 add_batch 帮助我们将服务层测试与模型解耦一样，添加一个 API 端点来添加批次将消除对丑陋的 add_stock fixture 的需求，并且我们的 E2E 测试可以摆脱那些硬编码的 SQL 查询以及对数据库的直接依赖。
感谢我们的服务函数，添加端点很容易，只需要一点 JSON 处理和一个函数调用
@app.route("/add_batch", methods=["POST"])
def add_batch():
session = get_session()
repo = repository.SqlAlchemyRepository(session)
eta = request.json["eta"]
if eta is not None:
eta = datetime.fromisoformat(eta).date()
services.add_batch(
request.json["ref"],
request.json["sku"],
request.json["qty"],
eta,
repo,
session,
)
return "OK", 201|
注意
|
您是否在想，POST 到/add_batch？这不是很 RESTful！您完全正确。我们很随意，但是如果您想使其更具 REST 风格，也许 POST 到/batches，那就放手去做吧！因为 Flask 是一个薄适配器，所以很容易。请参阅下一个侧边栏。 |
我们来自 conftest.py 的硬编码 SQL 查询被一些 API 调用替换，这意味着 API 测试除了 API 之外没有其他依赖项，这也很好
def post_to_add_batch(ref, sku, qty, eta):
url = config.get_api_url()
r = requests.post(
f"{url}/add_batch", json={"ref": ref, "sku": sku, "qty": qty, "eta": eta}
)
assert r.status_code == 201
@pytest.mark.usefixtures("postgres_db")
@pytest.mark.usefixtures("restart_api")
def test_happy_path_returns_201_and_allocated_batch():
sku, othersku = random_sku(), random_sku("other")
earlybatch = random_batchref(1)
laterbatch = random_batchref(2)
otherbatch = random_batchref(3)
post_to_add_batch(laterbatch, sku, 100, "2011-01-02")
post_to_add_batch(earlybatch, sku, 100, "2011-01-01")
post_to_add_batch(otherbatch, othersku, 100, None)
data = {"orderid": random_orderid(), "sku": sku, "qty": 3}
url = config.get_api_url()
r = requests.post(f"{url}/allocate", json=data)
assert r.status_code == 201
assert r.json()["batchref"] == earlybatch