producer_test.py 文件源码

def test_publish_pii_payload_data_message(
        self, pii_schema, example_payload_data, producer_instance
    ):
        with reconfigure(
            encryption_type='AES_MODE_CBC-1',
            skip_messages_with_pii=False
        ), producer_instance as producer:
            pii_message = CreateMessage(
                schema_id=pii_schema.schema_id,
                payload_data=example_payload_data
            )
            self._publish_and_assert_pii_message(pii_message, producer)
        assert len(multiprocessing.active_children()) == 0