diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7b2b6fb..e111822 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -232,6 +232,79 @@ jobs: # the sqlite job. - run: timeout -s ABRT 1500 python -m pytest -q + # The stored-row reader decodes a date and time by one rule on every driver. + # psycopg2 and mysqlclient read some of the values it refuses (PostgreSQL's + # 'infinity' as the largest datetime, a MySQL zero date as None), so the + # reader's tests run on each of them as well as on psycopg 3 and PyMySQL. + # Only the reader's file: it is the code the driver changes, and these legs + # install the driver the suite's other legs leave out. Django prefers + # psycopg 3 when both are installed, so psycopg is not installed here. + test-postgres-psycopg2: + runs-on: ubuntu-latest + timeout-minutes: 30 + name: postgres psycopg2 (stored-row reader) + services: + postgres: + image: postgres:16 + env: + POSTGRES_PASSWORD: ox + ports: + - 54329:5432 + options: >- + --health-cmd "pg_isready -U postgres" + --health-interval 5s + --health-timeout 5s + --health-retries 10 + env: + DJANGO_SETTINGS_MODULE: tests.settings_postgres + PYTHONFAULTHANDLER: "1" + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 + with: + persist-credentials: false + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7 + with: + python-version: "3.14" + - run: python -m pip install --upgrade pip + - run: python -m pip install -e . "Django~=6.1.0" pytest pytest-django psycopg2-binary prometheus-client "celery>=5.6" + - run: python -c "from django.db.backends.postgresql import base; assert base.Database.__name__ == 'psycopg2', base.Database" + - run: timeout -s ABRT 1500 python -m pytest -q tests/test_stored_read.py + + test-mysql-mysqlclient: + runs-on: ubuntu-latest + timeout-minutes: 30 + name: mysql mysqlclient (stored-row reader) + services: + mysql: + image: mysql:8 + env: + MYSQL_ROOT_PASSWORD: ox + MYSQL_DATABASE: oxtest + ports: + - 33069:3306 + options: >- + --health-cmd "mysqladmin ping -h 127.0.0.1 -uroot -pox" + --health-interval 5s + --health-timeout 5s + --health-retries 20 + env: + DJANGO_SETTINGS_MODULE: tests.settings_mysql + PYTHONFAULTHANDLER: "1" + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 + with: + persist-credentials: false + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7 + with: + python-version: "3.14" + - run: sudo apt-get update && sudo apt-get install -y libmysqlclient-dev pkg-config + - run: python -m pip install --upgrade pip + - run: python -m pip install -e . "Django~=6.1.0" pytest pytest-django mysqlclient prometheus-client "celery>=5.6" + # tests/settings_mysql.py registers PyMySQL only when mysqlclient is + # missing; this leg installs no PyMySQL, so MySQLdb is mysqlclient. + - run: python -c "import MySQLdb, sys; assert 'pymysql' not in sys.modules" + - run: timeout -s ABRT 1500 python -m pytest -q tests/test_stored_read.py + # A >= install resolves to the newest release and cannot test a floor. This # leg pins the floor with == so a regression at the bottom of the supported # range fails here. @@ -494,7 +567,7 @@ jobs: ci-ok: name: ci-ok if: always() - needs: [test-sqlite, test-postgres, test-postgres-naive, test-postgres-pooled, test-backport-floor, test-pool-floor, wagtail-smoke, test-mysql, lint, types, docs, packaging, zizmor] + needs: [test-sqlite, test-postgres, test-postgres-naive, test-postgres-pooled, test-backport-floor, test-pool-floor, wagtail-smoke, test-mysql, test-postgres-psycopg2, test-mysql-mysqlclient, lint, types, docs, packaging, zizmor] runs-on: ubuntu-latest timeout-minutes: 10 steps: diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 91bb926..4c242ca 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -25,9 +25,7 @@ docker run -d --name ox-pg -e POSTGRES_PASSWORD=ox -p 54329:5432 postgres:16 DJANGO_SETTINGS_MODULE=tests.settings_postgres .venv/bin/python -m pytest ``` -Run both before opening a PR; CI tests every supported Python and Django -version on SQLite and PostgreSQL, and the oldest and newest corners on -MySQL 8. +Run both before opening a PR. CI tests every supported Python and Django version on SQLite and PostgreSQL, and the oldest and newest corners on MySQL 8. Separate psycopg2 and mysqlclient jobs are configured to test only the stored-row reader. ## Style diff --git a/src/django_ox/_stored_read.py b/src/django_ox/_stored_read.py new file mode 100644 index 0000000..f07995f --- /dev/null +++ b/src/django_ox/_stored_read.py @@ -0,0 +1,1037 @@ +""" +Reading stored schedules and their tick log when a value in them may not +read. Not public API. + +A row written through django_ox.stored holds values Django reads back. A +row written around it, by SQL, a fixture or a data migration, holds +whatever its columns accept, and Django's read of such a value either +raises in the middle of a result, taking every row of it along, or on +SQLite reads something else without a word: a start that is not a date +reads as None, an `enabled` of 2 reads as False. A worker that dispatched +from either was wrong, and an admin page or an API call that read one could +not go on. + +So the values whose reading can fail or mislead are fetched as the database +holds them and decoded here, one value at a time. A date and time is +fetched as text, cast in the query on PostgreSQL and MySQL (SQLite holds +text already), and decoded by one rule on every driver, under USE_TZ and +the connection's time zone: the rule of psycopg 3 and PyMySQL with Django's +converters. What they read, this reads, and what they raise on or turn into +None is unreadable. The drivers Django also runs on read some of those +values as something: psycopg2 loads PostgreSQL's 'infinity' and '-infinity' +as the largest and smallest datetime, and mysqlclient loads a MySQL zero +date as None. They are unreadable here on those drivers too, so a stored +value means the same under any of them. An integer or a boolean is fetched +past Django's converters and has to be one, and a text column has to hold +text. A value that does not decode is reported by field, never raised from +the middle of a result and never read as something else. + +One value that reads is reported with them: a count of boundary writes +that is the most its column can hold. Every write that moves a schedule's +boundary adds one to the count in the database, and at the column's maximum +PostgreSQL and MySQL refuse that write each time it is tried, while SQLite +keeps the sum as a float, which no longer reads. A row holding it is +answered as one that does not read before any such write is tried +(`_refuse_a_full_count`). + +A read that fails as a whole, on a value the driver cannot hand over at all +(text SQLite cannot decode as UTF-8), is tried again in halves until the +row or the key that fails is alone. Only a failure known to be raised in +this process, while a value was being turned into Python, is narrowed down +that way (`is_unreadable_value` names them), and only while the connection +is known to take another statement (`can_read_on`). No read takes a +savepoint: a failure of that kind comes once the database has answered and +leaves a transaction as it was, so the reads that narrow it down run in the +same one. + +An error the database itself raised is raised as it is, with nothing read +after it, inside a transaction and with none open. It does not say that +one value was at fault, so no row or key is named for it, and inside a +PostgreSQL transaction no statement is taken after it until the +transaction is rolled back, which is for whoever owns it. +""" + +from __future__ import annotations + +import re +from collections.abc import Collection, Iterable, Mapping, Sequence +from dataclasses import dataclass, field +from datetime import UTC, datetime, timedelta, tzinfo +from datetime import timezone as fixed_offset +from typing import Any +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +from django.conf import settings +from django.db import DatabaseError, DataError, OperationalError, connections +from django.db.backends.base.base import BaseDatabaseWrapper +from django.db.models import ( + BooleanField, + CharField, + DateTimeField, + F, + Field, + Func, + IntegerField, + Max, + QuerySet, + TextField, +) +from django.db.models.functions import Cast +from django.db.transaction import TransactionManagementError +from django.utils import timezone +from django.utils.dateparse import parse_datetime + +from .exceptions import StoredValueUnreadable +from .models import OxSchedule, OxScheduleTick + +#: The most stored schedules one statement reads by key: enough that every +#: row of a large table is read in a few statements, ten thousand rows in +#: twenty, and a bound on the keys any one statement names. A connection +#: that takes fewer parameters than this is read in batches of what it +#: takes (`read_schedules`). Not a setting, and nothing outside this module +#: depends on the number. +_BATCH_SIZE = 500 + +#: How SQLite's driver reports text it cannot decode as UTF-8. It raises an +#: OperationalError, the class of a database that failed, for one value in +#: one row, so it is told apart by its message. +_SQLITE_UNDECODABLE = "Could not decode to UTF-8 column" + +#: The two of libpq's transaction states a session can be read in: no +#: transaction open, and one open in which no statement has failed. psycopg +#: and psycopg2 both report the state as `info.transaction_status`. +_PQTRANS_IDLE = 0 +_PQTRANS_INTRANS = 2 + +#: The most characters of a stored value a reason quotes. +_SHOWN_LIMIT = 60 + +#: The most characters of an error's message a reason quotes. +_ERROR_LIMIT = 200 + +# What is wrong with a value, all of it in this block. Each follows the +# field's name: "end_time holds '10000-01-01 00:00:00+00', which is not a +# date and time". `value` is the value as stored, quoted and cut to length. + +#: NULL in a column that needs a value. +_NULL = "is NULL, and this field needs a value" +#: Text a date and time cannot be read from, or a value that is not text. +_NOT_A_DATETIME = "holds {value}, which is not a date and time" +#: A date and time outside what Python's datetime holds, or one that does +#: not exist. `error` is what datetime said. +_DATETIME_OUT_OF_RANGE = ( + "holds {value}, which is not a date and time Python can hold ({error})" +) +#: A date and time with a zone, read while USE_TZ is off. +_ZONE_WITH_USE_TZ_OFF = "holds {value}, which has a time zone, and USE_TZ is off" +#: A date and time without a zone, read while USE_TZ is on. +_NO_ZONE_WITH_USE_TZ_ON = "holds {value}, which has no time zone, and USE_TZ is on" +#: Not an integer. +_NOT_A_WHOLE_NUMBER = "holds {value}, which is not a whole number" +#: Not one of the two values a boolean column holds. +_NOT_A_BOOLEAN = "holds {value}, which is neither true nor false" +#: A value that is not text in a column of text: bytes, which SQLite keeps in +#: any column. +_NOT_TEXT = "holds {value}, which is not text" +#: A count of boundary writes that is the most its column holds. +_COUNT_AT_ITS_MAXIMUM = ( + "holds {value}, the most its column can hold, so the count cannot be raised" +) +#: The driver could not hand the value over. `error` is what it raised. +_NOT_FETCHED = "could not be read: {error}" +#: A row that did not read, when no one field could be named. +_UNKNOWN_FIELD = "a value in it could not be read: {error}" + + +class UnreadableValue(ValueError): + """A stored value that does not decode. The message says why, safe to print.""" + + +def is_unreadable_value(exc: BaseException, *, using: str) -> bool: + """ + Whether a read on `using` failed in this process, on a value it was + turning into Python, rather than in the database. + + The failures known to be that, and no others: + + - a conversion error from a driver's or Django's converter, which is + no database error at all: ValueError, TypeError, AttributeError, + OverflowError; + - on PostgreSQL, a DataError that carries no SQLSTATE. The server + gives one to every error it raises, so this is the driver's own, on + a value it was sent and could not load: psycopg's, for a timestamp + outside the years a datetime holds; + - on SQLite, the driver's OperationalError for text it cannot decode + as UTF-8, told by its message and by carrying no SQLite error code, + which SQLite gives to every error of its own. + + Anything else is the database's and is raised as it is, whatever its + class. A DataError the server raised is among them on every database + (PostgreSQL's carries its SQLSTATE and MySQL's its error number), with + a lost connection, a lock, a deadlock, a cancelled statement, a missing + table and a refused permission. None of those says that one value was + at fault, so none is put down to a row or a key. + """ + if not isinstance(exc, DatabaseError): + return isinstance(exc, (ValueError, TypeError, AttributeError, OverflowError)) + if _database_code(exc) is not None: + return False + vendor = connections[using].vendor + if vendor == "postgresql": + return isinstance(exc, DataError) + if vendor == "sqlite": + return isinstance(exc, OperationalError) and _SQLITE_UNDECODABLE in str(exc) + return False + + +def _database_code(exc: BaseException) -> object: + """ + The code the database gave an error it raised, or None for one that + carries none. + + Django raises its own exception class with the driver's exception as + the cause. psycopg puts the SQLSTATE on its exception as `sqlstate` + and psycopg2 as `pgcode`, each None on an error the driver raised + itself; the SQLite driver puts SQLite's code on as `sqlite_errorcode`, + and leaves the attribute off an error of its own. + """ + for error in (exc, exc.__cause__): + for name in ("sqlstate", "pgcode", "sqlite_errorcode"): + code = getattr(error, name, None) + if code is not None: + return code + return None + + +def can_read_on(exc: BaseException, *, using: str) -> bool: + """ + Whether the read that raised `exc` may be followed by another on the + same connection: it failed in this process on a value + (`is_unreadable_value`), and the connection is known to take another + statement. + + What follows a failed read is a narrower one, to find the row or the + field it failed on. That is worth sending only after a failure a + narrower read can pin on one value, which an error the database raised + is never taken for, with a transaction open or without. And it must + not be sent into a transaction the failure has ended. No read here + takes a savepoint to come back to, so that second question is put to + the connection, without a query. + + Django is asked first: a transaction it has marked for rollback takes + no more statements. Then the driver, where it says. PostgreSQL's + driver reports the session's transaction state, and reading goes on + when no transaction is open or the one that is has not failed. A value + psycopg could not load fails in this process, once the server has + answered, and leaves the transaction as it was. + + MySQL's and SQLite's drivers report no such state. With no transaction + open there is nothing a failed statement could have left unfinished, + and reading goes on. Inside one it goes on only for the failure known + to be the driver's own over a statement the database answered: + SQLite's, refusing text that is not UTF-8. Anything else is raised as + it stands to whoever owns the transaction. Both would often take + another statement after it, and nothing here can tell when. + """ + if not is_unreadable_value(exc, using=using): + return False + connection = connections[using] + if connection.needs_rollback: + return False + if connection.vendor == "postgresql": + state = _postgresql_transaction_state(connection) + if state is not None: + return state in (_PQTRANS_IDLE, _PQTRANS_INTRANS) + if connection.get_autocommit() and not connection.in_atomic_block: + return True + # On SQLite the one database error that counts as a value's is the + # driver's own, on text it could not decode. + return connection.vendor == "sqlite" and isinstance(exc, DatabaseError) + + +def _postgresql_transaction_state(connection: BaseDatabaseWrapper) -> int | None: + """ + The transaction state libpq holds for this session, or None where the + driver does not give it. From the driver's own record, with no round + trip: a query could not be sent to ask whether a query can be sent. + """ + try: + return int(connection.connection.info.transaction_status) + except (AttributeError, TypeError, ValueError): + # No connection, or a driver without the attribute. + return None + + +class _Raw(Func): + """ + A column as the driver hands it over, past Django's converters. + + SQLite's driver converts a value by its column's declared type, so a + boolean column's 2 or 'abc' arrives as False before anything can look + at it. An expression has no declared type, and unary plus is the + expression that leaves every value as it is. Elsewhere it is the column. + + The output field is the bare Field, which no backend converts: Django + passes an expression with an integer output through int(), which turns + 60.5 into 60 and raises on 'abc'. + """ + + arity = 1 + template = "%(expressions)s" + output_field = Field() + + def as_sqlite(self, compiler: Any, connection: Any, **extra_context: Any) -> Any: + return self.as_sql( + compiler, connection, template="+%(expressions)s", **extra_context + ) + + +def raw(expression: Any) -> Func: + """An integer or boolean column, or an expression, as the driver hands it over.""" + return _Raw(expression) + + +def as_text(expression: Any) -> Cast: + """ + A date and time column, or an aggregate of one, as its text in the database. + + The cast is in the query, so no converter of the driver's or Django's + sees the value: PostgreSQL's text form, MySQL's CAST AS CHAR, and on + SQLite the text it holds. `decode_datetime` reads it. + """ + return Cast(expression, output_field=TextField()) + + +def _shown(value: object) -> str: + """A stored value as a reason quotes it: its repr, escaped, cut to length.""" + text = repr(value) + if len(text) > _SHOWN_LIMIT: + text = text[: _SHOWN_LIMIT - 3] + "..." + return text + + +def _error_text(exc: BaseException | None) -> str: + """An exception as a reason quotes it, escaped and cut to length.""" + text = "unknown error" if exc is None else f"{type(exc).__name__}: {exc}" + if not text.isprintable(): + text = ascii(text)[1:-1] + if len(text) > _ERROR_LIMIT: + text = text[: _ERROR_LIMIT - 3] + "..." + return text + + +# -- decoding --------------------------------------------------------------- + +#: PostgreSQL's text form of a timestamptz under the ISO DateStyle, which is +#: the only one psycopg reads: the pattern psycopg matches, so a value reads +#: here exactly when it reads there. 'infinity', '-infinity' and a date +#: before year 1 (which ends in ' BC') do not match. +_POSTGRESQL_TIMESTAMPTZ = re.compile( + r""" + (\d+) [^a-z0-9] (\d+) [^a-z0-9] (\d+) + (?: T | [^a-z0-9] ) + (\d+) [^a-z0-9] (\d+) [^a-z0-9] (\d+) + (?: \.(\d+) )? + ([-+]) (\d+) (?: : (\d+) )? (?: : (\d+) )? + """, + re.ASCII | re.IGNORECASE | re.VERBOSE, +) + +#: MySQL's text form of a DATETIME: the pattern PyMySQL converts, whole. +#: A zero date matches and then fails as a date. +_MYSQL_DATETIME = re.compile( + r"(\d{1,4})-(\d{1,2})-(\d{1,2})[T ](\d{1,2}):(\d{1,2}):(\d{1,2})(?:\.(\d{1,6}))?", + re.ASCII, +) + + +def _session_zone(connection: BaseDatabaseWrapper) -> tzinfo: + """ + The zone a PostgreSQL session renders a timestamptz in: the one Django + sets it to, and UTC where that zone is unknown here, as psycopg falls + back. + """ + try: + return ZoneInfo(connection.timezone_name) + except (ZoneInfoNotFoundError, ValueError): + return UTC + + +def _from_postgresql(text: str, connection: BaseDatabaseWrapper) -> datetime: + """ + A timestamptz's text as psycopg loads it and Django's loader relabels it. + + psycopg reads the instant from the text and moves it to the session's + zone; past the end of the years a datetime holds, it keeps the text's + own wall clock and offset instead. Django then replaces the zone with + the connection's: UTC, or none with USE_TZ off. + """ + match = _POSTGRESQL_TIMESTAMPTZ.fullmatch(text) + if match is None: + raise UnreadableValue(_NOT_A_DATETIME.format(value=_shown(text))) + year, month, day, hour, minute, second, fraction, sign, oh, om, os = match.groups() + micro = int(fraction) if fraction else 0 + if fraction and len(fraction) < 6: + micro *= 10 ** (6 - len(fraction)) + offset_seconds = 3600 * int(oh) + 60 * int(om or 0) + int(os or 0) + offset = timedelta(seconds=offset_seconds if sign == "+" else -offset_seconds) + instant = datetime( + int(year), + int(month), + int(day), + int(hour), + int(minute), + int(second), + micro, + tzinfo=UTC, + ) + try: + loaded = (instant - offset).astimezone(_session_zone(connection)) + except OverflowError: + loaded = instant.replace(tzinfo=fixed_offset(offset)) + return loaded.replace(tzinfo=connection.timezone) + + +def _from_mysql(text: str, connection: BaseDatabaseWrapper) -> datetime: + """A DATETIME's text as PyMySQL converts it and Django's converter zones it.""" + match = _MYSQL_DATETIME.fullmatch(text) + if match is None: + raise UnreadableValue(_NOT_A_DATETIME.format(value=_shown(text))) + year, month, day, hour, minute, second, fraction = match.groups() + loaded = datetime( + int(year), + int(month), + int(day), + int(hour), + int(minute), + int(second), + int((fraction or "").ljust(6, "0")), + ) + if settings.USE_TZ: + loaded = timezone.make_aware(loaded, connection.timezone) + return loaded + + +def _from_sqlite(text: str, connection: BaseDatabaseWrapper) -> datetime: + """ + SQLite's text as Django's converter reads it, except that text it + cannot read is unreadable rather than None. + """ + loaded = parse_datetime(text) + if loaded is None: + raise UnreadableValue(_NOT_A_DATETIME.format(value=_shown(text))) + if settings.USE_TZ and timezone.is_naive(loaded): + loaded = timezone.make_aware(loaded, connection.timezone) + return loaded + + +def decode_datetime( + value: object, using: str, *, null: bool = False +) -> datetime | None: + """ + Decode a date and time fetched as text through `as_text`. + + Use one decoding rule on every driver, following psycopg 3 and PyMySQL + with Django's converters, under USE_TZ and the connection's time zone. + For accepted values, preserve the instant and the time zone, or its + absence, that this rule produces. + + PostgreSQL 'infinity' and '-infinity' are unreadable even when psycopg2 + would return the largest or smallest datetime. MySQL zero dates are + unreadable even when mysqlclient would return None. + + Raise UnreadableValue for those values, dates outside years 1 through + 9999, nonexistent dates, and text that is not a date and time. + Also reject NULL unless `null`, and values whose time zone does not + match USE_TZ, including SQLite values with a zone when USE_TZ is off. + """ + if value is None: + if null: + return None + raise UnreadableValue(_NULL) + if not isinstance(value, str): + raise UnreadableValue(_NOT_A_DATETIME.format(value=_shown(value))) + connection = connections[using] + try: + if connection.vendor == "postgresql": + loaded = _from_postgresql(value, connection) + elif connection.vendor == "mysql": + loaded = _from_mysql(value, connection) + else: + loaded = _from_sqlite(value, connection) + except UnreadableValue: + raise + except (ValueError, OverflowError) as exc: + raise UnreadableValue( + _DATETIME_OUT_OF_RANGE.format(value=_shown(value), error=exc) + ) from exc + if timezone.is_aware(loaded) and not settings.USE_TZ: + raise UnreadableValue(_ZONE_WITH_USE_TZ_OFF.format(value=_shown(value))) + if timezone.is_naive(loaded) and settings.USE_TZ: + raise UnreadableValue(_NO_ZONE_WITH_USE_TZ_ON.format(value=_shown(value))) + return loaded + + +def decode_integer(value: object, *, null: bool = False) -> int | None: + """ + An integer column's value as fetched through `raw`. SQLite keeps text + and fractions in one, which raise UnreadableValue, as NULL does unless + `null`. + """ + if value is None: + if null: + return None + raise UnreadableValue(_NULL) + if type(value) is int: + return value + raise UnreadableValue(_NOT_A_WHOLE_NUMBER.format(value=_shown(value))) + + +def decode_boolean(value: object, *, null: bool = False) -> bool | None: + """ + A boolean column's value as fetched through `raw`: PostgreSQL's boolean, + or the 0 or 1 SQLite and MySQL keep. Anything else they keep (2, 'abc') + raises UnreadableValue, as NULL does unless `null`. + """ + if value is None: + if null: + return None + raise UnreadableValue(_NULL) + if type(value) is bool: + return value + if type(value) is int and value in (0, 1): + return bool(value) + raise UnreadableValue(_NOT_A_BOOLEAN.format(value=_shown(value))) + + +def _projection(model_field: Field[Any, Any]) -> Any: + """How one column is fetched: an expression, or None for the column itself.""" + if model_field.primary_key: + return None + if isinstance(model_field, DateTimeField): + return as_text(model_field.attname) + if isinstance(model_field, (BooleanField, IntegerField)): + return raw(model_field.attname) + return None + + +def _decode(model_field: Field[Any, Any], value: object, using: str) -> Any: + """One fetched column's value, decoded as `_projection` fetched it.""" + if isinstance(model_field, DateTimeField): + return decode_datetime(value, using, null=model_field.null) + if isinstance(model_field, BooleanField): + return decode_boolean(value, null=model_field.null) + if isinstance(model_field, IntegerField): + number = decode_integer(value, null=model_field.null) + if model_field.name == _BOUNDARY_COUNT: + _refuse_a_full_count(number, model_field, using) + return number + # SQLite keeps whatever it was given in a text column, and hands bytes + # back as they are. Django would put them in the field as the value, + # where every use of it as text fails. + if ( + isinstance(model_field, (CharField, TextField)) + and value is not None + and not isinstance(value, str) + ): + raise UnreadableValue(_NOT_TEXT.format(value=_shown(value))) + return value + + +#: The field that counts the writes to a schedule's activation boundary. +_BOUNDARY_COUNT = "boundary_generation" + + +def _refuse_a_full_count( + count: int | None, model_field: Field[Any, Any], using: str +) -> None: + """ + Raise UnreadableValue when boundary_generation cannot increase within + its column's range on `using`. + + The maximum is a readable integer. The guarded reader rejects it because + every write that moves the activation boundary increments the count. + These writes include a worker's boundary heal, a retime, and a pause + whose boundary is stale. + + PostgreSQL and MySQL refuse an increment at the maximum. SQLite stores + the sum as a float, which fails the integer read. SQL can set the count + to the maximum or one below it. From one below, the next boundary write + succeeds and reaches the maximum. Subsequent guarded reads reject the + row, so a worker skips it without trying to move its boundary. + Pause and delete by primary key remain available; other changes are + refused. Neither the API nor the admin can lower the count. + + This check rejects a decoded value. It does not attribute a database + error to a particular row. + + Use `ops.integer_field_range` for the column type on this connection: + 2147483647 on PostgreSQL, 4294967295 on MySQL, and 9223372036854775807 on + SQLite. Reject values at or above that maximum, including values in a + column altered to hold more than the model's declared type. + """ + if count is None: + return + _, highest = connections[using].ops.integer_field_range( + model_field.get_internal_type() + ) + if count >= highest: + raise UnreadableValue(_COUNT_AT_ITS_MAXIMUM.format(value=_shown(count))) + + +def _fetched( + query: QuerySet[Any], fields: Sequence[Field[Any, Any]] +) -> list[tuple[Any, ...]]: + """`fields` of every row of `query`, each as `_projection` fetches it.""" + names = [] + expressions = {} + for model_field in fields: + expression = _projection(model_field) + if expression is None: + names.append(model_field.attname) + else: + name = f"ox_stored_{model_field.attname}" + expressions[name] = expression + names.append(name) + return list(query.annotate(**expressions).values_list(*names)) + + +# -- schedule rows ----------------------------------------------------------- + + +@dataclass(frozen=True) +class UnreadableRow: + """ + A stored schedule that exists and holds a value that does not read. + + `unreadable` maps each field that did not read to why, safe to print, + and is empty when the read could not say which field it was. `values` + holds every field that did read, by attribute name, the primary key + always among them. `cause` is the first exception the read met. + + A read asked for some of the fields (`read_schedules`, `fields`) says + nothing about the others: they are in neither mapping. + """ + + alias: str + pk: int + unreadable: Mapping[str, str] + values: Mapping[str, Any] + cause: BaseException | None = field(default=None, compare=False, repr=False) + + @property + def fields(self) -> tuple[str, ...]: + """The names of the fields that did not read, in the model's order.""" + return tuple(self.unreadable) + + @property + def reason(self) -> str: + """Why the row did not read, in one line safe to print.""" + if self.unreadable: + return "; ".join(f"{name} {why}" for name, why in self.unreadable.items()) + return _UNKNOWN_FIELD.format(error=_error_text(self.cause)) + + def exception(self) -> StoredValueUnreadable: + """The StoredValueUnreadable that says so, with `cause` as its cause.""" + exc = StoredValueUnreadable( + alias=self.alias, + model=OxSchedule._meta.label, + pk=self.pk, + fields=self.fields, + reason=self.reason, + ) + exc.__cause__ = self.cause + return exc + + def deferred_instance(self) -> OxSchedule: + """ + An OxSchedule holding the fields that read, with every other field + deferred. Reading a deferred field asks Django's own read for it, + which raises as it always did; saving with `update_fields` that + name only fields set here writes nothing else. + """ + names = [ + model_field.attname + for model_field in OxSchedule._meta.concrete_fields + if model_field.attname in self.values + ] + return OxSchedule.from_db(self.alias, names, [self.values[n] for n in names]) + + +def read_schedules( + pks: Iterable[int] | QuerySet[OxSchedule], + *, + using: str, + lock: bool = False, + fields: Collection[str] | None = None, +) -> dict[int, OxSchedule | UnreadableRow]: + """ + The stored schedules with these primary keys, read from `using`. + + Each that exists maps to an OxSchedule equal to the one Django's own + read gives, or to an UnreadableRow naming the fields that did not read. + A key with no row is left out, so a caller can tell a row that is gone + from one that does not read. In the order the keys were given. + + A queryset is read for its primary keys alone, which always read, and + then as keys: `_BATCH_SIZE` rows a statement, or as many as the + connection takes parameters for where that is fewer. A batch that fails + as a whole, on a value the driver cannot hand over, is read again in + halves until the row that fails is alone, and that row's columns are + then read one at a time to name the ones that do not read. Those reads + are made only after a failure known to be a value's, and only while the + connection is known to take another statement (`can_read_on`): an error + the database raised is raised as it is. + + `lock` takes each row's lock as it reads it, where the database has row + locks, for a read that has to be current. On SQLite it does nothing: + there `lock_schedule` is what serialises. + + `fields` names the fields to read, for a caller that needs those alone + and must not depend on the rest of the row. The primary key is always + read. A column that is not named is not selected, so a value in it + that does not read cannot fail the read and is not reported: on the + OxSchedule handed back the field is deferred, and an UnreadableRow + speaks only of the fields that were read. Without `fields`, every one. + """ + if isinstance(pks, QuerySet): + pks = pks.values_list("pk", flat=True) + wanted = list(dict.fromkeys(pks)) + selected = _selected(fields) + # One parameter a key, and the statement has no other. + limit = connections[using].features.max_query_params + step = max(min(_BATCH_SIZE, limit) if limit else _BATCH_SIZE, 1) + found: dict[int, OxSchedule | UnreadableRow] = {} + for start in range(0, len(wanted), step): + batch = wanted[start : start + step] + found.update(_read_rows(batch, using, lock=lock, fields=selected)) + return {pk: found[pk] for pk in wanted if pk in found} + + +def _selected(fields: Collection[str] | None) -> list[Field[Any, Any]]: + """ + The model's fields a read fetches, in the model's order: all of them, + or the primary key and those named. A name that is not a field raises + FieldDoesNotExist, as the model's own lookup of it does. + """ + concrete = OxSchedule._meta.concrete_fields + if fields is None: + return list(concrete) + named = {OxSchedule._meta.get_field(name) for name in fields} + return [ + model_field + for model_field in concrete + if model_field.primary_key or model_field in named + ] + + +def _read_rows( + pks: list[int], using: str, *, lock: bool, fields: list[Field[Any, Any]] +) -> dict[int, OxSchedule | UnreadableRow]: + """One batch, or its halves when it does not read as a whole.""" + query = OxSchedule.objects.using(using).filter(pk__in=pks) + if lock and connections[using].features.has_select_for_update: + query = query.select_for_update() + try: + fetched = _fetched(query, fields) + except Exception as exc: + if not can_read_on(exc, using=using): + raise + if len(pks) == 1: + alone = _diagnose(pks[0], using, exc, fields) + return {} if alone is None else {pks[0]: alone} + middle = len(pks) // 2 + return { + **_read_rows(pks[:middle], using, lock=lock, fields=fields), + **_read_rows(pks[middle:], using, lock=lock, fields=fields), + } + rows = (_hydrate(dict(zip(fields, row, strict=True)), using) for row in fetched) + return {row.pk: row for row in rows} + + +def _hydrate( + fetched: Mapping[Field[Any, Any], Any], using: str +) -> OxSchedule | UnreadableRow: + """A fetched row as an OxSchedule, or as what about it did not read.""" + values: dict[str, Any] = {} + unreadable: dict[str, str] = {} + cause: BaseException | None = None + for model_field, value in fetched.items(): + try: + values[model_field.attname] = _decode(model_field, value, using) + except UnreadableValue as exc: + unreadable[model_field.name] = str(exc) + cause = cause or exc + pk = values[OxSchedule._meta.pk.attname] + if unreadable: + return UnreadableRow(using, pk, unreadable, values, cause) + names = [model_field.attname for model_field in fetched] + return OxSchedule.from_db(using, names, [values[name] for name in names]) + + +def _diagnose( + pk: int, using: str, cause: BaseException, fields: list[Field[Any, Any]] +) -> UnreadableRow | None: + """ + A row that failed to read as a whole, one column at a time, or None if + it is gone. Paid only for a row that failed: a statement for each + column the read was for. + """ + values: dict[str, Any] = {} + unreadable: dict[str, str] = {} + exists = False + for model_field in fields: + query = OxSchedule.objects.using(using).filter(pk=pk) + try: + fetched = _fetched(query, [model_field]) + except Exception as exc: + if not can_read_on(exc, using=using): + raise + exists = True + unreadable[model_field.name] = _NOT_FETCHED.format(error=_error_text(exc)) + continue + if not fetched: + continue + exists = True + try: + values[model_field.attname] = _decode(model_field, fetched[0][0], using) + except UnreadableValue as exc: + unreadable[model_field.name] = str(exc) + if not exists: + return None + return UnreadableRow(using, pk, unreadable, values, cause) + + +def lock_schedule( + pk: int, *, using: str, fields: Collection[str] | None = None +) -> OxSchedule | UnreadableRow | None: + """ + Take a stored schedule's row lock and read the row under it: the row, + what of it did not read, or None when it is gone. + + On PostgreSQL and MySQL the read is what locks, one statement: a + locking read waits for a concurrent writer and then reads the row as + that writer committed it, not as this transaction first saw it, and + holds the lock to the end of the transaction. It is read as + `read_schedules` reads, every value whose reading can fail fetched as + the database holds it and decoded once the statement has answered. So + the statement that takes the lock is not one a stored value fails, and + a row holding a value that does not read is locked like any other, to + be paused, repaired or deleted. + + SQLite has no row locks, and Django drops the locking clause there + without raising. What serialises SQLite is being the writer, and a + transaction that reads first starts as a reader, which is refused + rather than made to wait when it later writes. So there a no-op UPDATE + comes first, which makes the transaction the writer before it reads. + + `fields` names the fields to read, as for `read_schedules`: a caller + that names what it needs selects no other column, and a value in one + it did not name neither stops it nor is reported to it. + + Whether the row is there is what the read says, which leaves out a row + that is gone. No statement asks, and nothing depends on the UPDATE's + row count: a no-op UPDATE counts the rows it matched on SQLite, and on + MySQL the rows it changed unless Django's FOUND_ROWS flag is set, which + a project setting its own client_flag loses with no error. + + Raises TransactionManagementError outside a transaction, where the lock + would be released as soon as it was taken. + """ + connection = connections[using] + if connection.get_autocommit(): + raise TransactionManagementError( + "lock_schedule() holds a lock until the end of the transaction, " + "and there is no transaction." + ) + if not connection.features.has_select_for_update: + OxSchedule.objects.using(using).filter(pk=pk).update(name=F("name")) + return read_schedules([pk], using=using, lock=True, fields=fields).get(pk) + + +# -- the tick log ------------------------------------------------------------- + + +def _fetch_failed(exc: BaseException) -> str: + """Why a tick did not read, when the driver could not hand it over.""" + return f"scheduled_for {_NOT_FETCHED.format(error=_error_text(exc))}" + + +@dataclass(frozen=True) +class TickRead: + """ + What the tick log holds for one schedule key. + + One of three answers. A tick that reads: `at` is set. No tick at all, + within what was asked: `absent`. A tick that does not read: `reason` + says why, safe to print, and `pk` is the tick row's key where the read + that found it also found its key (`earliest_tick`). + `latest_ticks` does not look for it: `newest_tick_pk` does, for a caller + that is about to name the row. An unreadable tick is not an absent one, + and must not be taken for one. + """ + + key: str + at: datetime | None = None + pk: int | None = None + reason: str | None = None + cause: BaseException | None = field(default=None, compare=False, repr=False) + + @property + def absent(self) -> bool: + return self.at is None and self.reason is None + + @property + def unreadable(self) -> bool: + return self.reason is not None + + +def latest_ticks( + keys: Sequence[str], since: datetime | None = None, *, using: str +) -> dict[str, TickRead]: + """ + The newest tick for each schedule key, every key answered. + + With `since`, only ticks at or after it count, which is what lets an + index seek to them; a key whose ticks are all older is `absent`. Without + it, the newest of all, which a page of a few keys can afford and which + does not take a key whose only tick is unreadable and very old (a MySQL + zero date, PostgreSQL's -infinity) for one with no history. + + One grouped statement for as many keys as the connection takes + parameters for, with the newest tick fetched as text and decoded per + key, so one key's unreadable tick is that key's answer and the others + still read. A statement that fails as a whole, on a value the driver + cannot hand over, is tried again with half the keys, down to one, while + the connection is known to take another statement (`can_read_on`). + + The answer for a key whose newest tick does not read carries no `pk`. + Which row it is matters to the message that reports the skip, which is + written at most once a minute for a key, and the lookup is made then + (`newest_tick_pk`) rather than in every pass that skips the key. + """ + wanted = list(dict.fromkeys(keys)) + limit = connections[using].features.max_query_params + # One parameter a key, and one for `since`. + step = max(limit - 1, 1) if limit else max(len(wanted), 1) + found: dict[str, TickRead] = {} + for start in range(0, len(wanted), step): + found.update(_latest(wanted[start : start + step], since, using)) + return {key: found[key] for key in wanted} + + +def _ticks(using: str, since: datetime | None) -> QuerySet[OxScheduleTick]: + """The tick log on `using`, from `since` on where one is given.""" + ticks = OxScheduleTick.objects.using(using) + return ticks if since is None else ticks.filter(scheduled_for__gte=since) + + +def _latest(keys: list[str], since: datetime | None, using: str) -> dict[str, TickRead]: + """One slice of keys, or its halves when it does not read as a whole.""" + try: + fetched = { + row["schedule_name"]: row["latest"] + for row in _ticks(using, since) + .filter(schedule_name__in=keys) + .values("schedule_name") + .annotate(latest=as_text(Max("scheduled_for"))) + } + except Exception as exc: + if not can_read_on(exc, using=using): + raise + if len(keys) == 1: + key = keys[0] + return {key: TickRead(key, reason=_fetch_failed(exc), cause=exc)} + middle = len(keys) // 2 + return { + **_latest(keys[:middle], since, using), + **_latest(keys[middle:], since, using), + } + answers = {} + for key in keys: + if key not in fetched: + answers[key] = TickRead(key) + continue + try: + at = decode_datetime(fetched[key], using) + except UnreadableValue as exc: + answers[key] = TickRead(key, reason=f"scheduled_for {exc}", cause=exc) + else: + answers[key] = TickRead(key, at=at) + return answers + + +def newest_tick_pk( + key: str, since: datetime | None = None, *, using: str +) -> int | None: + """ + The primary key of the tick row the log holds newest for `key`, from + `since` on where one is given, or None when there is none or the read + that finds it fails on a value (`can_read_on`). + + One statement, for a message that names the row. It does not decide + anything: a caller that cannot have the key goes on without it. An + error the database raised is raised as it is. + """ + try: + return ( + _ticks(using, since) + .filter(schedule_name=key) + .order_by("-scheduled_for") + .values_list("pk", flat=True) + .first() + ) + except Exception as exc: + if not can_read_on(exc, using=using): + raise + return None + + +def earliest_tick( + key: str, *, using: str, exclude: Collection[int] = (), lock: bool = False +) -> TickRead: + """ + The earliest tick recorded for `key`, leaving out the rows in `exclude`. + + The read a first sighting is decided by, with its order and exclusions, + and its tick fetched as text and decoded. An unreadable earliest tick is + `unreadable`, never `absent`: taken for no history, it would make the + schedule anchor again over a tick that already has one. + + `lock` makes it a locking read where the database has one, skipping + rows another transaction holds where it can: the current read the + first-sighting latch repeats once it holds the latch. It is one + statement in the caller's transaction, with no savepoint around it, so + the locks it takes and the ones the caller already holds stay held. + + A read the database refuses inside that transaction is raised as it is + (`can_read_on`), for the caller to roll back. It is not answered as a + tick that does not read. + """ + query = OxScheduleTick.objects.using(using).filter(schedule_name=key) + if exclude: + query = query.exclude(pk__in=list(exclude)) + query = query.order_by("scheduled_for") + features = connections[using].features + if lock and features.has_select_for_update: + query = query.select_for_update( + skip_locked=features.has_select_for_update_skip_locked + ) + try: + fetched = list( + query.annotate( + ox_stored_scheduled_for=as_text("scheduled_for") + ).values_list("pk", "ox_stored_scheduled_for")[:1] + ) + except Exception as exc: + if not can_read_on(exc, using=using): + raise + return TickRead(key, reason=_fetch_failed(exc), cause=exc) + if not fetched: + return TickRead(key) + pk, text = fetched[0] + try: + at = decode_datetime(text, using) + except UnreadableValue as exc: + return TickRead(key, pk=pk, reason=f"scheduled_for {exc}", cause=exc) + return TickRead(key, at=at, pk=pk) diff --git a/src/django_ox/exceptions.py b/src/django_ox/exceptions.py index 8449ee4..de570fe 100644 --- a/src/django_ox/exceptions.py +++ b/src/django_ox/exceptions.py @@ -1,3 +1,8 @@ +from collections.abc import Sequence +from functools import partial +from typing import Any + + class TaskAbandoned(Exception): """ Recorded against a task whose worker stopped renewing its lease with no @@ -35,3 +40,58 @@ class TaskTimeout(TimeoutError): def __init__(self, message: str = "", *, timeout: float | None = None) -> None: super().__init__(message) self.timeout = timeout + + +class StoredValueUnreadable(Exception): + """ + A row django-ox stores holds a value that cannot be read back. + + Only a value written around django-ox's own write functions gets there: + by SQL, a fixture or a data migration. PostgreSQL keeps a timestamp past + year 9999 and 'infinity', SQLite keeps text in an integer column and a + date that does not exist, MySQL keeps a zero date outside strict mode. + Django's own read of such a row raises a different exception on each + database, and on SQLite can read some of them as something else without + raising at all, so django-ox reads these rows itself and raises this + where it cannot go on without the value. + + ``alias`` is the database the row was read from, ``model`` the model's + label, ``pk`` the row's primary key, ``fields`` the names of the fields + that did not read (empty when the read could not say which), and + ``reason`` what was wrong with them, safe to print. The exception the + read itself raised, where there was one, is the ``__cause__``. + """ + + def __init__( + self, + *, + alias: str = "", + model: str = "", + pk: Any = None, + fields: Sequence[str] = (), + reason: str = "", + ) -> None: + self.alias = alias + self.model = model + self.pk = pk + self.fields = tuple(fields) + self.reason = reason + super().__init__( + f"The {model} row with primary key {pk!r} in database {alias!r} " + f"holds a value that cannot be read: {reason}" + ) + + def __reduce__(self) -> tuple[Any, ...]: + # The arguments are keyword-only, which the default reduction, a + # call with self.args, cannot pass. + return ( + partial( + type(self), + alias=self.alias, + model=self.model, + pk=self.pk, + fields=self.fields, + reason=self.reason, + ), + (), + ) diff --git a/src/django_ox/models.py b/src/django_ox/models.py index edc515b..45229a3 100644 --- a/src/django_ox/models.py +++ b/src/django_ox/models.py @@ -319,7 +319,24 @@ class Meta: ] def __str__(self) -> str: - return self.name + """ + The schedule's name, or "pk N" where the name cannot be had as text: + bytes SQLite kept in the column, or a name the database holds that + the driver cannot decode. For display only. The fallback is never + assigned to the field and is no name to look the schedule up by. + """ + from ._stored_read import is_unreadable_value + + try: + name = self.name + except Exception as exc: + using = router.db_for_read(type(self), instance=self) + if not is_unreadable_value(exc, using=using): + raise + return f"pk {self.pk}" + if not isinstance(name, str): + return f"pk {self.pk}" + return name def clean(self) -> None: """ diff --git a/tests/test_stored_read.py b/tests/test_stored_read.py new file mode 100644 index 0000000..5779149 --- /dev/null +++ b/tests/test_stored_read.py @@ -0,0 +1,2680 @@ +""" +Reading stored schedules and the tick log when a value in them does not read. + +Django's own read of a stored value it cannot convert raises in the middle +of a result, and takes every row of that result with it: PostgreSQL keeps +year 10000, 'infinity' and dates before year 1, SQLite keeps a date that +does not exist and text that is not UTF-8, MySQL keeps zero dates. On SQLite +some values read as something else instead, with nothing raised: a start of +'banana' reads as None, an `enabled` of 2 reads as False, text in an integer +column reads as text. + +django_ox._stored_read fetches the values whose reading can fail or mislead +as the database holds them and decodes each on its own. A value Django reads +must decode to exactly what Django reads; one Django raises on or misreads +must be reported by field, beside healthy rows and keys that still read. + +Every value below is written by SQL, as only SQL can write it, in the form +the engine that keeps it keeps it, and the test first shows Django's own +read raising on it or misreading it. A test for another engine's value +skips and says why. + +No read takes a savepoint. One that fails in this process, on a value the +driver cannot hand over, is followed by narrower reads, in the same +transaction where there is one. One the database itself refuses is raised +as it is, with nothing read after it, inside a transaction and with none +open: its error is the database's, and no row or key is named for it. +""" + +import math +import pickle +import random +import sqlite3 +import threading +import time +from datetime import UTC, datetime, timedelta +from zoneinfo import ZoneInfo + +import pytest +from django.core.exceptions import FieldDoesNotExist +from django.db import ( + DatabaseError, + DataError, + OperationalError, + connection, + connections, + transaction, +) +from django.db.models import Max +from django.db.transaction import TransactionManagementError +from django.test.utils import CaptureQueriesContext +from django.utils import timezone + +from django_ox import _stored_read +from django_ox._stored_read import ( + TickRead, + UnreadableRow, + UnreadableValue, + as_text, + can_read_on, + decode_boolean, + decode_datetime, + decode_integer, + earliest_tick, + is_unreadable_value, + latest_ticks, + lock_schedule, + newest_tick_pk, + read_schedules, +) +from django_ox.exceptions import StoredValueUnreadable +from django_ox.models import OxSchedule, OxScheduleTick + +from .conftest import wait_for +from .dead_connection_tasks import from_another_connection +from .isolation import kill + +SCHEDULES = "django_ox_oxschedule" +TICKS = "django_ox_oxscheduletick" +ALIAS = "default" + + +def a_row(name, **over): + """A healthy stored schedule, written by the ORM with no validation.""" + now = timezone.now() + fields = { + "name": name, + "task_key": "report", + "trigger": "interval", + "every_seconds": 60, + "arguments": {"label": name}, + "start_time": now - timedelta(minutes=10), + "created_at": now, + "updated_at": now, + } + fields.update(over) + return OxSchedule.objects.create(**fields) + + +def a_tick(key, scheduled_for): + return OxScheduleTick.objects.create( + schedule_name=key, scheduled_for=scheduled_for, created_at=scheduled_for + ) + + +def many_rows(count): + """`count` healthy stored schedules, written in one go; their keys in order.""" + now = timezone.now() + OxSchedule.objects.bulk_create( + OxSchedule( + name=f"row-{i}", + task_key="report", + trigger="interval", + every_seconds=60, + arguments={"label": f"row-{i}"}, + start_time=now - timedelta(minutes=10), + created_at=now, + updated_at=now, + ) + for i in range(count) + ) + return list(OxSchedule.objects.order_by("pk").values_list("pk", flat=True)) + + +def by_sql(statement, params=()): + """ + Run one statement the way only SQL can write what it writes. + + MySQL refuses a zero date and a NULL in a NOT NULL column in the strict + mode the suite runs in, so its session is let off for the statement and + put back. + """ + with connection.cursor() as cursor: + if connection.vendor != "mysql": + cursor.execute(statement, params) + return + cursor.execute("SELECT @@SESSION.sql_mode") + (mode,) = cursor.fetchone() + cursor.execute("SET SESSION sql_mode = ''") + try: + cursor.execute(statement, params) + finally: + cursor.execute("SET SESSION sql_mode = %s", [mode]) + + +def set_column(pk, column, literal): + by_sql(f"UPDATE {SCHEDULES} SET {column} = {literal} WHERE id = {int(pk)}") # noqa: S608 + + +def insert_tick(key, literal): + """A tick row whose scheduled_for is `literal`, as SQL; its primary key.""" + created = { + "postgresql": "now()", + "mysql": "NOW(6)", + }.get(connection.vendor, "'2026-10-05 00:00:00'") + by_sql( + f"INSERT INTO {TICKS} (schedule_name, scheduled_for, task_id, created_at) " # noqa: S608 + f"VALUES (%s, {literal}, NULL, {created})", + [key], + ) + return ( + OxScheduleTick.objects.filter(schedule_name=key) + .order_by("-pk") + .values_list("pk", flat=True) + .first() + ) + + +def only_on(vendor, why): + if connection.vendor != vendor: + pytest.skip(f"a {vendor} value: {why}; this run is on {connection.vendor}") + + +def orm_read(pk): + """Django's own read of a stored row: the row, or the exception it raised.""" + try: + with transaction.atomic(): + return OxSchedule.objects.get(pk=pk) + except Exception as exc: # recorded, and asserted on by the caller + return exc + + +def misread(found): + """ + Django's own read of a date and time raised, or gave something else: + None, which reads as no value at all, or the stored text. Which of the + two depends on the engine and on USE_TZ. + """ + if isinstance(found, datetime) and found.replace(tzinfo=None) in ( + datetime.min, + datetime.max, + ): + # What psycopg2 hands over for PostgreSQL's '-infinity' and 'infinity'. + return True + return isinstance(found, Exception) or not isinstance(found, datetime) + + +def same(left, right): + """Equal, and for a date and time the same zone or none, and the same fold.""" + if isinstance(left, datetime) or isinstance(right, datetime): + return ( + isinstance(left, datetime) + and isinstance(right, datetime) + and left.replace(tzinfo=None) == right.replace(tzinfo=None) + and left.tzinfo == right.tzinfo + and left.utcoffset() == right.utcoffset() + and left.fold == right.fold + ) + return type(left) is type(right) and left == right + + +def assert_reads_as_django_does(read, pk): + assert isinstance(read, OxSchedule), read + expected = OxSchedule.objects.get(pk=pk) + for model_field in OxSchedule._meta.concrete_fields: + name = model_field.attname + assert same(getattr(read, name), getattr(expected, name)), name + assert read._state.db == ALIAS + assert not read._state.adding + + +def assert_all_read_as_django_does(found, leaving_out=()): + """Every stored row is in `found` as Django's own read of the table gives it.""" + expected = {row.pk: row for row in OxSchedule.objects.exclude(pk__in=leaving_out)} + assert found.keys() - set(leaving_out) == expected.keys() + for pk, row in expected.items(): + assert isinstance(found[pk], OxSchedule), found[pk] + for model_field in OxSchedule._meta.concrete_fields: + name = model_field.attname + assert same(getattr(found[pk], name), getattr(row, name)), (pk, name) + + +class Reads: + """connection.execute_wrapper() that keeps each SELECT on `table` as sent.""" + + def __init__(self, table): + self.table = table + self.sent = [] + + def __call__(self, execute, sql, params, many, context): + if self.table in sql and sql.lstrip().upper().startswith("SELECT"): + self.sent.append((sql, tuple(params or ()))) + return execute(sql, params, many, context) + + @property + def parameters(self): + """How many parameters each statement carried, in the order sent.""" + return [len(params) for _sql, params in self.sent] + + +@pytest.fixture +def parameter_limit(monkeypatch): + """ + Lowers how many parameters this connection takes in a statement. + + On the features class: SQLite's is a property that reads the live + connection, and PostgreSQL's is cached on the instance, which is cleared + so that the class's is what is read. SQLite's own limit is lowered with + it, so a statement over the limit is refused rather than only counted. + """ + lowered = [] + + def lower(limit): + monkeypatch.setattr(type(connection.features), "max_query_params", limit) + connection.features.__dict__.pop("max_query_params", None) + if connection.vendor == "sqlite": + raw = connection.connection + lowered.append((raw, raw.getlimit(sqlite3.SQLITE_LIMIT_VARIABLE_NUMBER))) + raw.setlimit(sqlite3.SQLITE_LIMIT_VARIABLE_NUMBER, limit) + + yield lower + for raw, before in lowered: + raw.setlimit(sqlite3.SQLITE_LIMIT_VARIABLE_NUMBER, before) + + +def selects(queries, table): + return [ + q["sql"] + for q in queries.captured_queries + if q["sql"].lstrip().upper().startswith("SELECT") and table in q["sql"] + ] + + +def columns_read(statement): + """The schedule columns a SELECT takes, by name.""" + taken = statement.split(" FROM ")[0] + return { + model_field.column + for model_field in OxSchedule._meta.concrete_fields + if connection.ops.quote_name(model_field.column) in taken + } + + +def savepoints(queries): + """The statements among `queries` that make, release or go back to a savepoint.""" + return [ + q["sql"] + for q in queries.captured_queries + if q["sql"] + .lstrip() + .upper() + .startswith(("SAVEPOINT", "RELEASE SAVEPOINT", "ROLLBACK TO")) + ] + + +#: Which PostgreSQL driver Django is running on. psycopg 3 raises a DataError +#: for a timestamp it cannot load; psycopg2 raises ValueError, and loads +#: 'infinity' and '-infinity' as the largest and smallest datetime. +USES_PSYCOPG_3 = connection.vendor != "postgresql" or "psycopg2" not in ( + connection.Database.__name__ +) + +#: The class Django gives an error the server raised on a value it refuses to +#: convert: a DataError, except through mysqlclient, which raises MySQL's 1367 +#: as an OperationalError. +REFUSED = ( + OperationalError + if connection.vendor == "mysql" and connection.Database.__name__ == "MySQLdb" + else DataError +) + +#: Text SQLite keeps that its driver cannot decode as UTF-8. +NOT_UTF_8 = "CAST(X'62616480FF' AS TEXT)" + + +class Refusing: + """ + connection.execute_wrapper() for a database that refuses one read. + + The `nth` SELECT that mentions `table` is replaced by a statement the + server itself raises a DataError on, so the session is left as a real + refusal leaves it: on PostgreSQL the transaction takes nothing more. + SQLite has no such statement, so there the error is raised in the + statement's place. Every statement sent after the refusal is kept in + `after`. + """ + + STATEMENT = {"postgresql": "SELECT 1/0", "mysql": "SELECT 1e400"} + #: What each database says, which is not what it says of a transaction + #: that has already failed. + SAYS = { + "postgresql": "division by zero", + "mysql": "Illegal double", + "sqlite": "string or blob too big", + } + + def __init__(self, table, nth=1): + self.table = table + self.nth = nth + self.refused = None + self.after = [] + + def __call__(self, execute, sql, params, many, context): + if self.refused is not None: + self.after.append(sql) + return execute(sql, params, many, context) + if self.table in sql and sql.lstrip().upper().startswith("SELECT"): + self.nth -= 1 + if not self.nth: + self.refused = sql + statement = self.STATEMENT.get(connection.vendor) + if statement is None: + raise DataError(self.SAYS["sqlite"]) + return execute(statement, None, many, context) + return execute(sql, params, many, context) + + def assert_raised_as_it_was(self, caught): + """The database's own error reached the caller, and nothing followed it.""" + assert self.refused is not None, "the read was never sent" + assert type(caught.value) is REFUSED + assert self.SAYS[connection.vendor] in str(caught.value) + assert self.after == [] + + +class Failing: + """ + connection.execute_wrapper() for a read that fails in this process, on + one value. + + Every SELECT on `table` that takes `column` and has `having` among its + parameters fails as a read does when a value in its answer cannot be + turned into Python: the database answers, and the failure comes after. + On PostgreSQL psycopg is sent a timestamp it cannot load, and on + SQLite the driver is sent text that is not UTF-8, so each raises its + own error for real. MySQL's driver hands every value over, and what + fails there is Django's converter on a zero date, whose error is + raised in the statement's place. The statements that failed are kept + in `failed`. + """ + + STATEMENT = { + "postgresql": "SELECT '10000-01-01 00:00:00+00'::timestamptz", + "sqlite": f"SELECT {NOT_UTF_8}", + } + #: The class of what each raises: the driver's own on PostgreSQL and + #: SQLite, and on MySQL what the converter raises. + RAISES = { + "postgresql": DataError if USES_PSYCOPG_3 else ValueError, + "sqlite": OperationalError, + "mysql": AttributeError, + } + + def __init__(self, table, column, having): + self.table = table + self.column = connection.ops.quote_name(column) + self.having = having + self.failed = [] + + def __call__(self, execute, sql, params, many, context): + taken = sql.split(" FROM ")[0] + if ( + self.table in sql + and sql.lstrip().upper().startswith("SELECT") + and self.column in taken + and self.having in (params or ()) + ): + self.failed.append(sql) + statement = self.STATEMENT.get(connection.vendor) + if statement is None: + raise AttributeError("'str' object has no attribute 'utcoffset'") + return execute(statement, None, many, context) + return execute(sql, params, many, context) + + +def raised_by(statement): + """ + What one statement raises on this connection, sent and fetched as a + read is. In a block of its own, so that a transaction around the test + outlives an error the server raised. + """ + try: + with transaction.atomic(), connection.cursor() as cursor: + cursor.execute(statement) + cursor.fetchall() + except Exception as exc: # handed to the caller, which asserts on it + return exc + raise AssertionError(f"{statement} raised nothing") + + +def refused_by_sqlite(): + """ + A DataError SQLite itself raises, for the tests that need the real + thing: the longest value the connection takes is lowered, a longer one + is asked for, and the limit is put back. + """ + raw = connection.connection + before = raw.getlimit(sqlite3.SQLITE_LIMIT_LENGTH) + raw.setlimit(sqlite3.SQLITE_LIMIT_LENGTH, 1000) + try: + return raised_by("SELECT zeroblob(2000)") + finally: + raw.setlimit(sqlite3.SQLITE_LIMIT_LENGTH, before) + + +# -- what Django's own read does with each value -------------------------- + +RAISES = "raises" + + +def raises_or_text(): + # PyMySQL hands a zero date over as text, which Django's converter, with + # USE_TZ on, cannot give a zone to. With it off, the text is the value. + from django.conf import settings + + return RAISES if settings.USE_TZ else str + + +#: (engine, column, SQL literal, how Django's read takes it, id). The value +#: each engine keeps that Django's read raises on or misreads. +UNREADABLE_ROW_VALUES = [ + pytest.param( + "sqlite", "end_time", "'10000-01-01 00:00:00'", None, id="sqlite-end-year-10000" + ), + pytest.param( + "sqlite", + "start_time", + "'10000-01-01 00:00:00'", + None, + id="sqlite-start-year-10000", + ), + pytest.param("sqlite", "start_time", "'banana'", None, id="sqlite-start-banana"), + pytest.param("sqlite", "end_time", "'banana'", None, id="sqlite-end-banana"), + pytest.param( + "sqlite", + "end_time", + "'2026-02-30 00:00:00'", + RAISES, + id="sqlite-end-does-not-exist", + ), + pytest.param( + "sqlite", + "created_at", + "'2026-02-30 00:00:00'", + RAISES, + id="sqlite-created-does-not-exist", + ), + pytest.param("sqlite", "updated_at", "'banana'", None, id="sqlite-updated-banana"), + pytest.param("sqlite", "start_time", "2.5", None, id="sqlite-start-a-fraction"), + pytest.param("sqlite", "every_seconds", "'abc'", "abc", id="sqlite-interval-text"), + pytest.param("sqlite", "phase_seconds", "'abc'", "abc", id="sqlite-phase-text"), + pytest.param( + "sqlite", "starting_deadline_seconds", "'abc'", "abc", id="sqlite-deadline-text" + ), + pytest.param( + "sqlite", "boundary_generation", "'abc'", "abc", id="sqlite-generation-text" + ), + pytest.param( + "sqlite", "every_seconds", "60.5", 60.5, id="sqlite-interval-fraction" + ), + pytest.param("sqlite", "enabled", "2", False, id="sqlite-enabled-2"), + pytest.param("sqlite", "enabled", "'abc'", False, id="sqlite-enabled-text"), + pytest.param( + "sqlite", + "name", + NOT_UTF_8, + RAISES, + id="sqlite-name-not-utf-8", + ), + pytest.param( + "sqlite", "name", "X'626c6f622d6e616d65'", b"blob-name", id="sqlite-name-blob" + ), + pytest.param( + "sqlite", "task_key", "X'626c6f62'", b"blob", id="sqlite-task-key-blob" + ), + pytest.param( + "sqlite", "boundary_for", "X'626c6f62'", b"blob", id="sqlite-boundary-for-blob" + ), + pytest.param( + "postgresql", + "end_time", + "'10000-01-01 00:00:00+00'", + RAISES, + id="postgresql-end-year-10000", + ), + pytest.param( + "postgresql", + "start_time", + "'10000-01-01 00:00:00+00'", + RAISES, + id="postgresql-start-year-10000", + ), + pytest.param( + "postgresql", + "updated_at", + "'10000-01-01 00:00:00+00'", + RAISES, + id="postgresql-updated-year-10000", + ), + pytest.param( + "postgresql", "end_time", "'infinity'", RAISES, id="postgresql-end-infinity" + ), + pytest.param( + "postgresql", + "start_time", + "'-infinity'", + RAISES, + id="postgresql-start-minus-infinity", + ), + pytest.param( + "postgresql", + "start_time", + "'0001-01-01 00:00:00+00 BC'", + RAISES, + id="postgresql-start-bc", + ), + pytest.param( + "mysql", + "end_time", + "'0000-00-00 00:00:00'", + raises_or_text, + id="mysql-end-zero-date", + ), + pytest.param( + "mysql", + "start_time", + "'2026-00-10 00:00:00'", + raises_or_text, + id="mysql-start-zero-month", + ), + pytest.param( + "mysql", + "start_time", + "NULL", + raises_or_text, + id="mysql-start-null-kept-as-zero", + ), + pytest.param( + "mysql", + "created_at", + "'0000-00-00 00:00:00'", + raises_or_text, + id="mysql-created-zero-date", + ), + pytest.param("mysql", "enabled", "2", 2, id="mysql-enabled-2"), +] + + +#: Values every engine keeps that read, and that validation, not the read, +#: has to refuse, as SET clauses. The reader hands them over as Django does. +READABLE_BUT_INVALID = [ + pytest.param( + "{trigger} = 'cron', cron = 'banana', every_seconds = NULL", id="cron-text" + ), + pytest.param("task_key = 'nope.missing'", id="unregistered-key"), + pytest.param("arguments = '[1, 2]'", id="arguments-not-a-mapping"), + pytest.param("every_seconds = 0", id="interval-zero"), + pytest.param("end_time = NULL", id="no-end"), + pytest.param("enabled = FALSE", id="disabled"), +] + + +#: One value per engine for the tests that only need one. +ONE_PER_ENGINE = { + "sqlite": ("end_time", "'2026-02-30 00:00:00'"), + "postgresql": ("end_time", "'10000-01-01 00:00:00+00'"), + "mysql": ("end_time", "'0000-00-00 00:00:00'"), +} + + +def corrupt(pk): + column, literal = ONE_PER_ENGINE[connection.vendor] + set_column(pk, column, literal) + return column + + +def show_django_misreads(pk, column, expected): + """ + Django's own read of the row raises, or gives the stored value as + `expected`: the evidence that this value is one the reader must catch. + """ + if callable(expected) and expected is not str: + expected = expected() + found = orm_read(pk) + if expected is RAISES: + # Or, on psycopg2, an 'infinity' read as the largest datetime. + assert isinstance(found, Exception) or misread(getattr(found, column)), getattr( + found, column + ) + return + assert isinstance(found, OxSchedule), found + if expected is str: + assert isinstance(getattr(found, column), str) + else: + assert same(getattr(found, column), expected), getattr(found, column) + + +@pytest.mark.django_db +class TestAStoredRowThatDoesNotRead: + @pytest.mark.parametrize( + ("vendor", "column", "literal", "django"), UNREADABLE_ROW_VALUES + ) + def test_is_named_by_field_and_the_rows_beside_it_read( + self, vendor, column, literal, django + ): + only_on(vendor, "kept by that engine only") + before, bad, after = a_row("before"), a_row("bad"), a_row("after") + set_column(bad.pk, column, literal) + show_django_misreads(bad.pk, column, django) + + found = read_schedules([before.pk, bad.pk, after.pk], using=ALIAS) + + assert list(found) == [before.pk, bad.pk, after.pk] + assert_reads_as_django_does(found[before.pk], before.pk) + assert_reads_as_django_does(found[after.pk], after.pk) + unreadable = found[bad.pk] + assert isinstance(unreadable, UnreadableRow) + assert unreadable.pk == bad.pk + assert unreadable.alias == ALIAS + assert unreadable.fields == (column,) + assert unreadable.reason.startswith(f"{column} ") + if column != "name": + # Decoded on its own, not found by a read that failed whole. + assert unreadable.reason.startswith(f"{column} holds ") + assert isinstance(unreadable.cause, UnreadableValue) + assert column not in unreadable.values + assert unreadable.values["id"] == bad.pk + if column != "name": + assert unreadable.values["name"] == "bad" + if column not in ("name", "task_key"): + assert unreadable.values["task_key"] == "report" + + @pytest.mark.parametrize( + ("vendor", "column", "literal", "django"), UNREADABLE_ROW_VALUES + ) + def test_raises_stored_value_unreadable_with_its_cause( + self, vendor, column, literal, django + ): + only_on(vendor, "kept by that engine only") + bad = a_row("bad") + set_column(bad.pk, column, literal) + + unreadable = read_schedules([bad.pk], using=ALIAS)[bad.pk] + with pytest.raises(StoredValueUnreadable) as caught: + raise unreadable.exception() + + exc = caught.value + assert (exc.alias, exc.model, exc.pk) == (ALIAS, "django_ox.OxSchedule", bad.pk) + assert exc.fields == (column,) + assert exc.reason == unreadable.reason + assert exc.__cause__ is unreadable.cause + assert isinstance(exc.__cause__, Exception) + assert not isinstance(exc, DatabaseError) + assert str(exc) == ( + f"The django_ox.OxSchedule row with primary key {bad.pk!r} in database " + f"'default' holds a value that cannot be read: {exc.reason}" + ) + + @pytest.mark.parametrize( + ("vendor", "column", "literal", "django"), UNREADABLE_ROW_VALUES + ) + def test_gives_an_instance_of_what_did_read(self, vendor, column, literal, django): + only_on(vendor, "kept by that engine only") + bad = a_row("bad") + set_column(bad.pk, column, literal) + + instance = read_schedules([bad.pk], using=ALIAS)[bad.pk].deferred_instance() + + assert instance.pk == bad.pk + assert instance._state.db == ALIAS + assert instance.get_deferred_fields() == {column} + + @pytest.mark.parametrize("assignments", READABLE_BUT_INVALID) + def test_a_value_that_reads_is_handed_over_as_django_reads_it(self, assignments): + row = a_row("row", end_time=timezone.now() + timedelta(days=1)) + clause = assignments.format(trigger=connection.ops.quote_name("trigger")) + by_sql(f"UPDATE {SCHEDULES} SET {clause} WHERE id = %s", [row.pk]) # noqa: S608 + + assert_reads_as_django_does( + read_schedules([row.pk], using=ALIAS)[row.pk], row.pk + ) + + def test_a_huge_interval_reads_and_is_left_to_validation(self): + only_on("sqlite", "only SQLite's integer column holds 2**62") + row = a_row("row") + set_column(row.pk, "every_seconds", str(2**62)) + + assert_reads_as_django_does( + read_schedules([row.pk], using=ALIAS)[row.pk], row.pk + ) + + def test_a_row_that_is_gone_is_left_out(self): + row = a_row("row") + gone = a_row("gone") + gone_pk = gone.pk + gone.delete() + + assert list(read_schedules([gone_pk, row.pk], using=ALIAS)) == [row.pk] + assert read_schedules([], using=ALIAS) == {} + + def test_a_queryset_is_read_for_its_keys(self): + rows = [a_row(f"row-{i}") for i in range(3)] + bad = a_row("bad") + corrupt(bad.pk) + + found = read_schedules(OxSchedule.objects.order_by("-pk"), using=ALIAS) + + assert list(found) == [bad.pk, *[r.pk for r in reversed(rows)]] + assert isinstance(found[bad.pk], UnreadableRow) + for row in rows: + assert_reads_as_django_does(found[row.pk], row.pk) + + +# -- a count of boundary writes that cannot be raised ----------------------- + +#: The most the count of boundary writes can be on each engine, written out +#: here so that what the reader goes by, Django's description of the column, +#: is checked against the number and against the database itself. +LARGEST_COUNT = { + "postgresql": 2147483647, + "mysql": 4294967295, + "sqlite": 9223372036854775807, +} + +COUNT = "boundary_generation" + + +def largest_count(): + return LARGEST_COUNT[connection.vendor] + + +def a_full_row(name="full"): + """A healthy stored schedule whose count of boundary writes SQL set to the most.""" + row = a_row(name) + set_column(row.pk, COUNT, str(largest_count())) + return row + + +def says_the_count_is_full(unreadable): + """The row is reported for its count alone, with the count's own reason.""" + assert isinstance(unreadable, UnreadableRow), unreadable + assert unreadable.fields == (COUNT,) + assert unreadable.reason == ( + f"{COUNT} holds {largest_count()}, the most its column can hold, so the " + "count cannot be raised" + ) + assert isinstance(unreadable.cause, UnreadableValue) + return True + + +@pytest.mark.django_db +class TestACountOfBoundaryWritesAtItsMaximum: + """ + The count of boundary writes is a whole number, and at the most its + column can hold it still reads as one: Django's own read hands it over. + What it cannot do is go up, and every write that moves a schedule's + boundary adds one to it. PostgreSQL and MySQL refuse that write, each + time it is tried, and SQLite keeps the sum as a float. Only SQL can put + the count there. + + The reader answers such a row as one that does not read, by its count, + before any write is tried. The maximum is the one Django gives for the + column on the connection, not a number kept in the reader. + """ + + def test_django_describes_the_column_as_each_database_holds_it(self): + field = OxSchedule._meta.get_field(COUNT) + + _, highest = connection.ops.integer_field_range(field.get_internal_type()) + + assert highest == largest_count() + + def test_django_reads_it_and_the_database_does_not_add_one_to_it(self): + row = a_full_row() + assert OxSchedule.objects.get(pk=row.pk).boundary_generation == largest_count() + adds_one = f"UPDATE {SCHEDULES} SET {COUNT} = {COUNT} + 1 WHERE id = {row.pk}" # noqa: S608 + + if connection.vendor != "sqlite": + refused = raised_by(adds_one) + # The database's own error, which is never put down to a row. + assert type(refused) is DataError + assert not is_unreadable_value(refused, using=ALIAS) + return + by_sql(adds_one) + kept_as = f"SELECT typeof({COUNT}) FROM {SCHEDULES} WHERE id = %s" # noqa: S608 + with connection.cursor() as cursor: + cursor.execute(kept_as, [row.pk]) + assert cursor.fetchone() == ("real",) + # What the sum left is a count that does not read, for its own reason. + left = read_schedules([row.pk], using=ALIAS)[row.pk] + assert left.fields == (COUNT,) + assert left.reason.endswith("which is not a whole number") + + def test_is_named_by_its_field_and_the_rows_beside_it_read(self): + before, full, after = a_row("before"), a_full_row(), a_row("after") + + with CaptureQueriesContext(connection) as queries: + found = read_schedules([before.pk, full.pk, after.pk], using=ALIAS) + + # Decided from what the one statement fetched, with nothing sent for it. + assert len(queries.captured_queries) == 1 + assert list(found) == [before.pk, full.pk, after.pk] + assert_reads_as_django_does(found[before.pk], before.pk) + assert_reads_as_django_does(found[after.pk], after.pk) + unreadable = found[full.pk] + assert says_the_count_is_full(unreadable) + assert (unreadable.pk, unreadable.alias) == (full.pk, ALIAS) + # Every other value of the row is handed over as it read. + every_other = { + model_field.attname for model_field in OxSchedule._meta.concrete_fields + } - {COUNT} + assert set(unreadable.values) == every_other + assert unreadable.values["name"] == "full" + assert unreadable.deferred_instance().get_deferred_fields() == {COUNT} + + def test_raises_stored_value_unreadable_naming_the_count(self): + full = a_full_row() + + unreadable = read_schedules([full.pk], using=ALIAS)[full.pk] + with pytest.raises(StoredValueUnreadable) as caught: + raise unreadable.exception() + + assert caught.value.fields == (COUNT,) + assert caught.value.reason == unreadable.reason + assert caught.value.__cause__ is unreadable.cause + + def test_one_below_it_reads_as_django_reads_it(self): + row = a_row("row") + set_column(row.pk, COUNT, str(largest_count() - 1)) + + found = read_schedules([row.pk], using=ALIAS)[row.pk] + + assert_reads_as_django_does(found, row.pk) + assert found.boundary_generation == largest_count() - 1 + + def test_the_maximum_is_the_one_django_gives_for_the_connection(self, monkeypatch): + at_it, below_it = a_row("at-it"), a_row("below-it") + set_column(at_it.pk, COUNT, "7") + set_column(below_it.pk, COUNT, "6") + asked = [] + + def holds_up_to_seven(internal_type): + asked.append(internal_type) + return (0, 7) + + monkeypatch.setattr(connection.ops, "integer_field_range", holds_up_to_seven) + found = read_schedules([at_it.pk, below_it.pk], using=ALIAS) + + unreadable = found[at_it.pk] + assert isinstance(unreadable, UnreadableRow) + assert unreadable.fields == (COUNT,) + assert unreadable.reason.startswith(f"{COUNT} holds 7, the most its column") + assert isinstance(found[below_it.pk], OxSchedule) + # Asked by the type of the count's own field, once for each row. + field = OxSchedule._meta.get_field(COUNT) + assert asked == [field.get_internal_type()] * 2 + + @pytest.mark.parametrize( + "column", ["every_seconds", "phase_seconds", "starting_deadline_seconds"] + ) + def test_no_other_whole_number_is_held_to_it(self, column): + row = a_row("row") + set_column(row.pk, column, str(largest_count())) + + assert_reads_as_django_does( + read_schedules([row.pk], using=ALIAS)[row.pk], row.pk + ) + + def test_beside_a_value_that_does_not_read_both_are_named(self): + full = a_full_row() + other = corrupt(full.pk) + + unreadable = read_schedules([full.pk], using=ALIAS)[full.pk] + + assert isinstance(unreadable, UnreadableRow) + assert unreadable.fields == (COUNT, other) + assert unreadable.unreadable[COUNT].startswith(f"holds {largest_count()}, ") + + def test_a_read_that_does_not_take_the_count_does_not_meet_it(self): + full = a_full_row() + + without = read_schedules([full.pk], using=ALIAS, fields=["task_key"])[full.pk] + with_it = read_schedules([full.pk], using=ALIAS, fields=["task_key", COUNT])[ + full.pk + ] + + assert isinstance(without, OxSchedule) + assert COUNT in without.get_deferred_fields() + assert without.task_key == "report" + assert says_the_count_is_full(with_it) + assert with_it.values == {"id": full.pk, "task_key": "report"} + + def test_the_locked_read_names_it_too(self): + full = a_full_row() + + with transaction.atomic(): + found = lock_schedule(full.pk, using=ALIAS) + some = lock_schedule(full.pk, using=ALIAS, fields=["enabled", COUNT]) + + assert says_the_count_is_full(found) + assert says_the_count_is_full(some) + assert some.values == {"id": full.pk, "enabled": True} + + +@pytest.mark.django_db(transaction=True) +def test_a_full_count_is_named_in_a_row_read_a_column_at_a_time(): + """ + A row the driver cannot hand over whole is read again a column at a + time, and its count is held to the same rule there. + """ + before, full = a_row("before"), a_full_row() + failing = Failing(SCHEDULES, "name", full.pk) + + with connection.execute_wrapper(failing): + found = read_schedules([before.pk, full.pk], using=ALIAS) + + assert failing.failed, "the read never failed" + assert_reads_as_django_does(found[before.pk], before.pk) + unreadable = found[full.pk] + assert isinstance(unreadable, UnreadableRow) + assert unreadable.fields == ("name", COUNT) + assert unreadable.unreadable["name"].startswith("could not be read: ") + assert unreadable.unreadable[COUNT] == ( + f"holds {largest_count()}, the most its column can hold, so the count " + "cannot be raised" + ) + + +@pytest.mark.django_db +class TestReadingRowsInBatches: + def test_healthy_rows_are_read_five_hundred_a_statement(self): + pks = many_rows(1200) + reads = Reads(SCHEDULES) + + with connection.execute_wrapper(reads): + found = read_schedules(OxSchedule.objects.order_by("pk"), using=ALIAS) + + # The keys, which take no parameter, and then the rows, a key a + # parameter: three statements for twelve hundred of them. + assert reads.parameters == [0, 500, 500, 200] + assert list(found) == pks + assert_all_read_as_django_does(found) + + def test_a_connection_that_takes_fewer_parameters_is_read_in_smaller_batches( + self, parameter_limit + ): + pks = many_rows(250) + parameter_limit(100) + reads = Reads(SCHEDULES) + + with connection.execute_wrapper(reads): + found = read_schedules(pks, using=ALIAS) + + assert reads.parameters == [100, 100, 50] + assert list(found) == pks + assert_all_read_as_django_does(found) + + def test_a_limit_above_five_hundred_leaves_the_batch_at_five_hundred( + self, parameter_limit + ): + pks = many_rows(501) + parameter_limit(999) + reads = Reads(SCHEDULES) + + with connection.execute_wrapper(reads): + found = read_schedules(pks, using=ALIAS) + + assert reads.parameters == [500, 1] + assert list(found) == pks + + @pytest.mark.parametrize("position", ["first", "middle", "last"]) + def test_a_bad_row_anywhere_loses_no_healthy_row(self, position): + rows = [a_row(f"row-{i}") for i in range(9)] + bad = {"first": rows[0], "middle": rows[4], "last": rows[-1]}[position] + column = corrupt(bad.pk) + + with CaptureQueriesContext(connection) as queries: + found = read_schedules([r.pk for r in rows], using=ALIAS) + + # The value is decoded on its own, so the batch still reads in one. + assert len(selects(queries, SCHEDULES)) == 1 + assert found[bad.pk].fields == (column,) + for row in rows: + if row is not bad: + assert_reads_as_django_does(found[row.pk], row.pk) + + @pytest.mark.parametrize("position", ["first", "middle", "last"]) + def test_a_row_the_driver_cannot_hand_over_is_found_by_halving(self, position): + only_on("sqlite", "SQLite keeps text its driver cannot decode as UTF-8") + rows = [a_row(f"row-{i}") for i in range(16)] + bad = {"first": rows[0], "middle": rows[7], "last": rows[-1]}[position] + set_column(bad.pk, "name", NOT_UTF_8) + assert isinstance(orm_read(bad.pk), OperationalError) + + # Inside this test's transaction, and in no savepoint: the failure + # is the driver's, raised once SQLite has answered, and the + # transaction takes the narrower reads as it took the first. + assert connection.in_atomic_block + with CaptureQueriesContext(connection) as queries: + found = read_schedules([r.pk for r in rows], using=ALIAS) + + assert not savepoints(queries) + assert found[bad.pk].fields == ("name",) + assert "UTF-8" in found[bad.pk].reason + assert isinstance(found[bad.pk].cause, OperationalError) + for row in rows: + if row is not bad: + assert_reads_as_django_does(found[row.pk], row.pk) + # Halving 16 rows down to one is 2 * log2(16) + 1 statements, then + # one a column to name the field. + columns = len(OxSchedule._meta.concrete_fields) + assert len(selects(queries, SCHEDULES)) <= 2 * 4 + 1 + columns + + def test_inside_a_transaction_the_transaction_still_works_after(self): + rows = [a_row(f"row-{i}") for i in range(3)] + corrupt(rows[1].pk) + if connection.vendor == "sqlite": + set_column(rows[2].pk, "name", NOT_UTF_8) + + with transaction.atomic(): + with CaptureQueriesContext(connection) as queries: + found = read_schedules([r.pk for r in rows], using=ALIAS) + # A statement after the reads, in the same transaction. + assert OxSchedule.objects.filter(pk=rows[0].pk).update(cron="") == 1 + + assert not savepoints(queries) + assert_reads_as_django_does(found[rows[0].pk], rows[0].pk) + assert isinstance(found[rows[1].pk], UnreadableRow) + if connection.vendor == "sqlite": + assert found[rows[2].pk].fields == ("name",) + + def test_a_read_the_driver_fails_is_narrowed_in_the_same_transaction(self): + """ + psycopg loads a timestamp in this process, and one it cannot load + raises a DataError there, with the statement answered and the + transaction as it was. The session says so, and the batch is read + again in halves. Made to happen by sending a typed read in the + first batch's place: the reads here cast every timestamp to text, + and none of them fails this way by itself. + """ + only_on("postgresql", "psycopg raises a DataError of its own") + rows = [a_row(f"row-{i}") for i in range(4)] + sent = [] + + def a_typed_read_first(execute, sql, params, many, context): + sent.append(sql) + if len(sent) == 1: + return execute( + "SELECT '10000-01-01 00:00:00+00'::timestamptz", None, many, context + ) + return execute(sql, params, many, context) + + with transaction.atomic(): + with connection.execute_wrapper(a_typed_read_first): + found = read_schedules([r.pk for r in rows], using=ALIAS) + assert OxSchedule.objects.filter(pk=rows[0].pk).update(cron="") == 1 + + # The batch, then its two halves. + assert len(sent) == 3 + assert not [sql for sql in sent if "SAVEPOINT" in sql.upper()] + for row in rows: + assert_reads_as_django_does(found[row.pk], row.pk) + + def test_a_locking_read_locks_where_the_database_can(self): + row = a_row("row") + + with transaction.atomic(), CaptureQueriesContext(connection) as queries: + found = read_schedules([row.pk], using=ALIAS, lock=True) + + assert_reads_as_django_does(found[row.pk], row.pk) + (statement,) = selects(queries, SCHEDULES) + assert ("FOR UPDATE" in statement) == connection.features.has_select_for_update + + +@pytest.mark.django_db +class TestReadingSomeOfTheFields: + """ + A caller that names the fields it needs is handed those, and its read + takes no other column: what another column holds neither stops the + read nor is reported by it. + """ + + FIELDS = ("task_key", "enabled", "every_seconds") + EVERY_FIELD = frozenset(f.attname for f in OxSchedule._meta.concrete_fields) + + def test_takes_the_key_and_the_named_columns_alone(self): + row = a_row("row") + + with CaptureQueriesContext(connection) as queries: + found = read_schedules([row.pk], using=ALIAS, fields=self.FIELDS)[row.pk] + + (statement,) = selects(queries, SCHEDULES) + assert columns_read(statement) == {"id", *self.FIELDS} + assert isinstance(found, OxSchedule) + expected = OxSchedule.objects.get(pk=row.pk) + for name in ("id", *self.FIELDS): + assert same(getattr(found, name), getattr(expected, name)), name + assert found.get_deferred_fields() == self.EVERY_FIELD - {"id", *self.FIELDS} + assert found._state.db == ALIAS + assert not found._state.adding + + @pytest.mark.parametrize( + ("vendor", "column", "literal", "django"), UNREADABLE_ROW_VALUES + ) + def test_a_value_in_a_column_it_did_not_name_is_not_its_concern( + self, vendor, column, literal, django + ): + only_on(vendor, "kept by that engine only") + row = a_row("row") + set_column(row.pk, column, literal) + assert isinstance(read_schedules([row.pk], using=ALIAS)[row.pk], UnreadableRow) + + with CaptureQueriesContext(connection) as queries: + found = read_schedules([row.pk], using=ALIAS, fields=["trigger", "cron"]) + + assert isinstance(found[row.pk], OxSchedule) + assert found[row.pk].trigger == "interval" + # One statement, for text SQLite's driver cannot decode as for any + # other value: the column is not taken, so nothing fails on it. + (statement,) = selects(queries, SCHEDULES) + assert columns_read(statement) == {"id", "trigger", "cron"} + + @pytest.mark.parametrize( + ("vendor", "column", "literal", "django"), UNREADABLE_ROW_VALUES + ) + def test_a_value_in_a_column_it_named_is_reported( + self, vendor, column, literal, django + ): + only_on(vendor, "kept by that engine only") + row = a_row("row") + set_column(row.pk, column, literal) + + with CaptureQueriesContext(connection) as queries: + found = read_schedules([row.pk], using=ALIAS, fields=["trigger", column]) + + unreadable = found[row.pk] + assert isinstance(unreadable, UnreadableRow) + assert unreadable.fields == (column,) + # It speaks of the fields that were read, and of no other. + assert set(unreadable.values) == {"id", "trigger"} + instance = unreadable.deferred_instance() + assert instance.get_deferred_fields() == self.EVERY_FIELD - {"id", "trigger"} + statements = selects(queries, SCHEDULES) + if literal == NOT_UTF_8: + # The driver could not hand the row over, so it is read again + # a column at a time: the three the read was for. + assert len(statements) == 1 + 3 + else: + assert len(statements) == 1 + for statement in statements: + assert columns_read(statement) <= {"id", "trigger", column} + + def test_a_name_that_is_not_a_field_is_refused(self): + row = a_row("row") + + with pytest.raises(FieldDoesNotExist): + read_schedules([row.pk], using=ALIAS, fields=["task_kee"]) + + +@pytest.mark.django_db +class TestTheLockedRead: + def test_locks_and_reads_a_row_that_does_not_read(self): + bad = a_row("bad") + column = corrupt(bad.pk) + found = orm_read(bad.pk) + assert isinstance(found, Exception) or misread(found.end_time) + + with transaction.atomic(): + locked = lock_schedule(bad.pk, using=ALIAS) + + assert isinstance(locked, UnreadableRow) + assert locked.fields == (column,) + assert locked.values["name"] == "bad" + + def test_a_row_that_reads_is_read_as_django_reads_it(self): + row = a_row("row") + + with transaction.atomic(): + locked = lock_schedule(row.pk, using=ALIAS) + + assert_reads_as_django_does(locked, row.pk) + + def test_the_read_says_whether_the_row_is_there(self): + """ + Nothing else does: on SQLite the lock is an UPDATE, whose row count + nothing here reads, and a statement sent only to ask would be one + more on every dispatch of a stored schedule. A row that is gone is + left out of the read, which is the answer. + """ + there = a_row("there") + gone = a_row("gone") + gone_pk = gone.pk + gone.delete() + + with transaction.atomic(), CaptureQueriesContext(connection) as queries: + assert lock_schedule(gone_pk, using=ALIAS) is None + found = lock_schedule(there.pk, using=ALIAS) + + assert_reads_as_django_does(found, there.pk) + assert not savepoints(queries) + statements = [ + q["sql"].lstrip().upper() + for q in queries.captured_queries + if SCHEDULES.upper() in q["sql"].upper() + ] + if connection.features.has_select_for_update: + # One statement a row: the read is what locks. + reads = statements + assert len(reads) == 2 + assert all("FOR UPDATE" in read for read in reads) + else: + # The writer first, then the row, and nothing between them. + assert len(statements) == 4 + assert all(write.startswith("UPDATE") for write in statements[::2]) + reads = statements[1::2] + for read in reads: + assert read.startswith("SELECT") and "OX_STORED_START_TIME" in read + assert not read.startswith("SELECT 1") + + def test_reads_only_the_fields_it_is_asked_for(self): + """ + A caller that names the fields it needs takes no other column, so + a value that does not read in a column it did not name neither + stops it nor is reported to it. + """ + row = a_row("row") + beside = corrupt(row.pk) + assert isinstance(read_schedules([row.pk], using=ALIAS)[row.pk], UnreadableRow) + + with transaction.atomic(), CaptureQueriesContext(connection) as queries: + locked = lock_schedule(row.pk, using=ALIAS, fields=["task_key", "enabled"]) + + assert isinstance(locked, OxSchedule) + assert (locked.pk, locked.task_key, locked.enabled) == (row.pk, "report", True) + assert beside in locked.get_deferred_fields() + (read,) = selects(queries, SCHEDULES) + assert columns_read(read) == {"id", "task_key", "enabled"} + assert ("FOR UPDATE" in read) == connection.features.has_select_for_update + + +@pytest.mark.django_db(transaction=True) +def test_the_locked_read_needs_a_transaction(): + row = a_row("row") + + with pytest.raises(TransactionManagementError, match="there is no transaction"): + lock_schedule(row.pk, using=ALIAS) + + +@pytest.mark.django_db(transaction=True) +def test_the_locked_read_holds_off_another_writer(): + """ + A second connection that writes the row while the lock is held waits + for it, here up to a short timeout that it then exceeds. + """ + bad = a_row("bad") + corrupt(bad.pk) + outcome = {} + locked = threading.Event() + finished = threading.Event() + + def writer(): + try: + # Not past a test that failed before it took the lock. + if not locked.wait(15) or "abandoned" in outcome: + return + with connection.cursor() as cursor: + if connection.vendor == "postgresql": + cursor.execute("SET lock_timeout = '300ms'") + elif connection.vendor == "mysql": + cursor.execute("SET SESSION innodb_lock_wait_timeout = 1") + else: + cursor.execute("PRAGMA busy_timeout = 300") + OxSchedule.objects.filter(pk=bad.pk).update(cron="") + outcome["wrote"] = True + except OperationalError as exc: + outcome["refused"] = exc + finally: + connection.close() + finished.set() + + thread = threading.Thread(target=writer) + thread.start() + try: + with transaction.atomic(): + assert isinstance(lock_schedule(bad.pk, using=ALIAS), UnreadableRow) + locked.set() + assert finished.wait(15) + finally: + if not locked.is_set(): + outcome["abandoned"] = True + locked.set() + thread.join(15) + + assert "refused" in outcome, outcome + # Released at the end of the transaction, the row takes the write. + assert OxSchedule.objects.filter(pk=bad.pk).update(cron="") == 1 + + +def seen_waiting_for_the_lock(limit=30.0): + """ + Whether a session came to wait for a lock within `limit` seconds. + + PostgreSQL and MySQL say when one does. They are asked from a connection + of its own: a session inside a transaction is shown the activity it saw + first for as long as the transaction lasts. And no more often than every + tenth of a second, because InnoDB renews its picture of the transactions + only once it has gone that long unread. + + SQLite does not say. Its writer waits on a busy timeout instead, so the + call is given a moment to reach it, and the answer is None. + """ + if connection.vendor == "sqlite": + time.sleep(0.3) + return None + query = { + "postgresql": ( + "SELECT count(*) FROM pg_stat_activity " + "WHERE wait_event_type = 'Lock' AND datname = current_database()" + ), + "mysql": ( + "SELECT count(*) FROM information_schema.innodb_trx " + "WHERE trx_state = 'LOCK WAIT'" + ), + }[connection.vendor] + + def waiting(): + seen = [] + + def count(other): + with other.cursor() as cursor: + cursor.execute(query) + seen.append(cursor.fetchone()[0]) + + from_another_connection(count) + return seen[0] > 0 + + return wait_for(waiting, timeout=limit, interval=0.1) + + +@pytest.mark.django_db(transaction=True) +@pytest.mark.parametrize("fields", [None, ["task_key", "enabled"]], ids=["row", "some"]) +def test_the_locked_read_waits_for_a_writer_and_reads_what_it_committed(fields): + """ + The read that takes the lock is a current read. A transaction that + has already looked at the row waits for the writer holding it, and is + then handed the row as that writer committed it: on MySQL a plain read + would go on answering from what the transaction saw first. The writer + here changes the row only once this read is seen waiting, so what comes + back was written after the read was sent. + + SQLite refuses a transaction that reads and then writes behind another + writer instead of making it wait, so there nothing is read first, and + the wait itself cannot be seen. + """ + row = a_row("row") + holding = threading.Event() + entering = threading.Event() + out = {} + + def writer(): + try: + with transaction.atomic(): + lock_schedule(row.pk, using=ALIAS) + holding.set() + assert entering.wait(30) + out["waited"] = seen_waiting_for_the_lock() + OxSchedule.objects.filter(pk=row.pk).update(task_key="written-since") + except BaseException as exc: # handed to the test, which fails on it + out["raised"] = exc + finally: + connections[ALIAS].close() + + thread = threading.Thread(target=writer) + thread.start() + try: + assert holding.wait(30) + with transaction.atomic(): + if connection.features.has_select_for_update: + first = OxSchedule.objects.values_list("task_key", flat=True) + assert first.get(pk=row.pk) == "report" + entering.set() + locked = lock_schedule(row.pk, using=ALIAS, fields=fields) + finally: + entering.set() + thread.join(60) + + assert not thread.is_alive() + assert "raised" not in out, out + assert out["waited"] is not False, "the read was not seen waiting for the lock" + assert locked.task_key == "written-since" + + +@pytest.mark.django_db(transaction=True) +@pytest.mark.parametrize("in_transaction", [False, True]) +@pytest.mark.parametrize("read", ["rows", "newest", "earliest"]) +def test_a_lost_connection_is_raised_not_put_down_to_a_value(read, in_transaction): + """ + A connection the server ended (SQLite: closed under Django) fails every + read that follows. That is the database failing, and it is raised for the + caller's database handling, never answered as a row or a tick that does + not read. + """ + row = a_row("row") + now = now_ish() + a_tick("k", now) + calls = { + "rows": lambda: read_schedules([row.pk], using=ALIAS), + "newest": lambda: latest_ticks(["k"], now - timedelta(hours=1), using=ALIAS), + "earliest": lambda: earliest_tick("k", using=ALIAS), + } + # Django's class for it, or on SQLite the driver's own where Django + # does not wrap the call: reading the parameter limit, for one. + failed = (DatabaseError, connection.Database.Error) + try: + with pytest.raises(failed) as caught: + if in_transaction: + with transaction.atomic(): + kill(connection) + calls[read]() + else: + kill(connection) + calls[read]() + assert not is_unreadable_value(caught.value, using=ALIAS) + assert not can_read_on(caught.value, using=ALIAS) + finally: + connection.close() + + +# -- a read the database refuses, and one the driver fails ------------------- + + +def the_reads(): + """ + Rows and ticks, and each read over more than one of them, so that a + read tried again in halves would have halves to try. + """ + rows = [a_row(f"row-{i}") for i in range(4)] + now = now_ish() + since = now - timedelta(hours=1) + keys = ["k0", "k1", "k2"] + for key in keys: + a_tick(key, now) + return { + "rows": (SCHEDULES, lambda: read_schedules([r.pk for r in rows], using=ALIAS)), + "newest": (TICKS, lambda: latest_ticks(keys, since, using=ALIAS)), + "earliest": (TICKS, lambda: earliest_tick("k0", using=ALIAS)), + } + + +@pytest.mark.django_db +@pytest.mark.parametrize("read", ["rows", "newest", "earliest"]) +def test_in_a_transaction_a_read_the_database_refuses_is_raised_as_it_was(read): + """ + A statement the server refuses ends a PostgreSQL transaction for every + statement after it. A narrower read sent into it raises "current + transaction is aborted", which is an error about the second read, and + no savepoint was taken to go back to. So the first error is raised, for + whoever owns the transaction to roll it back, and nothing is read after + it. It is never answered as a row or a tick that does not read, on any + database: where the driver cannot say the transaction still stands, + it is not assumed to. + """ + table, call = the_reads()[read] + refusing = Refusing(table) + + with ( + pytest.raises(REFUSED) as caught, + transaction.atomic(), + connection.execute_wrapper(refusing), + ): + call() + + refusing.assert_raised_as_it_was(caught) + # Rolled back by its owner, the connection reads again. + assert OxSchedule.objects.count() == 4 + + +@pytest.mark.django_db +def test_in_a_transaction_a_refused_lookup_of_an_unreadable_ticks_row_is_raised(): + """ + The read that names an unreadable newest tick's row is a statement of + its own, sent after a value did not decode. Refused, it is raised like + any other: the tick's row going unnamed is not worth a statement into a + transaction that may be over. + """ + literal = { + "postgresql": "'infinity'", + "mysql": "'9999-00-00 00:00:00'", + "sqlite": "'banana'", + }[connection.vendor] + now = now_ish() + since = now - timedelta(hours=1) + a_tick("k0", now) + insert_tick("k0", literal) + # The grouped read of the keys, then the lookup of the row. + refusing = Refusing(TICKS, nth=2) + + with ( + pytest.raises(REFUSED) as caught, + transaction.atomic(), + connection.execute_wrapper(refusing), + ): + assert latest_ticks(["k0"], since, using=ALIAS)["k0"].unreadable + newest_tick_pk("k0", since, using=ALIAS) + + refusing.assert_raised_as_it_was(caught) + + +@pytest.mark.django_db +def test_in_a_transaction_a_refused_read_of_one_column_is_raised(): + """ + A row the driver could not hand over is read a column at a time, and + each of those reads is held to the same rule as the first. + """ + only_on("sqlite", "SQLite keeps text its driver cannot decode as UTF-8") + bad = a_row("bad") + set_column(bad.pk, "name", NOT_UTF_8) + # The row whole, which the driver fails, and then its first column. + refusing = Refusing(SCHEDULES, nth=2) + + with ( + pytest.raises(REFUSED) as caught, + transaction.atomic(), + connection.execute_wrapper(refusing), + ): + read_schedules([bad.pk], using=ALIAS) + + refusing.assert_raised_as_it_was(caught) + + +@pytest.mark.django_db(transaction=True) +@pytest.mark.parametrize("read", ["rows", "newest", "earliest"]) +def test_with_no_transaction_open_a_read_the_database_refuses_is_raised_as_it_was(read): + """ + With no transaction open the refused statement was a transaction of its + own, and nothing is left for the next one to run into. A narrower read + is not sent all the same. What the database raised says that it + refused the statement, not that one value in an answer could not be + read, so there is no row or key for halves to find. Its error is + raised as it is, for the caller's handling of a database error, and is + never answered as a row or a tick that does not read. + """ + table, call = the_reads()[read] + refusing = Refusing(table) + + assert connection.get_autocommit() + assert not connection.in_atomic_block + with pytest.raises(REFUSED) as caught, connection.execute_wrapper(refusing): + call() + + refusing.assert_raised_as_it_was(caught) + # The connection reads again. + assert OxSchedule.objects.count() == 4 + + +@pytest.mark.django_db(transaction=True) +def test_with_no_transaction_open_a_read_that_fails_on_one_row_is_narrowed_to_it(): + """ + A failure raised in this process, while a value the database sent was + being turned into Python, is one row's. The batch is read again in + halves until that row is alone, its columns are read one at a time to + name the field, and every other row reads. + """ + rows = [a_row(f"row-{i}") for i in range(4)] + bad = rows[2] + failing = Failing(SCHEDULES, "name", bad.pk) + + assert connection.get_autocommit() + assert not connection.in_atomic_block + with connection.execute_wrapper(failing): + found = read_schedules([r.pk for r in rows], using=ALIAS) + + # The batch, the half the row is in, the row alone, and its one column. + assert len(failing.failed) == 4 + unreadable = found[bad.pk] + assert isinstance(unreadable, UnreadableRow) + assert unreadable.fields == ("name",) + assert unreadable.reason.startswith("name could not be read: ") + assert type(unreadable.cause) is Failing.RAISES[connection.vendor] + for row in rows: + if row is not bad: + assert_reads_as_django_does(found[row.pk], row.pk) + + +@pytest.mark.django_db(transaction=True) +@pytest.mark.parametrize("position", [0, 249, 250, 499]) +def test_a_row_that_fails_in_a_batch_of_five_hundred_is_narrowed_to_itself(position): + """ + A full batch is one statement, and one row of it that fails in this + process fails all five hundred. Read again in halves, the batch comes + down to that row in nine halvings at most: each sends the half the row + is in, which fails, and the half it is not in, which reads. Then the + row's columns, one at a time. Every other row of the batch reads. + """ + pks = many_rows(500) + bad = pks[position] + failing = Failing(SCHEDULES, "name", bad) + reads = Reads(SCHEDULES) + + assert connection.get_autocommit() + assert not connection.in_atomic_block + with connection.execute_wrapper(reads), connection.execute_wrapper(failing): + found = read_schedules(pks, using=ALIAS) + + assert reads.parameters[0] == 500 + halvings = math.ceil(math.log2(500)) + columns = len(OxSchedule._meta.concrete_fields) + # The batch and each half the row was in, and then its one column. + assert halvings + 1 <= len(failing.failed) <= halvings + 2 + assert len(reads.sent) <= 2 * halvings + 1 + columns + assert list(found) == pks + unreadable = found[bad] + assert isinstance(unreadable, UnreadableRow) + assert unreadable.fields == ("name",) + assert unreadable.reason.startswith("name could not be read: ") + assert type(unreadable.cause) is Failing.RAISES[connection.vendor] + assert_all_read_as_django_does(found, leaving_out=[bad]) + + +@pytest.mark.django_db(transaction=True) +def test_with_no_transaction_open_a_read_that_fails_on_one_key_is_narrowed_to_it(): + """ + The same for the newest ticks: the slice of keys is read again in + halves until the key that fails is alone, that key is answered as one + whose tick does not read, with the tick's row named, and the keys + beside it are answered as they stand. The earliest tick is read for + one key, which is its answer. + """ + now = now_ish() + keys = ["k0", "k1", "k2"] + ticks = {key: a_tick(key, now) for key in keys} + failing = Failing(TICKS, "scheduled_for", "k1") + + assert connection.get_autocommit() + assert not connection.in_atomic_block + with connection.execute_wrapper(failing): + answers = latest_ticks(keys, now - timedelta(hours=1), using=ALIAS) + earliest = earliest_tick("k1", using=ALIAS) + + # The slice, the half the key is in, the key alone, and its earliest. + assert len(failing.failed) == 4 + for answer in (answers["k1"], earliest): + assert answer.unreadable + assert answer.reason.startswith("scheduled_for could not be read: ") + assert type(answer.cause) is Failing.RAISES[connection.vendor] + assert answers["k1"].pk is None + assert newest_tick_pk("k1", now - timedelta(hours=1), using=ALIAS) == ticks["k1"].pk + assert answers["k0"] == TickRead("k0", at=now) + assert answers["k2"] == TickRead("k2", at=now) + + +@pytest.mark.django_db(transaction=True) +def test_with_no_transaction_open_text_the_driver_cannot_decode_is_narrowed_down(): + """ + The stored value that fails a whole read by itself: text SQLite keeps + and its driver cannot decode. It is found by halving with no + transaction open as it is inside one. + """ + only_on("sqlite", "SQLite keeps text its driver cannot decode as UTF-8") + rows = [a_row(f"row-{i}") for i in range(4)] + set_column(rows[1].pk, "name", NOT_UTF_8) + now = now_ish() + a_tick("k0", now) + a_tick("k1", now) + pk = insert_tick("k1", NOT_UTF_8) + + assert connection.get_autocommit() + assert not connection.in_atomic_block + found = read_schedules([r.pk for r in rows], using=ALIAS) + answers = latest_ticks(["k0", "k1"], now - timedelta(hours=1), using=ALIAS) + + assert found[rows[1].pk].fields == ("name",) + assert isinstance(found[rows[1].pk].cause, OperationalError) + for row in rows: + if row is not rows[1]: + assert_reads_as_django_does(found[row.pk], row.pk) + assert answers["k1"].unreadable + assert answers["k1"].pk is None + assert newest_tick_pk("k1", now - timedelta(hours=1), using=ALIAS) == pk + assert answers["k0"] == TickRead("k0", at=now) + + +@pytest.mark.django_db(transaction=True) +def test_a_transaction_keeps_what_it_wrote_before_a_read_the_driver_failed(): + """ + The whole of it, committed: the write before the read that failed, the + rows found by reading again in halves, and the write after them. No + savepoint is taken anywhere in it. + """ + only_on("sqlite", "SQLite keeps text its driver cannot decode as UTF-8") + rows = [a_row(f"row-{i}") for i in range(4)] + set_column(rows[2].pk, "name", NOT_UTF_8) + + with CaptureQueriesContext(connection) as queries, transaction.atomic(): + a_tick("k", now_ish()) + found = read_schedules([r.pk for r in rows], using=ALIAS) + OxSchedule.objects.filter(pk=rows[0].pk).update(starting_deadline_seconds=90) + + assert not savepoints(queries) + assert found[rows[2].pk].fields == ("name",) + assert_reads_as_django_does(found[rows[3].pk], rows[3].pk) + assert OxScheduleTick.objects.filter(schedule_name="k").count() == 1 + assert OxSchedule.objects.get(pk=rows[0].pk).starting_deadline_seconds == 90 + + +#: The one failure SQLite's driver raises in this process over a statement +#: the database answered, as its message alone. +UNDECODABLE = OperationalError("Could not decode to UTF-8 column 'name' with text 'b'") + + +def failed_in_the_driver(): + """ + The driver's own failure on a value it was sent, raised for real, or + None on MySQL, whose driver hands every value over. + """ + statement = Failing.STATEMENT.get(connection.vendor) + return None if statement is None else raised_by(statement) + + +def refused_by_the_database(): + """A DataError the database itself raised, for real, with its code.""" + statement = Refusing.STATEMENT.get(connection.vendor) + return refused_by_sqlite() if statement is None else raised_by(statement) + + +@pytest.mark.django_db +class TestWhetherAFailedReadMayBeFollowedByAnother: + def test_not_after_a_failure_that_is_the_databases(self): + for exc in ( + OperationalError("server closed the connection unexpectedly"), + OperationalError("database is locked"), + KeyError("x"), + refused_by_the_database(), + raised_by("SELECT * FROM no_such_table"), + ): + assert not can_read_on(exc, using=ALIAS), exc + + def test_in_a_transaction_the_driver_says_where_it_can(self): + # After the driver's own failure, and after a converter's. + expected = { + # The session reports that no statement in it has failed. + "postgresql": (True, True), + # Nothing is reported, so only the failure known to be the + # driver's own. + "sqlite": (True, False), + # Nothing is reported, and the driver has no failure of its own. + "mysql": (False, False), + }[connection.vendor] + + with transaction.atomic(): + assert OxSchedule.objects.count() == 0 + own = failed_in_the_driver() or DataError("a value that did not load") + answers = ( + can_read_on(own, using=ALIAS), + can_read_on(ValueError("x"), using=ALIAS), + ) + + assert answers == expected + + def test_not_once_a_postgresql_transaction_has_failed(self): + only_on("postgresql", "its session reports a transaction that has failed") + with pytest.raises(DataError), transaction.atomic(): + assert can_read_on(ValueError("x"), using=ALIAS) + try: + with connection.cursor() as cursor: + cursor.execute("SELECT 1/0") + except DataError as exc: + # Django has not marked the transaction: only the session + # knows the server has ended it. + assert not connection.needs_rollback + assert not can_read_on(exc, using=ALIAS) + assert not can_read_on(ValueError("x"), using=ALIAS) + raise + + def test_not_in_a_transaction_django_has_marked_for_rollback(self): + if connection.vendor == "mysql": + pytest.skip("on mysql no failure inside a transaction is read past") + with transaction.atomic(): + assert OxSchedule.objects.count() == 0 + own = failed_in_the_driver() + assert can_read_on(own, using=ALIAS) + transaction.set_rollback(True) + try: + assert not can_read_on(own, using=ALIAS) + finally: + transaction.set_rollback(False) + + +@pytest.mark.django_db(transaction=True) +def test_with_no_transaction_open_only_a_failure_on_a_value_is_read_past(): + connection.ensure_connection() + assert connection.get_autocommit() + assert not connection.in_atomic_block + # A converter's failure, on every database, and the driver's own where + # it has one. + assert can_read_on(ValueError("x"), using=ALIAS) + own = failed_in_the_driver() + if own is not None: + assert can_read_on(own, using=ALIAS) + # Nothing the database raised: a DataError no more than a lock or a + # table that is not there. + assert not can_read_on(refused_by_the_database(), using=ALIAS) + assert not can_read_on(OperationalError("database is locked"), using=ALIAS) + assert not can_read_on(raised_by("SELECT * FROM no_such_table"), using=ALIAS) + + +# -- the decoders ---------------------------------------------------------- + + +def generated_instants(rng, aware): + """ + Many dates and times a column can be given, the edges of the years a + datetime holds and the folds and gaps of the zones among them. + """ + zones = [ + UTC, + ZoneInfo("America/Chicago"), + ZoneInfo("Asia/Kolkata"), + ZoneInfo("Australia/Lord_Howe"), + ZoneInfo("Pacific/Kiritimati"), + ] + first = datetime(1, 1, 1, tzinfo=UTC) + last = datetime(9999, 12, 31, 23, 59, 59, 999999, tzinfo=UTC) + span = int((last - first).total_seconds()) + picked = [ + first, + last, + datetime(1970, 1, 1, tzinfo=UTC), + datetime(2026, 3, 8, 8, 30, tzinfo=UTC), + datetime(2026, 11, 1, 6, 30, tzinfo=UTC), + datetime(2026, 11, 1, 7, 30, tzinfo=UTC), + ] + for _ in range(300): + moment = first + timedelta(seconds=rng.randrange(span)) + micro = rng.choice([0, 0, 1, 500000, 120000, 999999, rng.randrange(1000000)]) + picked.append(moment.replace(microsecond=micro)) + for _ in range(60): + # Around now, where the zones' rules change most. + moment = datetime(2026, 1, 1, tzinfo=UTC) + timedelta( + seconds=rng.randrange(366 * 86400) + ) + picked.append(moment) + values = [] + for moment in picked: + zone = rng.choice(zones) + try: + local = moment.astimezone(zone) + except OverflowError: + local = moment + if aware: + values.append(local) + else: + values.append(local.replace(tzinfo=None, fold=rng.choice([0, 1]))) + return values + + +@pytest.mark.django_db +class TestTheDateAndTimeDecoder: + @pytest.mark.parametrize( + ("use_tz", "zone"), + [ + (True, "America/Chicago"), + (False, "America/Chicago"), + (False, "Pacific/Kiritimati"), + ], + ) + def test_decodes_every_time_the_orm_writes_to_what_the_orm_reads( + self, settings, use_tz, zone + ): + settings.USE_TZ = use_tz + settings.TIME_ZONE = zone + rng = random.Random(f"{connection.vendor}-{use_tz}-{zone}") + values = generated_instants(rng, aware=use_tz) + OxScheduleTick.objects.bulk_create( + OxScheduleTick(schedule_name=f"p{i}", scheduled_for=v, created_at=v) + for i, v in enumerate(values) + ) + + by_orm = dict(OxScheduleTick.objects.values_list("pk", "scheduled_for")) + by_text = dict( + OxScheduleTick.objects.annotate(t=as_text("scheduled_for")).values_list( + "pk", "t" + ) + ) + + assert len(by_orm) == len(values) + differ = { + pk: (by_text[pk], decode_datetime(by_text[pk], ALIAS), read) + for pk, read in by_orm.items() + if not same(decode_datetime(by_text[pk], ALIAS), read) + } + assert not differ + + @pytest.mark.parametrize("use_tz", [True, False]) + def test_a_schedule_row_reads_as_the_orm_reads_it(self, settings, use_tz): + settings.USE_TZ = use_tz + settings.TIME_ZONE = "America/Chicago" + rng = random.Random(f"rows-{connection.vendor}-{use_tz}") + values = generated_instants(rng, aware=use_tz)[:80] + rows = [ + a_row( + f"row-{i}", + start_time=start, + end_time=None if i % 3 == 0 else end, + created_at=start, + updated_at=end, + enabled=bool(i % 2), + every_seconds=None if i % 4 == 0 else 60 + i, + trigger="cron" if i % 4 == 0 else "interval", + cron="* * * * *" if i % 4 == 0 else "", + starting_deadline_seconds=None if i % 5 else i, + ) + for i, (start, end) in enumerate(zip(values, reversed(values), strict=True)) + ] + + found = read_schedules([r.pk for r in rows], using=ALIAS) + + for row in rows: + assert_reads_as_django_does(found[row.pk], row.pk) + + def test_with_use_tz_off_a_time_with_a_zone_is_unreadable(self, settings): + only_on( + "sqlite", "SQLite keeps an offset in the text, which USE_TZ off cannot read" + ) + settings.USE_TZ = False + pk = insert_tick("k", "'9999-01-01 00:00:00+00:00'") + # Django reads it with its zone, which a naive clock cannot be + # compared with. + assert timezone.is_aware(OxScheduleTick.objects.get(pk=pk).scheduled_for) + + answer = latest_ticks(["k"], datetime(2000, 1, 1), using=ALIAS)["k"] + + assert answer.unreadable + assert answer.pk is None + assert newest_tick_pk("k", datetime(2000, 1, 1), using=ALIAS) == pk + assert answer.reason == ( + "scheduled_for holds '9999-01-01 00:00:00+00:00', which has a time " + "zone, and USE_TZ is off" + ) + + def test_with_use_tz_on_a_time_with_an_offset_reads_as_django_reads_it( + self, settings + ): + only_on("sqlite", "SQLite keeps an offset in the text") + settings.USE_TZ = True + pk = insert_tick("k", "'2026-10-05 12:00:00+02:00'") + django = OxScheduleTick.objects.get(pk=pk).scheduled_for + + answer = latest_ticks(["k"], datetime(2000, 1, 1, tzinfo=UTC), using=ALIAS)["k"] + + assert same(answer.at, django) + assert answer.at == datetime(2026, 10, 5, 10, 0, tzinfo=UTC) + + +class TestTheTextFormsOfEachEngine: + """ + Each engine's text, decoded as its driver would. These need no database + of that engine: the text is what it sends, and the connection here + supplies only the zones. + """ + + @pytest.fixture + def conn(self, settings): + settings.USE_TZ = True + from django.db import connections + + return connections[ALIAS] + + def test_postgresql(self, conn): + read = _stored_read._from_postgresql + assert read("2026-10-05 12:34:56.5+00", conn) == datetime( + 2026, 10, 5, 12, 34, 56, 500000, tzinfo=UTC + ) + assert read("2026-10-05 12:34:56.123456+00", conn).microsecond == 123456 + assert read("1799-12-31 18:09:24-05:50:36", conn) == datetime( + 1800, 1, 1, tzinfo=UTC + ) + for text in ("infinity", "-infinity", "0001-01-01 00:00:00+00 BC", "banana"): + with pytest.raises(UnreadableValue): + read(text, conn) + # Python's own words, which differ between versions. + with pytest.raises(ValueError): + read("10000-01-01 00:00:00+00", conn) + + def test_mysql(self, conn): + read = _stored_read._from_mysql + assert read("2026-10-05 12:34:56.000000", conn) == datetime( + 2026, 10, 5, 12, 34, 56, tzinfo=UTC + ) + assert read("9999-12-31 23:59:59.999999", conn).microsecond == 999999 + for text in ("2026-00-10 00:00:00.000000", "0000-00-00 00:00:00.000000"): + with pytest.raises(ValueError): + read(text, conn) + with pytest.raises(UnreadableValue): + read("2026-10-05", conn) + + def test_sqlite(self, conn): + read = _stored_read._from_sqlite + assert read("2026-10-05 12:34:56", conn) == datetime( + 2026, 10, 5, 12, 34, 56, tzinfo=UTC + ) + for text in ("banana", "10000-01-01 00:00:00", "2.5"): + with pytest.raises(UnreadableValue): + read(text, conn) + with pytest.raises(ValueError): + read("2026-02-30 00:00:00", conn) + + +class TestTheNumberDecoders: + def test_integers(self): + assert decode_integer(0) == 0 + assert decode_integer(2**62) == 2**62 + assert decode_integer(None, null=True) is None + for value in ("abc", "60", 60.5, 60.0, True, b"60", None): + with pytest.raises(UnreadableValue): + decode_integer(value) + with pytest.raises( + UnreadableValue, match="holds 'abc', which is not a whole number" + ): + decode_integer("abc") + + def test_booleans(self): + assert decode_boolean(value=True) is True + assert decode_boolean(value=False) is False + assert decode_boolean(1) is True + assert decode_boolean(0) is False + assert decode_boolean(None, null=True) is None + for value in (2, -1, "abc", "1", 1.0, None): + with pytest.raises(UnreadableValue): + decode_boolean(value) + with pytest.raises( + UnreadableValue, match="holds 2, which is neither true nor false" + ): + decode_boolean(2) + + def test_a_value_is_quoted_to_a_length(self): + with pytest.raises(UnreadableValue) as caught: + decode_integer("x" * 500) + assert len(str(caught.value)) < 120 + + @pytest.mark.django_db + def test_null_where_a_value_is_needed(self): + with pytest.raises( + UnreadableValue, match="is NULL, and this field needs a value" + ): + decode_datetime(None, ALIAS) + assert decode_datetime(None, ALIAS, null=True) is None + + +@pytest.mark.django_db +class TestWhatCountsAsAnUnreadableValue: + def test_a_converters_failure_does(self): + for exc in ( + ValueError("x"), + TypeError("x"), + AttributeError("x"), + OverflowError("x"), + UnreadableValue("x"), + ): + assert is_unreadable_value(exc, using=ALIAS), exc + + def test_the_drivers_own_failure_on_a_value_does(self): + exc = failed_in_the_driver() + if exc is None: + pytest.skip("MySQL's driver hands every value over; none fails in it") + assert type(exc) is Failing.RAISES[connection.vendor] + assert is_unreadable_value(exc, using=ALIAS) + + def test_a_data_error_the_database_raised_does_not(self): + """ + The class is the same as the driver's own on PostgreSQL. What + tells them apart is the code the database gives every error it + raises, and that the driver's own has none. + """ + exc = refused_by_the_database() + assert type(exc) is REFUSED + assert Refusing.SAYS[connection.vendor] in str(exc) + assert not is_unreadable_value(exc, using=ALIAS) + + def test_each_drivers_failure_counts_on_its_own_database_alone(self): + without_a_code = DataError("timestamp too large (after year 10K)") + assert is_unreadable_value(UNDECODABLE, using=ALIAS) == ( + connection.vendor == "sqlite" + ) + assert is_unreadable_value(without_a_code, using=ALIAS) == ( + connection.vendor == "postgresql" + ) + + def test_the_database_failing_does_not(self): + from django.db import IntegrityError, InterfaceError, ProgrammingError + + for exc in ( + OperationalError("server closed the connection unexpectedly"), + OperationalError("database is locked"), + InterfaceError("connection already closed"), + ProgrammingError("relation does not exist"), + IntegrityError("duplicate key"), + KeyError("x"), + RuntimeError("x"), + raised_by("SELECT * FROM no_such_table"), + ): + assert not is_unreadable_value(exc, using=ALIAS), exc + + +def test_stored_value_unreadable_pickles(): + exc = StoredValueUnreadable( + alias="default", + model="django_ox.OxSchedule", + pk=5, + fields=("end_time",), + reason="end_time holds 'infinity', which is not a date and time", + ) + + copy = pickle.loads(pickle.dumps(exc)) # noqa: S301 - our own bytes + + assert type(copy) is StoredValueUnreadable + assert (copy.alias, copy.model, copy.pk, copy.fields, copy.reason) == ( + exc.alias, + exc.model, + exc.pk, + exc.fields, + exc.reason, + ) + assert str(copy) == str(exc) + + +# -- the tick log ------------------------------------------------------------ + + +def now_ish(): + return timezone.now().replace(microsecond=0) + + +#: (engine, SQL literal, id): a tick each engine keeps that Django's grouped +#: read of the newest tick raises on, and that sorts newest. +UNREADABLE_NEWEST_TICKS = [ + pytest.param("postgresql", "'10000-01-01 00:00:00+00'", id="postgresql-year-10000"), + pytest.param("postgresql", "'infinity'", id="postgresql-infinity"), + pytest.param("sqlite", "'banana'", id="sqlite-banana"), + pytest.param("sqlite", "'9999-02-30 00:00:00'", id="sqlite-does-not-exist"), + pytest.param("mysql", "'9999-00-00 00:00:00'", id="mysql-zero-month"), +] + +#: The same for the oldest tick, which the first-sighting read takes. +UNREADABLE_OLDEST_TICKS = [ + pytest.param("postgresql", "'-infinity'", id="postgresql-minus-infinity"), + pytest.param("postgresql", "'0001-01-01 00:00:00+00 BC'", id="postgresql-bc"), + pytest.param("sqlite", "'0000-02-30 00:00:00'", id="sqlite-year-zero"), + pytest.param("mysql", "'0000-00-00 00:00:00'", id="mysql-zero-date"), +] + + +def django_earliest(key): + try: + with transaction.atomic(): + return ( + OxScheduleTick.objects.filter(schedule_name=key) + .order_by("scheduled_for") + .values_list("scheduled_for", flat=True) + .first() + ) + except Exception as exc: # the caller asserts on it + return exc + + +def django_newest(key, since=None): + ticks = OxScheduleTick.objects.filter(schedule_name=key) + if since is not None: + ticks = ticks.filter(scheduled_for__gte=since) + try: + with transaction.atomic(): + return ( + ticks.values("schedule_name") + .annotate(latest=Max("scheduled_for")) + .get()["latest"] + ) + except Exception as exc: # the caller asserts on it + return exc + + +@pytest.mark.django_db +class TestTheNewestTickPerKey: + def test_answers_every_key_in_one_grouped_statement(self): + now = now_ish() + since = now - timedelta(hours=1) + for i in range(5): + for back in range(3): + a_tick(f"k{i}", now - timedelta(minutes=10 * back + i)) + a_tick("old", now - timedelta(days=1)) + keys = [f"k{i}" for i in range(5)] + ["old", "never"] + + with CaptureQueriesContext(connection) as queries: + answers = latest_ticks(keys, since, using=ALIAS) + + assert len(selects(queries, TICKS)) == 1 + assert list(answers) == keys + for i in range(5): + assert same(answers[f"k{i}"].at, django_newest(f"k{i}", since)) + assert not answers[f"k{i}"].unreadable + # Nothing at or after `since` is no history within what was asked. + assert answers["old"].absent + assert answers["never"].absent + assert answers["never"] == TickRead("never") + + def test_takes_one_statement_a_slice_of_the_parameter_limit(self): + now = now_ish() + since = now - timedelta(hours=1) + if connection.features.max_query_params is None: + # No limit: every key in one statement, however many. + keys = [f"k{i}" for i in range(2025)] + for key in keys[:25]: + a_tick(key, now) + with CaptureQueriesContext(connection) as queries: + answers = latest_ticks(keys, since, using=ALIAS) + assert len(selects(queries, TICKS)) == 1 + assert all(answers[key].at == now for key in keys[:25]) + assert all(answers[key].absent for key in keys[25:]) + return + # SQLite's own limit, lowered on this connection, so that a statement + # over it is refused rather than merely counted. A Django that + # declares a limit of its own instead of reading SQLite's is held to + # that one. + raw = connection.connection + before = raw.getlimit(sqlite3.SQLITE_LIMIT_VARIABLE_NUMBER) + raw.setlimit(sqlite3.SQLITE_LIMIT_VARIABLE_NUMBER, 10) + if connection.features.max_query_params != 10: + raw.setlimit(sqlite3.SQLITE_LIMIT_VARIABLE_NUMBER, before) + try: + limit = connection.features.max_query_params + keys = [f"k{i}" for i in range(2 * limit + 5)] + OxScheduleTick.objects.bulk_create( + OxScheduleTick(schedule_name=key, scheduled_for=now, created_at=now) + for key in keys + ) + with CaptureQueriesContext(connection) as queries: + answers = latest_ticks(keys, since, using=ALIAS) + finally: + raw.setlimit(sqlite3.SQLITE_LIMIT_VARIABLE_NUMBER, before) + # A key a parameter, and one for the bound. + assert len(selects(queries, TICKS)) == math.ceil(len(keys) / (limit - 1)) == 3 + assert all(answers[key].at == now for key in keys) + + def test_without_a_bound_every_tick_counts(self): + now = now_ish() + old = now - timedelta(days=4000) + a_tick("k", old) + a_tick("k", old - timedelta(days=1)) + + with CaptureQueriesContext(connection) as queries: + answers = latest_ticks(["k", "never"], using=ALIAS) + + assert len(selects(queries, TICKS)) == 1 + assert answers["k"] == TickRead("k", at=old) + assert same(answers["k"].at, django_newest("k")) + assert answers["never"].absent + + @pytest.mark.parametrize(("vendor", "literal"), UNREADABLE_OLDEST_TICKS) + def test_without_a_bound_an_only_tick_that_does_not_read_is_not_no_history( + self, vendor, literal + ): + only_on(vendor, "kept by that engine only") + pk = insert_tick("k", literal) + + answer = latest_ticks(["k"], using=ALIAS)["k"] + + assert answer.unreadable + assert answer.pk is None + assert newest_tick_pk("k", using=ALIAS) == pk + + @pytest.mark.parametrize(("vendor", "literal"), UNREADABLE_NEWEST_TICKS) + @pytest.mark.parametrize("position", ["first", "middle", "last"]) + def test_an_unreadable_newest_tick_is_its_keys_answer_alone( + self, vendor, literal, position + ): + only_on(vendor, "kept by that engine only") + now = now_ish() + since = now - timedelta(hours=1) + keys = [f"k{i}" for i in range(5)] + for key in keys: + a_tick(key, now) + bad = {"first": "k0", "middle": "k2", "last": "k4"}[position] + pk = insert_tick(bad, literal) + assert misread(django_newest(bad, since)) + + with CaptureQueriesContext(connection) as queries: + answers = latest_ticks(keys, since, using=ALIAS) + + assert answers[bad].unreadable + assert not answers[bad].absent + assert answers[bad].at is None + assert answers[bad].pk is None + assert newest_tick_pk(bad, since, using=ALIAS) == pk + assert answers[bad].reason.startswith("scheduled_for holds ") + assert isinstance(answers[bad].cause, UnreadableValue) + for key in keys: + if key != bad: + assert answers[key] == TickRead(key, at=now) + # The grouped statement alone: the bad tick's row is not looked up. + assert len(selects(queries, TICKS)) == 1 + + @pytest.mark.parametrize(("vendor", "literal"), UNREADABLE_NEWEST_TICKS) + def test_several_unreadable_newest_ticks_cost_the_one_grouped_statement( + self, vendor, literal + ): + only_on(vendor, "kept by that engine only") + now = now_ish() + since = now - timedelta(hours=1) + keys = [f"k{i}" for i in range(6)] + for key in keys: + a_tick(key, now) + bad = {key: insert_tick(key, literal) for key in ("k1", "k3", "k4")} + + with CaptureQueriesContext(connection) as queries: + answers = latest_ticks(keys, since, using=ALIAS) + + assert len(selects(queries, TICKS)) == 1 + for key in keys: + if key in bad: + assert answers[key].unreadable + assert answers[key].pk is None + else: + assert answers[key] == TickRead(key, at=now) + # The row is named by asking, for a key at a time. + for key, pk in bad.items(): + assert newest_tick_pk(key, since, using=ALIAS) == pk + + @pytest.mark.parametrize("position", ["first", "middle", "last"]) + def test_a_tick_the_driver_cannot_hand_over_is_found_by_halving_keys( + self, position + ): + only_on("sqlite", "SQLite keeps text its driver cannot decode as UTF-8") + now = now_ish() + keys = [f"k{i}" for i in range(8)] + for key in keys: + a_tick(key, now) + bad = {"first": "k0", "middle": "k3", "last": "k7"}[position] + pk = insert_tick(bad, NOT_UTF_8) + + # In this test's transaction, and in no savepoint. + assert connection.in_atomic_block + with CaptureQueriesContext(connection) as queries: + answers = latest_ticks(keys, now - timedelta(hours=1), using=ALIAS) + + assert not savepoints(queries) + assert answers[bad].unreadable + assert answers[bad].pk is None + assert newest_tick_pk(bad, now - timedelta(hours=1), using=ALIAS) == pk + assert "UTF-8" in answers[bad].reason + for key in keys: + if key != bad: + assert answers[key] == TickRead(key, at=now) + + def test_a_column_the_read_does_not_take_is_not_its_concern(self): + only_on("postgresql", "PostgreSQL keeps year 10000") + now = now_ish() + tick = a_tick("k", now) + by_sql( + f"UPDATE {TICKS} SET created_at = '10000-01-01 00:00:00+00' WHERE id = %s", # noqa: S608 + [tick.pk], + ) + + answers = latest_ticks(["k"], now - timedelta(hours=1), using=ALIAS) + + assert answers["k"] == TickRead("k", at=now) + + def test_inside_a_transaction_the_transaction_still_works_after(self): + now = now_ish() + a_tick("k0", now) + a_tick("k1", now) + literal = { + "postgresql": "'infinity'", + "mysql": "'9999-00-00 00:00:00'", + "sqlite": NOT_UTF_8, + }[connection.vendor] + insert_tick("k1", literal) + + with transaction.atomic(): + with CaptureQueriesContext(connection) as queries: + answers = latest_ticks( + ["k0", "k1"], now - timedelta(hours=1), using=ALIAS + ) + assert ( + OxScheduleTick.objects.filter(schedule_name="k0").update(task=None) == 1 + ) + + assert not savepoints(queries) + assert answers["k0"].at == now + assert answers["k1"].unreadable + + +@pytest.mark.django_db +class TestTheEarliestTick: + def test_is_the_first_by_time_leaving_out_the_given_rows(self): + now = now_ish() + first = a_tick("k", now - timedelta(hours=2)) + second = a_tick("k", now - timedelta(hours=1)) + a_tick("other", now - timedelta(days=1)) + + assert earliest_tick("k", using=ALIAS) == TickRead( + "k", at=first.scheduled_for, pk=first.pk + ) + assert earliest_tick("k", using=ALIAS, exclude=[first.pk]) == TickRead( + "k", at=second.scheduled_for, pk=second.pk + ) + assert earliest_tick("k", using=ALIAS, exclude=[first.pk, second.pk]).absent + assert earliest_tick("never", using=ALIAS).absent + found = earliest_tick("k", using=ALIAS).at + django = ( + OxScheduleTick.objects.filter(schedule_name="k") + .order_by("scheduled_for") + .values_list("scheduled_for", flat=True) + .first() + ) + assert same(found, django) + + @pytest.mark.parametrize(("vendor", "literal"), UNREADABLE_OLDEST_TICKS) + def test_an_unreadable_earliest_tick_is_not_no_history(self, vendor, literal): + only_on(vendor, "kept by that engine only") + now = now_ish() + a_tick("k", now) + pk = insert_tick("k", literal) + assert misread(django_earliest("k")) + + answer = earliest_tick("k", using=ALIAS) + + assert answer.unreadable + assert not answer.absent + assert answer.pk == pk + assert answer.reason.startswith("scheduled_for holds ") + # Left out, the next tick is the earliest again. + assert earliest_tick("k", using=ALIAS, exclude=[pk]).at == now + + def test_a_locking_read_where_the_database_has_one(self): + now = now_ish() + tick = a_tick("k", now) + + with transaction.atomic(), CaptureQueriesContext(connection) as queries: + answer = earliest_tick("k", using=ALIAS, lock=True) + + assert answer == TickRead("k", at=now, pk=tick.pk) + # One statement in the caller's transaction, and no savepoint. + assert not savepoints(queries) + (statement,) = selects(queries, TICKS) + features = connection.features + assert ("FOR UPDATE" in statement) == features.has_select_for_update + assert ("SKIP LOCKED" in statement) == ( + features.has_select_for_update + and features.has_select_for_update_skip_locked + ) + + +@pytest.mark.django_db(transaction=True) +def test_the_locking_earliest_read_skips_a_tick_another_transaction_holds(): + """ + The latch protocol's repeat read is a locking read that skips rows + another transaction holds. Here one transaction holds the earliest tick + through the same read, and a second one's read passes over it to the + next. + """ + if not connection.features.has_select_for_update_skip_locked: + pytest.skip(f"{connection.vendor} has no SKIP LOCKED") + now = now_ish() + first = a_tick("k", now - timedelta(hours=1)) + second = a_tick("k", now) + held = threading.Event() + release = threading.Event() + seen = {} + + def holder(): + try: + with transaction.atomic(): + seen["holder"] = earliest_tick("k", using=ALIAS, lock=True) + held.set() + release.wait(15) + finally: + connection.close() + + thread = threading.Thread(target=holder) + thread.start() + try: + assert held.wait(15) + with transaction.atomic(): + seen["other"] = earliest_tick("k", using=ALIAS, lock=True) + finally: + release.set() + thread.join(15) + + assert seen["holder"].pk == first.pk + assert seen["other"].pk == second.pk + + +@pytest.mark.django_db +class TestAScheduleNameThatIsNotText: + BLOB = "X'626c6f622d6e616d65'" + + def test_bytes_in_the_name_column_are_a_value_that_does_not_read(self): + only_on("sqlite", "SQLite keeps bytes in a text column") + row = a_row("row") + set_column(row.pk, "name", self.BLOB) + assert OxSchedule.objects.get(pk=row.pk).name == b"blob-name" + + unreadable = read_schedules([row.pk], using=ALIAS)[row.pk] + + assert isinstance(unreadable, UnreadableRow) + assert unreadable.fields == ("name",) + assert unreadable.reason == "name holds b'blob-name', which is not text" + assert isinstance(unreadable.cause, UnreadableValue) + + def test_bytes_that_are_text_in_a_name_are_not_taken_for_text(self): + only_on("sqlite", "SQLite keeps bytes in a text column") + row = a_row("row") + set_column(row.pk, "name", "CAST('plain' AS BLOB)") + + unreadable = read_schedules([row.pk], using=ALIAS)[row.pk] + + assert unreadable.reason == "name holds b'plain', which is not text" + + def test_a_name_that_reads_is_shown_as_it_is(self): + row = a_row("nightly") + + assert str(row) == "nightly" + assert str(OxSchedule.objects.get(pk=row.pk)) == "nightly" + assert str(OxSchedule(name="unsaved")) == "unsaved" + + def test_a_name_of_bytes_is_shown_as_the_primary_key(self): + only_on("sqlite", "SQLite keeps bytes in a text column") + row = a_row("row") + set_column(row.pk, "name", self.BLOB) + found = OxSchedule.objects.get(pk=row.pk) + + assert str(found) == f"pk {row.pk}" + assert f"pk {row.pk}" in repr(found) + # Shown, not assigned: the field still holds what the row holds. + assert found.name == b"blob-name" + + def test_a_name_the_driver_cannot_decode_is_shown_as_the_primary_key(self): + only_on("sqlite", "SQLite keeps text its driver cannot decode as UTF-8") + row = a_row("row") + set_column(row.pk, "name", NOT_UTF_8) + instance = read_schedules([row.pk], using=ALIAS)[row.pk].deferred_instance() + assert "name" in instance.get_deferred_fields() + + assert str(instance) == f"pk {row.pk}" + assert f"pk {row.pk}" in repr(instance) + # Still deferred, so the next read of it raises as it did. + assert "name" in instance.get_deferred_fields() + with pytest.raises(OperationalError): + instance.name # noqa: B018 + + def test_an_error_the_database_raised_is_not_hidden_by_the_display( + self, monkeypatch + ): + row = a_row("row") + instance = OxSchedule.objects.only("id").get(pk=row.pk) + assert "name" in instance.get_deferred_fields() + + def locked(*args, **kwargs): + raise OperationalError("database is locked") + + monkeypatch.setattr(OxSchedule, "refresh_from_db", locked) + + with pytest.raises(OperationalError, match="database is locked"): + str(instance) + + +#: The values a project can have around the API that psycopg2 and mysqlclient +#: read as something (the largest or smallest datetime, None) where psycopg 3 +#: and PyMySQL raise: (engine, column, SQL literal, how the reason quotes it). +LEGACY_ROW_VALUES = [ + pytest.param("postgresql", "end_time", "'infinity'", "'infinity'", id="pg-end"), + pytest.param( + "postgresql", "start_time", "'-infinity'", "'-infinity'", id="pg-start" + ), + pytest.param( + "mysql", "end_time", "'0000-00-00 00:00:00'", "'0000-00-00 ", id="zero" + ), + pytest.param( + "mysql", "start_time", "'2026-00-10 00:00:00'", "'2026-00-10", id="zero-month" + ), + pytest.param("mysql", "start_time", "NULL", "NULL", id="null-start"), +] + +LEGACY_TICK_VALUES = [ + pytest.param("postgresql", "'infinity'", id="pg-infinity"), + pytest.param("postgresql", "'-infinity'", id="pg-minus-infinity"), + pytest.param("mysql", "'0000-00-00 00:00:00'", id="zero"), + pytest.param("mysql", "'9999-00-00 00:00:00'", id="9999-zero-month"), +] + + +@pytest.mark.django_db +class TestTheLegacyValuesAreUnreadableOnEveryDriver: + """ + One answer on every driver: a stored value that psycopg2 or mysqlclient + would hand over as a datetime or None is unreadable here as it is with + psycopg 3 and PyMySQL, and the healthy rows beside it read. + """ + + @pytest.mark.parametrize( + ("vendor", "column", "literal", "shown"), LEGACY_ROW_VALUES + ) + def test_a_schedule_row(self, vendor, column, literal, shown): + only_on(vendor, "kept by that engine only") + before, bad, after = a_row("before"), a_row("bad"), a_row("after") + set_column(bad.pk, column, literal) + + found = read_schedules([before.pk, bad.pk, after.pk], using=ALIAS) + + assert_reads_as_django_does(found[before.pk], before.pk) + assert_reads_as_django_does(found[after.pk], after.pk) + unreadable = found[bad.pk] + assert isinstance(unreadable, UnreadableRow) + assert unreadable.fields == (column,) + assert unreadable.reason.startswith(f"{column} ") + assert shown in unreadable.reason or literal == "NULL" + assert isinstance(unreadable.cause, UnreadableValue) + + @pytest.mark.parametrize(("vendor", "literal"), LEGACY_TICK_VALUES) + def test_a_tick_row(self, vendor, literal): + only_on(vendor, "kept by that engine only") + now = now_ish() + a_tick("healthy", now) + insert_tick("legacy", literal) + + newest = latest_ticks(["healthy", "legacy"], using=ALIAS) + oldest = earliest_tick("legacy", using=ALIAS) + + assert newest["healthy"] == TickRead("healthy", at=now) + assert newest["legacy"].unreadable + assert oldest.unreadable + assert oldest.reason.startswith("scheduled_for holds ") + assert earliest_tick("healthy", using=ALIAS).at == now