-
Notifications
You must be signed in to change notification settings - Fork 432
Description
I am trying to use PynamoDB in unit tests with testcontainers and the dynamodb-local container.
In this setup, an ephemeral port is allocated for the dynamodb-local container. The ephemeral port is only available after the container starts. I use this ephemeral port to set the host URL by monkey-patching pyndamodb.settings.
This works well. My problem, however, is that the PynamoDB MetaModel class copies in the default host from its settings when the Model class is declared. This means I need to define the class WidgetModel in my test case after the test container fixture is started. While this is OK in this toy example, my actual Model classes are defined in application code and imported before the test container is started - therefore they get the incorrect host setting.
My preference would be for the settings inside the Meta class (host, region, etc.) to be left as None if not set and then resolved from the pynamodb.settings object lazily at the time the connection is created.
This pytest code should be runnable locally with the following libraries
dependencies = [
"boto3==1.38.16",
"pynamodb>=6.1.0",
"testcontainers[k3s]>=4.10.0",
]import logging
import os
from types import SimpleNamespace
from typing import cast
import pytest
from botocore.exceptions import BotoCoreError
from pynamodb import settings
from pynamodb.attributes import UnicodeAttribute
from pynamodb.connection import Connection
from pynamodb.models import Model
from testcontainers.core.container import DockerContainer
from testcontainers.core.waiting_utils import wait_container_is_ready
from types_boto3_dynamodb.client import DynamoDBClient
logger = logging.getLogger(__name__)
class DynamoDbContainer(DockerContainer):
"""
DynamoDB local container for testing.
"""
def __init__(
self, image: str = "amazon/dynamodb-local", port: int = 8000, **kwargs
) -> None:
super().__init__(image, **kwargs)
self.port = port
self.with_exposed_ports(self.port)
def start(self) -> "DynamoDbContainer":
super().start()
self._patch_pynamodb_settings()
self._connect()
return self
@wait_container_is_ready(ConnectionError, BotoCoreError)
def _connect(self) -> None:
try:
conn = self.connect()
boto_client: DynamoDBClient = cast(DynamoDBClient, conn.client)
boto_client.describe_limits()
logger.info("Connection to DynamoDB local container succeeded")
except Exception as e:
logger.error(f"Failed to connect to DynamoDB local container: {e}")
raise
def get_connection_url(self) -> str:
host = self.get_container_host_ip()
port = self.get_exposed_port(self.port)
return f"http://{host}:{port}"
def connect(self) -> Connection:
return Connection(
host=self.get_connection_url(),
)
def _patch_pynamodb_settings(self) -> None:
# https://github.com/pynamodb/PynamoDB/issues/1272
os.environ["AWS_ACCESS_KEY_ID"] = "bogus"
os.environ["AWS_SECRET_ACCESS_KEY"] = "bogus"
override = SimpleNamespace(
host=self.get_connection_url(),
region="bogus",
)
logger.info(f"Patching PynamoDB settings with: {override}")
# Monkey-patch PynamoDB override_settings with our config
# See https://github.com/pynamodb/PynamoDB/issues/555
settings.override_settings = override
@pytest.fixture(scope="session")
def dynamodb():
with DynamoDbContainer() as container:
yield container
def test_crud(dynamodb):
# Must be created after the container starts so that `host` and other attributes
# of the metaclass are set from our overridden PynamoDB settings.
class WidgetModel(Model):
class Meta: # pyright: ignore[reportIncompatibleVariableOverride]
table_name = "widgets"
read_capacity_units = 1
write_capacity_units = 1
id = UnicodeAttribute(hash_key=True)
name = UnicodeAttribute()
WidgetModel.create_table(wait=True)
widget = WidgetModel(id="id", name="Test Widget")
widget.save()
assert WidgetModel.get("id").name == "Test Widget"
widget.name = "Updated Widget"
widget.save()
assert WidgetModel.get("id").name == "Updated Widget"
widget.delete()
assert WidgetModel.count("id") == 0
WidgetModel.delete_table(wait=True)