diff --git a/.env.example b/.env.example index 0e27f27..6578363 100644 --- a/.env.example +++ b/.env.example @@ -223,6 +223,11 @@ DISCOVERY_SERVICE_ENDPOINT= NETWORK_SENDER_ID= # Publicly reachable base URL of this provider, sent as each catalog's bppUri. NETWORK_SENDER_URI= +NETWORK_PROVIDER_ID_TEMPLATE={kind}.{instance}.provider.oan.dev +NETWORK_ADVISORY_PROVIDER_NAME=Advisories from documents +NETWORK_SCHEMES_PROVIDER_NAME=Schemes from documents +NETWORK_GROUP_PROVIDER_NAME_TEMPLATE={group_name} {provider_name} +PROVIDER_CONFIG_OVERRIDE_PATH= # --- Keycloak container (only when running Keycloak inside docker-compose) --- diff --git a/AGENTS.md b/AGENTS.md index ef17b18..d9706ed 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -61,7 +61,8 @@ Core modules (all under `pipeline/`): - `db.py` — SQLite access layer; SQLite is the **canonical** metadata/review-state store (documents, pages, chunks, jobs, artifacts, audit log), not just a cache - `document_repository.py` — `DocumentRepository`, a domain layer over `db.py` for document reads; `db.py` stays one-function-per-query with no domain knowledge - `network_constants.py` — fixed values sent on the network (catalog/resource ids, topics, languages, Beckn version). The file v2 edits when the AI layer starts deriving them -- `catalog_builder.py` — pure builder mapping a document's knowledge kind (`advisory`/`scheme`) to the single `OnDemand` Beckn catalog announced for that kind; returns `None` for any other kind. No env, no I/O +- `catalog_builder.py` — pure builder mapping a document's knowledge kind (`advisory`/`scheme`) to the single `OnDemand` Beckn catalog announced for that kind; takes the resolved provider and prefixes catalog/resource ids for a Group; returns `None` for any other kind. No env, no I/O +- `network_provider.py` — pure provider resolution for Publish to Network: the provider id (overrides, then `{kind}.{instance}` template) and name for a Knowledge Kind and Group. `network_group.py` builds its `ProviderNaming` from env and the overrides file; `catalog_builder.py` takes the resolved provider - `document_validity.py` — pure module owning Document Validity: the period a document's chunks answer searches in, the upload-day start with no end by default, and the clock injection point. Shared by `api.py` (stamping the default at upload), `scheme_catalog.py` (validating a reviewer's edit), `activities.py` (stamping chunk payloads) and `vector_store/qdrant_store.py` (filtering at search time). `db.py` stores the dates it is handed and defaults nothing - `discovery_publish_service.py` — `DiscoveryPublishService`: owns the Publish to Network env vars and the HTTP call. Deliberately Temporal-free - `models.py` — Pydantic models, including `DocumentStage` enum and `PIPELINE_STAGES` (the stepper-UI stage list) diff --git a/CONTEXT.md b/CONTEXT.md index 9cf0d9f..4fd23e7 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -9,7 +9,7 @@ The operator action that approves a document out of `ready_for_ingestion` and le _Avoid_: "publish" alone — ambiguous with Publish to Network. **Publish to Network**: -The pipeline stage that runs only after a document's PROD promotion succeeds, and POSTs a Beckn-shaped `publish` envelope to an external Discovery Service. Skipped entirely when PROD is disabled (`DISABLE_PROD_SETTING=true`) — a document never reaches the network without first promoting to PROD. The envelope it sends is kind-level (see Knowledge Kind), not document-level: it does not make any single document searchable, only re-announces that this provider serves that kind at all. +The pipeline stage that runs only after a document's PROD promotion succeeds, and POSTs a Beckn-shaped `publish` envelope to an external Discovery Service. Skipped entirely when PROD is disabled (`DISABLE_PROD_SETTING=true`) — a document never reaches the network without first promoting to PROD. The envelope it sends is kind-level (see Knowledge Kind), not document-level: it does not make any single document searchable, only re-announces that the document's provider (its Group Provider, or the Shared Provider) serves that kind at all. _Avoid_: "publish" alone — ambiguous with Publish to Dev. **Master Catalog**: @@ -24,6 +24,18 @@ _Avoid_: "catalog" alone — ambiguous with Master Catalog and the Network Catal The Beckn wire-protocol `message.catalogs` object sent by Publish to Network to the Discovery Service. Structurally and semantically unrelated to the Master Catalog and Scheme Catalog tables — do not conflate the three. _Avoid_: "catalog" alone. +**Group**: +The state a document belongs to, named by its state code (`mh` for Maharashtra) and taken from the document's `instance`. Bharat Vistaar and `default` documents have no Group. New features say Group; existing screens, roles and data still say state or instance. +_Avoid_: "tenant"; "state" or "instance" in new features. Never call Bharat Vistaar a Group. + +**Group Provider**: +The network provider a Group's documents are announced under in Publish to Network, one per Knowledge Kind, each with its own catalog. Every Group has one; a deployment may give a Group its own provider id instead of the default. +_Avoid_: "state provider". + +**Shared Provider**: +The network provider used by documents with no Group (Bharat Vistaar, `default`), one per Knowledge Kind, keeping the original catalog ids. +_Avoid_: "national provider", "central provider". + **Knowledge Kind**: What a document *is* to the network — `advisory`, `scheme`, `video`, or an operator-entered slug — held in `documents.document_kind`. Asserted by a reviewer during the pipeline, never inferred from the file, and absent until then (every document starts as the default `document`). Only `advisory` and `scheme` map to a Network Catalog Envelope; the rest publish nothing. _Avoid_: "document type" — collides with `source_type`/`canonical_input_type`, which describe the input *format* (pdf, spreadsheet). Also avoid saying an advisory is "uploaded": what is uploaded is a file, which only becomes an advisory when someone classifies it. diff --git a/ENV.md b/ENV.md index 3e517c3..0ec38ff 100644 --- a/ENV.md +++ b/ENV.md @@ -249,10 +249,17 @@ Separate from the SQLite/Qdrant catalog above (which tracks PROD vector publicat First piece of the network-discovery push architecture: once a document's PROD promotion succeeds, the `publishing_to_network` pipeline stage POSTs a Beckn-shaped `publish` envelope to an external Discovery Service. It's a PROD-only stage (see `docs/ADR/0004-publishing-to-network-is-prod-only.md`) — skipped entirely when `DISABLE_PROD_SETTING=true`, since a document never reaches the network without first promoting to PROD. This is going to be the only way the experience layer discovers schemas going forward — the Master Scheme Catalog pull API above stays in place for backward compatibility. The catalog announces the *knowledge kind* the document was classified as — one catalog with exactly one `OnDemand` resource for `advisory`, another for `scheme`; any other `document_kind` publishes nothing. `DISCOVERY_SERVICE_ENDPOINT` and `NETWORK_SENDER_ID` are **required** whenever PROD promotion is enabled — there is no separate feature flag for this stage beyond `DISABLE_PROD_SETTING`. +Each catalog's provider is named per Knowledge Kind and instance: id and descriptor code `{kind}.{instance}.provider.oan.dev` (`advisory.bv.provider.oan.dev`, `scheme.mh.provider.oan.dev`), name `Advisories|Schemes from documents`. A document filed under a state (its **Group**, taken from the document's `instance`) uses the group code as the instance, prefixes catalog and resource ids with it (`mh-cat-oan-knowledge-provider-advisories`) and names catalog and provider ` ` (`Maharashtra Advisories from documents`). Documents under the portal instance (`PORTAL_INSTANCE`, `bv`) or the default instance (`DEFAULT_INSTANCE`) have no Group: they use the portal instance in the provider id and keep the unprefixed ids. Every Group publishes. An instance can replace the default provider id per Knowledge Kind in the provider id overrides file (`PROVIDER_CONFIG_OVERRIDE_PATH`); a kind it leaves out keeps the default. A missing or malformed file, or a template with an unknown placeholder, stops both the API and the worker from booting. The file is not mounted by `docker-compose.yml` — mount your own into the `api` and `worker` containers and point the variable at it; `deploy/network_group_providers.example.json` shows the shape. See `docs/ADR/0006-publish-to-network-under-per-group-provider.md`. + | Variable | Default | Purpose | |----------|---------|---------| | `DISCOVERY_SERVICE_ENDPOINT` | *(empty — required)* | Base URL of the external Discovery Service; the pipeline POSTs to `{DISCOVERY_SERVICE_ENDPOINT}/publish` | | `NETWORK_SENDER_ID` | *(empty — required)* | This instance's Beckn `senderId`, also sent as `networkId` | +| `PROVIDER_CONFIG_OVERRIDE_PATH` | unset (bundled `pipeline/network_group_providers.json`, `{}` — no overrides) | Path to a JSON object of instance code → `{kind: provider id}`, set per deployment, e.g. `{"ka": {"advisory": "advisory.ka.com", "scheme": "scheme.ka.com"}}`. Codes and kinds are case-insensitive; kinds are `advisory` or `scheme`. A missing file, invalid JSON, an unknown kind, or an empty or invalid id (same rule as the template) stop the API and worker from starting | +| `NETWORK_PROVIDER_ID_TEMPLATE` | `{kind}.{instance}.provider.oan.dev` | Default provider id and descriptor code, unless overridden in the file above. `{kind}` is `advisory` or `scheme`; `{instance}` is the group code, or `PORTAL_INSTANCE` for documents with no Group. Must use both and render a valid provider id (lowercase letters, digits, `-` and `.`, no leading, trailing or repeated `.`), or the API and worker refuse to start. A document whose instance code renders an invalid id fails its publish without retry | +| `NETWORK_ADVISORY_PROVIDER_NAME` | `Advisories from documents` | Provider name for advisory catalogs | +| `NETWORK_SCHEMES_PROVIDER_NAME` | `Schemes from documents` | Provider name for scheme catalogs | +| `NETWORK_GROUP_PROVIDER_NAME_TEMPLATE` | `{group_name} {provider_name}` | Provider and catalog name for a Group, e.g. `Maharashtra Advisories from documents`. Must use `{group_name}`, or the API and worker refuse to start | | `DISCOVERY_SERVICE_PUBLISH_INITIAL_INTERVAL_SECONDS` | `30` | Wait before the first retry of a failed publish | | `DISCOVERY_SERVICE_PUBLISH_BACKOFF_COEFFICIENT` | `2.0` | Multiplier applied to the wait after each retry | | `DISCOVERY_SERVICE_PUBLISH_MAX_INTERVAL_SECONDS` | `300` | Cap on the wait between retries | @@ -270,6 +277,7 @@ First piece of the network-discovery push architecture: once a document's PROD p | OCR / translation / chunking / domain tags | — | config / status | ✅ runs jobs | | Qdrant (`VECTOR_DB_*`) | — | ✅ | ✅ | | Publish to Network (`DISCOVERY_SERVICE_ENDPOINT`, `NETWORK_SENDER_ID`) | — | — | ✅ runs the publish activity | +| Network provider config (`PROVIDER_CONFIG_OVERRIDE_PATH`, `NETWORK_*_PROVIDER_*`) | — | validated at boot | ✅ validated at boot, used by the publish activity | --- diff --git a/deploy/network_group_providers.example.json b/deploy/network_group_providers.example.json new file mode 100644 index 0000000..23f63eb --- /dev/null +++ b/deploy/network_group_providers.example.json @@ -0,0 +1,6 @@ +{ + "ka": { + "advisory": "advisory.ka.com", + "scheme": "scheme.ka.com" + } +} diff --git a/docs/ADR/0006-publish-to-network-under-per-group-provider.md b/docs/ADR/0006-publish-to-network-under-per-group-provider.md new file mode 100644 index 0000000..5d9b0aa --- /dev/null +++ b/docs/ADR/0006-publish-to-network-under-per-group-provider.md @@ -0,0 +1,24 @@ +--- +status: accepted +--- + +# Publish to Network under a per-Group provider + +Context: ADR 0003 announced every document under one provider, with one catalog per Knowledge Kind. A state's experience (for example Maharashtra's, through DSS) needs to consume only its own state's knowledge from the network, and one shared provider gives it no way to do that (engineering-tracker#190). + +Decision: +- Every catalog has a provider per Knowledge Kind and instance. By default its id and descriptor code are `{kind}.{instance}.provider.oan.dev` (`advisory.mh.provider.oan.dev`, `scheme.bv.provider.oan.dev`). It is named `Advisories|Schemes from documents`, prefixed with the Group name for a Group (`Maharashtra Advisories from documents`). +- Documents with a Group use the group code as the instance. Catalog and resource ids are today's ids prefixed with the group code (`mh-cat-oan-knowledge-provider-advisories`, `mh-res-oan-knowledge-provider-advisories`, and the same for schemes), and the catalog takes the provider's name. +- The group code is derived from the document's `instance`, not from the uploader's group, so a Super Admin uploading for a state publishes under that state. +- Documents with no Group (Bharat Vistaar `bv`, `default`) use the portal instance `bv` in the provider id and keep ADR 0003's unprefixed catalog and resource ids. +- Every Group publishes; only an unmapped Knowledge Kind skips. A real publish failure stays fatal, per ADR 0004. +- The provider id template, the per-kind provider names and the Group name template are environment settings, so a deployment can change them without a code change. A template with an unknown placeholder, an id template missing `{kind}` or `{instance}`, a Group name template missing `{group_name}`, or a provider id (rendered or overridden) outside lowercase letters, digits, `-` and dot-separated parts stops the API and the worker from booting, since a published id can't be removed later. A document whose instance code still renders an invalid id fails its publish, without retry, before anything is sent. +- An instance that wants its own provider id overrides it per Knowledge Kind in a JSON file whose path is `PROVIDER_CONFIG_OVERRIDE_PATH`, e.g. `{"ka": {"advisory": "advisory.ka.com", "scheme": "scheme.ka.com"}}`. Instances and kinds it leaves out keep the template id. It defaults to a bundled empty `pipeline/network_group_providers.json`, so onboarding a Group needs no config. A missing or malformed file stops both the API and the worker from booting, so a bad deploy can't quietly publish under the wrong id. +- "No Group" follows the configured `PORTAL_INSTANCE` and `DEFAULT_INSTANCE` values rather than hardcoding `bv` and `default`. +- The publish endpoint stays the same for every Group. + +Consequences: +- Catalog and resource ids are permanent under `MERGE` (ADR 0003). State knowledge published before this change stays in the unprefixed catalogs; nothing moves it to the Group's `mh-` catalogs. +- The next `bv` publish re-announces the unprefixed catalogs under `advisory.bv`/`scheme.bv` instead of `knowledge-provider`, keeping their catalog and resource ids. Changing the id template or an override later does the same. Whether the network moves such a catalog to the new provider or keeps the old entry depends on the adapter, and is untested. +- Depends on the network adapter layer accepting the new provider ids, for `bv` as well as Groups, and the Groups' prefixed catalog and resource ids. +- Only the Publish to Network payload changes. Chunks and the search indexes gain no `group` field; they keep `instance` and `instance_name`, where Bharat Vistaar and `default` documents carry `bv`/`default` and state documents carry the state code, so #202 can filter on `instance`. diff --git a/pipeline/activities.py b/pipeline/activities.py index b29c479..2d30234 100644 --- a/pipeline/activities.py +++ b/pipeline/activities.py @@ -30,6 +30,7 @@ from .document_repository import DocumentRepository from .instances import instance_display_name from .network_constants import RESULT_REJECTED +from .network_provider import InvalidProviderIdError from .ocr import ocr_pdf as run_ocr_pdf from .ocr import ocr_pdf_in_segments as run_ocr_pdf_in_segments from .translation import load_translation_config, translate_pages @@ -1464,11 +1465,15 @@ async def publish_catalog_to_network(workflow_id: str, transaction_id: str) -> d repository = DocumentRepository() document_kind = normalize_document_kind(repository.get_document_kind(workflow_id)) + group = repository.get_document_group(workflow_id) service = DiscoveryPublishService() - result = await asyncio.to_thread( - service.publish, transaction_id, document_kind, workflow_id - ) + try: + result = await asyncio.to_thread( + service.publish, transaction_id, document_kind, workflow_id, group + ) + except InvalidProviderIdError as exc: + raise ApplicationError(str(exc), non_retryable=True) from exc if result["skipped"]: artifact_type = "network_publish_skipped" diff --git a/pipeline/api.py b/pipeline/api.py index 747704f..9506987 100644 --- a/pipeline/api.py +++ b/pipeline/api.py @@ -103,6 +103,7 @@ SearchSettingsUpdate, SettingsAuditResponse, ) +from .network_group import validate_network_provider_config from .workflows import ( ChunkingOnlyWorkflow, DocumentPipelineWorkflow, @@ -137,6 +138,8 @@ async def lifespan(app: FastAPI): """Initialize Temporal and MinIO clients on startup.""" global temporal_client, minio_client, MINIO_BUCKET + validate_network_provider_config() + auth_cfg = load_auth_config() validate_auth_config(auth_cfg) if auth_cfg.disabled: diff --git a/pipeline/auth/tenancy.py b/pipeline/auth/tenancy.py index 7b1d23a..e7813f6 100644 --- a/pipeline/auth/tenancy.py +++ b/pipeline/auth/tenancy.py @@ -26,6 +26,17 @@ def default_instance() -> str: return (os.environ.get("DEFAULT_INSTANCE") or "default").strip().lower() or "default" +def portal_instance() -> str: + return (os.environ.get("PORTAL_INSTANCE") or PORTAL_INSTANCE).strip().lower() or PORTAL_INSTANCE + + +def group_for_instance(instance: str | None) -> str | None: + code = (instance or "").strip().lower() + if not code or code in {portal_instance(), default_instance()}: + return None + return code + + def normalize_instance(value: str | None) -> str: text = (value or "").strip().lower() return text or default_instance() @@ -76,8 +87,7 @@ def resolve_create_instance(user: AuthUser, requested: str | None = None) -> str if requested_norm: return normalize_instance(requested_norm) # Prefer portal tag for platform operators when nothing chosen. - portal = (os.environ.get("PORTAL_INSTANCE") or PORTAL_INSTANCE).strip().lower() - return normalize_instance(portal or default_instance()) + return normalize_instance(portal_instance()) allowed = allowed_instances(user) or set() if not allowed: diff --git a/pipeline/catalog_builder.py b/pipeline/catalog_builder.py index 7b7da27..f1ac72d 100644 --- a/pipeline/catalog_builder.py +++ b/pipeline/catalog_builder.py @@ -2,12 +2,13 @@ Builds the Network Catalog Envelope's catalog for a document's knowledge kind. One catalog holding exactly one OnDemand resource per kind, identical for every -document of that kind - a seeker that matches the resource comes back to ask the -actual question, so there is nothing per-document to put on the wire. +document of that kind and Group - a seeker that matches the resource comes back +to ask the actual question, so there is nothing per-document to put on the wire. -Pure: no env, no I/O. Values live in network_constants.py and the transport in -discovery_publish_service.py. See -docs/ADR/0003-static-ondemand-network-catalog-per-knowledge-kind.md. +Pure: no env, no I/O. Values live in network_constants.py, the provider in +network_provider.py and the transport in discovery_publish_service.py. See +docs/ADR/0003-static-ondemand-network-catalog-per-knowledge-kind.md and +docs/ADR/0006-publish-to-network-under-per-group-provider.md. """ import copy @@ -20,7 +21,6 @@ ADVISORY_RESOURCE_NAME, ADVISORY_SUBJECT_CATEGORIES, ADVISORY_TOPICS, - CATALOG_PROVIDER, DEFAULT_DOCUMENT_KIND, INFORMATION_MODE, SCHEMA_CONTEXT_BASE, @@ -32,6 +32,7 @@ SCHEMES_RESOURCE_NAME, SERVED_LANGUAGES, ) +from .network_provider import Group _ADVISORY_RESOURCE_ATTRIBUTES = { "@context": f"{SCHEMA_CONTEXT_BASE}/KnowledgeAdvisory/v0.1/context.jsonld", @@ -80,7 +81,15 @@ def normalize_document_kind(document_kind: Optional[str]) -> str: return (document_kind or DEFAULT_DOCUMENT_KIND).strip().lower() -def build_catalog(document_kind: Optional[str]) -> Optional[dict]: +def is_publishable_kind(document_kind: Optional[str]) -> bool: + return normalize_document_kind(document_kind) in _KNOWLEDGE_KIND_CATALOGS + + +def build_catalog( + document_kind: Optional[str], + provider: dict, + group: Optional[Group] = None, +) -> Optional[dict]: """Return the catalog to announce for `document_kind`, or None. None means this kind has nothing to announce - `document`, `video` and any @@ -91,14 +100,24 @@ def build_catalog(document_kind: Optional[str]) -> Optional[dict]: if spec is None: return None + catalog_id = spec["catalog_id"] + catalog_name = spec["catalog_name"] + resource_id = spec["resource_id"] + provider = copy.deepcopy(provider) + if group is not None: + prefix = f"{group.code}-" + catalog_id = prefix + catalog_id + resource_id = prefix + resource_id + catalog_name = provider["descriptor"]["name"] + catalog = { - "id": spec["catalog_id"], - "descriptor": {"name": spec["catalog_name"]}, + "id": catalog_id, + "descriptor": {"name": catalog_name}, "isActive": True, - "provider": copy.deepcopy(CATALOG_PROVIDER), + "provider": provider, "resources": [ { - "id": spec["resource_id"], + "id": resource_id, "descriptor": {"name": spec["resource_name"]}, "resourceAttributes": copy.deepcopy(spec["resource_attributes"]), } diff --git a/pipeline/discovery_publish_service.py b/pipeline/discovery_publish_service.py index d34f2b8..554d9c6 100644 --- a/pipeline/discovery_publish_service.py +++ b/pipeline/discovery_publish_service.py @@ -16,8 +16,11 @@ import httpx -from .catalog_builder import build_catalog +from .catalog_builder import build_catalog, is_publishable_kind, normalize_document_kind +from .instances import instance_display_name from .network_constants import BECKN_VERSION, RESULT_ACCEPTED +from .network_group import load_provider_naming +from .network_provider import Group logger = logging.getLogger(__name__) @@ -41,12 +44,14 @@ def __init__( raise RuntimeError("NETWORK_SENDER_ID must be set to publish to the network.") self.timeout = timeout + self.provider_naming = load_provider_naming() def publish( self, transaction_id: str, document_kind: Optional[str], workflow_id: Optional[str] = None, + group: Optional[str] = None, ) -> dict: """ POST a catalog/publish envelope for `document_kind`. @@ -59,21 +64,20 @@ def publish( with `skipped=True`: the spec declares `message.catalogs` as `minItems: 1`, so there is no valid "publish nothing" request to send. """ - catalog = build_catalog(document_kind) - if catalog is None: + if not is_publishable_kind(document_kind): logger.info( "workflow_id=%s document_kind=%s network_publish_skipped=True", workflow_id, document_kind, ) - return { - "skipped": True, - "envelope": None, - "status_code": None, - "response_body": None, - "result_status": None, - "errors": [], - } + return self._skipped() + + catalog_group = None + if group is not None: + catalog_group = Group(code=group, name=instance_display_name(group)) + + provider = self.provider_naming.provider(normalize_document_kind(document_kind), catalog_group) + catalog = build_catalog(document_kind, provider, catalog_group) envelope = { "context": { @@ -155,6 +159,17 @@ def publish( "errors": errors, } + @staticmethod + def _skipped() -> dict: + return { + "skipped": True, + "envelope": None, + "status_code": None, + "response_body": None, + "result_status": None, + "errors": [], + } + def _read_result( self, response: httpx.Response, diff --git a/pipeline/document_repository.py b/pipeline/document_repository.py index 9f2e2c3..83ff1ce 100644 --- a/pipeline/document_repository.py +++ b/pipeline/document_repository.py @@ -9,6 +9,7 @@ from typing import Optional from . import db +from .auth.tenancy import group_for_instance class DocumentRepository: @@ -28,3 +29,9 @@ def get_document_kind(self, workflow_id: str) -> Optional[str]: if not doc: return None return doc.get("document_kind") + + def get_document_group(self, workflow_id: str) -> Optional[str]: + doc = self._db.get_document(workflow_id) + if not doc: + return None + return group_for_instance(doc.get("instance")) diff --git a/pipeline/network_constants.py b/pipeline/network_constants.py index 422a946..267575e 100644 --- a/pipeline/network_constants.py +++ b/pipeline/network_constants.py @@ -33,12 +33,8 @@ DEFAULT_DOCUMENT_KIND = "document" -# Same provider on every catalog - this pipeline publishes as one provider -# regardless of knowledge kind. -CATALOG_PROVIDER = { - "id": "knowledge-provider", - "descriptor": { - "code": "oan-knowledge-provider", - "name": "OAN Knowledge Provider", - }, -} +ADVISORY_PROVIDER_NAME = "Advisories from documents" +SCHEMES_PROVIDER_NAME = "Schemes from documents" + +PROVIDER_ID_TEMPLATE = "{kind}.{instance}.provider.oan.dev" +GROUP_PROVIDER_NAME_TEMPLATE = "{group_name} {provider_name}" diff --git a/pipeline/network_group.py b/pipeline/network_group.py new file mode 100644 index 0000000..f3464f5 --- /dev/null +++ b/pipeline/network_group.py @@ -0,0 +1,152 @@ +from __future__ import annotations + +import json +import os +import string +from pathlib import Path +from typing import Optional + +from .auth.tenancy import portal_instance +from .network_constants import ( + ADVISORY_PROVIDER_NAME, + GROUP_PROVIDER_NAME_TEMPLATE, + PROVIDER_ID_TEMPLATE, + SCHEMES_PROVIDER_NAME, +) +from .network_provider import PROVIDER_ID_RULE, PROVIDER_KINDS, ProviderNaming, is_valid_provider_id + +PROVIDER_CONFIG_OVERRIDE_PATH_ENV = "PROVIDER_CONFIG_OVERRIDE_PATH" +DEFAULT_PROVIDER_CONFIG_OVERRIDE_PATH = Path(__file__).with_name("network_group_providers.json") +PROVIDER_ID_TEMPLATE_ENV = "NETWORK_PROVIDER_ID_TEMPLATE" +GROUP_PROVIDER_NAME_TEMPLATE_ENV = "NETWORK_GROUP_PROVIDER_NAME_TEMPLATE" +ADVISORY_PROVIDER_NAME_ENV = "NETWORK_ADVISORY_PROVIDER_NAME" +SCHEMES_PROVIDER_NAME_ENV = "NETWORK_SCHEMES_PROVIDER_NAME" + + +class NetworkProviderConfigError(ValueError): + pass + + +def _env(name: str, default: str) -> str: + return os.environ.get(name, "").strip() or default + + +def _check_provider_id(provider_id: str, source: str) -> None: + if not is_valid_provider_id(provider_id): + raise NetworkProviderConfigError(f"{source} gives provider id {provider_id!r}; it must be {PROVIDER_ID_RULE}") + + +def parse_provider_id_overrides(raw: Optional[str], source: str) -> dict[str, dict[str, str]]: + if raw is None or not raw.strip(): + return {} + + try: + parsed = json.loads(raw) + except json.JSONDecodeError as exc: + raise NetworkProviderConfigError(f"{source} is not valid JSON: {exc}") from exc + + if not isinstance(parsed, dict): + raise NetworkProviderConfigError( + f"{source} must be a JSON object of instance code to {{kind: provider id}}" + ) + + overrides: dict[str, dict[str, str]] = {} + for key, kinds in parsed.items(): + code = key.strip().lower() if isinstance(key, str) else "" + if not code: + raise NetworkProviderConfigError(f"{source} entry {key!r} must be a non-empty instance code") + if code in overrides: + raise NetworkProviderConfigError(f"{source} names instance {code!r} more than once") + if not isinstance(kinds, dict) or not kinds: + raise NetworkProviderConfigError( + f"{source} entry {key!r} must map knowledge kinds to provider ids" + ) + provider_ids: dict[str, str] = {} + for kind, value in kinds.items(): + normalized_kind = kind.strip().lower() if isinstance(kind, str) else "" + provider_id = value.strip() if isinstance(value, str) else "" + if normalized_kind not in PROVIDER_KINDS: + raise NetworkProviderConfigError( + f"{source} entry {key!r} kind {kind!r} must be one of {list(PROVIDER_KINDS)}" + ) + if not provider_id: + raise NetworkProviderConfigError( + f"{source} entry {key!r} kind {kind!r} must have a non-empty provider id" + ) + _check_provider_id(provider_id, f"{source} entry {key!r} kind {kind!r}") + provider_ids[normalized_kind] = provider_id + overrides[code] = provider_ids + return overrides + + +def provider_config_override_path() -> Path: + configured = os.environ.get(PROVIDER_CONFIG_OVERRIDE_PATH_ENV, "").strip() + return Path(configured) if configured else DEFAULT_PROVIDER_CONFIG_OVERRIDE_PATH + + +def load_provider_id_overrides() -> dict[str, dict[str, str]]: + path = provider_config_override_path() + try: + raw = path.read_text(encoding="utf-8") + except OSError as exc: + raise NetworkProviderConfigError(f"Provider id overrides {path} cannot be read: {exc}") from exc + return parse_provider_id_overrides(raw, str(path)) + + +def _checked_template( + env_name: str, default: str, required: frozenset[str] = frozenset(), **placeholders: str +) -> str: + template = _env(env_name, default) + try: + rendered = template.format(**placeholders) + used = {field for _, field, _, _ in string.Formatter().parse(template) if field} + except (KeyError, IndexError, ValueError, AttributeError) as exc: + raise NetworkProviderConfigError( + f"{env_name}={template!r} may only use {sorted(placeholders)}: {exc}" + ) from exc + if not rendered.strip(): + raise NetworkProviderConfigError(f"{env_name} renders an empty value") + missing = required - used + if missing: + raise NetworkProviderConfigError(f"{env_name}={template!r} must use {sorted(missing)}") + return template + + +def _checked_id_template(shared_instance: str) -> str: + template = _checked_template( + PROVIDER_ID_TEMPLATE_ENV, + PROVIDER_ID_TEMPLATE, + frozenset({"kind", "instance"}), + kind="advisory", + instance=shared_instance, + ) + for kind in PROVIDER_KINDS: + for instance in (shared_instance, "mh"): + _check_provider_id( + template.format(kind=kind, instance=instance), f"{PROVIDER_ID_TEMPLATE_ENV}={template!r}" + ) + return template + + +def load_provider_naming() -> ProviderNaming: + shared_instance = portal_instance() + return ProviderNaming( + shared_instance=shared_instance, + id_template=_checked_id_template(shared_instance), + group_name_template=_checked_template( + GROUP_PROVIDER_NAME_TEMPLATE_ENV, + GROUP_PROVIDER_NAME_TEMPLATE, + frozenset({"group_name"}), + group_name="Maharashtra", + provider_name="Advisories from documents", + ), + provider_names={ + "advisory": _env(ADVISORY_PROVIDER_NAME_ENV, ADVISORY_PROVIDER_NAME), + "scheme": _env(SCHEMES_PROVIDER_NAME_ENV, SCHEMES_PROVIDER_NAME), + }, + provider_id_overrides=load_provider_id_overrides(), + ) + + +def validate_network_provider_config() -> None: + load_provider_naming() diff --git a/pipeline/network_group_providers.json b/pipeline/network_group_providers.json new file mode 100644 index 0000000..0967ef4 --- /dev/null +++ b/pipeline/network_group_providers.json @@ -0,0 +1 @@ +{} diff --git a/pipeline/network_provider.py b/pipeline/network_provider.py new file mode 100644 index 0000000..491aa78 --- /dev/null +++ b/pipeline/network_provider.py @@ -0,0 +1,56 @@ +import re +from dataclasses import dataclass, field +from typing import Mapping, Optional + +from .network_constants import ( + ADVISORY_PROVIDER_NAME, + GROUP_PROVIDER_NAME_TEMPLATE, + PROVIDER_ID_TEMPLATE, + SCHEMES_PROVIDER_NAME, +) + +PROVIDER_KINDS = ("advisory", "scheme") +PROVIDER_ID_PATTERN = re.compile(r"[a-z0-9-]+(?:\.[a-z0-9-]+)*") +PROVIDER_ID_RULE = "lowercase letters, digits, '-' and '.', with no leading, trailing or repeated '.'" + + +class InvalidProviderIdError(ValueError): + pass + + +def is_valid_provider_id(provider_id: str) -> bool: + return PROVIDER_ID_PATTERN.fullmatch(provider_id) is not None + + +@dataclass(frozen=True) +class Group: + code: str + name: str + + +@dataclass(frozen=True) +class ProviderNaming: + shared_instance: str + id_template: str = PROVIDER_ID_TEMPLATE + group_name_template: str = GROUP_PROVIDER_NAME_TEMPLATE + provider_names: Mapping[str, str] = field( + default_factory=lambda: { + "advisory": ADVISORY_PROVIDER_NAME, + "scheme": SCHEMES_PROVIDER_NAME, + } + ) + provider_id_overrides: Mapping[str, Mapping[str, str]] = field(default_factory=dict) + + def provider(self, kind: str, group: Optional[Group]) -> dict: + instance = group.code if group is not None else self.shared_instance + provider_id = self.provider_id_overrides.get(instance, {}).get(kind) or self.id_template.format( + kind=kind, instance=instance + ) + if not is_valid_provider_id(provider_id): + raise InvalidProviderIdError( + f"Provider id {provider_id!r} for {kind} on instance {instance!r} must be {PROVIDER_ID_RULE}" + ) + name = self.provider_names[kind] + if group is not None: + name = self.group_name_template.format(group_name=group.name, provider_name=name) + return {"id": provider_id, "descriptor": {"code": provider_id, "name": name}} diff --git a/pipeline/worker.py b/pipeline/worker.py index 0d39634..6cd996b 100644 --- a/pipeline/worker.py +++ b/pipeline/worker.py @@ -26,6 +26,7 @@ run_ocr_and_store, update_document_state, ) +from .network_group import validate_network_provider_config from .workflows import ( ChunkingOnlyWorkflow, DocumentPipelineWorkflow, @@ -53,6 +54,8 @@ async def main(): """Start the worker.""" + validate_network_provider_config() + translation_provider = os.environ.get("TRANSLATION_PROVIDER", "gemma_vllm").strip().lower() if translation_provider in {"gemma_vllm", "gemma4", "gemma"}: if not os.environ.get("TRANSLATION_VLLM_BASE_URL", "").strip(): diff --git a/tests/test_activities.py b/tests/test_activities.py index 79904c0..eb06734 100644 --- a/tests/test_activities.py +++ b/tests/test_activities.py @@ -526,7 +526,7 @@ class FakeService: def __init__(self): pass - def publish(self, transaction_id, document_kind, workflow_id=None): + def publish(self, transaction_id, document_kind, workflow_id=None, group=None): assert transaction_id == "txn-1" assert document_kind == "advisory" assert workflow_id == "wf-1" @@ -574,7 +574,7 @@ async def test_publish_failure_propagates_and_records_nothing(self, monkeypatch) import pipeline.db as db class FakeService: - def publish(self, transaction_id, document_kind, workflow_id=None): + def publish(self, transaction_id, document_kind, workflow_id=None, group=None): raise RuntimeError("discovery service unreachable") monkeypatch.setattr(activities, "DiscoveryPublishService", FakeService) @@ -596,7 +596,7 @@ async def test_unmapped_kind_records_a_skipped_artifact(self, monkeypatch): import pipeline.db as db class FakeService: - def publish(self, transaction_id, document_kind, workflow_id=None): + def publish(self, transaction_id, document_kind, workflow_id=None, group=None): return { "skipped": True, "envelope": None, @@ -641,7 +641,7 @@ async def test_non_rejected_results_complete_the_stage(self, monkeypatch, result import pipeline.db as db class FakeService: - def publish(self, transaction_id, document_kind, workflow_id=None): + def publish(self, transaction_id, document_kind, workflow_id=None, group=None): return { "skipped": False, "envelope": {"context": {"messageId": "msg-3"}}, @@ -675,7 +675,7 @@ async def test_rejected_result_records_the_errors_then_fails(self, monkeypatch): errors = [{"code": "SCH_SCHEMA_VALIDATION_FAILED", "message": "topics is required"}] class FakeService: - def publish(self, transaction_id, document_kind, workflow_id=None): + def publish(self, transaction_id, document_kind, workflow_id=None, group=None): return { "skipped": False, "envelope": {"context": {"messageId": "msg-4"}}, @@ -706,6 +706,108 @@ def publish(self, transaction_id, document_kind, workflow_id=None): assert recorded["metadata"]["errors"] == errors +class TestPublishCatalogToNetworkGroups: + @staticmethod + def _wire(monkeypatch, document, service_result, seen): + import pipeline.activities as activities + import pipeline.db as db + + class FakeService: + def publish(self, transaction_id, document_kind, workflow_id=None, group=None): + seen["group"] = group + return service_result + + monkeypatch.setattr(activities, "DiscoveryPublishService", FakeService) + monkeypatch.setattr( + activities, + "_upload_file_to_minio", + lambda *a, **k: ("minio://documents/x.json", 1, "application/json"), + ) + monkeypatch.setattr(db, "get_document", lambda workflow_id: document) + monkeypatch.setattr(db, "get_latest_document_job", lambda workflow_id: None) + recorded = {} + monkeypatch.setattr(db, "add_document_artifact", lambda **kwargs: recorded.update(kwargs)) + return recorded + + _PUBLISHED = { + "skipped": False, + "envelope": {"context": {"messageId": "msg-g"}}, + "status_code": 200, + "response_body": "{}", + "result_status": "ACCEPTED", + "errors": [], + } + + @pytest.mark.unit + @pytest.mark.asyncio + @pytest.mark.parametrize("instance,group", [("mh", "mh"), ("MH", "mh"), ("bv", None), ("default", None)]) + async def test_passes_the_documents_group(self, monkeypatch, instance, group): + import pipeline.activities as activities + + monkeypatch.delenv("PORTAL_INSTANCE", raising=False) + monkeypatch.delenv("DEFAULT_INSTANCE", raising=False) + seen = {} + self._wire( + monkeypatch, + {"document_kind": "advisory", "instance": instance}, + self._PUBLISHED, + seen, + ) + + await activities.publish_catalog_to_network("wf-g", "txn-g") + + assert seen["group"] == group + + @pytest.mark.unit + @pytest.mark.asyncio + async def test_unmapped_kind_records_a_skip(self, monkeypatch): + import pipeline.activities as activities + + skipped = { + "skipped": True, + "envelope": None, + "status_code": None, + "response_body": None, + "result_status": None, + "errors": [], + } + recorded = self._wire( + monkeypatch, + {"document_kind": "document", "instance": "mh"}, + skipped, + {}, + ) + + result = await activities.publish_catalog_to_network("wf-u", "txn-u") + + assert result["status"] == "skipped" + assert recorded["artifact_type"] == "network_publish_skipped" + assert recorded["metadata"]["reason"] == "no network catalog is mapped to this document kind" + + + @pytest.mark.unit + @pytest.mark.asyncio + async def test_an_invalid_provider_id_fails_without_retry_or_artifact(self, monkeypatch): + import pipeline.activities as activities + import pipeline.db as db + from pipeline.network_provider import InvalidProviderIdError + + class FakeService: + def publish(self, transaction_id, document_kind, workflow_id=None, group=None): + raise InvalidProviderIdError("Provider id 'advisory.ladakh_ut.provider.oan.dev' is invalid") + + monkeypatch.setattr(activities, "DiscoveryPublishService", FakeService) + monkeypatch.setattr(db, "get_document", lambda workflow_id: {"document_kind": "advisory", "instance": "ladakh_ut"}) + recorded = {} + monkeypatch.setattr(db, "add_document_artifact", lambda **kwargs: recorded.update(kwargs)) + + with pytest.raises(ApplicationError, match="ladakh_ut") as excinfo: + await activities.publish_catalog_to_network("wf-x", "txn-x") + + assert excinfo.value.non_retryable is True + assert recorded == {} + + class TestWorkerActivityRegistration: """ Every @activity.defn in activities.py must be registered on the worker diff --git a/tests/test_catalog_builder.py b/tests/test_catalog_builder.py index 24fbfa4..ee17169 100644 --- a/tests/test_catalog_builder.py +++ b/tests/test_catalog_builder.py @@ -4,15 +4,25 @@ from pipeline.catalog_builder import ( build_catalog, + is_publishable_kind, normalize_document_kind, ) from pipeline.network_constants import ( ADVISORY_CATALOG_ID, ADVISORY_RESOURCE_ID, - CATALOG_PROVIDER, SCHEMES_CATALOG_ID, SCHEMES_RESOURCE_ID, ) +from pipeline.network_provider import Group + +PROVIDER = { + "id": "advisory.bv.provider.oan.dev", + "descriptor": {"code": "advisory.bv.provider.oan.dev", "name": "Advisories from documents"}, +} +MH_PROVIDER = { + "id": "advisory.mh.provider.oan.dev", + "descriptor": {"code": "advisory.mh.provider.oan.dev", "name": "Maharashtra Advisories from documents"}, +} # Forbidden by the KnowledgeAdvisory schema under informationMode OnDemand. # Both advisory and scheme resources use this schema. @@ -27,7 +37,7 @@ def _attributes(kind): - return build_catalog(kind)["resources"][0]["resourceAttributes"] + return build_catalog(kind, PROVIDER)["resources"][0]["resourceAttributes"] class TestNormalizeDocumentKind: @@ -49,18 +59,18 @@ def test_normalizes_to_lookup_form(self, raw, expected): class TestBuildAdvisoryCatalog: @pytest.mark.unit def test_catalog_envelope_fields(self): - catalog = build_catalog("advisory") + catalog = build_catalog("advisory", PROVIDER) assert catalog["id"] == ADVISORY_CATALOG_ID assert "bppId" not in catalog assert "bppUri" not in catalog assert catalog["descriptor"] == {"name": "Agricultural advisory from documents"} assert catalog["isActive"] is True - assert catalog["provider"] == CATALOG_PROVIDER + assert catalog["provider"] == PROVIDER @pytest.mark.unit def test_exactly_one_resource_with_a_stable_id(self): - catalog = build_catalog("advisory") + catalog = build_catalog("advisory", PROVIDER) assert len(catalog["resources"]) == 1 resource = catalog["resources"][0] @@ -87,18 +97,18 @@ def test_omits_fields_forbidden_under_on_demand(self, field): class TestBuildSchemeCatalog: @pytest.mark.unit def test_catalog_envelope_fields(self): - catalog = build_catalog("scheme") + catalog = build_catalog("scheme", PROVIDER) assert catalog["id"] == SCHEMES_CATALOG_ID assert "bppId" not in catalog assert "bppUri" not in catalog assert catalog["descriptor"] == {"name": "Schemes from documents"} assert catalog["isActive"] is True - assert catalog["provider"] == CATALOG_PROVIDER + assert catalog["provider"] == PROVIDER @pytest.mark.unit def test_exactly_one_resource_with_a_stable_id(self): - catalog = build_catalog("scheme") + catalog = build_catalog("scheme", PROVIDER) assert len(catalog["resources"]) == 1 resource = catalog["resources"][0] @@ -128,24 +138,79 @@ class TestUnmappedKinds: @pytest.mark.unit @pytest.mark.parametrize("kind", ["document", "video", "how_to_faq", "", None]) def test_returns_none_so_the_caller_skips_publishing(self, kind): - assert build_catalog(kind) is None + assert build_catalog(kind, PROVIDER) is None + assert is_publishable_kind(kind) is False + + @pytest.mark.unit + @pytest.mark.parametrize("kind", ["advisory", " Scheme "]) + def test_knowledge_kinds_are_publishable(self, kind): + assert is_publishable_kind(kind) is True class TestBuilderIsolation: @pytest.mark.unit def test_mutating_a_built_catalog_does_not_leak_into_the_next(self): - first = build_catalog("advisory") + first = build_catalog("advisory", PROVIDER) first["resources"][0]["resourceAttributes"]["topics"].append("tampered") - second = build_catalog("advisory") + second = build_catalog("advisory", PROVIDER) assert second["resources"][0]["resourceAttributes"]["topics"] == ["Crop production"] @pytest.mark.unit def test_mutating_a_built_catalogs_provider_does_not_leak_into_the_next(self): - first = build_catalog("advisory") + first = build_catalog("advisory", PROVIDER) first["provider"]["descriptor"]["name"] = "tampered" - second = build_catalog("advisory") + second = build_catalog("advisory", PROVIDER) + + assert second["provider"] == PROVIDER + + +MAHARASHTRA = Group(code="mh", name="Maharashtra") + + +class TestGroupCatalog: + @pytest.mark.unit + @pytest.mark.parametrize( + "kind,catalog_id,resource_id", + [ + ( + "advisory", + "mh-cat-oan-knowledge-provider-advisories", + "mh-res-oan-knowledge-provider-advisories", + ), + ( + "scheme", + "mh-cat-oan-knowledge-provider-schemes", + "mh-res-oan-knowledge-provider-schemes", + ), + ], + ) + def test_ids_carry_the_group_and_the_catalog_takes_the_providers_name(self, kind, catalog_id, resource_id): + catalog = build_catalog(kind, MH_PROVIDER, MAHARASHTRA) + + assert catalog["id"] == catalog_id + assert catalog["descriptor"] == {"name": "Maharashtra Advisories from documents"} + assert catalog["resources"][0]["id"] == resource_id + assert catalog["provider"] == MH_PROVIDER + + @pytest.mark.unit + @pytest.mark.parametrize("kind", ["advisory", "scheme"]) + def test_resource_attributes_match_the_shared_catalog(self, kind): + grouped = build_catalog(kind, MH_PROVIDER, MAHARASHTRA)["resources"][0] + shared = build_catalog(kind, PROVIDER)["resources"][0] + + assert grouped["resourceAttributes"] == shared["resourceAttributes"] + assert grouped["descriptor"] == shared["descriptor"] - assert second["provider"] == CATALOG_PROVIDER + @pytest.mark.unit + def test_unmapped_kind_still_publishes_nothing(self): + assert build_catalog("document", MH_PROVIDER, MAHARASHTRA) is None + + @pytest.mark.unit + def test_no_group_keeps_the_unprefixed_ids(self): + catalog = build_catalog("advisory", PROVIDER, None) + + assert catalog["id"] == ADVISORY_CATALOG_ID + assert catalog["provider"] == PROVIDER diff --git a/tests/test_discovery_publish_service.py b/tests/test_discovery_publish_service.py index 30052a9..0318df7 100644 --- a/tests/test_discovery_publish_service.py +++ b/tests/test_discovery_publish_service.py @@ -439,3 +439,185 @@ def test_skipped_publish_logs_no_request(self, monkeypatch, caplog): messages = [r.getMessage() for r in caplog.records] assert any("network_publish_skipped=True" in m for m in messages) assert not any("network_publish_url=" in m for m in messages) + + +def _use_group_providers(monkeypatch, tmp_path, content): + path = tmp_path / "network_group_providers.json" + path.write_text(content) + monkeypatch.setenv("PROVIDER_CONFIG_OVERRIDE_PATH", str(path)) + + +@pytest.fixture(autouse=True) +def _default_provider_naming(monkeypatch): + for name in ( + "PORTAL_INSTANCE", + "NETWORK_PROVIDER_ID_TEMPLATE", + "NETWORK_GROUP_PROVIDER_NAME_TEMPLATE", + "NETWORK_ADVISORY_PROVIDER_NAME", + "NETWORK_SCHEMES_PROVIDER_NAME", + ): + monkeypatch.delenv(name, raising=False) + + +class TestDiscoveryPublishServiceGroupProvider: + @pytest.mark.unit + def test_configured_group_publishes_under_its_own_provider(self, monkeypatch, tmp_path): + from pipeline.discovery_publish_service import DiscoveryPublishService + + monkeypatch.setenv("DISCOVERY_SERVICE_ENDPOINT", "https://discovery.example.com") + monkeypatch.setenv("NETWORK_SENDER_ID", "docs-pipeline-bv") + _use_group_providers(monkeypatch, tmp_path, '{}') + service = DiscoveryPublishService() + client = _client_returning( + _on_publish_response("mh-cat-oan-knowledge-provider-advisories", "ACCEPTED") + ) + + with patch("pipeline.discovery_publish_service.httpx.Client", return_value=client): + result = service.publish(transaction_id="txn-mh", document_kind="advisory", group="mh") + + envelope = client.post.call_args.kwargs["json"] + catalog = envelope["message"]["catalogs"][0] + assert catalog["id"] == "mh-cat-oan-knowledge-provider-advisories" + assert catalog["resources"][0]["id"] == "mh-res-oan-knowledge-provider-advisories" + assert catalog["provider"] == { + "id": "advisory.mh.provider.oan.dev", + "descriptor": { + "code": "advisory.mh.provider.oan.dev", + "name": "Maharashtra Advisories from documents", + }, + } + assert envelope["message"]["publishDirectives"][0]["catalogId"] == catalog["id"] + assert client.post.call_args.args[0] == "https://discovery.example.com/publish" + assert result["skipped"] is False + assert result["result_status"] == "ACCEPTED" + + @pytest.mark.unit + def test_group_without_an_override_publishes_under_the_default_provider(self, monkeypatch, tmp_path): + from pipeline.discovery_publish_service import DiscoveryPublishService + + monkeypatch.setenv("DISCOVERY_SERVICE_ENDPOINT", "https://discovery.example.com") + monkeypatch.setenv("NETWORK_SENDER_ID", "docs-pipeline-bv") + _use_group_providers(monkeypatch, tmp_path, '{"ka": {"scheme": "scheme.ka.com"}}') + service = DiscoveryPublishService() + client = _client_returning( + _on_publish_response("mh-cat-oan-knowledge-provider-schemes", "ACCEPTED") + ) + + with patch("pipeline.discovery_publish_service.httpx.Client", return_value=client): + result = service.publish(transaction_id="txn-mh", document_kind="scheme", group="mh") + + catalog = client.post.call_args.kwargs["json"]["message"]["catalogs"][0] + assert catalog["provider"]["id"] == "scheme.mh.provider.oan.dev" + assert result["skipped"] is False + + @pytest.mark.unit + def test_overridden_group_publishes_under_its_configured_provider(self, monkeypatch, tmp_path): + from pipeline.discovery_publish_service import DiscoveryPublishService + + monkeypatch.setenv("DISCOVERY_SERVICE_ENDPOINT", "https://discovery.example.com") + monkeypatch.setenv("NETWORK_SENDER_ID", "docs-pipeline-bv") + _use_group_providers(monkeypatch, tmp_path, '{"KA": {"scheme": "scheme.ka.com"}}') + service = DiscoveryPublishService() + client = _client_returning( + _on_publish_response("ka-cat-oan-knowledge-provider-schemes", "ACCEPTED") + ) + + with patch("pipeline.discovery_publish_service.httpx.Client", return_value=client): + service.publish(transaction_id="txn-ka", document_kind="scheme", group="ka") + + catalog = client.post.call_args.kwargs["json"]["message"]["catalogs"][0] + assert catalog["id"] == "ka-cat-oan-knowledge-provider-schemes" + assert catalog["provider"] == { + "id": "scheme.ka.com", + "descriptor": {"code": "scheme.ka.com", "name": "Karnataka Schemes from documents"}, + } + + @pytest.mark.unit + def test_unmapped_kind_is_skipped_for_a_group(self, monkeypatch): + from pipeline.discovery_publish_service import DiscoveryPublishService + + monkeypatch.setenv("DISCOVERY_SERVICE_ENDPOINT", "https://discovery.example.com") + monkeypatch.setenv("NETWORK_SENDER_ID", "docs-pipeline-bv") + monkeypatch.delenv("PROVIDER_CONFIG_OVERRIDE_PATH", raising=False) + service = DiscoveryPublishService() + + result = service.publish(transaction_id="txn-doc", document_kind="document", group="ka") + + assert result["skipped"] is True + assert result["envelope"] is None + + @pytest.mark.unit + def test_no_group_publishes_under_the_shared_provider(self, monkeypatch, tmp_path): + from pipeline.discovery_publish_service import DiscoveryPublishService + + monkeypatch.setenv("DISCOVERY_SERVICE_ENDPOINT", "https://discovery.example.com") + monkeypatch.setenv("NETWORK_SENDER_ID", "docs-pipeline-bv") + _use_group_providers(monkeypatch, tmp_path, '{}') + service = DiscoveryPublishService() + client = _client_returning( + _on_publish_response("cat-oan-knowledge-provider-advisories", "ACCEPTED") + ) + + with patch("pipeline.discovery_publish_service.httpx.Client", return_value=client): + service.publish(transaction_id="txn-bv", document_kind="advisory", group=None) + + catalog = client.post.call_args.kwargs["json"]["message"]["catalogs"][0] + assert catalog["id"] == "cat-oan-knowledge-provider-advisories" + assert catalog["provider"] == { + "id": "advisory.bv.provider.oan.dev", + "descriptor": { + "code": "advisory.bv.provider.oan.dev", + "name": "Advisories from documents", + }, + } + + @pytest.mark.unit + def test_provider_naming_is_read_from_the_environment(self, monkeypatch, tmp_path): + from pipeline.discovery_publish_service import DiscoveryPublishService + + monkeypatch.setenv("DISCOVERY_SERVICE_ENDPOINT", "https://discovery.example.com") + monkeypatch.setenv("NETWORK_SENDER_ID", "docs-pipeline-bv") + monkeypatch.setenv("NETWORK_PROVIDER_ID_TEMPLATE", "{kind}.{instance}.provider.oan.prod") + monkeypatch.setenv("NETWORK_GROUP_PROVIDER_NAME_TEMPLATE", "{group_name} - {provider_name}") + monkeypatch.setenv("NETWORK_SCHEMES_PROVIDER_NAME", "Government schemes") + _use_group_providers(monkeypatch, tmp_path, '{}') + service = DiscoveryPublishService() + client = _client_returning( + _on_publish_response("mh-cat-oan-knowledge-provider-schemes", "ACCEPTED") + ) + + with patch("pipeline.discovery_publish_service.httpx.Client", return_value=client): + service.publish(transaction_id="txn-mh", document_kind="scheme", group="mh") + + provider = client.post.call_args.kwargs["json"]["message"]["catalogs"][0]["provider"] + assert provider == { + "id": "scheme.mh.provider.oan.prod", + "descriptor": { + "code": "scheme.mh.provider.oan.prod", + "name": "Maharashtra - Government schemes", + }, + } + + @pytest.mark.unit + def test_malformed_provider_map_fails_construction(self, monkeypatch, tmp_path): + from pipeline.discovery_publish_service import DiscoveryPublishService + from pipeline.network_group import NetworkProviderConfigError + + monkeypatch.setenv("DISCOVERY_SERVICE_ENDPOINT", "https://discovery.example.com") + monkeypatch.setenv("NETWORK_SENDER_ID", "docs-pipeline-bv") + _use_group_providers(monkeypatch, tmp_path, "{broken") + + with pytest.raises(NetworkProviderConfigError): + DiscoveryPublishService() + + @pytest.mark.unit + def test_bad_provider_id_template_fails_construction(self, monkeypatch): + from pipeline.discovery_publish_service import DiscoveryPublishService + from pipeline.network_group import NetworkProviderConfigError + + monkeypatch.setenv("DISCOVERY_SERVICE_ENDPOINT", "https://discovery.example.com") + monkeypatch.setenv("NETWORK_SENDER_ID", "docs-pipeline-bv") + monkeypatch.setenv("NETWORK_PROVIDER_ID_TEMPLATE", "{state}.provider.oan.dev") + + with pytest.raises(NetworkProviderConfigError, match="NETWORK_PROVIDER_ID_TEMPLATE"): + DiscoveryPublishService() diff --git a/tests/test_document_repository.py b/tests/test_document_repository.py index dbf6c4f..ae6dba4 100644 --- a/tests/test_document_repository.py +++ b/tests/test_document_repository.py @@ -76,3 +76,26 @@ def test_unclassified_document_reads_as_the_column_default(self, db_connection, @pytest.mark.db def test_missing_document_reads_as_none(self, db_connection): assert DocumentRepository().get_document_kind("no-such-workflow") is None + + +class TestGetDocumentGroup: + @pytest.fixture(autouse=True) + def _default_no_group_instances(self, monkeypatch): + monkeypatch.delenv("PORTAL_INSTANCE", raising=False) + monkeypatch.delenv("DEFAULT_INSTANCE", raising=False) + + @pytest.mark.unit + @pytest.mark.parametrize( + "instance,group", + [("mh", "mh"), ("MH", "mh"), ("bv", None), ("default", None), (None, None)], + ) + def test_group_comes_from_the_documents_instance(self, instance, group): + repo = DocumentRepository(FakeDb(document={"instance": instance})) + + assert repo.get_document_group("wf-1") == group + + @pytest.mark.unit + def test_returns_none_for_a_missing_document(self): + repo = DocumentRepository(FakeDb(document=None)) + + assert repo.get_document_group("wf-nope") is None diff --git a/tests/test_network_group.py b/tests/test_network_group.py new file mode 100644 index 0000000..77f6da8 --- /dev/null +++ b/tests/test_network_group.py @@ -0,0 +1,256 @@ +from pathlib import Path + +import pytest + +from pipeline.network_group import ( + DEFAULT_PROVIDER_CONFIG_OVERRIDE_PATH, + PROVIDER_CONFIG_OVERRIDE_PATH_ENV, + NetworkProviderConfigError, + load_provider_id_overrides, + load_provider_naming, + parse_provider_id_overrides, +) + + +@pytest.fixture(autouse=True) +def _default_instances(monkeypatch): + monkeypatch.delenv("PORTAL_INSTANCE", raising=False) + monkeypatch.delenv("DEFAULT_INSTANCE", raising=False) + + +class TestParseProviderIdOverrides: + @pytest.mark.unit + @pytest.mark.parametrize("raw", [None, "", " "]) + def test_unset_means_no_overrides(self, raw): + assert parse_provider_id_overrides(raw, "providers.json") == {} + + @pytest.mark.unit + def test_codes_and_kinds_are_case_insensitive_and_ids_trimmed(self): + parsed = parse_provider_id_overrides( + '{" KA ": {"Advisory": " advisory.ka.com ", "scheme": "scheme.ka.com"}, "tn": {"scheme": "s.tn"}}', + "providers.json", + ) + + assert parsed == { + "ka": {"advisory": "advisory.ka.com", "scheme": "scheme.ka.com"}, + "tn": {"scheme": "s.tn"}, + } + + @pytest.mark.unit + @pytest.mark.parametrize( + "raw", + [ + "not json", + '["ka"]', + '{"ka": "advisory.ka.com"}', + '{"ka": {}}', + '{"": {"advisory": "advisory.ka.com"}}', + '{"ka": {"video": "video.ka.com"}}', + '{"ka": {"advisory": ""}}', + '{"ka": {"advisory": 42}}', + '{"KA": {"advisory": "a"}, "ka": {"advisory": "b"}}', + '{"ka": {"advisory": "Advisory.KA.com"}}', + '{"ka": {"advisory": "advisory ka.com"}}', + '{"ka": {"advisory": ".advisory.ka.com"}}', + '{"ka": {"advisory": "advisory.ka.com."}}', + '{"ka": {"advisory": "advisory..ka.com"}}', + ], + ) + def test_malformed_overrides_are_rejected(self, raw): + with pytest.raises(NetworkProviderConfigError, match="providers.json"): + parse_provider_id_overrides(raw, "providers.json") + + +def _providers_file(monkeypatch, tmp_path, content): + path = tmp_path / "network_group_providers.json" + path.write_text(content) + monkeypatch.setenv(PROVIDER_CONFIG_OVERRIDE_PATH_ENV, str(path)) + return path + + +class TestLoadProviderIdOverrides: + @pytest.mark.unit + def test_reads_the_configured_file(self, monkeypatch, tmp_path): + _providers_file(monkeypatch, tmp_path, '{"ka": {"advisory": "advisory.ka.com"}}') + + assert load_provider_id_overrides() == {"ka": {"advisory": "advisory.ka.com"}} + + @pytest.mark.unit + def test_unset_path_uses_the_bundled_empty_default(self, monkeypatch): + monkeypatch.delenv(PROVIDER_CONFIG_OVERRIDE_PATH_ENV, raising=False) + + assert DEFAULT_PROVIDER_CONFIG_OVERRIDE_PATH.is_file() + assert load_provider_id_overrides() == {} + + @pytest.mark.unit + def test_missing_file_raises(self, monkeypatch, tmp_path): + missing = tmp_path / "absent.json" + monkeypatch.setenv(PROVIDER_CONFIG_OVERRIDE_PATH_ENV, str(missing)) + + with pytest.raises(NetworkProviderConfigError, match="absent.json"): + load_provider_id_overrides() + + @pytest.mark.unit + def test_malformed_file_raises(self, monkeypatch, tmp_path): + _providers_file(monkeypatch, tmp_path, "{broken") + + with pytest.raises(NetworkProviderConfigError, match="network_group_providers.json"): + load_provider_id_overrides() + + @pytest.mark.unit + def test_example_file_is_valid(self): + example = Path(__file__).parents[1] / "deploy" / "network_group_providers.example.json" + + assert parse_provider_id_overrides(example.read_text(), str(example))["ka"] == { + "advisory": "advisory.ka.com", + "scheme": "scheme.ka.com", + } + + +class TestLoadProviderNaming: + @pytest.fixture(autouse=True) + def _clear_naming_env(self, monkeypatch): + for name in ( + "NETWORK_PROVIDER_ID_TEMPLATE", + "NETWORK_GROUP_PROVIDER_NAME_TEMPLATE", + "NETWORK_ADVISORY_PROVIDER_NAME", + "NETWORK_SCHEMES_PROVIDER_NAME", + ): + monkeypatch.delenv(name, raising=False) + + @pytest.mark.unit + def test_defaults(self, monkeypatch): + monkeypatch.delenv(PROVIDER_CONFIG_OVERRIDE_PATH_ENV, raising=False) + + naming = load_provider_naming() + + assert naming.shared_instance == "bv" + assert naming.id_template == "{kind}.{instance}.provider.oan.dev" + assert naming.group_name_template == "{group_name} {provider_name}" + assert naming.provider_names == { + "advisory": "Advisories from documents", + "scheme": "Schemes from documents", + } + assert naming.provider_id_overrides == {} + + @pytest.mark.unit + def test_carries_the_provider_id_overrides(self, monkeypatch, tmp_path): + _providers_file(monkeypatch, tmp_path, '{"ka": {"scheme": "scheme.ka.com"}}') + + assert load_provider_naming().provider_id_overrides == {"ka": {"scheme": "scheme.ka.com"}} + + @pytest.mark.unit + def test_reads_overrides(self, monkeypatch): + monkeypatch.setenv("PORTAL_INSTANCE", "Portal") + monkeypatch.setenv("NETWORK_PROVIDER_ID_TEMPLATE", "{instance}.{kind}.example.net") + monkeypatch.setenv("NETWORK_GROUP_PROVIDER_NAME_TEMPLATE", "{provider_name} - {group_name}") + monkeypatch.setenv("NETWORK_ADVISORY_PROVIDER_NAME", "Advice") + monkeypatch.setenv("NETWORK_SCHEMES_PROVIDER_NAME", "Schemes") + + naming = load_provider_naming() + + assert naming.shared_instance == "portal" + assert naming.id_template == "{instance}.{kind}.example.net" + assert naming.group_name_template == "{provider_name} - {group_name}" + assert naming.provider_names == {"advisory": "Advice", "scheme": "Schemes"} + + @pytest.mark.unit + @pytest.mark.parametrize( + "env_name,template", + [ + ("NETWORK_PROVIDER_ID_TEMPLATE", "{state}.provider.oan.dev"), + ("NETWORK_PROVIDER_ID_TEMPLATE", "{0}.provider.oan.dev"), + ("NETWORK_PROVIDER_ID_TEMPLATE", "{kind.provider.oan.dev"), + ("NETWORK_PROVIDER_ID_TEMPLATE", "{kind.x}.{instance}.provider.oan.dev"), + ("NETWORK_GROUP_PROVIDER_NAME_TEMPLATE", "{group} {provider_name}"), + ], + ) + def test_unknown_placeholders_are_rejected(self, monkeypatch, env_name, template): + monkeypatch.setenv(env_name, template) + + with pytest.raises(NetworkProviderConfigError, match=env_name): + load_provider_naming() + + @pytest.mark.unit + @pytest.mark.parametrize( + "template", + ["{kind}.oan.dev", "{instance}.provider.oan.dev", "provider.oan.dev"], + ) + def test_id_template_must_name_both_kind_and_instance(self, monkeypatch, template): + monkeypatch.setenv("NETWORK_PROVIDER_ID_TEMPLATE", template) + + with pytest.raises(NetworkProviderConfigError, match="must use"): + load_provider_naming() + + @pytest.mark.unit + @pytest.mark.parametrize("template", ["{provider_name}", "Knowledge from documents"]) + def test_group_name_template_must_name_the_group(self, monkeypatch, template): + monkeypatch.setenv("NETWORK_GROUP_PROVIDER_NAME_TEMPLATE", template) + + with pytest.raises(NetworkProviderConfigError, match="must use"): + load_provider_naming() + + @pytest.mark.unit + def test_group_name_template_without_the_provider_name_is_accepted(self, monkeypatch): + monkeypatch.setenv("NETWORK_GROUP_PROVIDER_NAME_TEMPLATE", "{group_name} knowledge") + + assert load_provider_naming().group_name_template == "{group_name} knowledge" + + @pytest.mark.unit + @pytest.mark.parametrize( + "template", + [ + "{kind} {instance}.provider", + "{kind}.{instance}.", + ".{kind}.{instance}", + "{kind}..{instance}", + "{kind}.{instance}.Provider.oan.dev", + "{kind}/{instance}", + ], + ) + def test_id_template_must_render_a_valid_id(self, monkeypatch, template): + monkeypatch.setenv("NETWORK_PROVIDER_ID_TEMPLATE", template) + + with pytest.raises(NetworkProviderConfigError, match="provider id"): + load_provider_naming() + + @pytest.mark.unit + @pytest.mark.parametrize("template", ["{kind}-{instance}.provider.oan.dev", "{instance}.{kind}.example-1.net"]) + def test_id_template_with_a_valid_shape_is_accepted(self, monkeypatch, template): + monkeypatch.setenv("NETWORK_PROVIDER_ID_TEMPLATE", template) + + assert load_provider_naming().id_template == template + + +class TestBootRefusesMalformedProviderMap: + @pytest.fixture + def broken_map(self, monkeypatch, tmp_path): + _providers_file(monkeypatch, tmp_path, "{broken") + monkeypatch.setenv("TRANSLATION_PROVIDER", "none") + + @pytest.mark.unit + @pytest.mark.asyncio + async def test_worker_does_not_start(self, broken_map, monkeypatch): + import pipeline.worker as worker + + async def never_connect(*args, **kwargs): + raise AssertionError("worker connected to Temporal despite a bad provider map") + + monkeypatch.setattr(worker.Client, "connect", never_connect) + + with pytest.raises(NetworkProviderConfigError): + await worker.main() + + @pytest.mark.unit + @pytest.mark.asyncio + async def test_api_does_not_start(self, broken_map, monkeypatch): + import pipeline.api as api + + async def never_connect(*args, **kwargs): + raise AssertionError("API connected to Temporal despite a bad provider map") + + monkeypatch.setattr(api.Client, "connect", never_connect) + + with pytest.raises(NetworkProviderConfigError): + async with api.lifespan(api.app): + pass diff --git a/tests/test_network_provider.py b/tests/test_network_provider.py new file mode 100644 index 0000000..fb66840 --- /dev/null +++ b/tests/test_network_provider.py @@ -0,0 +1,107 @@ +import pytest + +from pipeline.network_provider import Group, InvalidProviderIdError, ProviderNaming + +MAHARASHTRA = Group(code="mh", name="Maharashtra") + + +def _provider(provider_id, name): + return {"id": provider_id, "descriptor": {"code": provider_id, "name": name}} + + +class TestDefaultProviderNaming: + @pytest.mark.unit + @pytest.mark.parametrize( + "kind,group,expected", + [ + ("advisory", None, _provider("advisory.bv.provider.oan.dev", "Advisories from documents")), + ("scheme", None, _provider("scheme.bv.provider.oan.dev", "Schemes from documents")), + ( + "advisory", + MAHARASHTRA, + _provider("advisory.mh.provider.oan.dev", "Maharashtra Advisories from documents"), + ), + ( + "scheme", + MAHARASHTRA, + _provider("scheme.mh.provider.oan.dev", "Maharashtra Schemes from documents"), + ), + ], + ) + def test_provider_per_kind_and_instance(self, kind, group, expected): + assert ProviderNaming(shared_instance="bv").provider(kind, group) == expected + + +class TestConfiguredProviderNaming: + NAMING = ProviderNaming( + shared_instance="portal", + id_template="{instance}-{kind}.example.net", + group_name_template="{provider_name} ({group_name})", + provider_names={"advisory": "Advice", "scheme": "Schemes"}, + ) + + @pytest.mark.unit + def test_shared_provider_follows_the_configured_naming(self): + assert self.NAMING.provider("scheme", None) == _provider("portal-scheme.example.net", "Schemes") + + @pytest.mark.unit + def test_group_provider_follows_the_configured_naming(self): + assert self.NAMING.provider("advisory", MAHARASHTRA) == _provider( + "mh-advisory.example.net", "Advice (Maharashtra)" + ) + + +class TestProviderIdOverrides: + NAMING = ProviderNaming( + shared_instance="bv", + provider_id_overrides={ + "ka": {"advisory": "advisory.ka.com", "scheme": "scheme.ka.com"}, + "tn": {"scheme": "scheme.tn.com"}, + "bv": {"advisory": "advisory.bharat.example"}, + } + ) + + @pytest.mark.unit + @pytest.mark.parametrize("kind,provider_id", [("advisory", "advisory.ka.com"), ("scheme", "scheme.ka.com")]) + def test_overridden_group_uses_its_own_provider_id(self, kind, provider_id): + provider = self.NAMING.provider(kind, Group(code="ka", name="Karnataka")) + + assert provider["id"] == provider_id + assert provider["descriptor"]["code"] == provider_id + + @pytest.mark.unit + def test_override_changes_only_the_id(self): + provider = self.NAMING.provider("scheme", Group(code="ka", name="Karnataka")) + + assert provider["descriptor"]["name"] == "Karnataka Schemes from documents" + + @pytest.mark.unit + def test_a_kind_without_an_override_falls_back_to_the_template(self): + provider = self.NAMING.provider("advisory", Group(code="tn", name="Tamil Nadu")) + + assert provider["id"] == "advisory.tn.provider.oan.dev" + + @pytest.mark.unit + def test_group_without_overrides_uses_the_defaults(self): + assert self.NAMING.provider("scheme", MAHARASHTRA) == _provider( + "scheme.mh.provider.oan.dev", "Maharashtra Schemes from documents" + ) + + @pytest.mark.unit + def test_shared_instance_can_be_overridden(self): + assert self.NAMING.provider("advisory", None)["id"] == "advisory.bharat.example" + + +class TestInvalidProviderIdAtPublish: + @pytest.mark.unit + @pytest.mark.parametrize("code", ["ladakh_ut", "u p", "MH."]) + def test_an_instance_that_renders_an_invalid_id_is_refused(self, code): + with pytest.raises(InvalidProviderIdError, match="must be lowercase"): + ProviderNaming(shared_instance="bv").provider("advisory", Group(code=code, name="Somewhere")) + + @pytest.mark.unit + def test_an_invalid_override_is_refused(self): + naming = ProviderNaming(shared_instance="bv", provider_id_overrides={"ka": {"scheme": "Scheme.KA"}}) + + with pytest.raises(InvalidProviderIdError): + naming.provider("scheme", Group(code="ka", name="Karnataka")) diff --git a/tests/test_tenancy.py b/tests/test_tenancy.py index 60148c2..a2149cb 100644 --- a/tests/test_tenancy.py +++ b/tests/test_tenancy.py @@ -12,6 +12,7 @@ allowed_instances, assert_document_instance_access, assert_instance_access, + group_for_instance, resolve_create_instance, user_can_access_instance, ) @@ -361,3 +362,41 @@ def test_api_helpers_hide_cross_tenant_mutations(db_connection, monkeypatch): api._require_document_for_user("wf-tenant-b-doc", user) assert exc.value.status_code == 404 assert api._document_for_user_or_none("wf-tenant-b-doc", user) is None + + +class TestGroupForInstance: + @pytest.fixture(autouse=True) + def _default_instances(self, monkeypatch): + monkeypatch.delenv("PORTAL_INSTANCE", raising=False) + monkeypatch.delenv("DEFAULT_INSTANCE", raising=False) + + @pytest.mark.unit + @pytest.mark.parametrize( + "instance,expected", + [ + ("mh", "mh"), + (" MH ", "mh"), + ("ka", "ka"), + ("bv", None), + ("BV", None), + ("default", None), + ("", None), + (None, None), + ], + ) + def test_states_are_groups_and_platform_instances_are_not(self, instance, expected): + assert group_for_instance(instance) == expected + + @pytest.mark.unit + def test_follows_a_configured_portal_instance(self, monkeypatch): + monkeypatch.setenv("PORTAL_INSTANCE", "portal") + + assert group_for_instance("portal") is None + assert group_for_instance("mh") == "mh" + + @pytest.mark.unit + def test_follows_a_configured_default_instance(self, monkeypatch): + monkeypatch.setenv("DEFAULT_INSTANCE", "local") + + assert group_for_instance("local") is None + assert group_for_instance("mh") == "mh"