Beam中assert_that与equal_to未正确断言，测试异常通过

你的测试用例未按预期失败，核心问题是**assert_that被放在了TestPipeline的with代码块外部**。

Beam的assert_that是一个PTransform，必须在管道上下文（即with TestPipeline() as p:的内部代码块）中应用到PCollection上，才能被纳入管道执行流程。你原代码里，with块结束后管道才会运行，但此时assert_that并未被添加到管道中，相当于根本没执行断言检查，所以测试直接通过了。

修正后的代码 #

import unittest

import apache_beam as beam
from apache_beam.testing.test_pipeline import TestPipeline
from apache_beam.testing.util import assert_that, equal_to


class TestBeam(unittest.TestCase):
    def test_fixed_window_grouping(self):
        with TestPipeline() as p:
            input = p | beam.Create(['a', 'b', 'c', 'd', 'e', 'f'])
            # 将assert_that移到with块内部，作为管道的一部分执行
            assert_that(input, equal_to([]))

修正后运行测试，assert_that会在管道运行时检查PCollection内容，发现实际内容['a','b','c','d','e','f']与预期[]不匹配，就会触发测试失败，符合你的预期。

内容的提问来源于stack exchange，提问作者Lydian