Mock Google Pub/Sub publish and pull in Python

import json
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Any, Callable


@dataclass
class Message:
    data: str
    attributes: dict[str, str] = field(default_factory=dict)
    message_id: str | None = None
    ack_id: str | None = None


class MockPublisher:
    def __init__(self) -> None:
        self._subscriptions: dict[str, deque[Message]] = {}

    def subscribe(self, topic: str, subscription: str) -> None:
        key = f"{topic}/{subscription}"
        self._subscriptions.setdefault(key, deque())

    def publish(self, topic: str, data: str, **attributes: Any) -> Message:
        msg = Message(data=data, attributes=attributes)
        for key in self._subscriptions:
            if key.startswith(f"{topic}/"):
                self._subscriptions[key].append(msg)
        return msg


class MockSubscriber:
    def __init__(self, publisher: MockPublisher) -> None:
        self._publisher = publisher

    def pull(self, topic: str, subscription: str, max_messages: int = 1) -> list[Message]:
        key = f"{topic}/{subscription}"
        queue = self._publisher._subscriptions.get(key)
        if not queue:
            return []
        messages = []
        for _ in range(min(max_messages, len(queue))):
            msg = queue.popleft()
            msg.ack_id = f"ack-{time.time()}-{len(messages)}"
            messages.append(msg)
        return messages

    def acknowledge(self, msg: Message) -> None:
        print(f"Acknowledged: {msg.ack_id}")


if __name__ == "__main__":
    publisher = MockPublisher()
    subscriber = MockSubscriber(publisher)

    publisher.subscribe("orders", "order-processor")
    publisher.subscribe("orders", "order-audit")

    publisher.publish("orders", '{"order": 1}', source="api")
    publisher.publish("orders", '{"order": 2}', source="cron")

    pulled = subscriber.pull("orders", "order-processor", max_messages=2)
    print(f"Pulled {len(pulled)} messages:")
    for msg in pulled:
        print(f"  ID={msg.message_id} Data={msg.data} Attr={msg.attributes}")
        subscriber.acknowledge(msg)

    remaining = subscriber.pull("orders", "order-audit", max_messages=10)
    print(f"Audit subscription remaining: {len(remaining)} messages")