4

I'm trying to unit test the following piece of code using pytest:

import json
from typing import Any, Dict

from confluent_kafka import Consumer


def get_message(config: Dict[str, Any]):
    consumer = Consumer(
        {
            "group.id": config["KAFKA_GROUP_ID"],
            "bootstrap.servers": config["KAFKA_BROKERS"],
            "default.topic.config": {"auto.offset.reset": "smallest"},
        }
    )
    consumer.subscribe([config["KAFKA_TOPIC"]])
    while True:
        collect = consumer.poll()
        if collect is None:
            continue

        try:
            message = json.loads(collect.value().decode("utf-8"))
        except json.JSONDecodeError:
            continue

        return message

But I can't mock subscribe function. I tried:

mock_subscribe = MagicMock(return_value='test')
monkeypatch.setattr('confluent_kafka.cimpl.Consumer.subscribe', mock_subscribe)

In result I got following error:

TypeError: can't set attributes of built-in/extension type

How to mock this function properly?

Gino Mempin
  • 25,369
  • 29
  • 96
  • 135
Pitirus
  • 117
  • 1
  • 13
  • 2
    Ok, so I can't mock this because of the implementation of this function in C/C++ – Pitirus Aug 09 '18 at 09:50
  • 3
    Yes, unfortunately. But you can create a simple Python class that wraps `Consumer` class (with same merthods calling original methods) and use it instead of the original one. This Python class you can mock then. – hoefling Aug 09 '18 at 10:12
  • 7
    This is not a duplicate. – Callam Delaney Sep 10 '20 at 14:11

0 Answers0