I'm interested in writing a unit test for my callback function. One thing in particular I want to be able to test is the attributes of the callback message argument. As an example, I want to make sure my logic to log the message.publish_time and message.attributes is correct.

It appears the callback receives a google.cloud.pubsub_v1.subscriber.message.Message. That appears to be a wrapper for google.pubsub_v1.PubsubMessage.

I've tried constructing the above with the following code:

from google.pubsub_v1 import PubsubMessage from google.cloud.pubsub_v1 import subscriber from google.protobuf.timestamp_pb2 import Timestamp from datetime import datetime, timezone publish_time = Timestamp() publish_time.FromDatetime(datetime.now(timezone.utc)) message_id = "test_message_id" data = b"test message" attributes = {"testattr1": "foo"} pubsub_message = PubsubMessage( data=data, attributes=attributes, publish_time=publish_time ) message = subscriber.message.Message(pubsub_message, 0, 0, 0) 

But that always results in this error message:

Traceback (most recent call last): File "/x/test_pubsub_basic.py", line 18, in <module> message = subscriber.message.Message(pubsub_message, 0, 0, 0) File "/x/.venv/lib/python3.9/site-packages/google/cloud/pubsub_v1/subscriber/message.py", line 114, in __init__ message.publish_time.seconds + message.publish_time.nanos / 1e9, AttributeError: 'DatetimeWithNanoseconds' object has no attribute 'seconds' 

I then noticed that in the documentation, it states not to construct the google.cloud.pubsub_v1.subscriber.message.Message class directly https://googleapis.dev/python/pubsub/latest/subscriber/api/message.html

Note

This class should not be constructed directly; it is the responsibility of BasePolicy subclasses to do so.

How can I construct a google.cloud.pubsub_v1.subscriber.message.Message with the correct types for its attributes like publish_time, and attributes.