| 3561 | |
| 3562 | @pytest.mark.parametrize("message_queue", ["kafka", "nats", "mqtt"]) |
| 3563 | def test_message_queue_topic_name_error(message_queue, tmp_path: pathlib.Path): |
| 3564 | input_path = tmp_path / "input.jsonl" |
| 3565 | write = MESSAGE_QUEUE_WRITE_METHOD[message_queue] |
| 3566 | write_kwargs = copy.deepcopy(MESSAGE_QUEUE_WRITE_KWARGS[message_queue]) |
| 3567 | |
| 3568 | class InputSchema(pw.Schema): |
| 3569 | k: int = pw.column_definition(primary_key=True) |
| 3570 | v: str |
| 3571 | |
| 3572 | table = pw.io.jsonlines.read(input_path, schema=InputSchema) |
| 3573 | if "topic" in write_kwargs: |
| 3574 | write_kwargs["topic"] = table.k |
| 3575 | if "topic_name" in write_kwargs: |
| 3576 | write_kwargs["topic_name"] = table.k |
| 3577 | with pytest.raises( |
| 3578 | ValueError, |
| 3579 | match="The topic name column must have a string type, however <class 'int'> is used", |
| 3580 | ): |
| 3581 | write(table, format="json", **write_kwargs) |
| 3582 | |
| 3583 | |
| 3584 | def test_output_column_sorting_by_references(tmp_path: pathlib.Path): |