diff --git a/docs/configuration/table.md b/docs/configuration/table.md index 0c4a8bf..91c6e8e 100644 --- a/docs/configuration/table.md +++ b/docs/configuration/table.md @@ -428,7 +428,7 @@ provenance: #### Manual provenance override -Use `provenance.override` when a table comes from another knowledge graph or source system whose Translator provenance cannot be derived from a PMC/PMID publication. The override is wired like the other Tablassert model classes and wins over the repo/publication auto-generation for that section's upstream sources, publications, and KL/AT. The primary entry of the edge `sources` list (`resource_role: primary_knowledge_source`) is **not** overridable per section; it always derives from the graph-level `infores` (see [Graph](graph.md)); put manual infores CURIEs in `upstream_resource_ids`. +Use `provenance.override` when a table comes from another knowledge graph or source system whose Translator provenance cannot be derived from a PMC/PMID publication. The override is wired like the other Tablassert model classes and wins over the repo/publication auto-generation for that section's upstream sources, publications, and KL/AT. The primary entry of the edge `sources` list (`resource_role: primary_knowledge_source`) is **not** re-targetable per section via `upstream_resource_ids`; it derives from the graph-level `infores` (see [Graph](graph.md)) unless an explicit [`sources` template](#explicit-sources-template) is given. Put manual infores CURIEs in `upstream_resource_ids`. ```yaml provenance: @@ -445,13 +445,49 @@ Override fields: | Field | Type | Required | Description | |-------|------|----------|-------------| +| `sources` | List[Object] | No | Explicit retrieval-`sources` entry templates replacing the derived primary/upstream emission entirely. Mutually exclusive with `upstream_resource_ids` and `upstream_source_record_urls` (the template subsumes both). See [Explicit sources template](#explicit-sources-template). | | `upstream_resource_ids` | List[String] | No | Manual upstream source infores CURIEs replacing the repo-derived `PMC`/`PMID` source map, the sanctioned place for manual infores. Each entry must start with `infores:`. | | `upstream_source_record_urls` | Map[String, List[URL]] | No | Per-upstream source record URLs keyed by infores CURIE; every key must appear in `upstream_resource_ids`. When set, the section's `source.url` values serve the RIG only and are NOT emitted on the primary `sources` entry; each listed upstream supporting entry carries its own `source_record_urls` instead. | | `publications` | List[String] | No | Manual publication CURIEs. Entries must currently start with `PMCID:`; PMID compatibility for manual overrides is intentionally deferred. | | `knowledge_level` | String | No | Override-specific KL value. Defaults to `statistical_association`. | | `agent_type` | String | No | Override-specific AT value. Defaults to `data_analysis_pipeline`. | -Tablassert emits the graph-level infores (or `infores:` when unset) as the primary entry of the Biolink `sources` list on each edge, `{resource_id: "infores:multiomics-kg", resource_role: "primary_knowledge_source", upstream_resource_ids: [...], source_record_urls: [...]}`, with one additional `supporting_data_source` entry per upstream. When `override.upstream_source_record_urls` is set, the primary entry emits no `source_record_urls` and each mapped supporting entry carries its own instead. No flat `primary_knowledge_source` scalar is emitted: current translator-ingests practice carries retrieval provenance only in `sources`, and the Biolink `RetrievalSource` class is where `resource_id` / `upstream_resource_ids` / `source_record_urls` are defined. Each retrieval-source entry uses `resource_id` as its sole identifier. The override cannot set a per-section primary source; manual infores CURIEs belong in `upstream_resource_ids`. Older flat `resource_id` / `primary_knowledge_source` output has been removed so generated KGX matches the Biolink edge contract. +Tablassert emits the graph-level infores (or `infores:` when unset) as the primary entry of the Biolink `sources` list on each edge, `{resource_id: "infores:multiomics-kg", resource_role: "primary_knowledge_source", upstream_resource_ids: [...], source_record_urls: [...]}`, with one additional `supporting_data_source` entry per upstream. When `override.upstream_source_record_urls` is set, the primary entry emits no `source_record_urls` and each mapped supporting entry carries its own instead. When `override.sources` is set, that derivation is skipped entirely and the template entries are emitted verbatim, in order. No flat `primary_knowledge_source` scalar is emitted: current translator-ingests practice carries retrieval provenance only in `sources`, and the Biolink `RetrievalSource` class is where `resource_id` / `upstream_resource_ids` / `source_record_urls` are defined. Each retrieval-source entry uses `resource_id` as its sole identifier. Without `override.sources`, the override cannot set a per-section primary source; manual infores CURIEs belong in `upstream_resource_ids`. Older flat `resource_id` / `primary_knowledge_source` output has been removed so generated KGX matches the Biolink edge contract. + +#### Explicit sources template + +`override.sources` replaces the derived `sources` list with an explicit, ordered template — one Biolink `RetrievalSource` struct per entry. Use it when the emitted provenance must differ structurally from the "graph infores as primary + one supporting entry per upstream" shape, e.g. to mark the graph as an `aggregator_knowledge_source` with the real primary elsewhere. + +```yaml +provenance: + override: + sources: + - resource_id: infores:multiomics-drugapprovals + resource_role: aggregator_knowledge_source + upstream_resource_ids: [infores:dailymed, infores:faers] + source_record_urls: + - "https://db.systemsbiology.net/gestalt/cgi-pub/KGinfo.pl?id={edge_id}" + - resource_id: infores:faers + resource_role: primary_knowledge_source + - resource_id: infores:dailymed + resource_role: supporting_data_source +``` + +Entry fields: + +| Field | Type | Required | Description | +|-------|------|----------|-------------| +| `resource_id` | String | Yes | Infores CURIE of this entry (its sole identifier on output). Must start with `infores:`, and must be unique within the template. | +| `resource_role` | String | Yes | One of the Biolink `ResourceRoleEnum` values: `primary_knowledge_source`, `aggregator_knowledge_source`, or `supporting_data_source`. Anything else fails KGX validation downstream, so it is rejected at config time. | +| `upstream_resource_ids` | List[String] | No | Upstream infores CURIEs carried by this entry; each must start with `infores:`. | +| `source_record_urls` | List[String] | No | Source record URLs carried by this entry. Each entry must be an absolute `http(s)://` URL, optionally containing the `{edge_id}` placeholder (see below). | + +Validation rules (all reported with error code `override-bad-sources`): + +- `sources` is **mutually exclusive** with `upstream_resource_ids` and `upstream_source_record_urls` — the template subsumes both, so combining them fails loudly. +- When set, the list must be **non-empty**, `resource_id` values must be **unique**, and at least one entry must carry role `primary_knowledge_source` or `aggregator_knowledge_source`. + +**The `{edge_id}` placeholder.** A `source_record_urls` entry may embed the literal string `{edge_id}` to build per-edge URLs (e.g. a drill-down page keyed by edge id). The edge `id` is a deterministic content hash assigned at the final dedup stage — *after* subgraphs are written — so the placeholder cannot be interpolated during the table build. Instead, the literal placeholder is emitted into the subgraph (and is what the content hash covers, keeping ids deterministic), and a post-dedup sweep of the final `*.edges.ndjson` replaces `{edge_id}` with each record's own `id` in every `sources[].source_record_urls` string. Files without the marker are left byte-identical. ### Annotations diff --git a/src/tablassert/errors.py b/src/tablassert/errors.py index cf9744c..af843dc 100644 --- a/src/tablassert/errors.py +++ b/src/tablassert/errors.py @@ -22,6 +22,7 @@ "encoding-bad-remove-entry", "graph-bad-infores", "override-bad-publication", + "override-bad-sources", "override-bad-upstream-infores", "override-bad-upstream-urls", "provenance-bad-pmc-id", diff --git a/src/tablassert/lib.py b/src/tablassert/lib.py index 9ea8e8d..206d200 100644 --- a/src/tablassert/lib.py +++ b/src/tablassert/lib.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json import math import operator import re @@ -44,7 +45,7 @@ from tablassert.enums import EncodingMethods, Files, InformationResources, Repositories, Tokens from tablassert.fullmap import ResolveSpec, fullmap_db_path, resolve, resolve_batch from tablassert.log import cat -from tablassert.models import Encoding, NodeEncoding, Qualifier, RIGConfig, Section +from tablassert.models import EDGE_ID_PLACEHOLDER, Encoding, NodeEncoding, Qualifier, RIGConfig, Section from tablassert.nlp import level_one, level_two from tablassert.qc import fullmap_audit from tablassert.rig import ( @@ -554,7 +555,12 @@ def inline_supporting_study(lf: pl.LazyFrame, study_id: str, study_name: str | N def retrieval_sources( - lf: pl.LazyFrame, primary: str, upstream: list[str], urls: list[str], upstream_urls: dict[str, list[str]] | None = None + lf: pl.LazyFrame, + primary: str, + upstream: list[str], + urls: list[str], + upstream_urls: dict[str, list[str]] | None = None, + explicit: list[dict[str, Any]] | None = None, ) -> pl.LazyFrame: """Add the Biolink ``sources`` retrieval-provenance column. @@ -572,18 +578,32 @@ def retrieval_sources( its own ``source_record_urls`` and the primary entry emits none (the primary is the transforming resource, not a downloadable record). + When ``explicit`` is given, exactly those entries are emitted, in order, and + ``primary``/``upstream``/``urls``/``upstream_urls`` are ignored. Each entry + template carries ``resource_id``, ``resource_role``, and optional + ``upstream_resource_ids``/``source_record_urls``; ``source_record_urls`` values + may contain the literal ``{edge_id}`` placeholder, which is NOT resolved here + (the edge id is only assigned at the final dedup stage) but in a post-dedup + sweep of the final edges NDJSON. + Args: lf: Source LazyFrame. primary: Infores CURIE of the primary knowledge source. upstream: Infores CURIEs of upstream/supporting data sources. urls: Source record URLs for the primary entry (ignored when ``upstream_urls`` is set). upstream_urls: Optional per-upstream source record URLs keyed by infores CURIE. + explicit: Optional explicit entry templates emitted verbatim, in order. Returns: LazyFrame with a ``sources`` ``list[struct]`` column appended. """ - if upstream_urls is not None: - entries: list[pl.Expr] = [_retrieval_source(primary, "primary_knowledge_source", upstream)] + if explicit is not None: + entries: list[pl.Expr] = [ + _retrieval_source(entry["resource_id"], entry["resource_role"], entry.get("upstream_resource_ids"), entry.get("source_record_urls")) + for entry in explicit + ] + elif upstream_urls is not None: + entries = [_retrieval_source(primary, "primary_knowledge_source", upstream)] entries.extend(_retrieval_source(x, "supporting_data_source", urls=upstream_urls.get(x)) for x in upstream) else: entries = [_retrieval_source(primary, "primary_knowledge_source", upstream, urls)] @@ -1253,6 +1273,12 @@ def _provenance_ops(self: Self) -> list[Any]: if override and override.upstream_source_record_urls is not None else None ) + # An explicit `sources` template replaces the derived primary/upstream + # emission entirely; the model already forbids combining it with + # `upstream_resource_ids`/`upstream_source_record_urls`. + explicit_sources: list[dict[str, Any]] | None = ( + [entry.model_dump(exclude_none=True) for entry in override.sources] if override and override.sources is not None else None + ) knowledge_level = override.knowledge_level if override else self.provenance.knowledge_level agent_type = override.agent_type if override else self.provenance.agent_type publication_values = override.publications if override else [publication_curie(self.provenance.repo, self.provenance.publication or "")] @@ -1278,8 +1304,9 @@ def _provenance_ops(self: Self) -> list[Any]: # RetrievalSource); current translator-ingests emits no flat # `primary_knowledge_source` scalar, so neither do we. A per-upstream URL # mapping (override.upstream_source_record_urls) re-homes the record URLs - # from the primary entry onto the matching supporting entries. - (retrieval_sources, (primary_knowledge_source, upstream_ids, [str(u) for u in self.source.url], upstream_urls)) + # from the primary entry onto the matching supporting entries; an explicit + # `sources` template (override.sources) replaces the whole derivation. + (retrieval_sources, (primary_knowledge_source, upstream_ids, [str(u) for u in self.source.url], upstream_urls, explicit_sources)) if primary_knowledge_source else None, (publications, (publication_values,)) if publication_values else None, @@ -1520,6 +1547,44 @@ def dedup_stream(p_in: Path, is_edges: bool) -> None: p_in.unlink() +def _resolve_edge_id_placeholders(edges_path: Path) -> None: + """Resolve ``{edge_id}`` placeholders in a final edges NDJSON file. + + The edge ``id`` is a deterministic content hash assigned by the Rust deduper + after subgraphs are written, so explicit ``override.sources`` record URLs + cannot embed it during the polars build: the literal placeholder is what + gets hashed, and this post-dedup sweep substitutes each record's own id into + every string inside every ``sources[].source_record_urls`` list. The pass is + skipped entirely when no line contains the marker (cheap substring precheck, + no full parse), leaving the file byte-identical. + + Args: + edges_path: Path to the deduplicated ``*.edges.ndjson`` file. + + Returns: + ``None``; rewrites ``edges_path`` in place via a temp file when any + placeholder was resolved. + """ + tmp_path: Path = edges_path.with_name(edges_path.name + ".placeholder.tmp") + resolved: bool = False + with edges_path.open("r", encoding="utf-8") as src, tmp_path.open("w", encoding="utf-8") as dst: + for line in src: + if EDGE_ID_PLACEHOLDER not in line: + dst.write(line) + continue + record: dict[str, Any] = json.loads(line) + for source in record.get("sources") or []: + urls: list[str] | None = source.get("source_record_urls") + if urls: + source["source_record_urls"] = [url.replace(EDGE_ID_PLACEHOLDER, record["id"]) for url in urls] + dst.write(json.dumps(record, ensure_ascii=False) + "\n") + resolved = True + if resolved: + tmp_path.replace(edges_path) + else: + tmp_path.unlink() + + def fold_unknown_to_supporting_text(lf: pl.LazyFrame) -> pl.LazyFrame: """Fold any non-Biolink edge column into ``supporting_text`` as ``col: value`` strings. @@ -1631,6 +1696,12 @@ def _write_ndjson( ``compile_graph``; each commented phase boundary below is a hook point for the US-009 ``on_phase`` progress callback. + Dedup assigns each edge ``id`` as the deterministic content hash of the + PRE-resolution record -- the literal ``{edge_id}`` placeholder in explicit + ``override.sources`` record URLs is what gets hashed -- and the placeholder + sweep that follows substitutes the assigned id into the final edges file + only, keeping ids deterministic. + Args: subnodes: Per-section node LazyFrames from ``_collect_subframes``. subedges: Per-section edge LazyFrames from ``_collect_subframes``. @@ -1665,6 +1736,9 @@ def _write_ndjson( if on_phase is not None: on_phase("dedup") dedup_stream(edges_tmp, is_edges=True) + # The deduper hashes the record WITH the literal `{edge_id}` placeholder still + # in place, so edge ids stay deterministic regardless of this resolution pass. + _resolve_edge_id_placeholders(edges_tmp.with_suffix("")) dedup_stream(nodes_tmp, is_edges=False) # Phase: rig. Summaries come from the FINAL deduplicated KGX files, and the diff --git a/src/tablassert/models.py b/src/tablassert/models.py index 0417595..795adbc 100644 --- a/src/tablassert/models.py +++ b/src/tablassert/models.py @@ -466,16 +466,99 @@ def validate_infores_curie(value: str, code: TablassertErrorCodes) -> str: return value +EDGE_ID_PLACEHOLDER: str = "{edge_id}" +"""Placeholder for the final edge id inside ``override.sources`` record URLs. + +The edge ``id`` is a deterministic content hash computed during the final dedup +stage -- after subgraphs are written -- so a per-edge URL cannot embed it during +the table build. The literal placeholder is emitted as-is and resolved against +the deduplicated ``*.edges.ndjson`` in a post-dedup sweep. +""" + +RESOURCE_ROLES: tuple[str, ...] = ("primary_knowledge_source", "aggregator_knowledge_source", "supporting_data_source") +"""Valid Biolink ``ResourceRoleEnum`` values for retrieval ``sources`` entries. + +Kept as literals instead of importing ``ResourceRoleEnum`` from the generated +``biolink_model`` Pydantic classes so config validation stays cheap; any other +value fails KGX validation downstream. +""" + + +class SourceOverride(TablaBase): + """One explicit retrieval-``sources`` entry template for manual provenance. + + Each entry becomes one Biolink ``RetrievalSource`` struct on the edge's + ``sources`` list, replacing the derived primary/upstream emission entirely. + """ + + resource_id: str = Field(description="Infores CURIE of this retrieval source entry.", examples=["infores:my-source"]) + resource_role: str = Field( + description="Biolink ResourceRoleEnum value for this entry.", + examples=["primary_knowledge_source", "aggregator_knowledge_source", "supporting_data_source"], + ) + upstream_resource_ids: list[str] | None = Field( + None, description="Upstream infores CURIEs carried by this entry.", examples=[["infores:my-upstream"]] + ) + source_record_urls: list[str] | None = Field( + None, + description=f"Source record URLs carried by this entry; `{EDGE_ID_PLACEHOLDER}` is replaced with the final edge id after dedup.", + examples=[["https://example.org/edge?id={edge_id}"]], + ) + + @field_validator("resource_id", mode="after") + @classmethod + def infores_resource_id(cls, value: str) -> str: + return validate_infores_curie(value, "override-bad-sources") + + @field_validator("resource_role", mode="after") + @classmethod + def biolink_resource_role(cls, value: str) -> str: + if value not in RESOURCE_ROLES: + raise TablassertValidationError(f"`resource_role` must be one of {list(RESOURCE_ROLES)}, got {value!r}.", code="override-bad-sources") + return value + + @field_validator("upstream_resource_ids", mode="after") + @classmethod + def infores_upstream_resource_ids(cls, values: list[str] | None) -> list[str] | None: + if values is None: + return None + for value in values: + validate_infores_curie(value, "override-bad-sources") + return values + + @field_validator("source_record_urls", mode="after") + @classmethod + def url_or_edge_id_template(cls, values: list[str] | None) -> list[str] | None: + # Typed as plain str (not HttpUrl) so the `{edge_id}` placeholder survives + # validation; after removing placeholder occurrences the rest must still be + # an absolute http(s) URL. + if values is None: + return None + for value in values: + stripped: str = value.replace(EDGE_ID_PLACEHOLDER, "") + if not stripped.startswith(("https://", "http://")): + raise TablassertValidationError( + f"`source_record_urls` entries must be http(s) URLs (optionally containing `{EDGE_ID_PLACEHOLDER}`), got {value!r}.", + code="override-bad-sources", + ) + return values + + class ManualProvenance(TablaBase): """Manually-specified provenance for non-PMID/PMC source graphs. When present under :class:`Provenance`, these values replace the legacy repo/publication-derived provenance while keeping the same KL/AT defaults. The primary ``sources`` entry (``resource_role: primary_knowledge_source``) - always derives from the graph-level ``rig.source_info.infores_id``; manual - infores CURIEs belong in ``upstream_resource_ids``. + derives from the graph-level ``rig.source_info.infores_id`` unless an + explicit ``sources`` template is given; manual infores CURIEs otherwise + belong in ``upstream_resource_ids``. """ + sources: list[SourceOverride] | None = Field( + None, + description="Explicit retrieval-`sources` entry templates replacing the derived primary/upstream emission entirely; mutually exclusive with `upstream_resource_ids` and `upstream_source_record_urls`, which it subsumes.", + ) upstream_resource_ids: list[str] = Field( default_factory=list, description="Manual upstream source infores CURIEs emitted instead of the repo-derived source map; the sanctioned place for manual infores.", @@ -515,6 +598,26 @@ def pmcid_publications(cls, values: list[str] | None) -> list[str] | None: ) return values + @model_validator(mode="after") + def sources_template_is_coherent(self: Self) -> Self: + if self.sources is None: + return self + if self.upstream_resource_ids or self.upstream_source_record_urls is not None: + raise TablassertValidationError( + "`sources` is mutually exclusive with `upstream_resource_ids` and `upstream_source_record_urls`; the explicit template subsumes both.", + code="override-bad-sources", + ) + if not self.sources: + raise TablassertValidationError("`sources` must contain at least one entry when set.", code="override-bad-sources") + resource_ids: list[str] = [entry.resource_id for entry in self.sources] + if len(set(resource_ids)) != len(resource_ids): + raise TablassertValidationError("`sources` entries must have unique `resource_id` values.", code="override-bad-sources") + if not any(entry.resource_role in ("primary_knowledge_source", "aggregator_knowledge_source") for entry in self.sources): + raise TablassertValidationError( + "`sources` must include at least one `primary_knowledge_source` or `aggregator_knowledge_source` entry.", code="override-bad-sources" + ) + return self + @model_validator(mode="after") def upstream_urls_match_resource_ids(self: Self) -> Self: if self.upstream_source_record_urls is None: diff --git a/tests/test_lib.py b/tests/test_lib.py index 78261a2..3aaa277 100644 --- a/tests/test_lib.py +++ b/tests/test_lib.py @@ -778,6 +778,92 @@ def test_tcode_collect_upstream_source_record_urls_rehome_urls(fixtures_path: Pa assert by_resource["infores:other-source"]["source_record_urls"] is None +def test_tcode_collect_explicit_sources_override_replaces_derivation(fixtures_path: Path) -> None: + """``override.sources`` is forwarded as the op's ``explicit`` arg and emitted verbatim, in order.""" + template: list[dict[str, Any]] = [ + { + "resource_id": "infores:multiomics-drugapprovals", + "resource_role": "aggregator_knowledge_source", + "upstream_resource_ids": ["infores:dailymed", "infores:faers"], + "source_record_urls": ["https://db.systemsbiology.net/gestalt/cgi-pub/KGinfo.pl?id={edge_id}"], + }, + {"resource_id": "infores:faers", "resource_role": "primary_knowledge_source"}, + {"resource_id": "infores:dailymed", "resource_role": "supporting_data_source"}, + ] + data: Any = from_yaml(fixtures_path / "minimal_section.yaml") + data["provenance"] = {"override": {"sources": template}} + store: Path = Path("/tmp/sectionhash.parquet") + tcode_model: Tcode = Tcode.model_validate( # pyright: ignore + {**data, "config": fixtures_path / "minimal_section.yaml", "store": store, "name": "MULTIOMICS_KG", "infores": "infores:multiomics-kg"} + ) + + collected: list[tuple[Any, tuple[Any]]] = tcode_model.collect(Path("/tmp/fullmap.redb")) # pyright: ignore + source_ops: list[tuple[Any, tuple[Any]]] = [op for op in collected if op[0] is retrieval_sources] + assert len(source_ops) == 1 + # The template is forwarded (with unset optional fields dropped) as the fifth op arg. + source_args: tuple[Any, ...] = source_ops[0][1] # pyright: ignore[reportAssignmentType] + assert source_args[4] == template + + result: pl.DataFrame = source_ops[0][0](pl.LazyFrame({"subject": ["A"]}), *source_args).collect() + sources: list[dict[str, Any]] = result["sources"].to_list()[0] + assert [s["resource_id"] for s in sources] == ["infores:multiomics-drugapprovals", "infores:faers", "infores:dailymed"] + assert [s["resource_role"] for s in sources] == ["aggregator_knowledge_source", "primary_knowledge_source", "supporting_data_source"] + # `resource_id` is the sole identifier on each entry (no `id` mirror, #115). + assert all("id" not in s for s in sources) + # The `{edge_id}` placeholder stays unresolved at this stage: the edge id is a + # content hash assigned by the final dedup stage, after subgraphs are written. + assert sources[0]["source_record_urls"] == ["https://db.systemsbiology.net/gestalt/cgi-pub/KGinfo.pl?id={edge_id}"] + assert sources[0]["upstream_resource_ids"] == ["infores:dailymed", "infores:faers"] + # Entries without urls/upstream emit typed nulls, like the default path. + assert sources[1]["source_record_urls"] is None + assert sources[1]["upstream_resource_ids"] is None + assert sources[2]["source_record_urls"] is None + + +def test_resolve_edge_id_placeholders(tmp_path: Path) -> None: + """The post-dedup sweep substitutes each record's own ``id`` for ``{edge_id}``, edges file only.""" + edges: Path = tmp_path / "graph.edges.ndjson" + with_placeholder: dict[str, Any] = { + "id": "uuid-1", + "subject": "CURIE:1", + "sources": [ + { + "id": "infores:multiomics-drugapprovals", + "resource_id": "infores:multiomics-drugapprovals", + "resource_role": "aggregator_knowledge_source", + "source_record_urls": ["https://db.systemsbiology.net/gestalt/cgi-pub/KGinfo.pl?id={edge_id}", "https://example.org/static"], + } + ], + } + without_placeholder: dict[str, Any] = { + "id": "uuid-2", + "subject": "CURIE:2", + "sources": [{"id": "infores:faers", "resource_id": "infores:faers", "resource_role": "primary_knowledge_source"}], + } + plain_without: str = json.dumps(without_placeholder) + edges.write_text(json.dumps(with_placeholder) + "\n" + plain_without + "\n", encoding="utf-8") + + lib._resolve_edge_id_placeholders(edges) + + lines: list[str] = edges.read_text(encoding="utf-8").splitlines() + resolved: dict[str, Any] = json.loads(lines[0]) + assert resolved["sources"][0]["source_record_urls"] == [ + "https://db.systemsbiology.net/gestalt/cgi-pub/KGinfo.pl?id=uuid-1", + "https://example.org/static", + ] + assert resolved["id"] == "uuid-1" + # Lines without the marker pass through untouched. + assert lines[1] == plain_without + assert not (tmp_path / "graph.edges.ndjson.placeholder.tmp").exists() + + # A file without the marker is left byte-identical. + clean: Path = tmp_path / "clean.edges.ndjson" + content: str = plain_without + "\n" + clean.write_text(content, encoding="utf-8") + lib._resolve_edge_id_placeholders(clean) + assert clean.read_text(encoding="utf-8") == content + + def test_tcode_original_value_before_regex_for_columns(fixtures_path: Path) -> None: """tcode captures original value before regex for column encoded nodes.""" data: Any = from_yaml(fixtures_path / "minimal_section.yaml") diff --git a/tests/test_models.py b/tests/test_models.py index f9ff2b2..a7c8e91 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -647,6 +647,109 @@ def test_manual_provenance_rejects_infores_key() -> None: assert "infores" in str(exc_info.value) +def test_manual_provenance_accepts_explicit_sources_template() -> None: + """an explicit ``sources`` template emits valid RetrievalSource entries verbatim.""" + override = ManualProvenance( # pyright: ignore + sources=[ + { + "resource_id": "infores:multiomics-drugapprovals", + "resource_role": "aggregator_knowledge_source", + "upstream_resource_ids": ["infores:dailymed", "infores:faers"], + "source_record_urls": ["https://db.systemsbiology.net/gestalt/cgi-pub/KGinfo.pl?id={edge_id}"], + }, + {"resource_id": "infores:faers", "resource_role": "primary_knowledge_source"}, + {"resource_id": "infores:dailymed", "resource_role": "supporting_data_source"}, + ] + ) + assert override.sources is not None + assert [entry.resource_id for entry in override.sources] == ["infores:multiomics-drugapprovals", "infores:faers", "infores:dailymed"] + assert override.sources[0].source_record_urls == ["https://db.systemsbiology.net/gestalt/cgi-pub/KGinfo.pl?id={edge_id}"] + assert override.sources[1].upstream_resource_ids is None + + +def test_manual_provenance_sources_rejects_bad_role() -> None: + """``sources.resource_role`` must be a Biolink ResourceRoleEnum value.""" + with pytest.raises(ValidationError) as exc_info: + ManualProvenance(sources=[{"resource_id": "infores:external-source", "resource_role": "knowledge_source"}]) # pyright: ignore + assert "override-bad-sources" in str(exc_info.value) + + +def test_manual_provenance_sources_rejects_non_infores_ids() -> None: + """``sources`` resource_id and upstream entries must be infores CURIEs.""" + with pytest.raises(ValidationError) as exc_info: + ManualProvenance(sources=[{"resource_id": "external-source", "resource_role": "primary_knowledge_source"}]) # pyright: ignore + assert "override-bad-sources" in str(exc_info.value) + + with pytest.raises(ValidationError) as exc_info: + ManualProvenance( # pyright: ignore + sources=[{"resource_id": "infores:external-source", "resource_role": "primary_knowledge_source", "upstream_resource_ids": ["dailymed"]}] + ) + assert "override-bad-sources" in str(exc_info.value) + + +def test_manual_provenance_sources_rejects_non_url_record_urls() -> None: + """``sources.source_record_urls`` entries must be http(s) URLs once ``{edge_id}`` is stripped.""" + with pytest.raises(ValidationError) as exc_info: + ManualProvenance( # pyright: ignore + sources=[{"resource_id": "infores:external-source", "resource_role": "primary_knowledge_source", "source_record_urls": ["not-a-url"]}] + ) + assert "override-bad-sources" in str(exc_info.value) + + with pytest.raises(ValidationError) as exc_info: + ManualProvenance( # pyright: ignore + sources=[ + { + "resource_id": "infores:external-source", + "resource_role": "primary_knowledge_source", + "source_record_urls": ["ftp://example.org/{edge_id}"], + } + ] + ) + assert "override-bad-sources" in str(exc_info.value) + + +def test_manual_provenance_sources_rejects_upstream_field_combinations() -> None: + """``sources`` is mutually exclusive with the upstream fields it subsumes.""" + with pytest.raises(ValidationError) as exc_info: + ManualProvenance( # pyright: ignore + sources=[{"resource_id": "infores:external-source", "resource_role": "primary_knowledge_source"}], + upstream_resource_ids=["infores:upstream-source"], + ) + assert "override-bad-sources" in str(exc_info.value) + + with pytest.raises(ValidationError) as exc_info: + ManualProvenance( # pyright: ignore + sources=[{"resource_id": "infores:external-source", "resource_role": "primary_knowledge_source"}], + upstream_source_record_urls={"infores:upstream-source": ["https://example.org/dataset"]}, + ) + assert "override-bad-sources" in str(exc_info.value) + + +def test_manual_provenance_sources_rejects_incoherent_templates() -> None: + """``sources`` must be non-empty, deduplicated, and carry a primary/aggregator entry.""" + with pytest.raises(ValidationError) as exc_info: + ManualProvenance(sources=[]) # pyright: ignore + assert "override-bad-sources" in str(exc_info.value) + + with pytest.raises(ValidationError) as exc_info: + ManualProvenance( # pyright: ignore + sources=[ + {"resource_id": "infores:external-source", "resource_role": "primary_knowledge_source"}, + {"resource_id": "infores:external-source", "resource_role": "supporting_data_source"}, + ] + ) + assert "override-bad-sources" in str(exc_info.value) + + with pytest.raises(ValidationError) as exc_info: + ManualProvenance( # pyright: ignore + sources=[ + {"resource_id": "infores:external-source", "resource_role": "supporting_data_source"}, + {"resource_id": "infores:other-source", "resource_role": "supporting_data_source"}, + ] + ) + assert "override-bad-sources" in str(exc_info.value) + + def test_provenance_override_replaces_publication_requirement() -> None: """publication is required unless manual provenance override is set.""" p = Provenance(override={"upstream_resource_ids": ["infores:external-source"], "publications": ["PMCID:PMC1234567"]}) # pyright: ignore