More tests

Author: palewire

manage.py

@@ -0,0 +1,21 @@
+#!/usr/bin/env python
+"""Django's command-line utility for administrative tasks."""
+
+import os
+import sys
+
+
+def main() -> None:
+    """Run administrative tasks."""
+    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "tests.settings")
+    try:
+        from django.core.management import execute_from_command_line
+    except ImportError as exc:
+        raise ImportError(
+            "Couldn't import Django. Are you sure it's installed?"
+        ) from exc
+    execute_from_command_line(sys.argv)
+
+
+if __name__ == "__main__":
+    main()

postgres_copy/psycopg_compat.py

@@ -82,10 +82,16 @@ def copy_to(
                 if is_text:
                     # For text destinations, we need to decode to str
                     text_dest = typing.cast(typing.TextIO, destination)
+                    # Handle both bytes and memoryview objects
+                    if isinstance(data, memoryview):
+                        data = data.tobytes()
                     text_dest.write(data.decode("utf-8"))
                 else:
                     # For binary destinations, we keep as bytes
                     binary_dest = typing.cast(typing.BinaryIO, destination)
+                    # Handle both bytes and memoryview objects
+                    if isinstance(data, memoryview):
+                        data = data.tobytes()
                     binary_dest.write(data)
 
     def copy_from(cursor: CursorProtocol, sql: str, source: FileObj) -> None:

pyproject.toml

@@ -105,6 +105,7 @@ docs = [
 addopts = "-p no:warnings --cov=postgres_copy --cov-branch -cov-report=term-missing:skip-covered --cov-context=test"
 testpaths = ["tests"]
 python_files = "test_*.py"
+DJANGO_SETTINGS_MODULE = "tests.settings"
 
 [tool.coverage.run]
 source = ["postgres_copy"]

tests/conftest.py

@@ -1,71 +1,10 @@
 import os
-from pathlib import Path
+import sys
 import pytest
-from django.conf import settings
 from django.db import connections
-import django
 
-ROOT_DIR = Path(__file__).parent.parent
-PG_USER = os.environ.get("PG_USER", "postgres")
-
-
-def pytest_configure():
-    settings.configure(
-        DATABASES={
-            "default": {
-                "HOST": "localhost",
-                "PORT": 5432,
-                "NAME": "postgres",
-                "USER": PG_USER,
-                "ENGINE": "django.db.backends.postgresql_psycopg2",
-            },
-            "other": {
-                "HOST": "localhost",
-                "PORT": 5432,
-                "NAME": "postgres",
-                "USER": PG_USER,
-                "ENGINE": "django.db.backends.postgresql_psycopg2",
-            },
-            "sqlite": {"NAME": "sqlite", "ENGINE": "django.db.backends.sqlite3"},
-            "secondary": {
-                "HOST": "localhost",
-                "PORT": 5432,
-                "NAME": "postgres",
-                "USER": PG_USER,
-                "ENGINE": "django.db.backends.postgresql_psycopg2",
-            },
-        },
-        INSTALLED_APPS=("tests",),
-        DATABASE_ROUTERS=["tests.test_router.CustomRouter"],
-        DEFAULT_AUTO_FIELD="django.db.models.BigAutoField",
-        LOGGING={
-            "version": 1,
-            "disable_existing_loggers": False,
-            "handlers": {
-                "file": {
-                    "level": "DEBUG",
-                    "class": "logging.FileHandler",
-                    "filename": ROOT_DIR / "tests.log",
-                },
-            },
-            "formatters": {
-                "verbose": {
-                    "format": "%(levelname)s|%(asctime)s|%(module)s|%(message)s",
-                    "datefmt": "%d/%b/%Y %H:%M:%S",
-                }
-            },
-            "loggers": {
-                "postgres_copy": {
-                    "handlers": ["file"],
-                    "level": "DEBUG",
-                    "propagate": True,
-                },
-            },
-        },
-    )
-
-    # Initialize Django
-    django.setup()
+# Add the parent directory to sys.path
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 
 @pytest.fixture(scope="session")
@@ -74,7 +13,6 @@ def django_db_setup(django_db_setup, django_db_blocker):
     Fixture to set up the test database with the required tables.
     This extends the built-in django_db_setup fixture.
     """
-
     # Import all test models
     from tests.test_models import (
         MockObject,

tests/settings.py

@@ -0,0 +1,99 @@
+import os
+from pathlib import Path
+
+# Build paths inside the project like this: BASE_DIR / 'subdir'.
+BASE_DIR = Path(__file__).resolve().parent.parent
+
+# Database settings
+PG_USER = os.environ.get("PG_USER", "postgres")
+
+# Use a unique database name for testing
+TEST_DB_NAME = "django_postgres_copy_test"
+
+DATABASES = {
+    "default": {
+        "HOST": "localhost",
+        "PORT": 5432,
+        "NAME": TEST_DB_NAME,
+        "USER": PG_USER,
+        "ENGINE": "django.db.backends.postgresql_psycopg2",
+        "TEST": {
+            "NAME": TEST_DB_NAME,
+        },
+    },
+    "other": {
+        "HOST": "localhost",
+        "PORT": 5432,
+        "NAME": "postgres",
+        "USER": PG_USER,
+        "ENGINE": "django.db.backends.postgresql_psycopg2",
+        "TEST": {
+            "NAME": "django_postgres_copy_test_other",
+        },
+    },
+    "sqlite": {
+        "NAME": "sqlite",
+        "ENGINE": "django.db.backends.sqlite3",
+        "TEST": {
+            "NAME": "django_postgres_copy_test_sqlite",
+        },
+    },
+    "secondary": {
+        "HOST": "localhost",
+        "PORT": 5432,
+        "NAME": "postgres",
+        "USER": PG_USER,
+        "ENGINE": "django.db.backends.postgresql_psycopg2",
+        "TEST": {
+            "NAME": "django_postgres_copy_test_secondary",
+        },
+    },
+}
+
+# Application definition
+INSTALLED_APPS = [
+    "tests",
+]
+
+# Database router
+DATABASE_ROUTERS = ["tests.test_router.CustomRouter"]
+
+# Default primary key field type
+DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField"
+
+# Logging configuration
+LOGGING = {
+    "version": 1,
+    "disable_existing_loggers": False,
+    "handlers": {
+        "file": {
+            "level": "DEBUG",
+            "class": "logging.FileHandler",
+            "filename": BASE_DIR / "tests.log",
+        },
+    },
+    "formatters": {
+        "verbose": {
+            "format": "%(levelname)s|%(asctime)s|%(module)s|%(message)s",
+            "datefmt": "%d/%b/%Y %H:%M:%S",
+        }
+    },
+    "loggers": {
+        "postgres_copy": {
+            "handlers": ["file"],
+            "level": "DEBUG",
+            "propagate": True,
+        },
+    },
+}
+
+# Secret key (required for Django but not used in tests)
+SECRET_KEY = "django-insecure-test-key-not-used-in-production"
+
+# Required Django settings (not used in tests)
+DEBUG = True
+ALLOWED_HOSTS = []
+MIDDLEWARE = []
+ROOT_URLCONF = "tests.urls"
+TEMPLATES = []
+WSGI_APPLICATION = ""

tests/test_psycopg_compat.py

@@ -27,11 +27,18 @@ def test_copy_to_with_psycopg2(self):
         """Test copy_to function with psycopg2."""
         # Mock the psycopg2 import to succeed and psycopg to fail
         with mock.patch.dict("sys.modules", {"psycopg": None}):
+            # Reload the module to ensure it uses the psycopg2 implementation
+            import importlib
+            import postgres_copy.psycopg_compat
+
+            importlib.reload(postgres_copy.psycopg_compat)
+            from postgres_copy.psycopg_compat import copy_to as reloaded_copy_to
+
             with mock.patch("psycopg2.extensions.adapt") as mock_adapt:
                 mock_adapt.return_value.getquoted.return_value = b"'test'"
 
                 # Call the function
-                copy_to(self.cursor, self.sql, (1, 2), self.destination)
+                reloaded_copy_to(self.cursor, self.sql, (1, 2), self.destination)
 
                 # Check that the psycopg2 version was called with the right parameters
                 self.cursor.copy_expert.assert_called_once()
@@ -45,8 +52,15 @@ def test_copy_from_with_psycopg2(self):
         """Test copy_from function with psycopg2."""
         # Mock the psycopg2 import to succeed and psycopg to fail
         with mock.patch.dict("sys.modules", {"psycopg": None}):
+            # Reload the module to ensure it uses the psycopg2 implementation
+            import importlib
+            import postgres_copy.psycopg_compat
+
+            importlib.reload(postgres_copy.psycopg_compat)
+            from postgres_copy.psycopg_compat import copy_from as reloaded_copy_from
+
             # Call the function
-            copy_from(self.cursor, self.sql, self.source)
+            reloaded_copy_from(self.cursor, self.sql, self.source)
 
             # Check that the psycopg2 version was called with the right parameters
             self.cursor.copy_expert.assert_called_once()

tests/urls.py

@@ -0,0 +1,2 @@
+# This file is required by Django but not used in tests
+urlpatterns = []