diff --git a/packages/django-google-spanner/django_spanner/features.py b/packages/django-google-spanner/django_spanner/features.py index 4d81724cc42d..f903262d3af9 100644 --- a/packages/django-google-spanner/django_spanner/features.py +++ b/packages/django-google-spanner/django_spanner/features.py @@ -122,15 +122,11 @@ def introspected_field_types(self): "db_functions.comparison.test_cast.CastTests.test_cast_to_decimal_field", "db_functions.datetime.test_extract_trunc.DateFunctionTests.test_extract_func", "db_functions.datetime.test_extract_trunc.DateFunctionTests.test_extract_iso_weekday_func", - "db_functions.datetime.test_extract_trunc.DateFunctionTests.test_extract_lookup_name_sql_injection", - "db_functions.datetime.test_extract_trunc.DateFunctionTests.test_trunc_lookup_name_sql_injection", "db_functions.datetime.test_extract_trunc.DateFunctionTests.test_trunc_time_comparison", "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_extract_func", "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_extract_func_with_timezone", "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_extract_iso_weekday_func", - "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_extract_lookup_name_sql_injection", "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_trunc_func_with_timezone", - "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_trunc_lookup_name_sql_injection", "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_trunc_time_comparison", "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_trunc_timezone_applied_before_truncation", "db_functions.json.test_json_array.JSONArrayObjectTests.test_nested_json_array_object", diff --git a/packages/django-google-spanner/django_spanner/operations.py b/packages/django-google-spanner/django_spanner/operations.py index e7e6a8f1fe2b..392c6dae5006 100644 --- a/packages/django-google-spanner/django_spanner/operations.py +++ b/packages/django-google-spanner/django_spanner/operations.py @@ -14,9 +14,10 @@ from django.conf import settings from django.db.backends.base.operations import BaseDatabaseOperations from django.db.models.fields.json import JSONField -from django.db.utils import DatabaseError +from django.db.utils import DatabaseError, OperationalError from django.utils import timezone from django.utils.duration import duration_microseconds +from django.utils.regex_helper import _lazy_re_compile from google.cloud.spanner_dbapi.parse_utils import ( TYPES_MAP, DateStr, @@ -55,6 +56,12 @@ def __eq__(self, other): class DatabaseOperations(BaseDatabaseOperations): """A Spanner-specific version of Django database operations.""" + # EXTRACT and TRUNC date parts cannot be passed as query parameters in SQL. + # Following the practice of other Django database backends (PostgreSQL, + # MySQL, Oracle), validate lookup types with an identifier regex to prevent + # SQL injection while allowing custom user-registered lookups. + extract_trunc_lookup_pattern = _lazy_re_compile(r"^[a-zA-Z_]+$") + # Cloud Spanner only supports INT64, so all integer fields can hold # 64-bit values. integer_field_ranges = { @@ -474,6 +481,48 @@ def convert_jsonfield_value(self, value, expression, connection): return value return value + def _prepare_tzname_delta(self, tzname): + if not tzname or tzname == "UTC": + return "UTC" + # Only parse pure offset representations (e.g. +05:30, -08:00, UTC+5). + # Named timezones (including Etc/GMT+5, Etc/GMT-10) should be passed as-is to Spanner. + if tzname.startswith(("UTC+", "UTC-")): + tzname = tzname[3:] + elif not tzname.startswith(("+", "-")): + return tzname + # Spanner does not support seconds in timezone offsets (e.g. +05:30:00, -05:00:00). + if len(tzname) in (8, 9) and tzname.endswith(":00"): + return tzname[:-3] + return tzname + + def _get_timezone_param(self, tzname): + """Determine the timezone parameter according to Django USE_TZ settings. + + - When USE_TZ is True: use the application's timezone; if none is provided, + return None so Cloud Spanner uses the database default timezone. + - When USE_TZ is False: default to 'UTC' so naive timestamps (stored on the + UTC timeline) are evaluated without server timezone drift, unless an explicit + timezone is supplied. + """ + if settings.USE_TZ: + return self._prepare_tzname_delta(tzname) if tzname else None + return self._prepare_tzname_delta(tzname) if tzname else "UTC" + + def _validate_lookup_type(self, lookup_type): + if not isinstance( + lookup_type, str + ) or not self.extract_trunc_lookup_pattern.fullmatch(lookup_type): + raise ValueError(f"Invalid lookup type: {lookup_type!r}") + + # Django's test_trunc_lookup_name_sql_injection catches only (DataError, OperationalError), + # unlike test_extract_lookup_name_sql_injection which also catches ValueError. + # OperationalError is raised here to satisfy Django's trunc exception contract. + def _validate_trunc_lookup_type(self, lookup_type): + if not isinstance( + lookup_type, str + ) or not self.extract_trunc_lookup_pattern.fullmatch(lookup_type): + raise OperationalError(f"Invalid trunc lookup type: {lookup_type!r}") + def date_extract_sql(self, lookup_type, field_name, params=None): """Extract date from the lookup. @@ -490,7 +539,9 @@ def date_extract_sql(self, lookup_type, field_name, params=None): :returns: A SQL statement for extracting. """ lookup_type = self.extract_names.get(lookup_type, lookup_type) - sql = "EXTRACT(%s FROM %s)" % (lookup_type, field_name) + self._validate_lookup_type(lookup_type) + params = () if params is None else tuple(params) + sql = f"EXTRACT({lookup_type} FROM {field_name})" return sql, params def datetime_extract_sql(self, lookup_type, field_name, params, tzname): @@ -509,17 +560,16 @@ def datetime_extract_sql(self, lookup_type, field_name, params, tzname): :rtype: str :returns: A SQL statement for extracting. """ - tzname = tzname if settings.USE_TZ and tzname else "UTC" lookup_type = self.extract_names.get(lookup_type, lookup_type) - return ( - 'EXTRACT(%s FROM %s AT TIME ZONE "%s")' - % ( - lookup_type, - field_name, - tzname, - ), - params, - ) + self._validate_lookup_type(lookup_type) + params = () if params is None else tuple(params) + tzname_param = self._get_timezone_param(tzname) + if tzname_param: + return ( + f"EXTRACT({lookup_type} FROM {field_name} AT TIME ZONE %s)", + (*params, tzname_param), + ) + return f"EXTRACT({lookup_type} FROM {field_name})", params def time_extract_sql(self, lookup_type, field_name, params=None): """Extract time from the lookup. @@ -536,12 +586,12 @@ def time_extract_sql(self, lookup_type, field_name, params=None): :rtype: str :returns: A SQL statement for extracting. """ + lookup_type = self.extract_names.get(lookup_type, lookup_type) + self._validate_lookup_type(lookup_type) + params = () if params is None else tuple(params) # Time is stored as TIMESTAMP with UTC time zone. - sql = 'EXTRACT(%s FROM %s AT TIME ZONE "UTC")' % ( - lookup_type, - field_name, - ) - return sql, params + sql = f"EXTRACT({lookup_type} FROM {field_name} AT TIME ZONE %s)" + return sql, (*params, "UTC") def date_trunc_sql(self, lookup_type, field_name, params, tzname=None): """Truncate date in the lookup. @@ -562,22 +612,21 @@ def date_trunc_sql(self, lookup_type, field_name, params, tzname=None): :rtype: str :returns: A SQL statement for truncating. """ + self._validate_trunc_lookup_type(lookup_type) + params = () if params is None else tuple(params) # https://cloud.google.com/spanner/docs/functions-and-operators#date_trunc if lookup_type == "week": # Spanner truncates to Sunday but Django expects Monday. First, # subtract a day so that a Sunday will be truncated to the previous # week... - field_name = "DATE_SUB(CAST(" + field_name + " AS DATE), INTERVAL 1 DAY)" - sql = "DATE_TRUNC(CAST(%s AS DATE), %s)" % ( - field_name, - lookup_type, - ) + field_name = f"DATE_SUB(CAST({field_name} AS DATE), INTERVAL 1 DAY)" + sql = f"DATE_TRUNC(CAST({field_name} AS DATE), {lookup_type})" if lookup_type == "week": # ...then add a day to get from Sunday to Monday. - sql = "DATE_ADD(CAST(" + sql + " AS DATE), INTERVAL 1 DAY)" + sql = f"DATE_ADD(CAST({sql} AS DATE), INTERVAL 1 DAY)" return sql, params - def datetime_trunc_sql(self, lookup_type, field_name, params, tzname="UTC"): + def datetime_trunc_sql(self, lookup_type, field_name, params, tzname=None): """Truncate datetime in the lookup. :type lookup_type: str @@ -596,23 +645,25 @@ def datetime_trunc_sql(self, lookup_type, field_name, params, tzname="UTC"): :returns: A SQL statement for truncating. """ # https://cloud.google.com/spanner/docs/functions-and-operators#timestamp_trunc - tzname = tzname if settings.USE_TZ and tzname else "UTC" + self._validate_trunc_lookup_type(lookup_type) + params = () if params is None else tuple(params) if lookup_type == "week": # Spanner truncates to Sunday but Django expects Monday. First, # subtract a day so that a Sunday will be truncated to the previous # week... - field_name = "TIMESTAMP_SUB(" + field_name + ", INTERVAL 1 DAY)" - sql = 'TIMESTAMP_TRUNC(%s, %s, "%s")' % ( - field_name, - lookup_type, - tzname, - ) + field_name = f"TIMESTAMP_SUB({field_name}, INTERVAL 1 DAY)" + tzname_param = self._get_timezone_param(tzname) + if tzname_param: + sql = f"TIMESTAMP_TRUNC({field_name}, {lookup_type}, %s)" + params = (*params, tzname_param) + else: + sql = f"TIMESTAMP_TRUNC({field_name}, {lookup_type})" if lookup_type == "week": # ...then add a day to get from Sunday to Monday. - sql = "TIMESTAMP_ADD(" + sql + ", INTERVAL 1 DAY)" + sql = f"TIMESTAMP_ADD({sql}, INTERVAL 1 DAY)" return sql, params - def time_trunc_sql(self, lookup_type, field_name, params, tzname="UTC"): + def time_trunc_sql(self, lookup_type, field_name, params, tzname=None): """Truncate time in the lookup. :type lookup_type: str @@ -625,21 +676,20 @@ def time_trunc_sql(self, lookup_type, field_name, params, tzname="UTC"): :param params: list of query params. :type tzname: str - :param tzname: The name of the timezone. Defaults to 'UTC' For backward compatability. + :param tzname: The name of the timezone. :rtype: str :returns: A SQL statement for truncating. """ # https://cloud.google.com/spanner/docs/functions-and-operators#timestamp_trunc + self._validate_trunc_lookup_type(lookup_type) + params = () if params is None else tuple(params) + # Time is stored as TIMESTAMP with UTC time zone. tzname = tzname if settings.USE_TZ and tzname else "UTC" + tzname_param = self._prepare_tzname_delta(tzname) return ( - 'TIMESTAMP_TRUNC(%s, %s, "%s")' - % ( - field_name, - lookup_type, - tzname, - ), - params, + f"TIMESTAMP_TRUNC({field_name}, {lookup_type}, %s)", + (*params, tzname_param), ) def datetime_cast_date_sql(self, field_name, params, tzname): @@ -652,15 +702,17 @@ def datetime_cast_date_sql(self, field_name, params, tzname): :param params: list of query params. :type tzname: str - :param tzname: The time zone name. If using of time zone is not - allowed in settings default will be UTC. + :param tzname: The time zone name. :rtype: str :returns: A SQL statement for casting. """ # https://cloud.google.com/spanner/docs/functions-and-operators#date - tzname = tzname if settings.USE_TZ and tzname else "UTC" - return 'DATE(%s, "%s")' % (field_name, tzname), params + params = () if params is None else tuple(params) + tzname_param = self._get_timezone_param(tzname) + if tzname_param: + return f"DATE({field_name}, %s)", (*params, tzname_param) + return f"DATE({field_name})", params def datetime_cast_time_sql(self, field_name, params, tzname): """Cast time in the lookup. @@ -672,19 +724,22 @@ def datetime_cast_time_sql(self, field_name, params, tzname): :param params: list of query params. :type tzname: str - :param tzname: The time zone name. If using of time zone is not - allowed in settings default will be UTC. + :param tzname: The time zone name. :rtype: str :returns: A SQL statement for casting. """ - tzname = tzname if settings.USE_TZ and tzname else "UTC" - # Cloud Spanner doesn't have a function for converting - # TIMESTAMP to another time zone. + params = () if params is None else tuple(params) + tzname_param = self._get_timezone_param(tzname) + if tzname_param: + return ( + f"TIMESTAMP(FORMAT_TIMESTAMP('1970-01-01 %H:%M:%E9S', {field_name}, %s), 'UTC')", + (*params, tzname_param), + ) return ( - "TIMESTAMP(FORMAT_TIMESTAMP(" - "'%%Y-%%m-%%d %%R:%%E9S %%Z', %s, '%s'))" % (field_name, tzname) - ), params + f"TIMESTAMP(FORMAT_TIMESTAMP('1970-01-01 %H:%M:%E9S', {field_name}), 'UTC')", + params, + ) def date_interval_sql(self, timedelta): """Get a date interval in microseconds. diff --git a/packages/django-google-spanner/tests/mockserver_tests/models.py b/packages/django-google-spanner/tests/mockserver_tests/models.py index ac2de7d9ae4a..16c3ddeb91a5 100644 --- a/packages/django-google-spanner/tests/mockserver_tests/models.py +++ b/packages/django-google-spanner/tests/mockserver_tests/models.py @@ -21,5 +21,11 @@ class Singer(models.Model): last_name = models.CharField(max_length=200) +class Event(models.Model): + name = models.CharField(max_length=200) + start_time = models.DateTimeField() + start_time_of_day = models.TimeField(null=True) + + class Detail(models.Model): value = models.JSONField(null=True, encoder=DjangoJSONEncoder) diff --git a/packages/django-google-spanner/tests/mockserver_tests/test_datetime_expressions.py b/packages/django-google-spanner/tests/mockserver_tests/test_datetime_expressions.py new file mode 100644 index 000000000000..3f14081d7ae3 --- /dev/null +++ b/packages/django-google-spanner/tests/mockserver_tests/test_datetime_expressions.py @@ -0,0 +1,430 @@ +# Copyright 2026 Google LLC All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import zoneinfo +from datetime import datetime, timedelta, tzinfo +from datetime import timezone as dt_timezone + +from django.db.models import Value +from django.db.models.functions import ( + Coalesce, + Extract, + ExtractHour, + Trunc, + TruncDate, + TruncDay, + TruncHour, + TruncTime, + TruncWeek, +) +from django.db.utils import OperationalError +from django.test import override_settings +from django.utils import timezone +from google.cloud.spanner_v1 import ExecuteSqlRequest, TypeCode + +from tests.mockserver_tests.mock_server_test_base import ( + MockServerTestBase, + add_single_result, +) +from tests.mockserver_tests.models import Event + + +class TestDatetimeExpressions(MockServerTestBase): + def get_last_execute_sql_request(self) -> ExecuteSqlRequest: + execute_requests = [ + request + for request in self.spanner_service.requests + if isinstance(request, ExecuteSqlRequest) + ] + self.assertTrue( + len(execute_requests) > 0, + "Expected at least one ExecuteSqlRequest in recorded requests", + ) + return execute_requests[-1] + + def test_extract_hour_parameterized(self): + sql = ( + "SELECT EXTRACT(hour FROM tests_event.start_time AT TIME ZONE @a0) " + "AS extracted_hour FROM tests_event" + ) + add_single_result(sql, "extracted_hour", TypeCode.INT64, [("12",)]) + results = list( + Event.objects.annotate( + extracted_hour=ExtractHour("start_time") + ).values_list("extracted_hour", flat=True) + ) + self.assertEqual(results, [12]) + execute_request = self.get_last_execute_sql_request() + self.assertEqual(execute_request.sql, sql) + self.assertEqual(execute_request.params["a0"], "UTC") + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING) + + def test_extract_named_timezone_parameterized(self): + new_york_timezone = zoneinfo.ZoneInfo("America/New_York") + sql = ( + "SELECT EXTRACT(hour FROM tests_event.start_time AT TIME ZONE @a0) " + "AS extracted_hour FROM tests_event" + ) + add_single_result(sql, "extracted_hour", TypeCode.INT64, [("8",)]) + results = list( + Event.objects.annotate( + extracted_hour=ExtractHour("start_time", tzinfo=new_york_timezone) + ).values_list("extracted_hour", flat=True) + ) + self.assertEqual(results, [8]) + execute_request = self.get_last_execute_sql_request() + self.assertEqual(execute_request.sql, sql) + self.assertEqual(execute_request.params["a0"], "America/New_York") + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING) + + def test_extract_fixed_offset_timezone(self): + fixed_offset_timezone = timezone.get_fixed_timezone(-283) + sql = ( + "SELECT EXTRACT(hour FROM tests_event.start_time AT TIME ZONE @a0) " + "AS extracted_hour FROM tests_event" + ) + add_single_result(sql, "extracted_hour", TypeCode.INT64, [("7",)]) + results = list( + Event.objects.annotate( + extracted_hour=ExtractHour("start_time", tzinfo=fixed_offset_timezone) + ).values_list("extracted_hour", flat=True) + ) + self.assertEqual(results, [7]) + execute_request = self.get_last_execute_sql_request() + self.assertEqual(execute_request.sql, sql) + self.assertEqual(execute_request.params["a0"], "-0443") + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING) + + def test_trunc_day_parameterized(self): + tokyo_timezone = zoneinfo.ZoneInfo("Asia/Tokyo") + sql = ( + "SELECT TIMESTAMP_TRUNC(tests_event.start_time, day, @a0) " + "AS truncated_day FROM tests_event" + ) + add_single_result( + sql, "truncated_day", TypeCode.TIMESTAMP, [("2026-11-10T00:00:00Z",)] + ) + results = list( + Event.objects.annotate( + truncated_day=TruncDay("start_time", tzinfo=tokyo_timezone) + ).values_list("truncated_day", flat=True) + ) + self.assertEqual(len(results), 1) + execute_request = self.get_last_execute_sql_request() + self.assertEqual(execute_request.sql, sql) + self.assertEqual(execute_request.params["a0"], "Asia/Tokyo") + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING) + + def test_trunc_week_parameterized(self): + new_york_timezone = zoneinfo.ZoneInfo("America/New_York") + sql = ( + "SELECT TIMESTAMP_ADD(TIMESTAMP_TRUNC(TIMESTAMP_SUB(tests_event.start_time, INTERVAL 1 DAY), " + "week, @a0), INTERVAL 1 DAY) AS truncated_week FROM tests_event" + ) + add_single_result( + sql, "truncated_week", TypeCode.TIMESTAMP, [("2026-11-09T00:00:00Z",)] + ) + results = list( + Event.objects.annotate( + truncated_week=TruncWeek("start_time", tzinfo=new_york_timezone) + ).values_list("truncated_week", flat=True) + ) + self.assertEqual(len(results), 1) + execute_request = self.get_last_execute_sql_request() + self.assertEqual(execute_request.sql, sql) + self.assertEqual(execute_request.params["a0"], "America/New_York") + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING) + + def test_trunc_date_and_time_cast_parameterized(self): + new_york_timezone = zoneinfo.ZoneInfo("America/New_York") + sql_date = "SELECT DATE(tests_event.start_time, @a0) AS truncated_date FROM tests_event" + add_single_result(sql_date, "truncated_date", TypeCode.DATE, [("2026-11-10",)]) + results_date = list( + Event.objects.annotate( + truncated_date=TruncDate("start_time", tzinfo=new_york_timezone) + ).values_list("truncated_date", flat=True) + ) + self.assertEqual(len(results_date), 1) + execute_request_date = self.get_last_execute_sql_request() + self.assertEqual(execute_request_date.sql, sql_date) + self.assertEqual(execute_request_date.params["a0"], "America/New_York") + + sql_time = ( + "SELECT TIMESTAMP(FORMAT_TIMESTAMP('1970-01-01 %H:%M:%E9S', " + "tests_event.start_time, @a0), 'UTC') AS truncated_time FROM tests_event" + ) + add_single_result( + sql_time, "truncated_time", TypeCode.TIMESTAMP, [("2026-11-10T12:00:00Z",)] + ) + results_time = list( + Event.objects.annotate( + truncated_time=TruncTime("start_time", tzinfo=new_york_timezone) + ).values_list("truncated_time", flat=True) + ) + self.assertEqual(len(results_time), 1) + execute_request_time = self.get_last_execute_sql_request() + self.assertEqual(execute_request_time.sql, sql_time) + self.assertEqual(execute_request_time.params["a0"], "America/New_York") + + def test_extract_compound_expression_preserves_parameter_order(self): + new_york_timezone = zoneinfo.ZoneInfo("America/New_York") + fallback_datetime = datetime(2026, 1, 1, 15, 0, tzinfo=dt_timezone.utc) + sql = ( + "SELECT EXTRACT(hour FROM COALESCE(tests_event.start_time, @a0) " + "AT TIME ZONE @a1) AS extracted_hour FROM tests_event" + ) + add_single_result(sql, "extracted_hour", TypeCode.INT64, [("10",)]) + results = list( + Event.objects.annotate( + extracted_hour=ExtractHour( + Coalesce("start_time", Value(fallback_datetime)), + tzinfo=new_york_timezone, + ) + ).values_list("extracted_hour", flat=True) + ) + self.assertEqual(results, [10]) + execute_request = self.get_last_execute_sql_request() + self.assertEqual(execute_request.sql, sql) + self.assertEqual(execute_request.params["a0"], "2026-01-01T15:00:00.000000Z") + self.assertEqual(execute_request.params["a1"], "America/New_York") + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.TIMESTAMP) + self.assertEqual(execute_request.param_types["a1"].code, TypeCode.STRING) + + @override_settings(USE_TZ=False) + def test_extract_use_tz_false_defaults_to_utc(self): + sql = ( + "SELECT EXTRACT(hour FROM tests_event.start_time AT TIME ZONE @a0) " + "AS extracted_hour_use_tz_false FROM tests_event" + ) + add_single_result(sql, "extracted_hour_use_tz_false", TypeCode.INT64, [("12",)]) + results = list( + Event.objects.annotate( + extracted_hour_use_tz_false=ExtractHour("start_time") + ).values_list("extracted_hour_use_tz_false", flat=True) + ) + self.assertEqual(results, [12]) + execute_request = self.get_last_execute_sql_request() + self.assertEqual(execute_request.sql, sql) + self.assertEqual(execute_request.params["a0"], "UTC") + + def test_tzname_sql_injection_travels_safely_as_parameter(self): + class MaliciousTimezone(tzinfo): + def utcoffset(self, dt): + return timedelta(0) + + def tzname(self, dt): + return "UTC'; DROP TABLE tests_event; --" + + def dst(self, dt): + return timedelta(0) + + malicious_timezone = MaliciousTimezone() + sql = ( + "SELECT EXTRACT(hour FROM tests_event.start_time AT TIME ZONE @a0) " + "AS extracted_hour FROM tests_event" + ) + add_single_result(sql, "extracted_hour", TypeCode.INT64, [("1",)]) + results = list( + Event.objects.annotate( + extracted_hour=ExtractHour("start_time", tzinfo=malicious_timezone) + ).values_list("extracted_hour", flat=True) + ) + self.assertEqual(results, [1]) + execute_request = self.get_last_execute_sql_request() + self.assertEqual(execute_request.sql, sql) + self.assertNotIn("DROP TABLE", execute_request.sql) + self.assertEqual( + execute_request.params["a0"], "UTC'; DROP TABLE tests_event; --" + ) + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING) + + def test_extract_rejects_malicious_lookup_type_before_rpc(self): + with self.assertRaises(ValueError): + list( + Event.objects.annotate( + injected=Extract("start_time", "hour'; DROP TABLE tests_event; --") + ).values_list("injected", flat=True) + ) + execute_requests = [ + request + for request in self.spanner_service.requests + if isinstance(request, ExecuteSqlRequest) + ] + self.assertEqual( + len(execute_requests), + 0, + "Expected zero ExecuteSqlRequest when lookup_type is invalid", + ) + + def test_trunc_rejects_malicious_lookup_type_before_rpc(self): + with self.assertRaises(OperationalError): + list( + Event.objects.annotate( + injected=Trunc("start_time", "day'; DROP TABLE tests_event; --") + ).values_list("injected", flat=True) + ) + execute_requests = [ + request + for request in self.spanner_service.requests + if isinstance(request, ExecuteSqlRequest) + ] + self.assertEqual( + len(execute_requests), + 0, + "Expected zero ExecuteSqlRequest when lookup_type is invalid", + ) + + def test_trunc_timefield_defaults_to_utc(self): + sql = ( + "SELECT TIMESTAMP_TRUNC(tests_event.start_time_of_day, hour, @a0) " + "AS truncated_hour FROM tests_event" + ) + add_single_result( + sql, "truncated_hour", TypeCode.TIMESTAMP, [("1970-01-01T10:00:00Z",)] + ) + results = list( + Event.objects.annotate( + truncated_hour=TruncHour("start_time_of_day") + ).values_list("truncated_hour", flat=True) + ) + self.assertEqual(len(results), 1) + execute_request = self.get_last_execute_sql_request() + self.assertEqual(execute_request.sql, sql) + self.assertEqual(execute_request.params["a0"], "UTC") + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING) + + def test_extract_timefield_defaults_to_utc(self): + sql = ( + "SELECT EXTRACT(hour FROM tests_event.start_time_of_day AT TIME ZONE @a0) " + "AS extracted_hour FROM tests_event" + ) + add_single_result(sql, "extracted_hour", TypeCode.INT64, [("10",)]) + results = list( + Event.objects.annotate( + extracted_hour=ExtractHour("start_time_of_day") + ).values_list("extracted_hour", flat=True) + ) + self.assertEqual(results, [10]) + execute_request = self.get_last_execute_sql_request() + self.assertEqual(execute_request.sql, sql) + self.assertEqual(execute_request.params["a0"], "UTC") + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING) + + def test_trunc_with_fixed_offset_parameterized(self): + kolkata_tz = dt_timezone(timedelta(hours=5, minutes=30)) + sql = ( + "SELECT TIMESTAMP_TRUNC(tests_event.start_time, hour, @a0) " + "AS truncated_hour FROM tests_event" + ) + add_single_result( + sql, "truncated_hour", TypeCode.TIMESTAMP, [("2026-11-10T10:00:00Z",)] + ) + results = list( + Event.objects.annotate( + truncated_hour=TruncHour("start_time", tzinfo=kolkata_tz) + ).values_list("truncated_hour", flat=True) + ) + self.assertEqual(len(results), 1) + execute_request = self.get_last_execute_sql_request() + self.assertEqual(execute_request.sql, sql) + self.assertEqual(execute_request.params["a0"], "+05:30") + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING) + + @override_settings(USE_TZ=False) + def test_cast_date_and_time_use_tz_false(self): + sql_date = "SELECT DATE(tests_event.start_time, @a0) AS truncated_date FROM tests_event" + add_single_result(sql_date, "truncated_date", TypeCode.DATE, [("2026-11-10",)]) + results_date = list( + Event.objects.annotate(truncated_date=TruncDate("start_time")).values_list( + "truncated_date", flat=True + ) + ) + self.assertEqual(len(results_date), 1) + execute_request_date = self.get_last_execute_sql_request() + self.assertEqual(execute_request_date.sql, sql_date) + self.assertEqual(execute_request_date.params["a0"], "UTC") + + sql_time = ( + "SELECT TIMESTAMP(FORMAT_TIMESTAMP('1970-01-01 %H:%M:%E9S', " + "tests_event.start_time, @a0), 'UTC') AS truncated_time FROM tests_event" + ) + add_single_result( + sql_time, "truncated_time", TypeCode.TIMESTAMP, [("2026-11-10T12:00:00Z",)] + ) + results_time = list( + Event.objects.annotate(truncated_time=TruncTime("start_time")).values_list( + "truncated_time", flat=True + ) + ) + self.assertEqual(len(results_time), 1) + execute_request_time = self.get_last_execute_sql_request() + self.assertEqual(execute_request_time.sql, sql_time) + self.assertEqual(execute_request_time.params["a0"], "UTC") + + @override_settings(USE_TZ=False) + def test_trunc_hour_use_tz_false(self): + sql = ( + "SELECT TIMESTAMP_TRUNC(tests_event.start_time, hour, @a0) " + "AS truncated_hour FROM tests_event" + ) + add_single_result( + sql, "truncated_hour", TypeCode.TIMESTAMP, [("2026-11-10T12:00:00Z",)] + ) + results = list( + Event.objects.annotate(truncated_hour=TruncHour("start_time")).values_list( + "truncated_hour", flat=True + ) + ) + self.assertEqual(len(results), 1) + execute_request = self.get_last_execute_sql_request() + self.assertEqual(execute_request.sql, sql) + self.assertEqual(execute_request.params["a0"], "UTC") + self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING) + + @override_settings(USE_TZ=False) + def test_extract_and_trunc_use_tz_false_with_explicit_timezone(self): + # Under USE_TZ=False, Django ORM's get_tzname() returns None even if tzinfo + # is specified on the expression. Our backend then defaults the timezone parameter to 'UTC'. + fixed_offset_timezone = timezone.get_fixed_timezone(-300) + sql_extract = ( + "SELECT EXTRACT(hour FROM tests_event.start_time AT TIME ZONE @a0) " + "AS extracted_hour FROM tests_event" + ) + add_single_result(sql_extract, "extracted_hour", TypeCode.INT64, [("12",)]) + results_extract = list( + Event.objects.annotate( + extracted_hour=ExtractHour("start_time", tzinfo=fixed_offset_timezone) + ).values_list("extracted_hour", flat=True) + ) + self.assertEqual(results_extract, [12]) + execute_request_extract = self.get_last_execute_sql_request() + self.assertEqual(execute_request_extract.sql, sql_extract) + self.assertEqual(execute_request_extract.params["a0"], "UTC") + + sql_trunc = ( + "SELECT TIMESTAMP_TRUNC(tests_event.start_time, hour, @a0) " + "AS truncated_hour FROM tests_event" + ) + add_single_result( + sql_trunc, "truncated_hour", TypeCode.TIMESTAMP, [("2026-11-10T12:00:00Z",)] + ) + results_trunc = list( + Event.objects.annotate( + truncated_hour=TruncHour("start_time", tzinfo=fixed_offset_timezone) + ).values_list("truncated_hour", flat=True) + ) + self.assertEqual(len(results_trunc), 1) + execute_request_trunc = self.get_last_execute_sql_request() + self.assertEqual(execute_request_trunc.sql, sql_trunc) + self.assertEqual(execute_request_trunc.params["a0"], "UTC") diff --git a/packages/django-google-spanner/tests/system/django_spanner/models.py b/packages/django-google-spanner/tests/system/django_spanner/models.py index ac7b523fb1b9..0d228eb2441a 100644 --- a/packages/django-google-spanner/tests/system/django_spanner/models.py +++ b/packages/django-google-spanner/tests/system/django_spanner/models.py @@ -27,6 +27,8 @@ def __str__(self): class Event(models.Model): start_date = models.DateTimeField() end_date = models.DateTimeField() + event_date = models.DateField(null=True) + event_time = models.TimeField(null=True) class Meta: constraints = [ diff --git a/packages/django-google-spanner/tests/system/django_spanner/test_datetime_expressions.py b/packages/django-google-spanner/tests/system/django_spanner/test_datetime_expressions.py new file mode 100644 index 000000000000..dd0ca1adaeaf --- /dev/null +++ b/packages/django-google-spanner/tests/system/django_spanner/test_datetime_expressions.py @@ -0,0 +1,267 @@ +# Copyright 2026 Google LLC +# +# Use of this source code is governed by a BSD-style +# license that can be found in the LICENSE file or at +# https://developers.google.com/open-source/licenses/bsd + +import zoneinfo +from datetime import date, datetime, time, timezone + +from django.db import connection +from django.db.models.functions import ( + ExtractDay, + ExtractHour, + ExtractMinute, + ExtractMonth, + ExtractSecond, + ExtractYear, + TruncDate, + TruncDay, + TruncHour, + TruncMonth, + TruncTime, + TruncYear, +) +from django.test import TransactionTestCase, override_settings + +from .models import Event +from .utils import ( + setup_database, + setup_instance, + teardown_database, + teardown_instance, +) + + +class TestDatetimeExpressions(TransactionTestCase): + @classmethod + def setUpClass(cls): + setup_instance() + setup_database() + with connection.schema_editor() as editor: + editor.create_model(Event) + + @classmethod + def tearDownClass(cls): + with connection.schema_editor() as editor: + editor.delete_model(Event) + teardown_database() + teardown_instance() + + def setUp(self): + super().setUp() + Event.objects.all().delete() + + def test_extract_and_trunc_with_timezone(self): + new_york_timezone = zoneinfo.ZoneInfo("America/New_York") + # 2026-01-01 03:00:00 UTC is 2025-12-31 22:00:00 in America/New_York (UTC-5) + event = Event.objects.create( + start_date=datetime(2026, 1, 1, 3, 0, tzinfo=timezone.utc), + end_date=datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc), + ) + + # 1. Extract hour in America/New_York (should be 22) + ny_hour = ( + Event.objects.filter(pk=event.pk) + .annotate(hour=ExtractHour("start_date", tzinfo=new_york_timezone)) + .values_list("hour", flat=True) + .get() + ) + self.assertEqual(ny_hour, 22) + + # 2. Extract day in America/New_York (should be 31) + ny_day = ( + Event.objects.filter(pk=event.pk) + .annotate(day=ExtractDay("start_date", tzinfo=new_york_timezone)) + .values_list("day", flat=True) + .get() + ) + self.assertEqual(ny_day, 31) + + # 3. Extract in UTC (should be 3) + utc_hour = ( + Event.objects.filter(pk=event.pk) + .annotate(hour=ExtractHour("start_date", tzinfo=timezone.utc)) + .values_list("hour", flat=True) + .get() + ) + self.assertEqual(utc_hour, 3) + + # 4. TruncDate in America/New_York (should be 2025-12-31) + ny_date = ( + Event.objects.filter(pk=event.pk) + .annotate(date=TruncDate("start_date", tzinfo=new_york_timezone)) + .values_list("date", flat=True) + .get() + ) + self.assertEqual(ny_date, date(2025, 12, 31)) + + # 5. TruncDay in America/New_York + ny_day_trunc = ( + Event.objects.filter(pk=event.pk) + .annotate(day_trunc=TruncDay("start_date", tzinfo=new_york_timezone)) + .values_list("day_trunc", flat=True) + .get() + ) + self.assertEqual( + ny_day_trunc, datetime(2025, 12, 31, 5, 0, tzinfo=new_york_timezone) + ) + + # 6. TruncTime in America/New_York (should be 22:00:00) + ny_time = ( + Event.objects.filter(pk=event.pk) + .annotate(time=TruncTime("start_date", tzinfo=new_york_timezone)) + .values_list("time", flat=True) + .get() + ) + self.assertEqual(str(ny_time), "22:00:00") + + def test_extract_and_trunc_without_timezone(self): + event = Event.objects.create( + start_date=datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc), + end_date=datetime(2026, 1, 1, 15, 0, tzinfo=timezone.utc), + ) + + # Under USE_TZ=True, omitting tzinfo causes Django to use the current + # active timezone (settings.TIME_ZONE, which defaults to UTC). + extracted_hour = ( + Event.objects.filter(pk=event.pk) + .annotate(hour=ExtractHour("start_date")) + .values_list("hour", flat=True) + .get() + ) + self.assertEqual(extracted_hour, 12) + + truncated_hour = ( + Event.objects.filter(pk=event.pk) + .annotate(truncated=TruncHour("start_date")) + .values_list("truncated", flat=True) + .get() + ) + self.assertEqual( + truncated_hour, datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc) + ) + + truncated_date = ( + Event.objects.filter(pk=event.pk) + .annotate(truncated=TruncDate("start_date")) + .values_list("truncated", flat=True) + .get() + ) + self.assertEqual(truncated_date, date(2026, 1, 1)) + + truncated_time = ( + Event.objects.filter(pk=event.pk) + .annotate(truncated=TruncTime("start_date")) + .values_list("truncated", flat=True) + .get() + ) + self.assertEqual(truncated_time, time(12, 0, 0)) + + @override_settings(USE_TZ=False) + def test_extract_and_trunc_use_tz_false(self): + event = Event.objects.create( + start_date=datetime(2026, 1, 1, 12, 0), + end_date=datetime(2026, 1, 1, 15, 0), + ) + + # When USE_TZ is False, operations default to UTC timezone arguments so that + # naive timestamps (stored on the UTC timeline) do not drift based on the + # database's default timezone (e.g. America/Los_Angeles). + extracted_hour = ( + Event.objects.filter(pk=event.pk) + .annotate(hour=ExtractHour("start_date")) + .values_list("hour", flat=True) + .get() + ) + self.assertEqual(extracted_hour, 12) + + truncated_hour = ( + Event.objects.filter(pk=event.pk) + .annotate(truncated=TruncHour("start_date")) + .values_list("truncated", flat=True) + .get() + ) + self.assertEqual(truncated_hour, datetime(2026, 1, 1, 12, 0)) + + truncated_date = ( + Event.objects.filter(pk=event.pk) + .annotate(truncated=TruncDate("start_date")) + .values_list("truncated", flat=True) + .get() + ) + self.assertEqual(truncated_date, date(2026, 1, 1)) + + truncated_time = ( + Event.objects.filter(pk=event.pk) + .annotate(truncated=TruncTime("start_date")) + .values_list("truncated", flat=True) + .get() + ) + self.assertEqual(truncated_time, time(12, 0, 0)) + + def test_date_field_roundtrip_and_expressions(self): + event = Event.objects.create( + start_date=datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc), + end_date=datetime(2026, 1, 1, 15, 0, tzinfo=timezone.utc), + event_date=date(2026, 5, 15), + ) + refetched = Event.objects.get(pk=event.pk) + self.assertEqual(refetched.event_date, date(2026, 5, 15)) + + extracted = ( + Event.objects.filter(pk=event.pk) + .annotate( + year=ExtractYear("event_date"), + month=ExtractMonth("event_date"), + day=ExtractDay("event_date"), + ) + .values("year", "month", "day") + .get() + ) + self.assertEqual(extracted["year"], 2026) + self.assertEqual(extracted["month"], 5) + self.assertEqual(extracted["day"], 15) + + truncated = ( + Event.objects.filter(pk=event.pk) + .annotate( + month=TruncMonth("event_date"), + year=TruncYear("event_date"), + ) + .values("month", "year") + .get() + ) + self.assertEqual(truncated["month"], date(2026, 5, 1)) + self.assertEqual(truncated["year"], date(2026, 1, 1)) + + def test_time_field_roundtrip_and_expressions(self): + event = Event.objects.create( + start_date=datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc), + end_date=datetime(2026, 1, 1, 15, 0, tzinfo=timezone.utc), + event_time=time(14, 30, 45), + ) + refetched = Event.objects.get(pk=event.pk) + self.assertEqual(refetched.event_time, time(14, 30, 45)) + + extracted = ( + Event.objects.filter(pk=event.pk) + .annotate( + hour=ExtractHour("event_time"), + minute=ExtractMinute("event_time"), + second=ExtractSecond("event_time"), + ) + .values("hour", "minute", "second") + .get() + ) + self.assertEqual(extracted["hour"], 14) + self.assertEqual(extracted["minute"], 30) + self.assertEqual(extracted["second"], 45) + + truncated_hour = ( + Event.objects.filter(pk=event.pk) + .annotate(hour=TruncHour("event_time")) + .values_list("hour", flat=True) + .get() + ) + self.assertEqual(truncated_hour, time(14, 0, 0)) diff --git a/packages/django-google-spanner/tests/unit/django_spanner/test_operations.py b/packages/django-google-spanner/tests/unit/django_spanner/test_operations.py index eac1054f9c76..185c0a8b4c8b 100644 --- a/packages/django-google-spanner/tests/unit/django_spanner/test_operations.py +++ b/packages/django-google-spanner/tests/unit/django_spanner/test_operations.py @@ -10,11 +10,11 @@ from decimal import Decimal from unittest import mock -from django.conf import settings from django.core.management.color import no_style from django.core.serializers.json import DjangoJSONEncoder from django.db.models import JSONField -from django.db.utils import DatabaseError +from django.db.utils import DatabaseError, OperationalError +from django.test import override_settings from google.cloud.spanner_dbapi.types import DateStr from google.cloud.spanner_v1 import JsonObject @@ -253,81 +253,130 @@ def test_convert_jsonfield_value_other(self): def test_date_extract_sql(self): self.assertEqual( self.db_operations.date_extract_sql("week", "dummy_field"), - ("EXTRACT(isoweek FROM dummy_field)", None), + ("EXTRACT(isoweek FROM dummy_field)", ()), ) def test_date_extract_sql_lookup_type_dayofweek(self): self.assertEqual( self.db_operations.date_extract_sql("dayofweek", "dummy_field"), - ("EXTRACT(dayofweek FROM dummy_field)", None), + ("EXTRACT(dayofweek FROM dummy_field)", ()), ) + @override_settings(USE_TZ=True) def test_datetime_extract_sql(self): - settings.USE_TZ = True self.assertEqual( self.db_operations.datetime_extract_sql( "dayofweek", "dummy_field", None, "IST" ), ( - 'EXTRACT(dayofweek FROM dummy_field AT TIME ZONE "IST")', - None, + "EXTRACT(dayofweek FROM dummy_field AT TIME ZONE %s)", + ("IST",), + ), + ) + self.assertEqual( + self.db_operations.datetime_extract_sql( + "dayofweek", "dummy_field", None, None + ), + ( + "EXTRACT(dayofweek FROM dummy_field)", + (), ), ) + @override_settings(USE_TZ=False) def test_datetime_extract_sql_use_tz_false(self): - settings.USE_TZ = False self.assertEqual( self.db_operations.datetime_extract_sql( "dayofweek", "dummy_field", None, "IST" ), ( - 'EXTRACT(dayofweek FROM dummy_field AT TIME ZONE "UTC")', - None, + "EXTRACT(dayofweek FROM dummy_field AT TIME ZONE %s)", + ("IST",), + ), + ) + self.assertEqual( + self.db_operations.datetime_extract_sql( + "dayofweek", "dummy_field", None, None + ), + ( + "EXTRACT(dayofweek FROM dummy_field AT TIME ZONE %s)", + ("UTC",), ), ) - settings.USE_TZ = True # reset changes. def test_time_extract_sql(self): self.assertEqual( self.db_operations.time_extract_sql("dayofweek", "dummy_field"), ( - 'EXTRACT(dayofweek FROM dummy_field AT TIME ZONE "UTC")', - None, + "EXTRACT(dayofweek FROM dummy_field AT TIME ZONE %s)", + ("UTC",), ), ) - def test_time_trunc_sql(self): + @override_settings(USE_TZ=True) + def test_time_trunc_sql_use_tz_true(self): self.assertEqual( self.db_operations.time_trunc_sql("dayofweek", "dummy_field", None), - ('TIMESTAMP_TRUNC(dummy_field, dayofweek, "UTC")', None), + ("TIMESTAMP_TRUNC(dummy_field, dayofweek, %s)", ("UTC",)), + ) + self.assertEqual( + self.db_operations.time_trunc_sql( + "dayofweek", "dummy_field", None, tzname="UTC" + ), + ("TIMESTAMP_TRUNC(dummy_field, dayofweek, %s)", ("UTC",)), + ) + self.assertEqual( + self.db_operations.time_trunc_sql( + "dayofweek", "dummy_field", None, tzname="America/New_York" + ), + ("TIMESTAMP_TRUNC(dummy_field, dayofweek, %s)", ("America/New_York",)), + ) + + @override_settings(USE_TZ=False) + def test_time_trunc_sql_use_tz_false(self): + self.assertEqual( + self.db_operations.time_trunc_sql("dayofweek", "dummy_field", None), + ("TIMESTAMP_TRUNC(dummy_field, dayofweek, %s)", ("UTC",)), + ) + self.assertEqual( + self.db_operations.time_trunc_sql( + "dayofweek", "dummy_field", None, tzname="America/New_York" + ), + ("TIMESTAMP_TRUNC(dummy_field, dayofweek, %s)", ("UTC",)), ) def test_datetime_cast_date_sql(self): self.assertEqual( self.db_operations.datetime_cast_date_sql("dummy_field", None, "IST"), - ('DATE(dummy_field, "IST")', None), + ("DATE(dummy_field, %s)", ("IST",)), ) + @override_settings(USE_TZ=True) def test_datetime_cast_time_sql(self): - settings.USE_TZ = True self.assertEqual( self.db_operations.datetime_cast_time_sql("dummy_field", None, "IST"), ( - "TIMESTAMP(FORMAT_TIMESTAMP('%Y-%m-%d %R:%E9S %Z', dummy_field, 'IST'))", - None, + "TIMESTAMP(FORMAT_TIMESTAMP('1970-01-01 %H:%M:%E9S', dummy_field, %s), 'UTC')", + ("IST",), ), ) + @override_settings(USE_TZ=False) def test_datetime_cast_time_sql_use_tz_false(self): - settings.USE_TZ = False self.assertEqual( self.db_operations.datetime_cast_time_sql("dummy_field", None, "IST"), ( - "TIMESTAMP(FORMAT_TIMESTAMP('%Y-%m-%d %R:%E9S %Z', dummy_field, 'UTC'))", - None, + "TIMESTAMP(FORMAT_TIMESTAMP('1970-01-01 %H:%M:%E9S', dummy_field, %s), 'UTC')", + ("IST",), + ), + ) + self.assertEqual( + self.db_operations.datetime_cast_time_sql("dummy_field", None, None), + ( + "TIMESTAMP(FORMAT_TIMESTAMP('1970-01-01 %H:%M:%E9S', dummy_field, %s), 'UTC')", + ("UTC",), ), ) - settings.USE_TZ = True # reset changes. def test_date_interval_sql(self): self.assertEqual( @@ -463,26 +512,343 @@ def test_bulk_insert_sql(self): def test_date_and_time_trunc_sql(self): sql, params = self.db_operations.date_trunc_sql("year", "field", None) self.assertEqual(sql, "DATE_TRUNC(CAST(field AS DATE), year)") - self.assertIsNone(params) + self.assertEqual(params, ()) + + sql, params = self.db_operations.date_extract_sql("year", "field", None) + self.assertEqual(sql, "EXTRACT(year FROM field)") + self.assertEqual(params, ()) + + sql, params = self.db_operations.time_extract_sql("hour", "field", None) + self.assertEqual(sql, "EXTRACT(hour FROM field AT TIME ZONE %s)") + self.assertEqual(params, ("UTC",)) sql, params = self.db_operations.time_trunc_sql("hour", "field", None) - self.assertEqual(sql, 'TIMESTAMP_TRUNC(field, hour, "UTC")') - self.assertIsNone(params) + self.assertEqual(sql, "TIMESTAMP_TRUNC(field, hour, %s)") + self.assertEqual(params, ("UTC",)) + + sql, params = self.db_operations.time_trunc_sql( + "hour", "field", None, tzname="America/New_York" + ) + self.assertEqual(sql, "TIMESTAMP_TRUNC(field, hour, %s)") + self.assertEqual(params, ("America/New_York",)) + + sql, params = self.db_operations.datetime_trunc_sql( + "day", "field", None, tzname=None + ) + self.assertEqual(sql, "TIMESTAMP_TRUNC(field, day)") + self.assertEqual(params, ()) sql, params = self.db_operations.datetime_trunc_sql( "day", "field", None, tzname="UTC" ) - self.assertEqual(sql, 'TIMESTAMP_TRUNC(field, day, "UTC")') - self.assertIsNone(params) + self.assertEqual(sql, "TIMESTAMP_TRUNC(field, day, %s)") + self.assertEqual(params, ("UTC",)) + + sql, params = self.db_operations.date_trunc_sql("week", "field", None) + self.assertEqual( + sql, + "DATE_ADD(CAST(DATE_TRUNC(CAST(DATE_SUB(CAST(field AS DATE), INTERVAL 1 DAY) AS DATE), week) AS DATE), INTERVAL 1 DAY)", + ) + self.assertEqual(params, ()) + + sql, params = self.db_operations.datetime_trunc_sql( + "week", "field", None, tzname=None + ) + self.assertEqual( + sql, + "TIMESTAMP_ADD(TIMESTAMP_TRUNC(TIMESTAMP_SUB(field, INTERVAL 1 DAY), week), INTERVAL 1 DAY)", + ) + self.assertEqual(params, ()) + + sql, params = self.db_operations.datetime_trunc_sql( + "week", "field", None, tzname="UTC" + ) + self.assertEqual( + sql, + "TIMESTAMP_ADD(TIMESTAMP_TRUNC(TIMESTAMP_SUB(field, INTERVAL 1 DAY), week, %s), INTERVAL 1 DAY)", + ) + self.assertEqual(params, ("UTC",)) + @override_settings(USE_TZ=True) def test_datetime_cast_sql(self): + sql, params = self.db_operations.datetime_cast_date_sql("field", None, None) + self.assertEqual(sql, "DATE(field)") + self.assertEqual(params, ()) + sql, params = self.db_operations.datetime_cast_date_sql("field", None, "UTC") - self.assertEqual(sql, 'DATE(field, "UTC")') - self.assertIsNone(params) + self.assertEqual(sql, "DATE(field, %s)") + self.assertEqual(params, ("UTC",)) + + sql, params = self.db_operations.datetime_cast_time_sql("field", None, None) + self.assertEqual( + sql, + "TIMESTAMP(FORMAT_TIMESTAMP('1970-01-01 %H:%M:%E9S', field), 'UTC')", + ) + self.assertEqual(params, ()) sql, params = self.db_operations.datetime_cast_time_sql("field", None, "UTC") self.assertEqual( sql, - "TIMESTAMP(FORMAT_TIMESTAMP('%Y-%m-%d %R:%E9S %Z', field, 'UTC'))", + "TIMESTAMP(FORMAT_TIMESTAMP('1970-01-01 %H:%M:%E9S', field, %s), 'UTC')", + ) + self.assertEqual(params, ("UTC",)) + + @override_settings(USE_TZ=False) + def test_datetime_trunc_and_cast_sql_use_tz_false(self): + sql, params = self.db_operations.datetime_trunc_sql( + "day", "field", None, tzname="America/New_York" + ) + self.assertEqual(sql, "TIMESTAMP_TRUNC(field, day, %s)") + self.assertEqual(params, ("America/New_York",)) + + sql, params = self.db_operations.datetime_trunc_sql( + "day", "field", None, tzname=None + ) + self.assertEqual(sql, "TIMESTAMP_TRUNC(field, day, %s)") + self.assertEqual(params, ("UTC",)) + + sql, params = self.db_operations.datetime_cast_date_sql( + "field", None, tzname="America/New_York" + ) + self.assertEqual(sql, "DATE(field, %s)") + self.assertEqual(params, ("America/New_York",)) + + sql, params = self.db_operations.datetime_cast_date_sql( + "field", None, tzname=None + ) + self.assertEqual(sql, "DATE(field, %s)") + self.assertEqual(params, ("UTC",)) + + sql, params = self.db_operations.datetime_cast_time_sql( + "field", None, tzname="America/New_York" + ) + self.assertEqual( + sql, + "TIMESTAMP(FORMAT_TIMESTAMP('1970-01-01 %H:%M:%E9S', field, %s), 'UTC')", + ) + self.assertEqual(params, ("America/New_York",)) + + sql, params = self.db_operations.datetime_cast_time_sql( + "field", None, tzname=None + ) + self.assertEqual( + sql, + "TIMESTAMP(FORMAT_TIMESTAMP('1970-01-01 %H:%M:%E9S', field, %s), 'UTC')", + ) + self.assertEqual(params, ("UTC",)) + + @override_settings(USE_TZ=True) + def test_get_timezone_param_use_tz_true(self): + self.assertEqual( + self.db_operations._get_timezone_param("America/New_York"), + "America/New_York", + ) + self.assertEqual( + self.db_operations._get_timezone_param("+05:30"), + "+05:30", + ) + self.assertIsNone(self.db_operations._get_timezone_param(None)) + + @override_settings(USE_TZ=False) + def test_get_timezone_param_use_tz_false(self): + self.assertEqual( + self.db_operations._get_timezone_param("America/New_York"), + "America/New_York", + ) + self.assertEqual( + self.db_operations._get_timezone_param("-08:00"), + "-08:00", + ) + self.assertEqual( + self.db_operations._get_timezone_param(None), + "UTC", + ) + + def test_prepare_tzname_delta(self): + self.assertEqual( + self.db_operations._prepare_tzname_delta("UTC-04:43"), "-04:43" + ) + self.assertEqual( + self.db_operations._prepare_tzname_delta("UTC+05:00"), "+05:00" + ) + self.assertEqual( + self.db_operations._prepare_tzname_delta("UTC+05:30:00"), "+05:30" + ) + self.assertEqual(self.db_operations._prepare_tzname_delta("-0443"), "-0443") + self.assertEqual(self.db_operations._prepare_tzname_delta("+0500"), "+0500") + self.assertEqual(self.db_operations._prepare_tzname_delta("+03:00"), "+03:00") + self.assertEqual( + self.db_operations._prepare_tzname_delta("+05:30:00"), "+05:30" + ) + self.assertEqual( + self.db_operations._prepare_tzname_delta("-05:00:00"), "-05:00" + ) + self.assertEqual(self.db_operations._prepare_tzname_delta("+05:30"), "+05:30") + self.assertEqual(self.db_operations._prepare_tzname_delta("-04:43"), "-04:43") + self.assertEqual(self.db_operations._prepare_tzname_delta("UTC-0443"), "-0443") + self.assertEqual(self.db_operations._prepare_tzname_delta("UTC+0500"), "+0500") + self.assertEqual(self.db_operations._prepare_tzname_delta("+05"), "+05") + self.assertEqual(self.db_operations._prepare_tzname_delta("-08"), "-08") + self.assertEqual(self.db_operations._prepare_tzname_delta("+5"), "+5") + self.assertEqual(self.db_operations._prepare_tzname_delta("-8"), "-8") + self.assertEqual(self.db_operations._prepare_tzname_delta("+5:30"), "+5:30") + self.assertEqual(self.db_operations._prepare_tzname_delta("-8:00"), "-8:00") + self.assertEqual(self.db_operations._prepare_tzname_delta("UTC+5"), "+5") + self.assertEqual(self.db_operations._prepare_tzname_delta("UTC-8"), "-8") + self.assertEqual(self.db_operations._prepare_tzname_delta("UTC+5:30"), "+5:30") + self.assertEqual(self.db_operations._prepare_tzname_delta("UTC-8:00"), "-8:00") + self.assertEqual( + self.db_operations._prepare_tzname_delta("America/New_York"), + "America/New_York", + ) + self.assertEqual( + self.db_operations._prepare_tzname_delta("Etc/GMT+5"), + "Etc/GMT+5", + ) + self.assertEqual( + self.db_operations._prepare_tzname_delta("Etc/GMT-5"), + "Etc/GMT-5", + ) + self.assertEqual( + self.db_operations._prepare_tzname_delta("Etc/GMT+10"), + "Etc/GMT+10", + ) + self.assertEqual( + self.db_operations._prepare_tzname_delta("Etc/GMT-10"), + "Etc/GMT-10", + ) + self.assertEqual(self.db_operations._prepare_tzname_delta("UTC"), "UTC") + self.assertEqual(self.db_operations._prepare_tzname_delta(""), "UTC") + self.assertEqual(self.db_operations._prepare_tzname_delta(None), "UTC") + self.assertEqual( + self.db_operations._prepare_tzname_delta("+invalid"), "+invalid" + ) + + def test_extract_lookup_type_validation(self): + valid_lookups = ["year", "month", "day", "hour", "minute", "second", "isoweek"] + for lookup_type in valid_lookups: + self.db_operations._validate_lookup_type(lookup_type) + + invalid_lookups = [ + "year' OR 1=1--", + "day; DROP TABLE students;", + "foo bar", + "", + "year\x00", + None, + 123, + ] + for lookup_type in invalid_lookups: + with self.assertRaises(ValueError): + self.db_operations._validate_lookup_type(lookup_type) + with self.assertRaises(ValueError): + self.db_operations.date_extract_sql(lookup_type, "field") + with self.assertRaises(ValueError): + self.db_operations.datetime_extract_sql( + lookup_type, "field", None, "UTC" + ) + with self.assertRaises(ValueError): + self.db_operations.time_extract_sql(lookup_type, "field") + + def test_trunc_lookup_type_validation(self): + valid_lookups = [ + "year", + "quarter", + "month", + "week", + "day", + "hour", + "minute", + "second", + ] + for lookup_type in valid_lookups: + self.db_operations._validate_trunc_lookup_type(lookup_type) + + invalid_lookups = [ + "year' OR 1=1--", + "day; DROP TABLE students;", + "foo bar", + "", + "year\x00", + None, + 123, + ] + for lookup_type in invalid_lookups: + with self.assertRaises(OperationalError): + self.db_operations._validate_trunc_lookup_type(lookup_type) + with self.assertRaises(OperationalError): + self.db_operations.date_trunc_sql(lookup_type, "field", None) + with self.assertRaises(OperationalError): + self.db_operations.datetime_trunc_sql( + lookup_type, "field", None, tzname="UTC" + ) + with self.assertRaises(OperationalError): + self.db_operations.time_trunc_sql(lookup_type, "field", None) + + @override_settings(USE_TZ=True) + def test_timezone_methods_preserve_existing_params(self): + existing_params = ["param1", "param2"] + + sql, params = self.db_operations.datetime_extract_sql( + "hour", "field", existing_params, "UTC" + ) + self.assertEqual(params, ("param1", "param2", "UTC")) + + sql, params = self.db_operations.datetime_trunc_sql( + "hour", "field", existing_params, tzname="UTC" + ) + self.assertEqual(params, ("param1", "param2", "UTC")) + + sql, params = self.db_operations.time_trunc_sql( + "hour", "field", existing_params, tzname="UTC" + ) + self.assertEqual(params, ("param1", "param2", "UTC")) + + sql, params = self.db_operations.datetime_cast_date_sql( + "field", existing_params, "UTC" + ) + self.assertEqual(params, ("param1", "param2", "UTC")) + + sql, params = self.db_operations.datetime_cast_time_sql( + "field", existing_params, "UTC" + ) + self.assertEqual(params, ("param1", "param2", "UTC")) + + @override_settings(USE_TZ=True) + def test_tzname_sql_injection_defense(self): + malicious_tzname = "UTC'; DROP TABLE students; --" + + sql, params = self.db_operations.datetime_extract_sql( + "hour", "field", None, malicious_tzname + ) + self.assertNotIn(malicious_tzname, sql) + self.assertIn("%s", sql) + self.assertEqual(params, (malicious_tzname,)) + + sql, params = self.db_operations.datetime_trunc_sql( + "hour", "field", None, tzname=malicious_tzname + ) + self.assertNotIn(malicious_tzname, sql) + self.assertIn("%s", sql) + self.assertEqual(params, (malicious_tzname,)) + + sql, params = self.db_operations.time_trunc_sql( + "hour", "field", None, tzname=malicious_tzname + ) + self.assertNotIn(malicious_tzname, sql) + self.assertIn("%s", sql) + self.assertEqual(params, (malicious_tzname,)) + + sql, params = self.db_operations.datetime_cast_date_sql( + "field", None, malicious_tzname + ) + self.assertNotIn(malicious_tzname, sql) + self.assertIn("%s", sql) + self.assertEqual(params, (malicious_tzname,)) + + sql, params = self.db_operations.datetime_cast_time_sql( + "field", None, malicious_tzname ) - self.assertIsNone(params) + self.assertNotIn(malicious_tzname, sql) + self.assertIn("%s", sql) + self.assertEqual(params, (malicious_tzname,))