diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 55e00b78d..6a24bce13 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -80,6 +80,7 @@ flowchart LR | `temporal_expressions.py` | Pure Korean relative-time resolver for Global Ask (ADR 0150) | | `ask_time_axis.py` | Event-time vs ingestion-time clock choice for that window (ADR 0202) | | `ontology.py` | Loads the governed Turtle source tree (`lineageweave-kg.ttl` plus generated fragments), the formal OWL 2/RDFS/SKOS vocabulary for the Knowledge Graph's node/edge types, source taxonomies, and published O*NET linkages (ADR 0004, ADR 0252, ADR 0255, ADR 0256) | +| `backend/app/occupation_rating_ingestion.py` | Projects authenticated, bounded occupation-rating source evidence with exact decimal and artifact provenance semantics (ADR 0258) | | `ontology_neighborhood.py` | Bounded typed ontology/provenance neighborhood (ADR 0184); PostgreSQL stays authoritative, OWL subclass is not an instance edge | | `ontology_source_cursor.py` | Opaque HMAC source-window continuation (ADR 0124); keyset pagination, never OFFSET | | `period_report.py` | Fit GRM/GPCM on persisted IRT rows, FIPC-select, EAP-score a period (ADR 0003 slice 3; Bock & Mislevy, 1982) | diff --git a/CHANGELOG.md b/CHANGELOG.md index d8a2f0f2b..15193b4e0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,13 +8,21 @@ All notable changes to this project are documented here. Format follows ### Added +- Authenticated occupation profiles can now read one exact imported + release/source with rating and scale artifact provenance, exact decimal + strings, uncertainty, suppression/relevance flags, and explicit unavailable + versus empty outcomes. The endpoint derives no ranking or recommendation + (ADR 0258). - O*NET occupation-rating source evidence now has a replay-safe PostgreSQL contract with normalized release, source-table, scale, occupation, element, and observation tables. Exact release/source LIST partitions fail closed; nullable categories remain idempotent identities, divergent duplicates and truncation fail closed, and no source value is promoted to a local weight or - person score. Task Ratings remain outside this content-model-element store - pending their own normalized identity contract (ADR 0257). + person score. A pinned CSV importer validates rating and scale-reference + digests, row counts, source identities, scale bounds, uncertainty, flags, + and exact update months before immutable transactional insertion. Task + Ratings remain outside this content-model-element store pending their own + normalized identity contract (ADR 0257). - All eight O*NET 31.0 published linkage tables now contribute 1,417 directed Ability/Essential Skill/Transferable Skill/Work Style relations to Work Activities and Work Context. Every direct relation has an exact reified diff --git a/backend/app/main.py b/backend/app/main.py index 6457bbde1..60dd9e1a5 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -30,7 +30,7 @@ import asyncpg import redis.asyncio as redis -from fastapi import Depends, FastAPI, HTTPException, Query, status +from fastapi import Depends, FastAPI, HTTPException, Path, Query, status from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel @@ -98,6 +98,7 @@ upsert_commitment_ticket, ) from backend.app.operations_dashboard import fetch_operations_dashboard +from backend.app.occupation_rating_ingestion import fetch_occupation_ratings from backend.app.keyman_ingestion import ingest_post_keymen from backend.app.knowledge_graph import ( corporate_entity_exists, @@ -2275,6 +2276,32 @@ async def read_ontology_neighborhood( return payload +@app.get("/api/occupations/{onetsoc_code}/ratings") +async def read_occupation_ratings( + onetsoc_code: str = Path(..., pattern=r"^[0-9]{2}-[0-9]{4}\.[0-9]{2}$"), + data_release_code: str = Query( + ..., min_length=1, max_length=63, pattern=r"^[a-z0-9][a-z0-9.-]*$" + ), + source_table_code: str = Query( + ..., min_length=1, max_length=63, pattern=r"^[a-z][a-z0-9_]*$" + ), + limit: int = Query(100, ge=1, le=500), + offset: int = Query(0, ge=0, le=10000), + _account: CurrentAccount = Depends(get_current_account), + pool: asyncpg.Pool = Depends(get_pool), +) -> dict[str, object]: + """Return one authenticated, provenance-bearing occupation source profile.""" + async with pool.acquire() as conn: + return await fetch_occupation_ratings( + conn, + data_release_code=data_release_code, + source_table_code=source_table_code, + onetsoc_code=onetsoc_code, + limit=limit, + offset=offset, + ) + + @app.get("/api/posts/{post_id}/counterparties") async def read_post_counterparties( post_id: str, diff --git a/backend/app/occupation_rating_ingestion.py b/backend/app/occupation_rating_ingestion.py new file mode 100644 index 000000000..320fe248d --- /dev/null +++ b/backend/app/occupation_rating_ingestion.py @@ -0,0 +1,127 @@ +"""Read exact imported occupation ratings without deriving a score or weight.""" + +from __future__ import annotations + +from decimal import Decimal +from typing import Any, Protocol + + +class RatingReadConnection(Protocol): + """Small asyncpg-compatible surface used by the rating read projection.""" + + async def fetchrow(self, query: str, *args: object) -> Any: + """Return one row or ``None``.""" + + async def fetch(self, query: str, *args: object) -> list[Any]: + """Return ordered rows.""" + + +def _decimal_text(value: Decimal | None) -> str | None: + """Return the exact database decimal representation or honest absence.""" + return str(value) if value is not None else None + + +async def fetch_occupation_ratings( + conn: RatingReadConnection, + *, + data_release_code: str, + source_table_code: str, + onetsoc_code: str, + limit: int, + offset: int, +) -> dict[str, object]: + """Return one bounded source profile and explicit artifact availability.""" + source = await conn.fetchrow( + """select rating_source.source_table_name, + rating_source.source_artifact_url, + rating_source.source_artifact_sha256, + rating_source.source_row_count, + scale_source.source_artifact_url as scale_artifact_url, + scale_source.source_artifact_sha256 as scale_artifact_sha256, + scale_source.source_row_count as scale_source_row_count + from occupational_source_table rating_source + left join occupational_source_table scale_source + on scale_source.data_release_code = rating_source.data_release_code + and scale_source.source_table_code = 'scales_reference' + where rating_source.data_release_code = $1 + and rating_source.source_table_code = $2""", + data_release_code, + source_table_code, + ) + if source is None: + return { + "data_release_code": data_release_code, + "source_table_code": source_table_code, + "onetsoc_code": onetsoc_code, + "source_available": False, + "source": None, + "items": [], + "next_offset": None, + } + rows = await conn.fetch( + """select observation.element_id, element.element_name, + observation.scale_id, scale.scale_name, + scale.minimum_value, scale.maximum_value, + observation.category_value, observation.data_value, + observation.sample_size, observation.standard_error, + observation.lower_ci_bound, observation.upper_ci_bound, + observation.recommend_suppress, observation.not_relevant, + observation.source_updated_month, observation.domain_source_code + from occupational_rating_observation observation + join occupational_element_definition element + on element.data_release_code = observation.data_release_code + and element.element_id = observation.element_id + join occupational_scale_definition scale + on scale.data_release_code = observation.data_release_code + and scale.scale_id = observation.scale_id + where observation.data_release_code = $1 + and observation.source_table_code = $2 + and observation.onetsoc_code = $3 + order by observation.element_id, observation.scale_id, + observation.category_value nulls first + limit $4 offset $5""", + data_release_code, + source_table_code, + onetsoc_code, + limit + 1, + offset, + ) + page = rows[:limit] + items = [ + { + "element_id": row["element_id"], + "element_name": row["element_name"], + "scale_id": row["scale_id"], + "scale_name": row["scale_name"], + "minimum_value": _decimal_text(row["minimum_value"]), + "maximum_value": _decimal_text(row["maximum_value"]), + "category_value": row["category_value"], + "data_value": _decimal_text(row["data_value"]), + "sample_size": row["sample_size"], + "standard_error": _decimal_text(row["standard_error"]), + "lower_ci_bound": _decimal_text(row["lower_ci_bound"]), + "upper_ci_bound": _decimal_text(row["upper_ci_bound"]), + "recommend_suppress": row["recommend_suppress"], + "not_relevant": row["not_relevant"], + "source_updated_month": row["source_updated_month"], + "domain_source_code": row["domain_source_code"], + } + for row in page + ] + return { + "data_release_code": data_release_code, + "source_table_code": source_table_code, + "onetsoc_code": onetsoc_code, + "source_available": True, + "source": { + "source_table_name": source["source_table_name"], + "source_artifact_url": source["source_artifact_url"], + "source_artifact_sha256": source["source_artifact_sha256"], + "source_row_count": source["source_row_count"], + "scale_artifact_url": source["scale_artifact_url"], + "scale_artifact_sha256": source["scale_artifact_sha256"], + "scale_source_row_count": source["scale_source_row_count"], + }, + "items": items, + "next_offset": offset + limit if len(rows) > limit else None, + } diff --git a/docs/adr/0257-onet-occupation-rating-observation-store.md b/docs/adr/0257-onet-occupation-rating-observation-store.md index aae24d860..8e00e13a8 100644 --- a/docs/adr/0257-onet-occupation-rating-observation-store.md +++ b/docs/adr/0257-onet-occupation-rating-observation-store.md @@ -57,13 +57,22 @@ also violate LineageWeave's externalized-compute boundary. truncation fail closed. Any later aggregation, comparison, temporal model, multilevel model, or occupational recommendation belongs to TEPP/fast-mlsirm or another owning Rust service and must cite these rows. +10. `scripts/import_onet_ratings.py` accepts one caller-pinned official CSV and + Scales Reference file. It verifies both artifact SHA-256 values and row + counts, exact scale names and bounds, reference-name consistency, finite + decimals, Y/N/blank flags, optional Category and Not Relevant columns, + exact `MM/YYYY` source months, and observation-key uniqueness before opening the + target connection. A transaction-scoped advisory lock serializes one + release's partition DDL. ## Consequences LineageWeave can import the governed public O*NET content-model rating corpus without manufacturing semantics or embedding large production datasets in git. Release/source partitions localize hot imports and permit exact detach/archive -operations. A separate API/UI ADR is still required before exposing ratings. +operations. The same pinned artifact is idempotent; a reused release, source, +scale, occupation, or element identity with different source metadata fails +closed. A separate API/UI ADR is still required before exposing ratings. ## References diff --git a/docs/adr/0258-occupation-rating-read-api.md b/docs/adr/0258-occupation-rating-read-api.md new file mode 100644 index 000000000..7af31c817 --- /dev/null +++ b/docs/adr/0258-occupation-rating-read-api.md @@ -0,0 +1,50 @@ +# ADR 0258: Authenticated occupation-rating source read API + +**Status:** Accepted + +**Date:** 2026-08-27 +**Extends:** ADR 0120, ADR 0184, ADR 0257 + +## Context + +ADR 0257 preserves released occupation-to-element observations, but a database +import alone does not let a product user inspect what a job profile says. A +read contract must distinguish an unimported source from an imported source +with no row for one occupation, preserve low-precision and not-relevant flags, +and avoid presenting a published rating as a local weight or recommendation. + +## Decision + +1. Add an authenticated, read-only occupation-rating endpoint. O*NET source + observations are licensed public reference data and are not tenant records; + any authenticated LineageWeave account may read an imported artifact. +2. Require exact release, source-table, and O*NET-SOC codes. Return + `source_available=false` when that pinned artifact is not imported; return + `source_available=true` with an empty item list when it is imported but has + no observation for the requested occupation. +3. Return the rating and Scales Reference artifact URLs, SHA-256 values, and + row counts. Every observation retains element/scale identity, declared + bounds, optional category, exact decimal strings, sample/error/interval, + suppression, relevance, source month, and domain source. +4. Order by element, scale, and category and use bounded offset pagination. + Per-occupation source partitions bound this projection; a cursor needs a + later decision only if measured production latency requires it. +5. Do not aggregate, rank, normalize, infer person traits, or recommend an + occupation. Suppressed values remain visible with the suppression flag so a + user can audit the source without mistaking low precision for absence. +6. API and frontend copy describe the evidence and the user's next action, + never importer, partition, model-provider, or orchestration internals. + +## Consequences + +The semantic layer gains an honest product read boundary without duplicating +psychometric arithmetic. An accessible UI and its Storybook states remain a +separate delivery step after this API has authenticated runtime evidence. + +## References + +National Center for O*NET Development. (2026). *O*NET 31.0 database* [Data +set]. https://www.onetcenter.org/database.html + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: +Queries—limit and offset*. https://www.postgresql.org/docs/current/queries-limit.html diff --git a/docs/adr/README.md b/docs/adr/README.md index 74e74592c..1a36fa1a8 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -34,6 +34,7 @@ decision from them. | [`SOC_2018_HIERARCHY_REFERENCES.md`](../doctoring/SOC_2018_HIERARCHY_REFERENCES.md) | [0252](0252-complete-2018-soc-hierarchy.md) | | [`ONET_31_LINKAGE_REFERENCES.md`](../doctoring/ONET_31_LINKAGE_REFERENCES.md) | [0256](0256-onet-content-model-published-linkages.md) | | [`ONET_RATING_STORE_REFERENCES.md`](../doctoring/ONET_RATING_STORE_REFERENCES.md) | [0257](0257-onet-occupation-rating-observation-store.md) | +| Occupation-rating authenticated read projection | [0258](0258-occupation-rating-read-api.md) | [0011](0011-prov-o-standard-relations.md) and [0065](0065-prov-o-provenance-boundary.md) cite the dated W3C PROV-O and PROV-DM Recommendations (https://www.w3.org/TR/2013/REC-prov-o-20130430/ and https://www.w3.org/TR/2013/REC-prov-dm-20130430/). diff --git a/docs/product-requirements.md b/docs/product-requirements.md index dae982b4f..d84b284b0 100644 --- a/docs/product-requirements.md +++ b/docs/product-requirements.md @@ -140,10 +140,26 @@ persistence and UI remain unavailable until their separate ADR acceptance. Ratings remain unavailable until their integer Task IDs and statements have a separate normalized source-target contract. -Acceptance: the replay-safe migration creates the normalized store; PostgreSQL -integration proves missing partitions fail closed, null-category UPSERT is -idempotent, and invalid uncertainty is rejected. Corpus import, API, UI, and -derived modeling remain unavailable until separate accepted delivery records. +Acceptance: the replay-safe migration creates the normalized store; the pinned +CSV importer validates both rating and scale-reference digests and row counts, +reference identity, source scale, uncertainty, flags, and dates before +persistence; PostgreSQL integration proves missing partitions fail closed and +repeated null-category UPSERT is idempotent. +API, UI, and derived modeling remain unavailable until separate accepted +delivery records. + +### PRD-FR-2E — Occupation-rating evidence read + +- Let an authenticated user open one exact release/source/occupation profile + with both rating and scale artifact provenance (ADR 0258). +- Distinguish an unavailable imported source from an available source with no + observation for the occupation. +- Preserve exact decimal text, uncertainty, suppression, relevance, source + month, domain source, and declared bounds; derive no ranking or recommendation. + +Acceptance: invalid identifiers and unbounded pages are rejected; an unavailable +source never appears as a negative profile; pagination is deterministic; and a +suppressed observation retains its value and warning flag together. ### PRD-FR-3 — Bounded ontology exploration diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index c99d17e75..bd7e5a884 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -1,7 +1,7 @@ # Product & Technical Gap Baseline > Current queue overlay: 2026-08-27 KST. Protected `main` was -> `ff7431bd1851c03e737808d22c6a2d43968582f9`; 35 PRs and 10 issues were +> `ff7431bd1851c03e737808d22c6a2d43968582f9`; 38 PRs and 10 issues were > open. This overlay supersedes the older queue count and exact-head table > below, which remain historical evidence. Re-fetch the head, checks, reviews, > threads, applicable rulesets, and merge SHA immediately before any lifecycle @@ -26,18 +26,20 @@ explicit unavailable state, not a reason to infer mappings from labels. | Classification depth | ADR 0252's candidate branch imports all 1,447 official 2018 SOC nodes with source-declared hierarchy, pinned XLSX/CSV digests, a deterministic Turtle renderer, and fail-closed read model | Pass exact-head review/checks and protected merge; add ISCO/ESCO crosswalks only where the publishing authority supplies them | | Construct granularity | ADR 0255's candidate branch publishes all 3,006 O*NET 31.0 Content Model Reference concepts with exact IDs, names, descriptions, and source-defined parents; it imports no ratings or person assertions | Pass exact-head review/checks and protected merge; import separately released occupation-element observations only through the provenance contract below | | Construct-to-work relations | ADR 0256's candidate branch imports all 1,417 official Ability/Essential Skill/Transferable Skill/Work Style linkages to Work Activities and Work Context with assertion-level source provenance and no invented weights | Pass exact-head review/checks and protected merge; keep these published relevance links distinct from occupation ratings and causal claims | -| Occupation-to-construct relations | ADR 0257 defines a candidate 3NF, release/source-partitioned immutable observation store preserving value, optional category, sample/error/CI, suppression, relevance, exact `MM/YYYY` source update month, source digest, and domain provenance; no corpus is imported | Pass exact-head review/checks and protected merge; add a deterministic source importer for pinned public artifacts; never invent a date, weight, or local normalization | +| Occupation-to-construct relations | ADR 0257 defines a candidate 3NF, release/source-partitioned immutable observation store and deterministic pinned-CSV importer preserving value, optional category, sample/error/CI, suppression, relevance, exact `MM/YYYY` source update month, source digest, and domain provenance. The official O*NET 31.0 Abilities file (94,640 rows, 910 occupations, 52 elements; SHA-256 `7e9cd79791ce6014e1d26d0a449ae5b1e7aa7ef52d39b3934c3bb8d438104b88`) and all 33 Scales Reference rows (SHA-256 `bcba23858ce21ecaacbde303a8993e35d46724b4afb8c9ec2b10e04f42adcfc9`) imported into a throwaway local PostgreSQL database with all 94,640 observations, 55 suppression flags, 7,572 not-relevant flags, and source months from `12/2004` through `08/2026`; every scale retained `scales_reference` artifact provenance, the database was dropped afterward, and no corpus is committed or claimed deployed | Pass exact-head review/checks and protected merge; validate and import every selected official rating artifact through an authorized runtime, returning only aggregate evidence; never invent or locally normalize a weight | | Job-family and job-series semantics | No authoritative employer-specific job architecture is present | Define an organization-neutral import contract that preserves the authorized source hierarchy and distinguishes standard occupation codes from employer job families/series; no label-based binding | | Temporal and multilevel interpretation | Static vocabulary only; no person-level inference is asserted | Version valid and transaction time, preserve occupation/organization/unit nesting and multiple membership, and require TEPP or the owning Rust psychometric service before any calibrated temporal or multilevel result | -| Product consumption | The read model has no persisted semantic-layer consumer or authenticated UI evidence | Add a provenance-bearing API and accessible ontology exploration flow, then verify synthetic Storybook edge states plus authenticated aggregate runtime evidence without exposing identifying records | +| Product consumption | ADR 0258 defines a candidate authenticated occupation-rating API that distinguishes unavailable source from empty profile and returns exact decimal/uncertainty/warning/artifact provenance without ranking; accessible UI and authenticated runtime evidence remain absent | Pass exact-head review/checks and protected merge; add the accessible exploration UI, synthetic Storybook unavailable/empty/suppressed/paginated states, screenshots, and authenticated aggregate runtime evidence without exposing identifying records | ### Current exact-head PR queue | PR | Exact observed head | Base | Observed gate state | |---:|---|---|---| -| #732 | `1c622545` | `feat/onet-31-content-model-ontology` | unstable; full tests in progress; frontend, CodeRabbit, and Devin Review passed; one informational review thread open | -| #731 | `2c848373` | `feat/soc-2018-full-hierarchy` | unstable; full tests, frontend, and Devin Review queued | -| #724 | `f5ee37b0` | `feat/io-occupational-taxonomy` | clean; full tests, frontend, CodeRabbit, and Devin Review passed; parent-first stack gate remains | +| #735 | `56d8cbfa` | `feat/onet-occupation-ratings-contract` | clean immediately after exact parent merge; hosted checks had not yet registered on this head, so this is not merge-readiness evidence | +| #734 | `0dedf33b` | `feat/onet-content-model-linkages` | unstable; two hosted gates pending after exact parent reconciliation | +| #732 | `3e30402d` | `feat/onet-31-content-model-ontology` | unstable; one hosted gate pending after exact parent reconciliation | +| #731 | `9654bb8b` | `feat/soc-2018-full-hierarchy` | unstable; one hosted gate pending after exact parent reconciliation | +| #724 | `fdec8f65` | `feat/io-occupational-taxonomy` | clean; hosted gates passed; parent-first stack gate remains | | #719 | `6ee2278a` | `feat/io-psych-construct-ontology` | unstable; full tests, frontend, CodeRabbit, and Devin Review passed | | #718 | `2723fea3` | `feat/fja-worker-function-ontology` | unstable; full tests, frontend, CodeRabbit, and Devin Review passed | | #717 | `771a8edf` | `feat/voice-of-x-complete-taxonomy` | unstable; 1 pending check(s) | diff --git a/scripts/import_onet_ratings.py b/scripts/import_onet_ratings.py new file mode 100644 index 000000000..7887f740b --- /dev/null +++ b/scripts/import_onet_ratings.py @@ -0,0 +1,537 @@ +"""Validate and import one official O*NET occupation-rating CSV artifact.""" + +from __future__ import annotations + +import argparse +import asyncio +import csv +import hashlib +import json +import re +from dataclasses import dataclass +from datetime import UTC, date, datetime +from decimal import Decimal, InvalidOperation +from pathlib import Path +from urllib.parse import urlsplit + +import asyncpg + +_RATING_FIELDS = { + "O*NET-SOC Code", + "Title", + "Element ID", + "Element Name", + "Scale ID", + "Scale Name", + "Data Value", + "Date", + "Domain Source", +} +_SCALE_FIELDS = {"Scale ID", "Scale Name", "Minimum", "Maximum"} +_SOURCE_CODE = re.compile(r"^[a-z][a-z0-9_]{0,62}$") +_SCALES_SOURCE_CODE = "scales_reference" +_SHA256 = re.compile(r"^[0-9a-fA-F]{64}$") + + +@dataclass(frozen=True) +class ScaleDefinition: + """One exact O*NET scale identity and its declared numeric bounds.""" + + scale_id: str + scale_name: str + minimum_value: Decimal + maximum_value: Decimal + + +@dataclass(frozen=True) +class RatingObservation: + """One validated O*NET occupation-to-element source observation.""" + + onetsoc_code: str + occupation_title: str + element_id: str + element_name: str + scale_id: str + category_value: int | None + data_value: Decimal + sample_size: int | None + standard_error: Decimal | None + lower_ci_bound: Decimal | None + upper_ci_bound: Decimal | None + recommend_suppress: bool | None + not_relevant: bool | None + source_updated_month: str + domain_source_code: str + + +def _decimal(value: str, field: str, *, optional: bool = False) -> Decimal | None: + """Parse one finite source decimal while preserving an honest blank.""" + text = value.strip() + if optional and not text: + return None + try: + parsed = Decimal(text) + except InvalidOperation as exc: + raise ValueError(f"invalid {field}: {value!r}") from exc + if not parsed.is_finite(): + raise ValueError(f"invalid {field}: {value!r}") + return parsed + + +def _integer(value: str, field: str, *, optional: bool = False) -> int | None: + """Parse one source integer while preserving an honest blank.""" + text = value.strip() + if optional and not text: + return None + try: + return int(text) + except ValueError as exc: + raise ValueError(f"invalid {field}: {value!r}") from exc + + +def _flag(value: str, field: str) -> bool | None: + """Parse the official Y/N/blank tri-state flag vocabulary.""" + text = value.strip() + if not text: + return None + if text == "Y": + return True + if text == "N": + return False + raise ValueError(f"invalid {field} flag: {value!r}") + + +def _updated_month(value: str, today: date) -> str: + """Validate and preserve one exact O*NET ``MM/YYYY`` source month.""" + text = value.strip() + if not re.fullmatch(r"(0[1-9]|1[0-2])/[0-9]{4}", text): + raise ValueError(f"invalid source update date: {value!r}") + try: + month_text, year_text = text.split("/") + parsed = date(int(year_text), int(month_text), 1) + except (TypeError, ValueError) as exc: + raise ValueError(f"invalid source update date: {value!r}") from exc + if parsed > today.replace(day=1): + raise ValueError(f"future source update date: {value!r}") + return text + + +def _rows(path: Path, required: set[str]) -> list[dict[str, str]]: + """Read one UTF-8 CSV only when its authoritative columns are present.""" + with path.open(encoding="utf-8-sig", newline="") as handle: + reader = csv.DictReader(handle) + fields = set(reader.fieldnames or ()) + missing = sorted(required - fields) + if missing: + raise ValueError(f"missing CSV columns: {', '.join(missing)}") + rows = [] + for line_number, row in enumerate(reader, start=2): + if None in row or any(value is None for value in row.values()): + raise ValueError(f"malformed CSV row: {line_number}") + rows.append(dict(row)) + return rows + + +def read_scale_file(path: Path) -> dict[str, ScaleDefinition]: + """Return exact scale definitions from the official Scales Reference CSV.""" + scales: dict[str, ScaleDefinition] = {} + for row in _rows(path, _SCALE_FIELDS): + scale_id = row["Scale ID"].strip() + definition = ScaleDefinition( + scale_id=scale_id, + scale_name=row["Scale Name"].strip(), + minimum_value=_decimal(row["Minimum"], "scale minimum"), # type: ignore[arg-type] + maximum_value=_decimal(row["Maximum"], "scale maximum"), # type: ignore[arg-type] + ) + if not scale_id or not definition.scale_name: + raise ValueError("empty scale identity") + if definition.minimum_value > definition.maximum_value: + raise ValueError(f"inverted scale bounds: {scale_id}") + if scale_id in scales and scales[scale_id] != definition: + raise ValueError(f"conflicting scale definition: {scale_id}") + scales[scale_id] = definition + if not scales: + raise ValueError("scale file has no rows") + return scales + + +def read_rating_file( + path: Path, + scales: dict[str, ScaleDefinition], + *, + today: date | None = None, +) -> list[RatingObservation]: + """Validate an official rating CSV and return exact source observations.""" + observed_today = today or datetime.now(UTC).date() + occupations: dict[str, str] = {} + elements: dict[str, str] = {} + observations: list[RatingObservation] = [] + identities: set[tuple[str, str, str, int | None]] = set() + for row in _rows(path, _RATING_FIELDS): + onetsoc_code = row["O*NET-SOC Code"].strip() + occupation_title = row["Title"].strip() + element_id = row["Element ID"].strip() + element_name = row["Element Name"].strip() + scale_id = row["Scale ID"].strip() + scale = scales.get(scale_id) + if scale is None or row["Scale Name"].strip() != scale.scale_name: + raise ValueError(f"unknown or conflicting scale identity: {scale_id}") + if ( + onetsoc_code in occupations + and occupations[onetsoc_code] != occupation_title + ): + raise ValueError(f"conflicting occupation title: {onetsoc_code}") + if element_id in elements and elements[element_id] != element_name: + raise ValueError(f"conflicting element name: {element_id}") + occupations[onetsoc_code] = occupation_title + elements[element_id] = element_name + data_value = _decimal(row["Data Value"], "data value") + if not scale.minimum_value <= data_value <= scale.maximum_value: # type: ignore[operator] + raise ValueError(f"data value outside scale {scale_id}") + sample_size = _integer(row.get("N", ""), "sample size", optional=True) + standard_error = _decimal( + row.get("Standard Error", ""), "standard error", optional=True + ) + lower = _decimal(row.get("Lower CI Bound", ""), "lower CI bound", optional=True) + upper = _decimal(row.get("Upper CI Bound", ""), "upper CI bound", optional=True) + if sample_size is not None and sample_size <= 0: + raise ValueError("sample size must be positive") + if standard_error is not None and standard_error < 0: + raise ValueError("standard error must be non-negative") + if (lower is None) != (upper is None) or (lower is not None and lower > upper): + raise ValueError("invalid confidence interval") + category = _integer(row.get("Category", ""), "category", optional=True) + identity = (onetsoc_code, element_id, scale_id, category) + if identity in identities: + raise ValueError(f"duplicate rating identity: {identity}") + identities.add(identity) + domain_source = row["Domain Source"].strip() + if ( + not onetsoc_code + or not occupation_title + or not element_id + or not element_name + or not domain_source + ): + raise ValueError("empty rating identity") + observations.append( + RatingObservation( + onetsoc_code=onetsoc_code, + occupation_title=occupation_title, + element_id=element_id, + element_name=element_name, + scale_id=scale_id, + category_value=category, + data_value=data_value, # type: ignore[arg-type] + sample_size=sample_size, + standard_error=standard_error, + lower_ci_bound=lower, + upper_ci_bound=upper, + recommend_suppress=_flag( + row.get("Recommend Suppress", ""), "recommend suppress" + ), + not_relevant=_flag(row.get("Not Relevant", ""), "not relevant"), + source_updated_month=_updated_month(row["Date"], observed_today), + domain_source_code=domain_source, + ) + ) + if not observations: + raise ValueError("rating file has no rows") + return observations + + +def _parser() -> argparse.ArgumentParser: + """Build the explicit, provenance-bearing importer command contract.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--target-dsn", required=True) + parser.add_argument("--release-code", required=True) + parser.add_argument("--release-version", required=True) + parser.add_argument("--source-table-code", required=True) + parser.add_argument("--source-table-name", required=True) + parser.add_argument("--source-url", required=True) + parser.add_argument("--source-sha256", required=True) + parser.add_argument("--source-row-count", type=int, required=True) + parser.add_argument("--publisher", default="National Center for O*NET Development") + parser.add_argument( + "--license-url", default="https://creativecommons.org/licenses/by/4.0/" + ) + parser.add_argument("--scales-file", type=Path, required=True) + parser.add_argument("--scales-url", required=True) + parser.add_argument("--scales-sha256", required=True) + parser.add_argument("--scales-row-count", type=int, required=True) + parser.add_argument("--ratings-file", type=Path, required=True) + return parser + + +def _validate_args(args: argparse.Namespace) -> None: + """Reject ambiguous provenance and unsafe source identities before I/O.""" + if ( + not _SOURCE_CODE.fullmatch(args.source_table_code) + or args.source_table_code == _SCALES_SOURCE_CODE + ): + raise ValueError("source table code must be non-reserved lower snake case") + for field in ("release_code", "release_version", "source_table_name", "publisher"): + if not str(getattr(args, field)).strip(): + raise ValueError(f"{field} must not be blank") + for field in ("source_sha256", "scales_sha256"): + if not _SHA256.fullmatch(str(getattr(args, field))): + raise ValueError(f"{field} must be one SHA-256 digest") + for field in ("source_row_count", "scales_row_count"): + if getattr(args, field) <= 0: + raise ValueError(f"{field} must be positive") + for field in ("source_url", "scales_url", "license_url"): + parsed = urlsplit(str(getattr(args, field))) + if ( + parsed.scheme != "https" + or not parsed.hostname + or parsed.username is not None + or parsed.password is not None + ): + raise ValueError(f"{field} must be an HTTPS URL without userinfo") + for field in ("ratings_file", "scales_file"): + if not getattr(args, field).is_file(): + raise ValueError(f"{field} must be a regular file") + + +async def _reject_reference_conflicts( + conn: asyncpg.Connection, + args: argparse.Namespace, + scales: dict[str, ScaleDefinition], + observations: list[RatingObservation], +) -> None: + """Reject a reused source identity whose immutable source labels differ.""" + release = await conn.fetchrow( + """select release_version, source_publisher_name, source_license_url + from occupational_data_release where data_release_code = $1""", + args.release_code, + ) + if release is not None and tuple(release) != ( + args.release_version, + args.publisher, + args.license_url, + ): + raise ValueError("conflicting release identity") + sources = ( + ( + args.source_table_code, + args.source_table_name, + args.source_url, + args.source_sha256.lower(), + args.source_row_count, + ), + ( + _SCALES_SOURCE_CODE, + "Scales Reference", + args.scales_url, + args.scales_sha256.lower(), + args.scales_row_count, + ), + ) + for source_code, source_name, source_url, source_digest, source_rows in sources: + source = await conn.fetchrow( + """select source_table_name, source_artifact_url, + source_artifact_sha256, source_row_count + from occupational_source_table + where data_release_code = $1 and source_table_code = $2""", + args.release_code, + source_code, + ) + if source is not None and tuple(source) != ( + source_name, + source_url, + source_digest, + source_rows, + ): + raise ValueError(f"conflicting source-table identity: {source_code}") + expected = { + "scale": { + item.scale_id: ( + _SCALES_SOURCE_CODE, + item.scale_name, + item.minimum_value, + item.maximum_value, + ) + for item in scales.values() + }, + "occupation": { + item.onetsoc_code: (item.occupation_title,) for item in observations + }, + "element": {item.element_id: (item.element_name,) for item in observations}, + } + queries = { + "scale": "select scale_id, source_table_code, scale_name, minimum_value, maximum_value from occupational_scale_definition where data_release_code = $1", + "occupation": "select onetsoc_code, occupation_title from occupational_classification_entry where data_release_code = $1", + "element": "select element_id, element_name from occupational_element_definition where data_release_code = $1", + } + for kind, query in queries.items(): + for row in await conn.fetch(query, args.release_code): + source_id, *values = tuple(row) + if ( + source_id in expected[kind] + and tuple(values) != expected[kind][source_id] + ): + raise ValueError(f"conflicting {kind} identity: {source_id}") + + +async def import_ratings(args: argparse.Namespace) -> dict[str, object]: + """Validate one pinned source artifact, then transactionally UPSERT its rows.""" + _validate_args(args) + digest = hashlib.sha256(args.ratings_file.read_bytes()).hexdigest() + if digest != args.source_sha256.lower(): + raise ValueError("rating artifact SHA-256 mismatch") + scales_digest = hashlib.sha256(args.scales_file.read_bytes()).hexdigest() + if scales_digest != args.scales_sha256.lower(): + raise ValueError("scales artifact SHA-256 mismatch") + scales = read_scale_file(args.scales_file) + if len(scales) != args.scales_row_count: + raise ValueError("scales artifact row-count mismatch") + observations = read_rating_file(args.ratings_file, scales) + if len(observations) != args.source_row_count: + raise ValueError("rating artifact row-count mismatch") + conn = await asyncpg.connect(args.target_dsn) + release_partition = f"occupational_rating_release_{hashlib.sha256(args.release_code.encode()).hexdigest()[:16]}" + source_partition = f"occupational_rating_source_{hashlib.sha256(f'{args.release_code}\0{args.source_table_code}'.encode()).hexdigest()[:16]}" + try: + async with conn.transaction(): + await conn.execute( + "select pg_advisory_xact_lock(hashtextextended($1, 0))", + args.release_code, + ) + await _reject_reference_conflicts(conn, args, scales, observations) + release_literal = await conn.fetchval( + "select quote_literal($1)", args.release_code + ) + source_literal = await conn.fetchval( + "select quote_literal($1)", args.source_table_code + ) + await conn.execute( + """insert into occupational_data_release + (data_release_code, release_version, source_publisher_name, source_license_url) + values ($1, $2, $3, $4) + on conflict (data_release_code) do nothing""", + args.release_code, + args.release_version, + args.publisher, + args.license_url, + ) + await conn.execute( + f"create table if not exists {release_partition} partition of occupational_rating_observation for values in ({release_literal}) partition by list (source_table_code)" + ) + await conn.execute( + f"create table if not exists {source_partition} partition of {release_partition} for values in ({source_literal})" + ) + await conn.execute( + """insert into occupational_source_table + (data_release_code, source_table_code, source_table_name, + source_artifact_url, source_artifact_sha256, source_row_count) + values ($1, $2, $3, $4, $5, $6) + on conflict (data_release_code, source_table_code) do nothing""", + args.release_code, + args.source_table_code, + args.source_table_name, + args.source_url, + digest, + len(observations), + ) + await conn.execute( + """insert into occupational_source_table + (data_release_code, source_table_code, source_table_name, + source_artifact_url, source_artifact_sha256, source_row_count) + values ($1, $2, 'Scales Reference', $3, $4, $5) + on conflict (data_release_code, source_table_code) do nothing""", + args.release_code, + _SCALES_SOURCE_CODE, + args.scales_url, + scales_digest, + len(scales), + ) + await conn.executemany( + """insert into occupational_scale_definition + (data_release_code, source_table_code, scale_id, scale_name, + minimum_value, maximum_value) + values ($1, $2, $3, $4, $5, $6) + on conflict (data_release_code, scale_id) do nothing""", + [ + ( + args.release_code, + _SCALES_SOURCE_CODE, + item.scale_id, + item.scale_name, + item.minimum_value, + item.maximum_value, + ) + for item in scales.values() + ], + ) + await conn.executemany( + """insert into occupational_classification_entry + (data_release_code, onetsoc_code, occupation_title) + values ($1, $2, $3) on conflict (data_release_code, onetsoc_code) do nothing""", + [ + (args.release_code, code, title) + for code, title in { + item.onetsoc_code: item.occupation_title + for item in observations + }.items() + ], + ) + await conn.executemany( + """insert into occupational_element_definition + (data_release_code, element_id, element_name) + values ($1, $2, $3) on conflict (data_release_code, element_id) do nothing""", + [ + (args.release_code, code, name) + for code, name in { + item.element_id: item.element_name for item in observations + }.items() + ], + ) + await conn.executemany( + """insert into occupational_rating_observation + (data_release_code, source_table_code, onetsoc_code, element_id, + scale_id, category_value, data_value, sample_size, standard_error, + lower_ci_bound, upper_ci_bound, recommend_suppress, not_relevant, + source_updated_month, domain_source_code) + values ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15) + on conflict on constraint occupational_rating_identity_key do nothing""", + [ + ( + args.release_code, + args.source_table_code, + item.onetsoc_code, + item.element_id, + item.scale_id, + item.category_value, + item.data_value, + item.sample_size, + item.standard_error, + item.lower_ci_bound, + item.upper_ci_bound, + item.recommend_suppress, + item.not_relevant, + item.source_updated_month, + item.domain_source_code, + ) + for item in observations + ], + ) + finally: + await conn.close() + return { + "release_code": args.release_code, + "source_table_code": args.source_table_code, + "imported_rows": len(observations), + "source_sha256": digest, + "scales_sha256": scales_digest, + } + + +def main() -> None: + """Run the command-line importer and print aggregate, non-identifying evidence.""" + print( + json.dumps(asyncio.run(import_ratings(_parser().parse_args())), sort_keys=True) + ) + + +if __name__ == "__main__": + main() diff --git a/tests/test_import_onet_ratings.py b/tests/test_import_onet_ratings.py new file mode 100644 index 000000000..74fdc509e --- /dev/null +++ b/tests/test_import_onet_ratings.py @@ -0,0 +1,233 @@ +"""Contracts for the official O*NET occupation-rating CSV importer.""" + +import asyncio +import hashlib +from datetime import date +from decimal import Decimal +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from scripts.import_onet_ratings import ( + import_ratings, + read_rating_file, + read_scale_file, +) + + +def _write(path: Path, text: str) -> Path: + path.write_text(text, encoding="utf-8", newline="") + return path + + +def test_official_csv_preserves_decimal_missingness_and_uncertainty( + tmp_path: Path, +) -> None: + scales = read_scale_file( + _write( + tmp_path / "scales.csv", + "Scale ID,Scale Name,Minimum,Maximum\nIM,Importance,1,5\n", + ) + ) + ratings = read_rating_file( + _write( + tmp_path / "abilities.csv", + "O*NET-SOC Code,Title,Element ID,Element Name,Scale ID,Scale Name,Data Value,N,Standard Error,Lower CI Bound,Upper CI Bound,Recommend Suppress,Not Relevant,Date,Domain Source\n" + "15-1252.00,Synthetic occupation,1.A.1.a.1,Oral Comprehension,IM,Importance,4.10,120,0.08,3.94,4.26,N,,08/2026,Analyst\n", + ), + scales, + today=date(2026, 8, 27), + ) + + assert len(ratings) == 1 + assert ratings[0].data_value == Decimal("4.10") + assert ratings[0].data_value.as_tuple().exponent == -2 + assert ratings[0].standard_error == Decimal("0.08") + assert ratings[0].lower_ci_bound == Decimal("3.94") + assert ratings[0].upper_ci_bound == Decimal("4.26") + assert ratings[0].not_relevant is None + assert ratings[0].source_updated_month == "08/2026" + + +@pytest.mark.parametrize( + ("field", "value", "message"), + [ + ("Data Value", "5.01", "outside scale"), + ("Standard Error", "-0.01", "standard error"), + ("Date", "09/2026", "future"), + ("Date", "8/2026", "invalid source update date"), + ("Recommend Suppress", "maybe", "flag"), + ], +) +def test_invalid_source_measurement_fails_before_persistence( + tmp_path: Path, + field: str, + value: str, + message: str, +) -> None: + scales = read_scale_file( + _write( + tmp_path / "scales.csv", + "Scale ID,Scale Name,Minimum,Maximum\nIM,Importance,1,5\n", + ) + ) + row = { + "O*NET-SOC Code": "15-1252.00", + "Title": "Synthetic occupation", + "Element ID": "1.A.1.a.1", + "Element Name": "Oral Comprehension", + "Scale ID": "IM", + "Scale Name": "Importance", + "Data Value": "4.10", + "N": "120", + "Standard Error": "0.08", + "Lower CI Bound": "3.94", + "Upper CI Bound": "4.26", + "Recommend Suppress": "N", + "Not Relevant": "", + "Date": "08/2026", + "Domain Source": "Analyst", + } + row[field] = value + path = tmp_path / "invalid.csv" + import csv + + with path.open("w", encoding="utf-8", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=row) + writer.writeheader() + writer.writerow(row) + + with pytest.raises(ValueError, match=message): + read_rating_file(path, scales, today=date(2026, 8, 27)) + + +def test_conflicting_reference_name_fails_closed(tmp_path: Path) -> None: + scales = read_scale_file( + _write( + tmp_path / "scales.csv", + "Scale ID,Scale Name,Minimum,Maximum\nIM,Importance,1,5\n", + ) + ) + path = _write( + tmp_path / "conflict.csv", + "O*NET-SOC Code,Title,Element ID,Element Name,Scale ID,Scale Name,Data Value,N,Standard Error,Lower CI Bound,Upper CI Bound,Recommend Suppress,Not Relevant,Date,Domain Source\n" + "15-1252.00,Synthetic occupation,1.A.1.a.1,Oral Comprehension,IM,Importance,4.10,,,,,N,,08/2026,Analyst\n" + "15-1252.00,Conflicting title,1.A.1.a.1,Oral Comprehension,IM,Importance,4.20,,,,,N,,08/2026,Analyst\n", + ) + + with pytest.raises(ValueError, match="conflicting occupation title"): + read_rating_file(path, scales, today=date(2026, 8, 27)) + + +def test_short_csv_row_fails_with_import_error(tmp_path: Path) -> None: + scales = read_scale_file( + _write( + tmp_path / "scales.csv", + "Scale ID,Scale Name,Minimum,Maximum\nIM,Importance,1,5\n", + ) + ) + path = _write( + tmp_path / "short.csv", + "O*NET-SOC Code,Title,Element ID,Element Name,Scale ID,Scale Name,Data Value,Date,Domain Source\n" + "15-1252.00,Synthetic occupation,1.A.1.a.1\n", + ) + + with pytest.raises(ValueError, match="malformed CSV row: 2"): + read_rating_file(path, scales, today=date(2026, 8, 27)) + + +def test_category_table_may_omit_not_relevant_without_inventing_false( + tmp_path: Path, +) -> None: + scales = read_scale_file( + _write( + tmp_path / "scales.csv", + "Scale ID,Scale Name,Minimum,Maximum\nPT,Percent,0,100\n", + ) + ) + rows = read_rating_file( + _write( + tmp_path / "education.csv", + "O*NET-SOC Code,Title,Element ID,Element Name,Scale ID,Scale Name,Category,Data Value,N,Standard Error,Lower CI Bound,Upper CI Bound,Recommend Suppress,Date,Domain Source\n" + "15-1252.00,Synthetic occupation,2.D.1,Education,PT,Percent,6,42.50,100,1.2,40.1,44.9,N,08/2026,Incumbent\n", + ), + scales, + today=date(2026, 8, 27), + ) + + assert rows[0].category_value == 6 + assert rows[0].not_relevant is None + + +def test_machine_generated_profile_keeps_unpublished_uncertainty_missing( + tmp_path: Path, +) -> None: + scales = read_scale_file( + _write( + tmp_path / "scales.csv", + "Scale ID,Scale Name,Minimum,Maximum\nDR,Distinctiveness Rank,0,7\n", + ) + ) + rows = read_rating_file( + _write( + tmp_path / "work_styles.csv", + "O*NET-SOC Code,Title,Element ID,Element Name,Scale ID,Scale Name,Data Value,Date,Domain Source\n" + "15-1252.00,Synthetic occupation,1.D.1.a,Innovation,DR,Distinctiveness Rank,7.00,08/2026,AI/Expert\n", + ), + scales, + today=date(2026, 8, 27), + ) + + assert rows[0].sample_size is None + assert rows[0].standard_error is None + assert rows[0].recommend_suppress is None + + +def test_scales_digest_fails_before_database_connection( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + scales = _write( + tmp_path / "scales.csv", + "Scale ID,Scale Name,Minimum,Maximum\nIM,Importance,1,5\n", + ) + ratings = _write( + tmp_path / "abilities.csv", + "O*NET-SOC Code,Title,Element ID,Element Name,Scale ID,Scale Name,Data Value,N,Standard Error,Lower CI Bound,Upper CI Bound,Recommend Suppress,Not Relevant,Date,Domain Source\n" + "15-1252.00,Synthetic occupation,1.A.1.a.1,Oral Comprehension,IM,Importance,4.10,120,0.08,3.94,4.26,N,,08/2026,Analyst\n", + ) + connected = False + + async def fake_connect(_dsn: str) -> None: + nonlocal connected + connected = True + + monkeypatch.setattr("scripts.import_onet_ratings.asyncpg.connect", fake_connect) + args = SimpleNamespace( + target_dsn="postgresql://unused", + release_code="onet-31.0-synthetic", + release_version="31.0-synthetic", + source_table_code="abilities", + source_table_name="Abilities", + source_url="https://example.test/abilities.csv", + source_sha256=hashlib.sha256(ratings.read_bytes()).hexdigest(), + source_row_count=1, + publisher="Synthetic publisher", + license_url="https://example.test/license", + scales_file=scales, + scales_url="https://example.test/scales.csv", + scales_sha256="0" * 64, + scales_row_count=1, + ratings_file=ratings, + ) + + args.source_url = "https://:secret@example.test/abilities.csv" + with pytest.raises(ValueError, match="without userinfo"): + asyncio.run(import_ratings(args)) + assert connected is False + + args.source_url = "https://example.test/abilities.csv" + with pytest.raises(ValueError, match="scales artifact SHA-256 mismatch"): + asyncio.run(import_ratings(args)) + assert connected is False diff --git a/tests/test_occupation_rating_ingestion.py b/tests/test_occupation_rating_ingestion.py new file mode 100644 index 000000000..3f2244d7f --- /dev/null +++ b/tests/test_occupation_rating_ingestion.py @@ -0,0 +1,162 @@ +"""Tests for the provenance-bearing occupation-rating read projection.""" + +import asyncio +from decimal import Decimal + +from backend.app.main import read_occupation_ratings +from backend.app.occupation_rating_ingestion import fetch_occupation_ratings + + +class FakeConnection: + """Minimal ordered asyncpg stand-in for one projection query.""" + + def __init__(self, source, rows=()): + self.source = source + self.rows = list(rows) + self.fetch_called = False + + async def fetchrow(self, _query: str, *_args: object): + """Return configured source metadata.""" + return self.source + + async def fetch(self, _query: str, *_args: object): + """Return configured observation rows.""" + self.fetch_called = True + return self.rows + + +class FakeAcquire: + """Async pool-acquire context for route wiring.""" + + def __init__(self, conn: FakeConnection): + self.conn = conn + + async def __aenter__(self) -> FakeConnection: + """Return the configured connection.""" + return self.conn + + async def __aexit__(self, *_args: object) -> None: + """Release without external state.""" + + +class FakePool: + """Minimal pool exposing one acquisition context.""" + + def __init__(self, conn: FakeConnection): + self.conn = conn + + def acquire(self) -> FakeAcquire: + """Return one deterministic acquisition context.""" + return FakeAcquire(self.conn) + + +def test_unimported_source_is_not_an_empty_observed_profile() -> None: + conn = FakeConnection(None) + + result = asyncio.run( + fetch_occupation_ratings( + conn, + data_release_code="onet-31.0", + source_table_code="abilities", + onetsoc_code="15-1252.00", + limit=100, + offset=0, + ) + ) + + assert result["source_available"] is False + assert result["items"] == [] + assert conn.fetch_called is False + + +def test_rating_projection_preserves_exact_decimal_and_warning_flags() -> None: + source = { + "source_table_name": "Abilities", + "source_artifact_url": "https://example.test/abilities.csv", + "source_artifact_sha256": "a" * 64, + "source_row_count": 2, + "scale_artifact_url": "https://example.test/scales.csv", + "scale_artifact_sha256": "b" * 64, + "scale_source_row_count": 33, + } + row = { + "element_id": "1.A.1.a.1", + "element_name": "Oral Comprehension", + "scale_id": "IM", + "scale_name": "Importance", + "minimum_value": Decimal("1.00"), + "maximum_value": Decimal("5.00"), + "category_value": None, + "data_value": Decimal("4.10"), + "sample_size": 8, + "standard_error": Decimal("0.1830"), + "lower_ci_bound": Decimal("3.7414"), + "upper_ci_bound": Decimal("4.4586"), + "recommend_suppress": True, + "not_relevant": None, + "source_updated_month": "08/2026", + "domain_source_code": "Analyst", + } + conn = FakeConnection(source, (row, row)) + + result = asyncio.run( + fetch_occupation_ratings( + conn, + data_release_code="onet-31.0", + source_table_code="abilities", + onetsoc_code="15-1252.00", + limit=1, + offset=0, + ) + ) + + item = result["items"][0] + assert item["data_value"] == "4.10" + assert item["standard_error"] == "0.1830" + assert item["recommend_suppress"] is True + assert item["not_relevant"] is None + assert result["source"]["scale_artifact_sha256"] == "b" * 64 + assert result["next_offset"] == 1 + + +def test_empty_profile_keeps_imported_scale_provenance() -> None: + source = { + "source_table_name": "Abilities", + "source_artifact_url": "https://example.test/abilities.csv", + "source_artifact_sha256": "a" * 64, + "source_row_count": 2, + "scale_artifact_url": "https://example.test/scales.csv", + "scale_artifact_sha256": "b" * 64, + "scale_source_row_count": 33, + } + + result = asyncio.run( + fetch_occupation_ratings( + FakeConnection(source), + data_release_code="onet-31.0", + source_table_code="abilities", + onetsoc_code="15-9999.99", + limit=100, + offset=0, + ) + ) + + assert result["source_available"] is True + assert result["items"] == [] + assert result["source"]["scale_artifact_sha256"] == "b" * 64 + + +def test_authenticated_route_delegates_to_bounded_projection() -> None: + result = asyncio.run( + read_occupation_ratings( + onetsoc_code="15-1252.00", + data_release_code="onet-31.0", + source_table_code="abilities", + limit=100, + offset=0, + _account=object(), + pool=FakePool(FakeConnection(None)), + ) + ) + + assert result["source_available"] is False diff --git a/tests/test_schema.py b/tests/test_schema.py index 4a689dd76..d12fa59de 100644 --- a/tests/test_schema.py +++ b/tests/test_schema.py @@ -15,10 +15,12 @@ from __future__ import annotations import asyncio +import hashlib import os import uuid from decimal import Decimal from pathlib import Path +from types import SimpleNamespace from urllib.parse import urlsplit, urlunsplit import asyncpg @@ -26,7 +28,9 @@ import psycopg2.errors import pytest +from backend.app.occupation_rating_ingestion import fetch_occupation_ratings from backend.app.post_chat_ingestion import gather_global_chat_sources +from scripts.import_onet_ratings import import_ratings _ADMIN_DSN = os.environ.get( "LINEAGEWEAVE_TEST_POSTGRES_ADMIN_DSN", "postgresql://localhost/postgres" @@ -478,6 +482,72 @@ def test_onet_rating_store_partitions_upserts_and_rejects_invalid_error(schema_d cur.execute("rollback to savepoint invalid_standard_error") +def test_onet_rating_importer_is_idempotent_against_postgresql( + schema_db, + tmp_path: Path, +) -> None: + """A pinned synthetic artifact imports twice as one exact observation.""" + scales = tmp_path / "scales.csv" + scales.write_text( + "Scale ID,Scale Name,Minimum,Maximum\nIM,Importance,1,5\n", + encoding="utf-8", + ) + ratings = tmp_path / "abilities.csv" + ratings.write_text( + "O*NET-SOC Code,Title,Element ID,Element Name,Scale ID,Scale Name,Data Value,N,Standard Error,Lower CI Bound,Upper CI Bound,Recommend Suppress,Not Relevant,Date,Domain Source\n" + "15-1252.00,Synthetic occupation,1.A.1.a.1,Oral Comprehension,IM,Importance,4.10,120,0.08,3.94,4.26,N,,08/2026,Analyst\n", + encoding="utf-8", + ) + args = SimpleNamespace( + target_dsn=urlunsplit( + urlsplit(_ADMIN_DSN)._replace(path=f"/{schema_db.info.dbname}") + ), + release_code="onet-31.0-synthetic", + release_version="31.0-synthetic", + source_table_code="abilities", + source_table_name="Abilities", + source_url="https://example.test/abilities.csv", + source_sha256=hashlib.sha256(ratings.read_bytes()).hexdigest(), + source_row_count=1, + publisher="Synthetic publisher", + license_url="https://example.test/license", + scales_file=scales, + scales_url="https://example.test/scales.csv", + scales_sha256=hashlib.sha256(scales.read_bytes()).hexdigest(), + scales_row_count=1, + ratings_file=ratings, + ) + + assert asyncio.run(import_ratings(args))["imported_rows"] == 1 + assert asyncio.run(import_ratings(args))["imported_rows"] == 1 + with schema_db.cursor() as cur: + cur.execute( + """select count(*), min(data_value), bool_or(not_relevant is null) + from occupational_rating_observation + where data_release_code = 'onet-31.0-synthetic'""" + ) + assert cur.fetchone() == (1, Decimal("4.10"), True) + + async def read_imported_profile() -> dict[str, object]: + conn = await asyncpg.connect(args.target_dsn) + try: + return await fetch_occupation_ratings( + conn, + data_release_code=args.release_code, + source_table_code=args.source_table_code, + onetsoc_code="15-1252.00", + limit=100, + offset=0, + ) + finally: + await conn.close() + + profile = asyncio.run(read_imported_profile()) + assert profile["source_available"] is True + assert profile["items"][0]["data_value"] == "4.10" + assert profile["source"]["scale_artifact_sha256"] == args.scales_sha256 + + def test_global_ask_evidence_search_indexes_exist_on_normalized_tables(schema_db) -> None: """The real PostgreSQL schema owns all nine evidence-search indexes.""" with schema_db.cursor() as cur: