diff --git a/src/datajoint/adapters/base.py b/src/datajoint/adapters/base.py index 737807731..32f591a53 100644 --- a/src/datajoint/adapters/base.py +++ b/src/datajoint/adapters/base.py @@ -1340,6 +1340,25 @@ def job_metadata_columns(self) -> list[str]: """ ... + @abstractmethod + def provenance_columns(self) -> list[str]: + """ + Return the hidden extrinsic-provenance column for Entry tables. + + Returns + ------- + list[str] + List of column definition strings (fully formatted with quotes). + + Examples + -------- + MySQL: + ["`_prov` json DEFAULT NULL"] + PostgreSQL: + ['"_prov" jsonb DEFAULT NULL'] + """ + ... + # ========================================================================= # Error Translation # ========================================================================= diff --git a/src/datajoint/adapters/mysql.py b/src/datajoint/adapters/mysql.py index 4d2d4ca73..082b55041 100644 --- a/src/datajoint/adapters/mysql.py +++ b/src/datajoint/adapters/mysql.py @@ -1013,6 +1013,17 @@ def job_metadata_columns(self) -> list[str]: "`_job_version` varchar(64) DEFAULT ''", ] + def provenance_columns(self) -> list[str]: + """ + Return the MySQL extrinsic-provenance column definition. + + Examples + -------- + >>> adapter.provenance_columns() + ["`_prov` json DEFAULT NULL"] + """ + return ["`_prov` json DEFAULT NULL"] + # ========================================================================= # Error Translation # ========================================================================= diff --git a/src/datajoint/adapters/postgres.py b/src/datajoint/adapters/postgres.py index ed7a557f7..a08b887f8 100644 --- a/src/datajoint/adapters/postgres.py +++ b/src/datajoint/adapters/postgres.py @@ -1366,6 +1366,17 @@ def job_metadata_columns(self) -> list[str]: "\"_job_version\" varchar(64) DEFAULT ''", ] + def provenance_columns(self) -> list[str]: + """ + Return the PostgreSQL extrinsic-provenance column definition. + + Examples + -------- + >>> adapter.provenance_columns() + ['"_prov" jsonb DEFAULT NULL'] + """ + return ['"_prov" jsonb DEFAULT NULL'] + # ========================================================================= # Error Translation # ========================================================================= diff --git a/src/datajoint/autopopulate.py b/src/datajoint/autopopulate.py index ed2493714..61b36c5c9 100644 --- a/src/datajoint/autopopulate.py +++ b/src/datajoint/autopopulate.py @@ -11,6 +11,7 @@ import traceback from typing import TYPE_CHECKING, Any, Generator +from . import provenance from .errors import DataJointError, LostConnectionError from .expression import AndList, QueryExpression @@ -684,6 +685,13 @@ def _populate1( self._upstream_key = dict(key) self._upstream = None + # Rows this make() writes into Entry tables that carry no foreign key back + # here -- the fan-out ingestion pattern -- record the ingesting table and + # key, which is what makes such a row traceable without one. + from .jobs import _get_job_version + + prov_token = provenance.set_ingesting(self.full_table_name, key, _get_job_version(self.connection._config)) + try: if not is_generator: make(dict(key), **(make_kwargs or {})) @@ -740,6 +748,7 @@ def _populate1( jobs.complete(key, duration=duration) return True finally: + provenance.reset_ingesting(prov_token) self.__class__._allow_insert = False # Clear the per-make() upstream state: `_upstream = None` invalidates # the memoized Diagram; `_upstream_key = None` restores the "outside diff --git a/src/datajoint/declare.py b/src/datajoint/declare.py index 1ba1f5a6b..c63bb6ebb 100644 --- a/src/datajoint/declare.py +++ b/src/datajoint/declare.py @@ -521,6 +521,20 @@ def declare( job_metadata_sql = adapter.job_metadata_columns() attribute_sql.extend(job_metadata_sql) + # Add the hidden extrinsic-provenance slot to Entry tables, where rows enter + # from outside the pipeline. Computed and Imported tables have no use for + # it -- their provenance is entailed by the foreign-key graph -- and a part + # inherits its master's. + # Matched against the Manual tier itself, not by excluding the other tiers' + # prefixes: enumerating exclusions makes every tier added later an Entry + # table by default, which is how job tables (`~`) first acquired the slot. + # Imported here rather than at module scope: user_tables imports table, + # which imports this module. + from .user_tables import Manual + + if config.provenance.capture and re.fullmatch(Manual.tier_regexp, table_name): + attribute_sql.extend(adapter.provenance_columns()) + if not primary_key: # Singleton table: add hidden sentinel attribute primary_key = ["_singleton"] diff --git a/src/datajoint/deploy.py b/src/datajoint/deploy.py index 73c4ab30b..bb6a0245a 100644 --- a/src/datajoint/deploy.py +++ b/src/datajoint/deploy.py @@ -12,7 +12,8 @@ ``add_job_metadata_columns``, ``rebuild_lineage``. - :mod:`datajoint.deploy` — configure an environment for a consumer's requirements (CDC tools, replication, role grants, performance tuning). - Cadence: re-runnable, idempotent. Examples: :func:`set_replica_identity`. + Cadence: re-runnable, idempotent. Examples: :func:`set_replica_identity`, + :func:`add_prov_column`. Functions in this module should be safe to call repeatedly from a deploy hook without accumulating side effects. @@ -183,3 +184,114 @@ def set_replica_identity( connection.query(ddl) result["tables_modified"] += 1 return result + + +def add_prov_column(target: "TargetType", dry_run: bool = True) -> dict: + """ + Add the hidden ``_prov`` attribute to Entry (``dj.Manual``) tables that lack it. + + Capture defaults on, so tables declared from 2.3.4 onward already carry the + slot. Two populations do not: tables declared before 2.3.4, and tables + declared while ``config.provenance.capture`` was off. Inserts into those + record nothing, silently, and this brings them in line. + + It belongs here rather than in :mod:`datajoint.migrate` because it is not a + one-shot correction of legacy state. It is idempotent — a table that already + has the column is reported and left alone — and it stays useful for as long + as capture can be turned off, which outlives the migration module. + + Parameters + ---------- + target : Schema, Table class, or Table instance + Given a Schema, every Entry table in it is processed. + dry_run : bool, optional + If True, report what would change without altering anything. Default True. + + Returns + ------- + dict + ``tables_analyzed``, ``tables_modified``, ``columns_added``, ``ddl``, + and ``details`` — a per-table list of dicts. + + Examples + -------- + >>> from datajoint.deploy import add_prov_column + >>> add_prov_column(schema, dry_run=True)["ddl"] + >>> add_prov_column(schema, dry_run=False)["tables_modified"] + + Notes + ----- + - Only Entry tables are touched. Computed and Imported tables have no use + for the slot — their provenance is entailed by the foreign-key graph — and + a part table inherits its master's. + - Rows already present keep ``NULL``. Provenance is recorded at insert and + is never reconstructed after the fact. + """ + import re + + from . import provenance + from .schemas import _Schema + from .table import Table + from .user_tables import Manual + + if isinstance(target, _Schema): + connection = target.connection + if connection is None or not target.database: + raise DataJointError("Schema is not activated. Call schema.activate(...) before add_prov_column().") + database = target.database + table_names = list(target.list_tables()) + elif isinstance(target, type) and issubclass(target, Table): + instance = target() + connection = instance.connection + if connection is None: + raise DataJointError(f"Table {target.__name__} has no active connection.") + database, table_names = instance.database, [instance.table_name] + elif isinstance(target, Table): + connection = target.connection + if connection is None: + raise DataJointError(f"Table {type(target).__name__} has no active connection.") + database, table_names = target.database, [target.table_name] + else: + raise DataJointError(f"target must be a Schema or Table class/instance; got {type(target).__name__}") + + if not database: + raise DataJointError("Cannot add the provenance column: the target has no database.") + + adapter = connection.adapter + column_sql = adapter.provenance_columns()[0] + + result: dict[str, Any] = { + "tables_analyzed": 0, + "tables_modified": 0, + "columns_added": 0, + "ddl": [], + "details": [], + } + + for table_name in table_names: + if not re.fullmatch(Manual.tier_regexp, table_name): + continue + result["tables_analyzed"] += 1 + + existing = { + row[0] + for row in connection.query( + "SELECT COLUMN_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s", + args=(database, table_name), + ).fetchall() + } + if provenance.PROV_ATTRIBUTE in existing: + result["details"].append({"table": f"{database}.{table_name}", "status": "already_present"}) + continue + + ddl = ( + f"ALTER TABLE {adapter.quote_identifier(database)}.{adapter.quote_identifier(table_name)} ADD COLUMN {column_sql}" + ) + result["ddl"].append(ddl) + result["details"].append({"table": f"{database}.{table_name}", "status": "pending" if dry_run else "added"}) + if not dry_run: + connection.query(ddl) + result["tables_modified"] += 1 + result["columns_added"] += 1 + + return result diff --git a/src/datajoint/heading.py b/src/datajoint/heading.py index 2816a22dc..33f2d6920 100644 --- a/src/datajoint/heading.py +++ b/src/datajoint/heading.py @@ -397,7 +397,11 @@ def quote(name): return adapter.quote_identifier(name) if adapter else f'"{name}"' def render_field(name): - attr = self.attributes[name] + # `attributes` hides underscore-prefixed names, so a caller that asks + # for one by name -- copying `_prov` through an INSERT ... SELECT -- + # falls back to the full set. Default field lists are unaffected: + # they are built from `attributes` and never contain hidden names. + attr = self.attributes.get(name) or self._attributes[name] if attr.attribute_expression is None: return quote(name) else: diff --git a/src/datajoint/provenance.py b/src/datajoint/provenance.py new file mode 100644 index 000000000..6f949d185 --- /dev/null +++ b/src/datajoint/provenance.py @@ -0,0 +1,140 @@ +"""Extrinsic provenance for rows that enter the pipeline from outside. + +Inside the pipeline, provenance is structural: a Computed table's row cannot +exist unless its declared upstream exists, so the foreign-key graph *is* the +lineage and nothing has to be recorded for it to hold. + +At the boundary the structure runs out. Rows arrive in Entry tables from a +person, an instrument, or a feed, and the framework has no way to say where +they came from. This module supplies the slot and fills it. + +The attribute is **framework-owned: no author ever writes it.** ``insert`` +takes no provenance argument, and its content comes from three places, none of +them the call site: + +* **configuration** -- ``config.provenance.source``, set per deployment, naming + the external system this process draws from; +* **ambient connection state** -- the connecting user, host and database, the + insert time, and the code version; +* **ambient execution state** -- the ingesting table and key, when the insert + runs inside a ``make()``. + +That ownership is the point. A field an operator can set is weaker evidence +than one the system sets, and nothing is left for a pipeline to neglect. +Anything an author wants to record deliberately belongs in the data model as a +visible attribute, where queries can reach it. +""" + +import contextlib +import contextvars +import datetime +import json +from typing import Any + +#: Name of the hidden attribute. Hidden attributes are excluded from +#: ``heading.attributes``, so this never appears in a query heading. +PROV_ATTRIBUTE = "_prov" + +# Set by autopopulate around a make() call so that rows written to Entry tables +# from inside an ingesting make() record what wrote them, which is what makes a +# fanned-out row traceable without a foreign key. +_ingesting: contextvars.ContextVar = contextvars.ContextVar("dj_ingesting", default=None) + + +def set_ingesting(table_name, key, version=None): + """Record the ``make()`` now executing; returns a token for ``reset_ingesting``. + + Parameters + ---------- + table_name : str + Full table name of the ingesting table. + key : dict + The key ``make()`` was called with. + version : str, optional + Code version, as resolved for the job. + """ + value = { + "table": table_name, + "key": {k: _jsonable(v) for k, v in (key or {}).items()}, + } + if version: + value["version"] = version + return _ingesting.set(value) + + +def reset_ingesting(token): + """Restore the ingesting context saved by :func:`set_ingesting`.""" + if token is not None: + _ingesting.reset(token) + + +@contextlib.contextmanager +def ingesting(table_name, key, version=None): + """Scope :func:`set_ingesting` to a block.""" + token = set_ingesting(table_name, key, version) + try: + yield + finally: + reset_ingesting(token) + + +def _jsonable(value): + """Render a key value in a form ``json.dumps`` accepts.""" + if isinstance(value, (str, int, float, bool)) or value is None: + return value + if isinstance(value, (datetime.datetime, datetime.date, datetime.time)): + return value.isoformat() + if isinstance(value, bytes): + return value.hex() + return str(value) + + +def build_payload(connection, config=None): + """Assemble the provenance record for rows inserted on this connection. + + Returns ``None`` when there is nothing worth recording, so that a row is + left with ``NULL`` rather than an empty object. + """ + if config is None: + from .settings import config as _config + + config = _config + + payload: dict[str, Any] = {"time": datetime.datetime.now(datetime.timezone.utc).isoformat()} + + conn_info = getattr(connection, "conn_info", None) or {} + agent = {key: conn_info[key] for key in ("user", "host", "database_name") if conn_info.get(key) is not None} + if agent: + payload["agent"] = agent + + try: + from .jobs import _get_job_version + + version = _get_job_version(getattr(connection, "_config", None) or config) + except Exception: # version capture must never break an insert + version = "" + if version: + payload["version"] = version + + source = config.provenance.source + if source: + payload["source"] = source + + context = _ingesting.get() + if context: + payload["context"] = context + + # Time alone says nothing about origin; without any of the other three this + # is noise rather than a record. + return payload if len(payload) > 1 else None + + +def serialize(payload): + """Render a payload for the ``json`` column. + + ``default=str`` because ``config.provenance.source`` is deployment-supplied + and typed ``dict[str, Any]``: a ``date`` or a ``Path`` in it would otherwise + raise from inside every insert into every Entry table, with an error naming + neither provenance nor the setting that caused it. + """ + return json.dumps(payload, default=str) diff --git a/src/datajoint/settings.py b/src/datajoint/settings.py index 7a4d99c1c..d99b2d8f7 100644 --- a/src/datajoint/settings.py +++ b/src/datajoint/settings.py @@ -312,6 +312,49 @@ class JobsSettings(BaseSettings): ) +class ProvenanceSettings(BaseSettings): + """Extrinsic provenance capture at pipeline entry points.""" + + model_config = SettingsConfigDict( + env_prefix="DJ_PROVENANCE_", + case_sensitive=False, + extra="forbid", + validate_assignment=True, + ) + + capture: bool = Field( + default=True, + description="Add the hidden `_prov` attribute to Entry (dj.Manual) tables at declaration " + "and fill it on insert. Tables declared while this is False never receive the column; " + "use datajoint.deploy.add_prov_column to add it to an existing table.", + ) + source: dict[str, Any] = Field( + default_factory=dict, + description="External source identity recorded on every row this process enters, e.g. " + "{'system': 'PyRat', 'endpoint': 'https://...'}. Set per deployment via " + "DJ_PROVENANCE_SOURCE (as JSON), the config file, or the secrets directory. " + "No author supplies this at the insert call site.", + ) + + @field_validator("source") + @classmethod + def _source_must_be_json_serializable(cls, value: dict) -> dict: + """Reject a source that cannot be recorded, at the point it is set. + + Every insert into an Entry table serializes this. Without the check the + failure surfaces from inside an unrelated insert, naming neither + provenance nor the setting responsible. + """ + try: + json.dumps(value, default=str) + except (TypeError, ValueError) as error: + raise ValueError( + f"provenance.source must be JSON-serializable; it is recorded on every row " + f"entering an Entry table. {error.__class__.__name__}: {error}" + ) from error + return value + + class Config(BaseSettings): """ Main DataJoint configuration. @@ -348,6 +391,7 @@ class Config(BaseSettings): connection: ConnectionSettings = Field(default_factory=ConnectionSettings) display: DisplaySettings = Field(default_factory=DisplaySettings) jobs: JobsSettings = Field(default_factory=JobsSettings) + provenance: ProvenanceSettings = Field(default_factory=ProvenanceSettings) # Unified stores configuration (replaces external and object_storage) # ``validation_alias`` redirects pydantic-settings' env source away from the diff --git a/src/datajoint/table.py b/src/datajoint/table.py index 5c1c84598..16737da07 100644 --- a/src/datajoint/table.py +++ b/src/datajoint/table.py @@ -12,6 +12,7 @@ import numpy as np import pandas +from . import provenance from .condition import make_condition from .declare import alter, declare from .dependencies import extract_master @@ -877,6 +878,12 @@ def insert( except StopIteration: pass fields = list(name for name in rows.heading if name in self.heading) + # Carry provenance across rather than leaving the copies NULL. A row + # copied from another table did not originate here, so the source's + # record is the true one; re-stamping it with this moment would claim + # an origin that is not where the data came from. + if self._has_prov_attribute() and provenance.PROV_ATTRIBUTE in (rows.heading._attributes or {}): + fields.append(provenance.PROV_ATTRIBUTE) quoted_fields = ",".join(self.adapter.quote_identifier(f) for f in fields) # Duplicate handling (backend-agnostic) @@ -921,6 +928,7 @@ def _insert_rows(self, rows, replace, skip_duplicates, ignore_extra_fields): # collects the field list from first row (passed by reference) field_list = [] rows = list(self.__make_row_to_insert(row, field_list, ignore_extra_fields) for row in rows) + self._attach_provenance(rows, field_list) if rows: try: # Handle empty field_list (all-defaults insert) @@ -947,6 +955,46 @@ def _insert_rows(self, rows, replace, skip_duplicates, ignore_extra_fields): except DuplicateError as err: raise err.suggest("To ignore duplicate entries in insert, set skip_duplicates=True") + def _has_prov_attribute(self): + """Whether this table carries the hidden extrinsic-provenance attribute.""" + self.heading.attributes # force lazy load; hidden attributes are filtered out of it + all_attrs = self.heading._attributes + return all_attrs is not None and provenance.PROV_ATTRIBUTE in all_attrs + + def _attach_provenance(self, rows, field_list): + """Append the framework-owned `_prov` value to every row of an insert. + + No author supplies this: the content comes from configuration, from the + connection, and -- inside a `make()` -- from the ingesting table and key. + Rows are modified in place, and `field_list` gains the attribute so the + column list matches. + """ + if not rows or not self.connection._config.provenance.capture: + return + if not self._has_prov_attribute(): + # Declared before capture was enabled. datajoint.deploy.add_prov_column + # adds the slot to such a table. + return + try: + payload = provenance.build_payload(self.connection, self.connection._config) + value = provenance.serialize(payload) if payload is not None else None + except Exception as error: + # Recording where a row came from must never stop it being written. + # `source` is deployment-supplied and typed `dict[str, Any]`, so this + # is reachable from configuration alone; the validator on that field + # catches the common case at assignment, and this covers the rest. + logger.warning( + f"Provenance not recorded for insert into {self.full_table_name}: " f"{error.__class__.__name__}: {error}" + ) + return + if value is None: + return + for row in rows: + row["names"] = list(row["names"]) + [provenance.PROV_ATTRIBUTE] + row["placeholders"] = list(row["placeholders"]) + ["%s"] + row["values"] = list(row["values"]) + [value] + field_list.append(provenance.PROV_ATTRIBUTE) + def insert_dataframe(self, df, index_as_pk=None, **insert_kwargs): """ Insert DataFrame with explicit index handling. diff --git a/tests/integration/test_entry_provenance.py b/tests/integration/test_entry_provenance.py new file mode 100644 index 000000000..ffec5d1f5 --- /dev/null +++ b/tests/integration/test_entry_provenance.py @@ -0,0 +1,271 @@ +"""Integration tests for the hidden `_prov` attribute on Entry tables. + +Covers what only a real database shows: which tiers receive the column, that the +value is written without any author involvement, that it stays out of the +heading and out of query composition, and that a fan-out write from inside a +`make()` records the ingesting table and key. +""" + +import json + +import pytest + +import datajoint as dj +from datajoint import provenance +from datajoint.deploy import add_prov_column + + +@pytest.fixture +def schema_prov(connection_test, prefix): + """A schema declared with provenance capture on and a source configured.""" + original_capture = dj.config.provenance.capture + original_source = dj.config.provenance.source + dj.config.provenance.capture = True + dj.config.provenance.source = {"system": "PyRat", "endpoint": "https://pyrat.example.org"} + + schema = dj.Schema(prefix + "_entry_prov", connection=connection_test) + + class Subject(dj.Manual): + definition = """ + subject_id : int16 + --- + species : varchar(32) + """ + + class Param(dj.Lookup): + definition = """ + param_id : int16 + --- + value : float32 + """ + contents = [(1, 1.0)] + + class RecordingFile(dj.Manual): + definition = """ + file_id : int16 + --- + path : varchar(255) + """ + + # The fan-out shape: one make() parsing a source into an Entry table that + # carries no foreign key back to the ingesting table. + class Ingest(dj.Imported): + definition = """ + -> RecordingFile + --- + n_rows : int16 + """ + + def make(self, key): + Session.insert1({"session_id": 100 + key["file_id"], "note": "fanned out"}) + self.insert1({**key, "n_rows": 1}) + + class Session(dj.Manual): + definition = """ + session_id : int16 + --- + note : varchar(64) + """ + + class Analysis(dj.Computed): + definition = """ + -> Subject + --- + score : float32 + """ + + def make(self, key): + self.insert1({**key, "score": 1.0}) + + schema(Subject) + schema(Param) + schema(RecordingFile) + schema(Session) + schema(Ingest) + schema(Analysis) + + yield ( + schema, + dict( + Subject=Subject, + Param=Param, + RecordingFile=RecordingFile, + Session=Session, + Ingest=Ingest, + Analysis=Analysis, + ), + ) + + schema.drop() + dj.config.provenance.capture = original_capture + dj.config.provenance.source = original_source + + +def _raw_prov(table): + """Read the hidden attribute directly; it is not reachable through a heading.""" + rows = table.connection.query( + f"SELECT {table.adapter.quote_identifier(provenance.PROV_ATTRIBUTE)} FROM {table.full_table_name}" + ).fetchall() + return [json.loads(r[0]) if isinstance(r[0], (str, bytes)) else r[0] for r in rows] + + +def test_only_entry_tables_get_the_column(schema_prov): + """Computed and Imported provenance is entailed; Lookup and Part need none.""" + _, t = schema_prov + + def has_prov(table): + table.heading.attributes # force load + return provenance.PROV_ATTRIBUTE in table.heading._attributes + + assert has_prov(t["Subject"]()) + assert has_prov(t["Session"]()) + assert has_prov(t["RecordingFile"]()) + assert not has_prov(t["Param"]()) + assert not has_prov(t["Ingest"]()) + assert not has_prov(t["Analysis"]()) + + +def test_insert_records_provenance_without_the_author(schema_prov): + """No `prov=` argument exists; the row carries a record regardless.""" + _, t = schema_prov + t["Subject"].insert1({"subject_id": 1, "species": "mouse"}) + + (record,) = _raw_prov(t["Subject"]()) + assert record["source"] == {"system": "PyRat", "endpoint": "https://pyrat.example.org"} + assert "time" in record + assert record["agent"]["user"] + assert "context" not in record # not inside a make() + + +def test_prov_stays_out_of_the_heading_and_of_queries(schema_prov): + """A hidden attribute must not leak into fetches or joins.""" + _, t = schema_prov + t["Subject"].insert1({"subject_id": 2, "species": "rat"}) + + assert provenance.PROV_ATTRIBUTE not in t["Subject"]().heading.names + row = t["Subject"]().to_dicts()[0] + assert provenance.PROV_ATTRIBUTE not in row + # a join must not trip over the hidden column + joined = (t["Subject"] * t["Param"]).to_dicts() + assert len(joined) == 1 + assert provenance.PROV_ATTRIBUTE not in joined[0] + + +def test_author_cannot_write_prov(schema_prov): + """The slot is framework-owned: passing it is an unknown attribute.""" + _, t = schema_prov + with pytest.raises(Exception): + t["Subject"].insert1({"subject_id": 3, "species": "mouse", "_prov": {"forged": True}}) + + +def test_fan_out_write_records_the_ingesting_make(schema_prov): + """The Session row has no foreign key to Ingest, yet records what wrote it.""" + _, t = schema_prov + t["RecordingFile"].insert1({"file_id": 7, "path": "/data/a.tif"}) + t["Ingest"].populate() + + (record,) = _raw_prov(t["Session"]()) + assert record["context"]["key"] == {"file_id": 7} + assert "_ingest" in record["context"]["table"] + # and the file row itself, inserted outside any make(), carries no context + (file_record,) = _raw_prov(t["RecordingFile"]()) + assert "context" not in file_record + + +def test_capture_off_declares_no_column_and_migration_adds_it(connection_test, prefix): + """The retrofit path for a table declared before capture was on.""" + original = dj.config.provenance.capture + dj.config.provenance.capture = False + schema = dj.Schema(prefix + "_entry_prov_off", connection=connection_test) + + class Legacy(dj.Manual): + definition = """ + legacy_id : int16 + --- + note : varchar(32) + """ + + schema(Legacy) + try: + Legacy().heading.attributes + assert provenance.PROV_ATTRIBUTE not in Legacy().heading._attributes + + # An insert while the column is absent must still succeed, silently. + dj.config.provenance.capture = True + Legacy.insert1({"legacy_id": 1, "note": "before"}) + + preview = add_prov_column(Legacy, dry_run=True) + assert preview["columns_added"] == 1 and preview["ddl"] + + applied = add_prov_column(Legacy, dry_run=False) + assert applied["columns_added"] == 1 + + # idempotent + assert add_prov_column(Legacy, dry_run=False)["columns_added"] == 0 + + Legacy().heading._init_from_database() + assert provenance.PROV_ATTRIBUTE in Legacy().heading._attributes + + # the pre-existing row keeps NULL; a new row carries a record + Legacy.insert1({"legacy_id": 2, "note": "after"}) + values = [ + r[0] + for r in Legacy() + .connection.query(f"SELECT `{provenance.PROV_ATTRIBUTE}` FROM {Legacy().full_table_name} ORDER BY legacy_id") + .fetchall() + ] + assert values[0] is None + assert values[1] is not None + finally: + schema.drop() + dj.config.provenance.capture = original + + +def test_job_tables_do_not_get_prov(schema_prov): + """Regression for the #1555 review: `~` passed the old prefix test. + + The job table is only materialised by a refresh, so a plain populate() does + not surface this -- which is how it survived the first round of tests. + """ + schema, t = schema_prov + t["RecordingFile"].insert1({"file_id": 11, "path": "/data/b.tif"}) + t["Ingest"].jobs.refresh() + t["Ingest"].populate(reserve_jobs=True) + + conn = t["Ingest"]().connection + tables = [r[0] for r in conn.query(f"SHOW TABLES IN `{schema.database}`").fetchall()] + job_tables = [name for name in tables if name.startswith("~")] + assert job_tables, "no job table was created; the test would pass vacuously" + + for name in job_tables: + columns = { + r[0] + for r in conn.query( + "SELECT COLUMN_NAME FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s", + args=(schema.database, name), + ).fetchall() + } + assert provenance.PROV_ATTRIBUTE not in columns, f"{name} carries {provenance.PROV_ATTRIBUTE}" + + +def test_insert_from_query_carries_provenance_across(schema_prov): + """`insert(QueryExpression)` builds INSERT ... SELECT and used to leave NULL. + + A copied row did not originate in the destination, so the source's record is + the true one; re-stamping it here would claim an origin that is not where + the data came from. + """ + schema, t = schema_prov + t["Subject"].insert1({"subject_id": 40, "species": "mouse"}) + (original,) = _raw_prov(t["Subject"]() & "subject_id = 40") + assert original is not None + + class SubjectCopy(dj.Manual): + definition = t["Subject"].definition + + schema(SubjectCopy) + SubjectCopy.insert(t["Subject"]() & "subject_id = 40") + + (copied,) = _raw_prov(SubjectCopy()) + assert copied is not None, "copied row lost its provenance" + assert copied == original, "copied row was re-stamped instead of carrying its origin" diff --git a/tests/unit/test_provenance.py b/tests/unit/test_provenance.py new file mode 100644 index 000000000..c2c250fa6 --- /dev/null +++ b/tests/unit/test_provenance.py @@ -0,0 +1,188 @@ +"""Unit tests for extrinsic provenance payload assembly and tier selection. + +These cover the parts that need no database: which tables get the slot, what +goes into the payload, and that the payload is JSON-serializable. +""" + +import json +import re + +import pytest + +from datajoint import provenance +from datajoint.settings import Config + + +class FakeConnection: + """Stands in for a Connection: only conn_info and _config are read.""" + + def __init__(self, config, **conn_info): + self.conn_info = conn_info + self._config = config + + +@pytest.fixture +def config(): + cfg = Config() + cfg.provenance.capture = True + cfg.provenance.source = {} + cfg.jobs.version_method = None + return cfg + + +@pytest.mark.parametrize( + "table_name, expected", + [ + ("subject", True), # Entry + ("session_note", True), # Entry, single underscores are ordinary + ("#param", False), # Lookup + ("_ingest", False), # Imported + ("__analysis", False), # Computed + ("subject__detail", False), # Part of an Entry master + ("__analysis__unit", False), # Part of a Computed master + ("_ingest__row", False), # Part of an Imported master + ("~jobs", False), # job table + ("~~analysis", False), # per-table job queue + ("~lineage", False), # lineage table + ], +) +def test_is_entry_table(table_name, expected): + """Only Entry tables get the slot; parts inherit their master's. + + Exercises the predicate `declare` and `deploy` use: a match against the + Manual tier itself, rather than a list of prefixes to exclude. + """ + from datajoint.user_tables import Manual + + assert bool(re.fullmatch(Manual.tier_regexp, table_name)) is expected + + +def test_payload_is_none_without_anything_to_say(config): + """A bare timestamp is noise, so the row is left NULL instead.""" + conn = FakeConnection(config) + assert provenance.build_payload(conn, config) is None + + +def test_payload_records_the_connection(config): + conn = FakeConnection(config, user="alice", host="db.example.org", database_name="lab") + payload = provenance.build_payload(conn, config) + assert payload["agent"] == {"user": "alice", "host": "db.example.org", "database_name": "lab"} + assert "time" in payload + assert "source" not in payload + assert "context" not in payload + + +def test_payload_records_the_configured_source(config): + config.provenance.source = {"system": "PyRat", "endpoint": "https://pyrat.example.org"} + conn = FakeConnection(config) + payload = provenance.build_payload(conn, config) + assert payload["source"] == {"system": "PyRat", "endpoint": "https://pyrat.example.org"} + + +def test_payload_records_the_ingesting_make(config): + """A fan-out write records what wrote it, without any foreign key.""" + conn = FakeConnection(config, user="worker") + with provenance.ingesting("`lab`.`_ingest`", {"file_id": 7}, version="abc1234"): + payload = provenance.build_payload(conn, config) + assert payload["context"] == { + "table": "`lab`.`_ingest`", + "key": {"file_id": 7}, + "version": "abc1234", + } + # and the context does not leak past the block + assert "context" not in provenance.build_payload(conn, config) + + +def test_ingesting_context_nests_and_restores(config): + conn = FakeConnection(config, user="worker") + with provenance.ingesting("`lab`.`_outer`", {"a": 1}): + with provenance.ingesting("`lab`.`_inner`", {"b": 2}): + assert provenance.build_payload(conn, config)["context"]["table"] == "`lab`.`_inner`" + assert provenance.build_payload(conn, config)["context"]["table"] == "`lab`.`_outer`" + + +def test_payload_survives_unserializable_key_values(config): + """A key value json cannot render is stringified rather than raising.""" + import datetime + import uuid + + conn = FakeConnection(config, user="worker") + key = { + "when": datetime.datetime(2026, 9, 30, 12, 0), + "who": uuid.UUID("12345678-1234-5678-1234-567812345678"), + "raw": b"\xde\xad", + } + with provenance.ingesting("`lab`.`_ingest`", key): + payload = provenance.build_payload(conn, config) + rendered = json.loads(provenance.serialize(payload)) + assert rendered["context"]["key"]["when"] == "2026-09-30T12:00:00" + assert rendered["context"]["key"]["raw"] == "dead" + assert rendered["context"]["key"]["who"] == "12345678-1234-5678-1234-567812345678" + + +def test_settings_defaults_to_capturing(): + """A slot nobody can rely on is a slot nobody codes against.""" + assert Config().provenance.capture is True + assert Config().provenance.source == {} + + +def test_settings_come_from_the_environment(monkeypatch): + """The Platform sets provenance per project through the usual channels.""" + monkeypatch.setenv("DJ_PROVENANCE_CAPTURE", "false") + monkeypatch.setenv("DJ_PROVENANCE_SOURCE", '{"system": "PyRat", "endpoint": "https://x"}') + from datajoint.settings import ProvenanceSettings + + settings = ProvenanceSettings() + assert settings.capture is False + assert settings.source == {"system": "PyRat", "endpoint": "https://x"} + + +def test_entry_test_follows_the_tier_definition_not_a_prefix_list(): + """Regression for #1555 review: `~` tables were Entry tables. + + Excluding the other tiers' prefixes one by one means any tier added later is + an Entry table until someone remembers this function. Matching + `Manual.tier_regexp` inverts that: a name is an Entry table only if the + library says it is. + """ + from datajoint.user_tables import Computed, Imported, Lookup, Manual, Part + + def grants_prov(name): + return re.fullmatch(Manual.tier_regexp, name) is not None + + assert grants_prov("subject") + for tier in (Lookup, Imported, Computed, Part): + sample = {Lookup: "#param", Imported: "_ingest", Computed: "__analysis", Part: "subject__detail"}[tier] + assert re.fullmatch(tier.tier_regexp, sample), f"{sample} is not a {tier.__name__}" + assert not grants_prov(sample) + # The job prefix belongs to no user tier at all, which is how it slipped through. + assert not any(re.fullmatch(t.tier_regexp, "~~analysis") for t in (Manual, Lookup, Imported, Computed, Part)) + assert not grants_prov("~~analysis") + + +def test_serialize_survives_a_deployment_supplied_source(config): + """`source` is dict[str, Any]; a date in it must not break an insert.""" + import datetime + import pathlib + + where = pathlib.Path("/mnt/raw") + payload = { + "time": "t", + "source": {"when": datetime.date(2026, 1, 1), "where": where}, + } + rendered = json.loads(provenance.serialize(payload)) + # Compare against str(Path), not a literal: the separator is platform-specific + # and the contract under test is that the value is stringified at all. + assert rendered["source"] == {"when": "2026-01-01", "where": str(where)} + + +def test_source_must_be_serializable_at_assignment(): + """The error belongs where the setting is made, not inside an unrelated insert.""" + from pydantic import ValidationError + + from datajoint.settings import ProvenanceSettings + + cyclic: dict = {} + cyclic["self"] = cyclic + with pytest.raises(ValidationError, match="JSON-serializable"): + ProvenanceSettings(source=cyclic)