From 7617e09e897ed3a5dc69960c5ba3a852c50efcfd Mon Sep 17 00:00:00 2001 From: Georgie Kennedy Date: Wed, 14 Oct 2026 13:33:14 +1100 Subject: [PATCH 1/3] refactor of testing suite --- .github/workflows/ci.yml | 48 ++- docs/usage/testing.md | 55 ++-- pyproject.toml | 3 + .../services/source_planning/assisted.py | 6 +- tests/conftest.py | 81 +++++ tests/integration/test_omop_emb_adapter.py | 10 +- tests/integration/test_omop_emb_contract.py | 147 +++++++++ tests/integration/test_omop_graph_adapter.py | 294 +++++++----------- .../integration/test_omop_postgres_runtime.py | 209 +++++++++++++ tests/support/embedding_contract_fixture.py | 55 ++++ tests/support/seed_postgres_fixture.py | 29 +- tests/support/wheel_smoke.py | 90 ++++++ tests/test_rest_api.py | 33 ++ tests/unit/test_plugins.py | 51 ++- .../test_required_integration_preflight.py | 80 +++++ tests/unit/test_required_test_gate.py | 16 + ...est_source_planning_assisted_classifier.py | 200 ++++++++++++ .../test_source_planning_document_formats.py | 200 ++++++++++++ tests/unit/test_source_planning_tools.py | 36 +++ tests/unit/test_tui_app.py | 16 +- 20 files changed, 1441 insertions(+), 218 deletions(-) create mode 100644 tests/integration/test_omop_emb_contract.py create mode 100644 tests/integration/test_omop_postgres_runtime.py create mode 100644 tests/support/embedding_contract_fixture.py create mode 100644 tests/support/wheel_smoke.py create mode 100644 tests/unit/test_required_integration_preflight.py create mode 100644 tests/unit/test_required_test_gate.py create mode 100644 tests/unit/test_source_planning_assisted_classifier.py create mode 100644 tests/unit/test_source_planning_document_formats.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5dac962..b5ddb5c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -35,7 +35,7 @@ jobs: --password postgres \ --database-name test_db \ --test-only false - uv run omop-config databases add cdm test_cdm_db --connection pg_test --cdm-schema public + uv run omop-config databases add cdm test_cdm_db --connection pg_test --cdm-schema gw_primary --vocab-schema gw_vocab uv run omop-config databases add generic test_vector_db --connection pg_test --schema-name embeddings uv run omop-config vector-stores add test_embeddings \ --backend-type pgvector \ @@ -44,3 +44,49 @@ jobs: --cdm-db test_cdm_db \ --vector-store-name test_embeddings uv run python -m tests.support.seed_postgres_fixture + test-evidence: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - uses: astral-sh/setup-uv@v6 + - name: Run hermetic suite with coverage and skip reasons + run: | + uv sync --frozen --all-extras --dev + export GROUNDWORKERS_REQUIRE_FORMAT_TESTS=1 + echo "GROUNDWORKERS_REQUIRE_FORMAT_TESTS=1" >> "$GITHUB_ENV" + COVERAGE_FILE="$RUNNER_TEMP/groundworkers.coverage" uv run pytest -q -ra \ + --cov=groundworkers --cov-branch \ + --cov-report=xml:"$RUNNER_TEMP/coverage.xml" \ + --junitxml="$RUNNER_TEMP/pytest.xml" + - uses: actions/upload-artifact@v4 + if: always() + with: + name: groundworkers-test-evidence + path: | + ${{ runner.temp }}/coverage.xml + ${{ runner.temp }}/pytest.xml + if-no-files-found: warn + wheel-smoke: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - uses: astral-sh/setup-uv@v6 + - name: Build and smoke-test base and optional wheel installs + run: | + uv build --wheel + wheel="$(find dist -maxdepth 1 -name '*.whl' -print -quit)" + wheel="$(pwd)/$wheel" + uv venv "$RUNNER_TEMP/gw-wheel-base" --python 3.12 + uv pip install --python "$RUNNER_TEMP/gw-wheel-base/bin/python" "$wheel" + mkdir -p "$RUNNER_TEMP/gw-wheel-run" + cd "$RUNNER_TEMP/gw-wheel-run" + "$RUNNER_TEMP/gw-wheel-base/bin/python" "$GITHUB_WORKSPACE/tests/support/wheel_smoke.py" base + uv venv "$RUNNER_TEMP/gw-wheel-extras" --python 3.12 + uv pip install --python "$RUNNER_TEMP/gw-wheel-extras/bin/python" "$wheel[all_source,tui]" httpx + "$RUNNER_TEMP/gw-wheel-extras/bin/python" "$GITHUB_WORKSPACE/tests/support/wheel_smoke.py" extras diff --git a/docs/usage/testing.md b/docs/usage/testing.md index 233d7e0..89dbf3c 100644 --- a/docs/usage/testing.md +++ b/docs/usage/testing.md @@ -1,34 +1,45 @@ # Testing +## Test lanes + +Groundworkers uses distinct lanes because an isolated unit run cannot prove the database, optional parser, plugin-entry-point, or built-wheel contracts. + +| Lane | What it proves | CI behavior | +|---|---|---| +| Unit and service | Deterministic policy, adapter boundaries, source planning, transport behavior, and controlled provider responses | Runs without a live database or model service | +| Optional dependency | Real XLSX, DOCX, and PDF parsing plus missing-extra guidance | Required format tests must pass in the all-extras CI lane; missing-extra behavior is simulated | +| PostgreSQL integration | Seeded graph retrieval, real pgvector storage and reader behavior, model selection, filters, and embedding-tier grounding | Required: every required_integration case must be collected and pass | +| Wheel smoke | Base and optional-extra installations, package entry-point discovery, CLI startup, and successful MCP/REST planning operations | Builds once and runs outside the checkout | +| Optional live provider | Real configured model-provider connectivity | Opt-in developer smoke only; never substitutes for seeded embedding tests | + ## Running tests -```bash -# all tests (unit + integration; integration tests are skipped without a live DB) -pytest +~~~bash +# Hermetic suite, including real optional-format parsers +uv run pytest -q -ra + +# Unit tests only +uv run pytest tests/unit/ -# unit tests only (no database required) -pytest tests/unit/ +# Required PostgreSQL cases; provide a seeded test stack first +GROUNDWORKERS_CONFIG_PATH=/path/to/stack.toml \ +GROUNDWORKERS_REQUIRE_INTEGRATION=1 \ +GROUNDWORKERS_TEST_DATABASE_URL=postgresql+psycopg://... \ +uv run pytest -m required_integration -q -ra -# single file -pytest tests/unit/test_concept_tools.py -``` +# Coverage for local review +COVERAGE_FILE=/tmp/groundworkers.coverage uv run pytest -q -ra \ + --cov=groundworkers --cov-branch --cov-report=term-missing +~~~ -## Test configuration +The required integration gate derives its cases from the `required_integration` marker, so every marked case is checked for collection and a passing result. Missing configuration, unavailable adapters, and absent seeded rows fail in that lane. The all-extras CI job applies the same execution check to `required_format` cases, so a missing parser cannot silently turn the format contract into a skip. Without the corresponding `GROUNDWORKERS_REQUIRE_*` flag, PostgreSQL and optional-format fixtures remain opt-in and skip with a reason. -Integration tests that query the OMOP database require a `.env` file at the repo root (or an environment-variable equivalent): +## Configuration and optional provider smoke -```dotenv -OA_CONFIG_PATH=/path/to/stack.toml -``` +Set GROUNDWORKERS_CONFIG_PATH (or GROUNDWORKERS_CONFIG) to the explicit Groundworkers stack file for integration tests. OA_CONFIG_PATH selects the oa-configurator default stack, but does not by itself opt the graph integration tests into a required run. PostgreSQL CI sets both paths and seeds a compact fixture before pytest starts. -Unit tests mock the adapter layer and run without a live database. +The external-provider smoke tests are separate from seeded deterministic embedding tests. To run them locally, set GROUNDWORKERS_RUN_EMBEDDING_INTEGRATION=1, GROUNDWORKERS_CONFIG_PATH, and GROUNDWORKERS_TEST_CONCEPT_ID; they require a configured provider and a concept with stored vectors. -## What is tested +## Coverage and skip evidence -| Area | Scope | -|---|---| -| Concept tools | Deterministic concept lookup, hierarchy, relationship, path, neighborhood, classified-edge, and standard-mapping tools: correct responses, `NOT_FOUND` errors, `INVALID_INPUT` guards, and depth clamping | -| Resolver tools | `concept_ground`: empty query rejection, result ordering by score, match_kind on exact hits | -| Embedding tools | `embedding_index_status`, `embedding_neighbours`, `embedding_search`, `embedding_search_batch`, `embedding_encode`: correct responses, clamped limits, aligned batch rows, error propagation | -| Server registry | `create_server` with and without each adapter; system tools always present; adapter-conditional tool registration | -| Config validation | oa-configurator stack validation, Groundworkers package fields, and extra field rejection | +CI publishes coverage.xml and pytest.xml from the hermetic all-extras run. The JUnit report carries named skip reasons. PostgreSQL CI separately enforces the required test list, so optional local skips cannot satisfy that gate. Coverage percentages are review evidence, not a global pass threshold. Prioritize unexecuted behavior at database, provider, transport, and packaging boundaries. diff --git a/pyproject.toml b/pyproject.toml index 254edaf..8675d0b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -112,5 +112,8 @@ include = ["src"] testpaths = ["tests"] markers = [ "integration: requires a live backend or database", + "required_integration: deterministic integration contract required by CI", + "required_format: optional document-format contract required in all-extras CI", + "optional_integration: environment-dependent integration smoke test", "requires_client: requires an embedding API client", ] diff --git a/src/groundworkers/services/source_planning/assisted.py b/src/groundworkers/services/source_planning/assisted.py index 740020a..d3aa79e 100644 --- a/src/groundworkers/services/source_planning/assisted.py +++ b/src/groundworkers/services/source_planning/assisted.py @@ -9,7 +9,7 @@ from dataclasses import replace -from pydantic import BaseModel, Field, ValidationError +from pydantic import BaseModel, ConfigDict, Field, ValidationError from groundworkers.adapters.llm import LLMAdapter from groundworkers.base.errors import GroundworkersError @@ -31,6 +31,8 @@ class AssistedColumnDecision(BaseModel): + model_config = ConfigDict(extra="ignore") + header: str role: str | None = None confidence: float = Field(ge=0.0, le=1.0) @@ -40,6 +42,8 @@ class AssistedColumnDecision(BaseModel): class AssistedClassificationResult(BaseModel): + model_config = ConfigDict(extra="ignore") + decisions: list[AssistedColumnDecision] = Field(default_factory=list) diff --git a/tests/conftest.py b/tests/conftest.py index b5a869b..3b8f40d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,5 +1,6 @@ from __future__ import annotations +import os import sys from pathlib import Path @@ -8,6 +9,13 @@ ROOT = Path(__file__).resolve().parents[1] SRC = ROOT / "src" +_REQUIRED_TEST_MARKERS = { + "GROUNDWORKERS_REQUIRE_INTEGRATION": "required_integration", + "GROUNDWORKERS_REQUIRE_FORMAT_TESTS": "required_format", +} +_required_test_ids: dict[str, set[str]] = {} +_required_test_outcomes: dict[str, dict[str, str]] = {} + if str(SRC) not in sys.path: sys.path.insert(0, str(SRC)) @@ -31,3 +39,76 @@ def _isolated_env_file(tmp_path_factory, monkeypatch): # it leaves every later test looking like it was launched with a broken # OA_CONFIG_PATH -- which makes the setup page open the location wizard. monkeypatch.setattr(_env, "_rejected", None) + + +def _missing_required_tests(expected: set[str], selected: set[str]) -> list[str]: + return sorted(expected - selected) + + +def _required_test_failures( + expected: set[str], outcomes: dict[str, str] +) -> list[str]: + return sorted(nodeid for nodeid in expected if outcomes.get(nodeid) != "passed") + + +@pytest.hookimpl(hookwrapper=True, tryfirst=True) +def pytest_collection_modifyitems(session: pytest.Session, items: list[pytest.Item]): + enabled = { + marker + for variable, marker in _REQUIRED_TEST_MARKERS.items() + if os.getenv(variable) == "1" + } + expected = { + marker: {item.nodeid for item in items if item.get_closest_marker(marker)} + for marker in enabled + } + yield + selected_items = {item.nodeid for item in session.items} + _required_test_ids.clear() + _required_test_ids.update(expected) + for marker in enabled: + _required_test_outcomes[marker] = {} + session.config._required_test_ids = expected + session.config._required_test_collection_errors = { + marker: _missing_required_tests(ids, selected_items) + for marker, ids in expected.items() + } + session.config._required_test_outcomes = { + marker: _required_test_outcomes.setdefault(marker, {}) for marker in enabled + } + + +def pytest_runtest_logreport(report: pytest.TestReport) -> None: + if report.when not in {"setup", "call", "teardown"}: + return + for marker, outcomes in _required_test_outcomes.items(): + if report.when == "call" or report.outcome != "passed": + if report.nodeid in _required_test_ids.get(marker, set()): + outcomes[report.nodeid] = report.outcome + + +def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None: + required = getattr(session.config, "_required_test_ids", {}) + collection_errors = getattr(session.config, "_required_test_collection_errors", {}) + outcomes = getattr(session.config, "_required_test_outcomes", {}) + problems = { + marker: sorted( + set(collection_errors.get(marker, [])) + | set(_required_test_failures(ids, outcomes.get(marker, {}))) + ) + for marker, ids in required.items() + } + for marker, ids in required.items(): + if not ids: + problems[marker].append(f"no tests collected with {marker} marker") + failures = {marker: entries for marker, entries in problems.items() if entries} + if not failures: + return + reporter = session.config.pluginmanager.get_plugin("terminalreporter") + if reporter is not None: + for marker, entries in failures.items(): + reporter.write_sep("!", f"required {marker} execution gate failed") + for nodeid in entries: + outcome = outcomes.get(marker, {}).get(nodeid, "not executed or deselected") + reporter.write_line(f"{nodeid}: {outcome}") + session.exitstatus = pytest.ExitCode.TESTS_FAILED diff --git a/tests/integration/test_omop_emb_adapter.py b/tests/integration/test_omop_emb_adapter.py index 3114ca3..9a81330 100644 --- a/tests/integration/test_omop_emb_adapter.py +++ b/tests/integration/test_omop_emb_adapter.py @@ -42,7 +42,7 @@ def _load_integration_adapter(): return adapter -@pytest.mark.integration +@pytest.mark.optional_integration def test_index_status_returns_at_least_one_model(): adapter = _load_integration_adapter() @@ -59,7 +59,7 @@ def _get_test_concept_id() -> int: return int(value) -@pytest.mark.integration +@pytest.mark.optional_integration def test_get_neighbours_returns_sorted_results(): adapter = _load_integration_adapter() status = adapter.index_status() @@ -71,11 +71,12 @@ def test_get_neighbours_returns_sorted_results(): model_name=status["models"][0]["model_name"], ) + assert result["results"], "configured embedding smoke test returned no neighbours" similarities = [item["similarity"] for item in result["results"]] assert similarities == sorted(similarities, reverse=True) -@pytest.mark.integration +@pytest.mark.optional_integration def test_get_neighbours_excludes_query_concept(): adapter = _load_integration_adapter() status = adapter.index_status() @@ -87,10 +88,11 @@ def test_get_neighbours_excludes_query_concept(): model_name=status["models"][0]["model_name"], ) + assert result["results"], "configured embedding smoke test returned no neighbours" assert all(item["concept_id"] != concept_id for item in result["results"]) -@pytest.mark.integration +@pytest.mark.optional_integration def test_get_neighbours_unknown_concept_raises_not_found(): adapter = _load_integration_adapter() status = adapter.index_status() diff --git a/tests/integration/test_omop_emb_contract.py b/tests/integration/test_omop_emb_contract.py new file mode 100644 index 0000000..d6b2adb --- /dev/null +++ b/tests/integration/test_omop_emb_contract.py @@ -0,0 +1,147 @@ +from __future__ import annotations + +import os + +import pytest +from oa_configurator import ResolvedModel, ResolvedProvider +from omop_emb.backends import open_vector_store_reader +from omop_emb.config import MetricType +from omop_emb.interface import EmbeddingReaderInterface + +from groundworkers.adapters.omop_emb import OmopEmbAdapter +from groundworkers.adapters.omop_graph import OmopGraphAdapter +from groundworkers.base.errors import GroundworkersError +from groundworkers.bootstrap import build_app_config +from groundworkers.services.graph import GraphService +from groundworkers.services.grounding import ConceptGroundingService +from tests.support.embedding_contract_fixture import ( + MODEL_NAME, + seed_embedding_contract, +) + + +class FixedEncoder: + provider = "ollama" + model = MODEL_NAME + + def embed_texts(self, texts, *, role, batch_size=None): + return [[1.0, 0.0, 0.0] if text != "orthogonal query" else [0.0, 1.0, 0.0] for text in texts] + + async def async_embed_texts(self, texts, *, role, batch_size=None): + return self.embed_texts(texts, role=role, batch_size=batch_size) + + +@pytest.fixture(scope="module") +def embedding_contract(): + if os.getenv("GROUNDWORKERS_REQUIRE_INTEGRATION") != "1": + pytest.skip("deterministic pgvector embedding contract runs in required PostgreSQL CI") + config_path = os.getenv("GROUNDWORKERS_CONFIG_PATH") or os.getenv("GROUNDWORKERS_CONFIG") + if not config_path: + raise RuntimeError("required embedding contract needs GROUNDWORKERS_CONFIG_PATH") + config = build_app_config(config_path=config_path) + if config.vector_store is None: + raise AssertionError("required embedding contract has no configured vector store") + seed_embedding_contract(config) + adapter = OmopEmbAdapter( + backend_factory=lambda: open_vector_store_reader(config.vector_store), + backend_type=config.vector_store.backend_type, + cdm_session_factory=config.cdm_session_factory, + ) + adapter._model_backend_factory = lambda: FixedEncoder() + try: + yield config, adapter + finally: + adapter.close() + + +@pytest.mark.required_integration +def test_real_reader_returns_seeded_neighbours_and_keeps_batch_alignment(embedding_contract): + _config, adapter = embedding_contract + neighbours = adapter.get_neighbours(201826, limit=5, model_name=MODEL_NAME) + assert [item["concept_id"] for item in neighbours["results"]][:2] == [201827, 201830] + assert 201826 not in {item["concept_id"] for item in neighbours["results"]} + + record = adapter._resolve_model_record(MODEL_NAME) + reader = EmbeddingReaderInterface( + model=record.model_name, + backend=adapter._get_backend(), + metric_type=record.metric_type or MetricType.COSINE, + provider_type=record.provider_type, + ) + batch = reader.get_nearest_concepts_from_query_texts( + ["nearest to diabetes", "orthogonal query"], + model_backend=FixedEncoder(), + k=1, + faiss_index_config=record.index_config, + ) + assert [[match.concept_id for match in row] for row in batch] == [[201826], [201828]] + + +@pytest.mark.required_integration +def test_real_model_registry_handles_missing_and_ambiguous_selection(embedding_contract): + _, adapter = embedding_contract + assert adapter.resolve_model_name(MODEL_NAME) == MODEL_NAME + with pytest.raises(GroundworkersError) as missing: + adapter.resolve_model_name("not-registered:v1") + assert missing.value.code == "NOT_FOUND" + with pytest.raises(GroundworkersError) as ambiguous: + adapter.resolve_model_name(None) + assert ambiguous.value.code == "BACKEND_UNAVAIL" + + +@pytest.mark.required_integration +def test_real_text_search_applies_domain_vocabulary_standard_and_active_filters(embedding_contract): + _, adapter = embedding_contract + result = adapter.search( + "a deterministic unrelated phrase", + limit=10, + domain="Condition", + vocabulary="SNOMED", + standard_only=True, + active_only=True, + model_name=MODEL_NAME, + ) + assert {item["concept_id"] for item in result["results"]} == {201826, 201827, 201828} + + +@pytest.mark.required_integration +def test_graph_grounding_executes_real_embedding_tier_with_seeded_reader(embedding_contract): + config, _adapter = embedding_contract + assert config.vector_store is not None + resolved_model = ResolvedModel( + name="groundworkers_fixture", + provider=ResolvedProvider( + name="fixture_provider", provider="ollama", base_url=None, api_key=None + ), + model=MODEL_NAME, + embedding_dim=3, + document_prefix=None, + query_prefix=None, + embeddings=True, + tool_use=False, + structured_output=False, + extended_thinking=False, + configuration={}, + ) + graph_adapter = OmopGraphAdapter( + engine=config.cdm_engine, + cdm_session_factory=config.cdm_session_factory, + embedding_backend_factory=lambda: open_vector_store_reader(config.vector_store), + resolved_embedding_model=resolved_model, + model_backend_factory=FixedEncoder, + ) + assert graph_adapter.is_available() is True + graph = GraphService(graph_adapter) + assert graph.embedding_resolver_active is True + + result = ConceptGroundingService(graph).ground( + "a deterministic unrelated phrase", + limit=5, + domain="Condition", + vocabulary_id=None, + ) + + assert result["grounding_explanation"]["matched_tier"] == "EMBEDDING_NEAREST" + assert result["grounding_explanation"]["used_embedding"] is True + assert [hit["concept_id"] for hit in result["results"]] + assert result["results"][0]["concept_id"] == 201826 diff --git a/tests/integration/test_omop_graph_adapter.py b/tests/integration/test_omop_graph_adapter.py index 4a42df5..750fabf 100644 --- a/tests/integration/test_omop_graph_adapter.py +++ b/tests/integration/test_omop_graph_adapter.py @@ -1,11 +1,6 @@ import os import pytest -from oa_configurator import Role, physical_schema_of -from omop_alchemy.cdm.model.vocabulary import Concept, Concept_Relationship -from sqlalchemy import exists, func, select -from sqlalchemy import inspect as sa_inspect -from sqlalchemy.orm import aliased from groundworkers.app import build_adapters from groundworkers.bootstrap import build_app_config @@ -13,11 +8,22 @@ from groundworkers.services.grounding import ConceptGroundingService +def _integration_prerequisite_failure(message, error=None): + if os.getenv("GROUNDWORKERS_REQUIRE_INTEGRATION") == "1": + if error is not None: + raise error + raise AssertionError(message) + pytest.skip(message) + + def _load_graph_adapter(): try: import omop_graph # noqa: F401 except ImportError: - pytest.skip("omop_graph is not installed in this environment") + _integration_prerequisite_failure( + "omop_graph is not installed in this environment", + RuntimeError("required graph integration needs omop-graph installed"), + ) config_path = os.getenv("GROUNDWORKERS_CONFIG_PATH") or os.getenv( "GROUNDWORKERS_CONFIG" @@ -27,97 +33,38 @@ def _load_graph_adapter(): "an explicit integration config path is required; set " "GROUNDWORKERS_CONFIG_PATH" ) - if os.getenv("GROUNDWORKERS_REQUIRE_INTEGRATION") == "1": - raise RuntimeError(message) - pytest.skip(message) + _integration_prerequisite_failure(message, RuntimeError(message)) try: config = build_app_config(config_path=config_path) except (FileNotFoundError, ValueError) as exc: - if os.getenv("GROUNDWORKERS_REQUIRE_INTEGRATION") == "1": - raise - pytest.skip(f"shared stack config is unavailable: {exc}") + _integration_prerequisite_failure( + f"shared stack config is unavailable: {exc}", exc + ) adapter = build_adapters(config).omop_graph if adapter is None: - pytest.skip("omop_graph adapter was not built") + _integration_prerequisite_failure("omop_graph adapter was not built") if not adapter.is_available(): - pytest.skip("omop_graph backend is not available in this environment") + _integration_prerequisite_failure( + "omop_graph backend is not available in this environment" + ) adapter._integration_vocab_engine = config.cdm_vocab_engine return adapter -def _find_nonstandard_condition_term(adapter) -> str: - standard = aliased(Concept) - same_name = aliased(Concept) - stmt = ( - select(Concept.concept_name) - .join( - Concept_Relationship, - (Concept_Relationship.concept_id_1 == Concept.concept_id) - & (func.lower(Concept_Relationship.relationship_id) == "maps to") - & Concept_Relationship.invalid_reason.is_(None), - ) - .join( - standard, - (standard.concept_id == Concept_Relationship.concept_id_2) - & (standard.standard_concept == "S") - & (standard.domain_id == "Condition") - & standard.invalid_reason.is_(None), - ) - .where( - Concept.domain_id == "Condition", - Concept.standard_concept.is_(None), - Concept.invalid_reason.is_(None), - ~exists( - select(1).where( - func.lower(same_name.concept_name) - == func.lower(Concept.concept_name), - same_name.domain_id == "Condition", - same_name.standard_concept.in_(("S", "C")), - same_name.invalid_reason.is_(None), - ) - ), - ) - .order_by(Concept.concept_id) - .limit(1) - ) - with adapter._cdm_session_factory() as session: - row = session.execute(stmt).first() - if row is None: - pytest.skip( - "no uniquely named non-standard Condition concept with a standard mapping was found" - ) - return str(row[0]) +def _require_seeded_concepts(adapter, concept_ids): + missing = [concept_id for concept_id in concept_ids if adapter.get_concept(concept_id) is None] + if missing: + raise AssertionError(f"required OMOP fixture concepts are missing: {missing}") -def _skip_if_parent_lookup_is_unindexed(adapter) -> None: - """Skip hierarchy smoke checks when the live vocabulary DB is not shaped for them.""" +@pytest.mark.required_integration +def test_required_graph_fixture_concepts_are_seeded(): + adapter = _load_graph_adapter() - try: - schema = physical_schema_of( - adapter._integration_vocab_engine, schema_tag=Role.VOCAB - ) - indexes = sa_inspect(adapter._integration_vocab_engine).get_indexes( - "concept_ancestor", schema=schema - ) - except Exception as exc: - # Broad except: integration environment preflight. - pytest.skip(f"could not inspect concept_ancestor indexes: {exc}") - - if not any( - "descendant_concept_id" in (index.get("column_names") or []) - or any( - "descendant_concept_id" in str(expression).lower() - for expression in (index.get("expressions") or ()) - ) - for index in indexes - ): - pytest.skip( - "concept_ancestor lacks a descendant-leading index; " - "parent hierarchy lookup is an environment smoke test, not an application regression test" - ) + _require_seeded_concepts(adapter, (201826, 201827, 201828, 201837, 201838)) -@pytest.mark.integration +@pytest.mark.required_integration def test_get_concept_by_known_id(): adapter = _load_graph_adapter() @@ -127,7 +74,7 @@ def test_get_concept_by_known_id(): assert concept["concept_name"] == "Type 2 diabetes mellitus" -@pytest.mark.integration +@pytest.mark.required_integration def test_get_concept_returns_correct_fields(): adapter = _load_graph_adapter() @@ -150,33 +97,29 @@ def test_get_concept_returns_correct_fields(): } -@pytest.mark.integration -def test_ancestors_depth_is_monotonically_increasing(): +@pytest.mark.required_integration +def test_ancestors_match_seeded_relationships_and_depths(): adapter = _load_graph_adapter() - _skip_if_parent_lookup_is_unindexed(adapter) graph = GraphService(adapter) ancestors = graph.get_ancestors(201826, max_depth=5) - depths = [item["depth"] for item in ancestors] - assert depths == sorted(depths) + assert [(item["concept_id"], item["depth"]) for item in ancestors] == [(201827, 1)] -@pytest.mark.integration -def test_descendants_returns_list_and_respects_depth(): +@pytest.mark.required_integration +def test_descendants_match_seeded_relationships_and_depths(): adapter = _load_graph_adapter() graph = GraphService(adapter) - # 201826 has descendants; this test verifies shape and depth constraint. - # A true leaf concept is vocab-specific — use depth=0 equivalent via max_depth=1 - # and verify every result has depth <= 1. - descendants = graph.get_descendants(201826, max_depth=1) + descendants = graph.get_descendants(201827, max_depth=1) - assert isinstance(descendants, list) - assert all(item["depth"] <= 1 for item in descendants) + assert [(item["concept_id"], item["depth"]) for item in descendants] == [ + (201826, 1), (201828, 1), (201829, 1) + ] -@pytest.mark.integration +@pytest.mark.required_integration def test_concept_by_code_snomed(): adapter = _load_graph_adapter() @@ -186,7 +129,7 @@ def test_concept_by_code_snomed(): assert concepts[0]["vocabulary_id"] == "SNOMED" -@pytest.mark.integration +@pytest.mark.required_integration def test_ground_exact_match(): adapter = _load_graph_adapter() service = ConceptGroundingService(GraphService(adapter)) @@ -199,12 +142,13 @@ def test_ground_exact_match(): assert "grounding_explanation" in result results = result["results"] assert results + assert results[0]["concept_id"] == 201826 assert results[0]["match_kind"] == "EXACT" assert results[0]["concept_name"].lower() == "type 2 diabetes mellitus" assert results[0]["standard_concept"] is True -@pytest.mark.integration +@pytest.mark.required_integration def test_ground_partial_match(): adapter = _load_graph_adapter() service = ConceptGroundingService(GraphService(adapter)) @@ -213,10 +157,11 @@ def test_ground_partial_match(): results = result["results"] assert results + assert 201826 in {hit["concept_id"] for hit in results} assert all(r["standard_concept"] is True for r in results) -@pytest.mark.integration +@pytest.mark.required_integration def test_ground_returns_standard_concepts_only(): adapter = _load_graph_adapter() service = ConceptGroundingService(GraphService(adapter)) @@ -225,16 +170,17 @@ def test_ground_returns_standard_concepts_only(): results = result["results"] assert results + assert {201826, 201827} & {hit["concept_id"] for hit in results} assert all(r["standard_concept"] is True for r in results) scores = [r["total_score"] for r in results] assert scores == sorted(scores, reverse=True) -@pytest.mark.integration +@pytest.mark.required_integration def test_ground_parentless_condition_term_standardizes_nonstandard_source(): adapter = _load_graph_adapter() service = ConceptGroundingService(GraphService(adapter)) - query = _find_nonstandard_condition_term(adapter) + query = "Source diabetes mellitus" result = service.ground( query, @@ -246,91 +192,57 @@ def test_ground_parentless_condition_term_standardizes_nonstandard_source(): results = result["results"] assert results + assert results[0]["concept_id"] == 201826 assert results[0]["standard_concept"] is True assert results[0]["standardized_from"] is not None assert result["grounding_explanation"]["parent_ids_source"] == "none" assert result["grounding_explanation"]["effective_parent_ids"] == [] -@pytest.mark.integration +@pytest.mark.required_integration def test_standard_flags_distinguish_s_from_c(): """The strict flag contract distinguishes raw OMOP ``S`` and ``C`` rows.""" adapter = _load_graph_adapter() - stmt = ( - select(Concept.standard_concept, func.min(Concept.concept_id)) - .where(Concept.standard_concept.in_(("S", "C"))) - .group_by(Concept.standard_concept) - ) - with adapter._cdm_session_factory() as session: - sampled = {row[0]: int(row[1]) for row in session.execute(stmt).all()} - if set(sampled) != {"S", "C"}: - pytest.skip("vocabulary does not contain both 'S' and 'C' concepts") - - views = adapter.concept_views(list(sampled.values())) - assert views[sampled["S"]]["standard_concept"] is True - assert views[sampled["S"]]["classification_concept"] is False - assert views[sampled["C"]]["standard_concept"] is False - assert views[sampled["C"]]["classification_concept"] is True + views = adapter.concept_views([201826, 201834]) + assert views[201826]["standard_concept"] is True + assert views[201826]["classification_concept"] is False + assert views[201834]["standard_concept"] is False + assert views[201834]["classification_concept"] is True -@pytest.mark.integration +@pytest.mark.required_integration def test_classification_concept_is_not_standard(): """A classification concept is never exposed as a standard mapping target.""" adapter = _load_graph_adapter() - other = aliased(Concept) - stmt = ( - select(Concept.concept_name) - .where( - Concept.standard_concept == "C", - Concept.domain_id == "Drug", - Concept.invalid_reason.is_(None), - ~exists( - select(1).where( - func.lower(other.concept_name) == func.lower(Concept.concept_name), - other.standard_concept == "S", - ) - ), - ) - .order_by(Concept.concept_id) - .limit(1) - ) - with adapter._cdm_session_factory() as session: - row = session.execute(stmt).first() - if row is None: - pytest.skip("no uniquely named classification Drug concept was found") - - with adapter._cdm_session_factory() as session: - concept_id = session.execute( - select(Concept.concept_id).where(Concept.concept_name == row[0]) - ).scalar_one() - classification = adapter.get_concept(int(concept_id)) + classification = adapter.get_concept(201834) assert classification is not None assert classification["classification_concept"] is True assert classification["standard_concept"] is False -@pytest.mark.integration +@pytest.mark.required_integration def test_grounding_flags_are_mutually_exclusive_across_a_query_set(): adapter = _load_graph_adapter() service = ConceptGroundingService(GraphService(adapter)) - for query, domain in ( - ("Type 2 diabetes mellitus", None), - ("metformin", "Drug"), - ("appendectomy", "Procedure"), - ("asthma", None), + for query, domain, expected_id in ( + ("Type 2 diabetes mellitus", None, 201826), + ("metformin", "Drug", 201830), + ("appendectomy", "Procedure", 201831), + ("asthma", None, 201829), ): result = service.ground(query, limit=10, domain=domain, vocabulary_id=None) + assert expected_id in {hit["concept_id"] for hit in result["results"]} for hit in result["results"]: assert not (hit["standard_concept"] and hit["classification_concept"]), ( f"{query!r} -> {hit['concept_id']} reported as both standard and classification" ) -@pytest.mark.integration +@pytest.mark.optional_integration def test_embedding_tier_produces_candidates_when_configured(): """Read-only (write=False) grounding must still be able to use the embedding tier. @@ -356,31 +268,47 @@ def test_embedding_tier_produces_candidates_when_configured(): assert all(hit["embedding_score"] is not None for hit in result["results"]) -@pytest.mark.integration -def test_standard_only_and_active_only_narrow_candidate_resolution(): +@pytest.mark.required_integration +def test_standard_only_and_active_only_exclude_seeded_decoys(): adapter = _load_graph_adapter() service = ConceptGroundingService(GraphService(adapter)) - unfiltered = service.ground( - "diabetes", limit=20, domain="Condition", vocabulary_id=None + source_query = "Active nonstandard fixture source" + source_unfiltered = service.ground( + source_query, limit=20, domain="Condition", vocabulary_id=None ) - filtered = service.ground( - "diabetes", + source_standard = service.ground( + source_query, limit=20, domain="Condition", vocabulary_id=None, standard_only=True, - active_only=True, ) - - assert filtered["grounding_explanation"]["standard_only"] is True - assert filtered["grounding_explanation"]["active_only"] is True - # Filtering can only narrow the candidate pool, never widen it. - assert len(filtered["results"]) <= len(unfiltered["results"]) - - -@pytest.mark.integration -def test_get_edges_returns_predicate_kinds(): + assert [hit["concept_id"] for hit in source_unfiltered["results"]] == [201826] + assert source_unfiltered["results"][0]["standardized_from"]["concept_id"] == 201837 + assert source_standard["results"] == [] + + query = "Type 2 diabetes mellitus" + unfiltered = service.ground(query, limit=20, domain="Condition", vocabulary_id=None) + standard = service.ground(query, limit=20, domain="Condition", vocabulary_id=None, + standard_only=True) + active = service.ground(query, limit=20, domain="Condition", vocabulary_id=None, active_only=True) + both = service.ground(query, limit=20, domain="Condition", vocabulary_id=None, + standard_only=True, active_only=True) + + def ids(result): + return {hit["concept_id"] for hit in result["results"]} + + assert ids(unfiltered) == {201826, 201838} + assert ids(standard) == {201826, 201838} + assert ids(active) == {201826} + assert ids(both) == {201826} + assert standard["grounding_explanation"]["standard_only"] is True + assert active["grounding_explanation"]["active_only"] is True + + +@pytest.mark.required_integration +def test_get_edges_match_seeded_relationships(): adapter = _load_graph_adapter() graph = GraphService(adapter) @@ -388,24 +316,21 @@ def test_get_edges_returns_predicate_kinds(): assert "outbound" in edges assert "inbound" in edges - valid_kinds = {"HIERARCHY", "IDENTITY", "COMPOSITION", "ASSOCIATION", "ATTRIBUTE"} - for edge in edges["outbound"]: - assert edge["predicate_kind"] in valid_kinds - assert "relationship_id" in edge - assert "target_concept_id" in edge + assert [(e["relationship_id"], e["target_concept_id"]) for e in edges["outbound"]] == [ + ("Is a", 201827) + ] + assert [(e["relationship_id"], e["source_concept_id"]) for e in edges["inbound"]] == [ + ("Maps to", 201832), ("Maps to", 201833), ("Maps to", 201837) + ] + assert edges["outbound"][0]["predicate_kind"] == "HIERARCHY" -@pytest.mark.integration +@pytest.mark.required_integration def test_find_path_between_concept_and_its_ancestor(): adapter = _load_graph_adapter() graph = GraphService(adapter) - ancestors = graph.get_ancestors(201826, max_depth=1) - if not ancestors: - pytest.skip("concept has no ancestors in this dataset") - parent_id = ancestors[0]["concept_id"] - - result = graph.find_path(201826, parent_id, max_depth=5) + result = graph.find_path(201826, 201827, max_depth=5) assert result["found"] is True assert len(result["paths"]) > 0 @@ -413,10 +338,10 @@ def test_find_path_between_concept_and_its_ancestor(): assert path["length"] == len(path["steps"]) if path["steps"]: assert path["steps"][0]["subject_id"] == 201826 - assert path["steps"][-1]["object_id"] == parent_id + assert path["steps"][-1]["object_id"] == 201827 -@pytest.mark.integration +@pytest.mark.required_integration def test_map_icd_code_to_standard_snomed(): adapter = _load_graph_adapter() graph = GraphService(adapter) @@ -427,4 +352,5 @@ def test_map_icd_code_to_standard_snomed(): assert "standard_concepts" in result assert result["source"]["vocabulary_id"] == "ICD10CM" assert result["source"]["concept_code"] == "E11.9" - assert all(c["standard_concept"] is True for c in result["standard_concepts"]) + assert result["source"]["concept_id"] == 201832 + assert [c["concept_id"] for c in result["standard_concepts"]] == [201826] diff --git a/tests/integration/test_omop_postgres_runtime.py b/tests/integration/test_omop_postgres_runtime.py new file mode 100644 index 0000000..0b50cfb --- /dev/null +++ b/tests/integration/test_omop_postgres_runtime.py @@ -0,0 +1,209 @@ +from __future__ import annotations + +import os + +import pytest +from oa_configurator import Role +from sqlalchemy import create_engine, event, inspect, text +from sqlalchemy.engine import Engine + +from groundworkers.app import build_adapters, build_application +from groundworkers.application.setup.configuration import load_configuration +from groundworkers.application.setup.performance_maintenance import ( + populate_normalized_indexes, + populate_trigram_indexes, +) +from groundworkers.base.errors import GroundworkersError +from groundworkers.bootstrap import build_app_config, build_app_config_from_stack + + +@pytest.fixture(scope="module") +def indexed_runtime_config(): + config_path = os.getenv("GROUNDWORKERS_CONFIG_PATH") or os.getenv( + "GROUNDWORKERS_CONFIG" + ) + if os.getenv("GROUNDWORKERS_REQUIRE_INTEGRATION") != "1": + pytest.skip("PostgreSQL schema and runtime checks run in the required database lane") + if not config_path: + raise RuntimeError("required PostgreSQL checks need GROUNDWORKERS_CONFIG_PATH") + + snapshot = load_configuration(config_path=config_path) + populate_trigram_indexes(snapshot) + populate_normalized_indexes(snapshot) + config = build_app_config(config_path=config_path) + schema = config.cdm_database.schema_for_role(Role.VOCAB) + indexes = { + index["name"] + for index in inspect(config.cdm_vocab_engine).get_indexes( + "concept", schema=schema + ) + } + assert "idx_concept_name_normkey" in indexes + assert "idx_concept_lower_name_trgm" in indexes + yield config, schema + config.cdm_engine.dispose() + if config.cdm_vocab_engine is not config.cdm_engine: + config.cdm_vocab_engine.dispose() + + +@pytest.mark.required_integration +def test_postgres_created_indexes_reflect_and_serve_normalized_search( + indexed_runtime_config, +): + config, schema = indexed_runtime_config + app = build_application(config) + + matches, plan = app.services.vocab.search_normalized_with_plan( + "Type 2 diabetes mellitus", domain="Condition" + ) + + assert plan == "indexed" + assert 201826 in {match.concept_id for match in matches} + assert config.cdm_database.schema_for_role(Role.VOCAB) == schema + + +@pytest.mark.required_integration +def test_postgres_restricted_runtime_role_reads_split_schema_without_write_grants( + indexed_runtime_config, +): + config, vocab_schema = indexed_runtime_config + from tests.support.embedding_contract_fixture import ( + MODEL_NAME, + seed_embedding_contract, + ) + + seed_embedding_contract(config) + admin_engine = create_engine(os.environ["GROUNDWORKERS_TEST_DATABASE_URL"]) + runtime = None + adapters = None + try: + with admin_engine.begin() as connection: + connection.execute( + text( + "DO $$ BEGIN " + "IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'gw_runtime') THEN " + "CREATE ROLE gw_runtime LOGIN PASSWORD 'gw_runtime_fixture'; " + "ELSE ALTER ROLE gw_runtime LOGIN PASSWORD 'gw_runtime_fixture'; END IF; " + "END $$" + ) + ) + connection.execute(text("GRANT CONNECT ON DATABASE test_db TO gw_runtime")) + schemas = connection.execute( + text( + "SELECT schema_name FROM information_schema.schemata " + "WHERE schema_name NOT LIKE 'pg_%' AND schema_name <> 'information_schema'" + ) + ).scalars().all() + quote = connection.dialect.identifier_preparer.quote + for schema in schemas: + quoted_schema = quote(schema) + connection.execute(text(f"GRANT USAGE ON SCHEMA {quoted_schema} TO gw_runtime")) + connection.execute(text(f"GRANT SELECT ON ALL TABLES IN SCHEMA {quoted_schema} TO gw_runtime")) + + stack = config.stack.model_copy(deep=True) + connection = stack.connections["pg_test"] + stack.connections["pg_test"] = connection.model_copy( + update={"user": "gw_runtime", "password": "gw_runtime_fixture"} + ) + runtime = build_app_config_from_stack(stack) + assert runtime.cdm_database.schema_for_role(Role.PRIMARY) == "gw_primary" + assert runtime.cdm_database.schema_for_role(Role.VOCAB) == vocab_schema + adapters = build_adapters(runtime) + graph = adapters.omop_graph + assert graph is not None and graph.is_available() + assert adapters.omop_emb is not None + + statements = [] + + def record_statement(_conn, _cursor, statement, *_args): + statements.append(statement) + + event.listen(Engine, "before_cursor_execute", record_statement) + try: + concept = graph.get_concept(201826) + assert concept is not None and concept["concept_name"] == "Type 2 diabetes mellitus" + neighbours = adapters.omop_emb.get_neighbours( + 201826, limit=5, model_name=MODEL_NAME + ) + assert neighbours["results"] + finally: + event.remove(Engine, "before_cursor_execute", record_statement) + + assert statements + assert not any( + statement.lstrip().split(None, 1)[0].upper() + in {"INSERT", "UPDATE", "DELETE", "MERGE", "CREATE", "ALTER", "DROP", "TRUNCATE"} + for statement in statements + ) + with runtime.cdm_vocab_engine.connect() as connection: + can_insert = connection.execute( + text("SELECT has_table_privilege(current_user, :table_name, 'INSERT')"), + {"table_name": f"{vocab_schema}.concept"}, + ).scalar_one() + assert can_insert is False + finally: + if adapters is not None: + if adapters.omop_graph is not None: + adapters.omop_graph.close() + if adapters.omop_emb is not None: + adapters.omop_emb.close() + if runtime is not None: + runtime.cdm_engine.dispose() + if runtime.cdm_vocab_engine is not runtime.cdm_engine: + runtime.cdm_vocab_engine.dispose() + try: + with admin_engine.begin() as connection: + role_exists = connection.execute( + text("SELECT 1 FROM pg_roles WHERE rolname = 'gw_runtime'") + ).scalar_one_or_none() + if role_exists: + connection.execute(text("DROP OWNED BY gw_runtime")) + connection.execute(text("DROP ROLE gw_runtime")) + finally: + admin_engine.dispose() + + +@pytest.mark.required_integration +def test_postgres_statement_timeout_rollback_keeps_session_usable( + indexed_runtime_config, + monkeypatch, +): + config, _schema = indexed_runtime_config + from groundworkers.services import vocab + + monkeypatch.setattr(vocab, "NORMALIZED_SEARCH_TIMEOUT_MS", 1) + app = build_application(config) + assert ( + app.services.vocab._normalized_search_plan( + domain="Condition", + vocabulary_id=None, + parent_ids=None, + include_synonyms=False, + allow_unindexed_scan=False, + ) + == "indexed" + ) + injected = False + + def delay_app_search(_conn, cursor, statement, *_args): + nonlocal injected + if not injected and "set_config" in statement.lower(): + injected = True + cursor.execute("SELECT pg_sleep(0.05)") + + event.listen(config.cdm_vocab_engine, "after_cursor_execute", delay_app_search) + try: + with pytest.raises(GroundworkersError) as excinfo: + app.services.vocab.search_normalized_with_plan( + "Type 2 diabetes mellitus", domain="Condition" + ) + assert excinfo.value.code == "QUERY_TIMEOUT" + finally: + event.remove(config.cdm_vocab_engine, "after_cursor_execute", delay_app_search) + + results, plan = app.services.vocab.search_normalized_with_plan( + "Type 2 diabetes mellitus", domain="Condition" + ) + assert injected + assert plan == "indexed" + assert 201826 in {match.concept_id for match in results} diff --git a/tests/support/embedding_contract_fixture.py b/tests/support/embedding_contract_fixture.py new file mode 100644 index 0000000..cbbfb59 --- /dev/null +++ b/tests/support/embedding_contract_fixture.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +import numpy as np +from omop_emb.backends import open_vector_store_writer +from omop_emb.backends.embedding_table import ConceptEmbeddingRecord +from omop_emb.backends.index_config import FlatIndexConfig + +MODEL_NAME = "groundworkers-contract:v1" +OTHER_MODEL_NAME = "groundworkers-secondary:v1" + + +def seed_embedding_contract(config) -> None: + if config.vector_store is None: + raise AssertionError("embedding fixture needs a configured vector store") + records = ( + ConceptEmbeddingRecord(201826, "Condition", "SNOMED", True, True), + ConceptEmbeddingRecord(201827, "Condition", "SNOMED", True, True), + ConceptEmbeddingRecord(201828, "Condition", "SNOMED", True, True), + ConceptEmbeddingRecord(201830, "Drug", "RxNorm", True, True), + ConceptEmbeddingRecord(201834, "Drug", "ATC", False, True), + ConceptEmbeddingRecord(201837, "Condition", "LOCAL", False, True), + ConceptEmbeddingRecord(201838, "Condition", "SNOMED", True, False), + ) + vectors = np.asarray( + [ + [1.0, 0.0, 0.0], + [0.95, 0.05, 0.0], + [0.0, 1.0, 0.0], + [0.85, 0.15, 0.0], + [0.8, 0.2, 0.0], + [0.9, 0.1, 0.0], + [0.99, 0.01, 0.0], + ], + dtype=np.float32, + ) + writer = open_vector_store_writer(config.vector_store) + try: + for model_name, model_records, model_vectors in ( + (MODEL_NAME, records, vectors), + (OTHER_MODEL_NAME, records[:1], vectors[:1]), + ): + if not writer.get_registered_models(model_name=model_name): + writer.register_model( + model_name=model_name, + provider_type="ollama", + index_config=FlatIndexConfig(), + dimensions=3, + ) + writer.upsert_embeddings( + model_name=model_name, + records=model_records, + embeddings=model_vectors, + ) + finally: + writer.close() diff --git a/tests/support/seed_postgres_fixture.py b/tests/support/seed_postgres_fixture.py index f2f4f89..7cc94d2 100644 --- a/tests/support/seed_postgres_fixture.py +++ b/tests/support/seed_postgres_fixture.py @@ -10,7 +10,7 @@ import os from datetime import date -from oa_configurator import Resolver, load_stack_config +from oa_configurator import Resolver, Role, load_stack_config from omop_alchemy.cdm.model.vocabulary import ( # noqa: F401 Concept, Concept_Ancestor, @@ -45,6 +45,12 @@ def seed() -> None: cdm = Resolver( load_stack_config(os.environ["GROUNDWORKERS_CONFIG_PATH"]) ).resolve_database("test_cdm_db") + primary_schema = cdm.schema_for_role(Role.PRIMARY) + vocab_schema = cdm.schema_for_role(Role.VOCAB) + quote = engine.dialect.identifier_preparer.quote + with engine.begin() as connection: + for schema in {primary_schema, vocab_schema, "embeddings"}: + connection.execute(text(f"CREATE SCHEMA IF NOT EXISTS {quote(schema)}")) bootstrap(cdm, create=True) concepts = [ @@ -105,6 +111,7 @@ def seed() -> None: (201829, 201827, "Is a"), (201832, 201826, "Maps to"), (201833, 201826, "Maps to"), + (201837, 201826, "Maps to"), ] ancestors = [ (201826, 201826, 0, 0), @@ -124,7 +131,9 @@ def seed() -> None: # at domain/vocabulary/class rows, while those rows point at concepts. # The real vocabulary loader handles this cycle; this tiny fixture uses # the PostgreSQL test superuser to load the same compact data safely. - connection.execute(text("CREATE EXTENSION IF NOT EXISTS vector")) + connection.execute(text("SET LOCAL search_path TO " + f"{quote(primary_schema)}, {quote(vocab_schema)}, public")) + connection.execute(text("CREATE EXTENSION IF NOT EXISTS vector WITH SCHEMA public")) connection.execute(text("SET session_replication_role = replica")) try: connection.execute( @@ -251,6 +260,22 @@ def seed() -> None: for concept_id, concept_name, domain_id, vocabulary_id, concept_class_id, standard_concept, concept_code in concepts ], ) + # Exact-label decoys make both candidate filters observable in + # PostgreSQL integration tests. The first is active but non-standard; + # the second is standard but invalidated. + connection.execute( + text( + "INSERT INTO concept " + "(concept_id, concept_name, domain_id, vocabulary_id, concept_class_id, " + "standard_concept, concept_code, valid_start_date, valid_end_date, invalid_reason) " + "VALUES " + "(201837, 'Active nonstandard fixture source', 'Condition', 'LOCAL', " + "'Clinical Finding', NULL, 'ACTIVE-NONSTANDARD', :start, :end, NULL), " + "(201838, 'Type 2 diabetes mellitus', 'Condition', 'SNOMED', " + "'Clinical Finding', 'S', 'INVALID-STANDARD', :start, :end, 'D')" + ), + {"start": start, "end": end}, + ) connection.execute( text( "INSERT INTO concept_relationship " diff --git a/tests/support/wheel_smoke.py b/tests/support/wheel_smoke.py new file mode 100644 index 0000000..9c16279 --- /dev/null +++ b/tests/support/wheel_smoke.py @@ -0,0 +1,90 @@ +"""Smoke-test an installed wheel while running outside the source checkout.""" + +from __future__ import annotations + +import importlib.metadata +import subprocess +import sys +from pathlib import Path + + +def main() -> None: + mode = sys.argv[1] + import groundworkers + + checkout = Path(__file__).resolve().parents[2] + installed = Path(groundworkers.__file__).resolve() + assert not installed.is_relative_to(checkout), ( + f"wheel smoke imported checkout source: {installed}" + ) + entries = importlib.metadata.entry_points(group="omop.config") + assert any( + entry.name == "groundworkers" + and entry.value == "groundworkers.config:GroundworkersConfig" + for entry in entries + ) + subprocess.run( + [sys.executable, "-m", "groundworkers.server", "--help"], + check=True, + stdout=subprocess.DEVNULL, + ) + if mode == "extras": + _exercise_transports() + + +def _exercise_transports() -> None: + from fastapi.testclient import TestClient + from oa_configurator import CDMDatabaseConfig, ConnectionConfig, StackConfig + + from groundworkers.app import Adapters, GroundworkersApp, Services + from groundworkers.base.server import GroundworkersMCPServer + from groundworkers.bootstrap import build_app_config_from_stack + from groundworkers.services.source_planning.assisted import ( + AssistedColumnRoleClassifier, + ) + from groundworkers.services.source_planning.service import SourcePlanningService + from groundworkers.tools.source_planning_tools import register_source_planning_tools + from groundworkers.transports.rest import create_rest_app + + class FixedModel: + def complete_structured(self, _prompt, _schema, **_kwargs): + return {"decisions": [{"header": "unknown_a", "role": "values", "confidence": 0.9}]} + + service = SourcePlanningService( + assisted_classifier=AssistedColumnRoleClassifier(FixedModel()) + ) + server = GroundworkersMCPServer("wheel-smoke") + register_source_planning_tools(server, service) + mcp_result = server.call( + "source_plan_assisted", + content="unknown_a,unknown_b\nA,B\n", + filename="fixture.csv", + include_intermediate=True, + ) + assert mcp_result["llm_tier_used"] is True + + stack = StackConfig( + connections={"test": ConnectionConfig(dialect="sqlite", database_name=":memory:")}, + databases={"cdm": CDMDatabaseConfig(connection="test")}, + tools={"groundworkers": {"cdm_db": "cdm"}}, + ) + app = GroundworkersApp( + config=build_app_config_from_stack(stack), + adapters=Adapters(), + services=Services(source_planning=service), + ) + client = TestClient(create_rest_app(app)) + response = client.post( + "/v1/source-planning/assisted-plan", + json={ + "content": "unknown_a,unknown_b\nA,B\n", + "filename": "fixture.csv", + "include_intermediate": True, + }, + ) + assert response.status_code == 200 + assert response.json()["llm_tier_used"] is True + + +if __name__ == "__main__": + main() diff --git a/tests/test_rest_api.py b/tests/test_rest_api.py index 5da16b2..3f088e5 100644 --- a/tests/test_rest_api.py +++ b/tests/test_rest_api.py @@ -6,11 +6,13 @@ from groundworkers.app import Adapters, GroundworkersApp, Services from groundworkers.base.errors import GroundworkersError from groundworkers.bootstrap import build_app_config_from_stack +from groundworkers.services.source_planning.assisted import AssistedColumnRoleClassifier from groundworkers.services.source_planning.models import ( IngestionPlan, PreIngestBundle, SourceFormat, ) +from groundworkers.services.source_planning.service import SourcePlanningService from groundworkers.services.source_planning.warnings import PlanningWarning from groundworkers.transports.rest import create_rest_app from groundworkers.transports.rest.api import _status_code_for_error @@ -162,6 +164,37 @@ def test_assisted_plan_endpoint_decodes_and_serializes_bundle() -> None: ] +def test_assisted_plan_endpoint_runs_real_service_and_classifier() -> None: + class FixedModel: + def complete_structured(self, _prompt, _schema, **_kwargs): + return { + "decisions": [ + {"header": "unknown_a", "role": "values", "confidence": 0.93} + ] + } + + service = SourcePlanningService( + assisted_classifier=AssistedColumnRoleClassifier(FixedModel()) + ) + client = _client(mapping=FakeMappingService(), source_planning=service) + + response = client.post( + "/v1/source-planning/assisted-plan", + json={ + "content": "unknown_a,unknown_b\nA,B\n", + "filename": "diagnoses.csv", + "include_intermediate": True, + }, + ) + + assert response.status_code == 200 + result = response.json() + assert result["llm_tier_used"] is True + annotation = result["annotated_tables"][0]["column_annotations"]["unknown_a"] + assert annotation["role"] == "values" + assert annotation["detection_tier"] == "LLM" + + def test_candidate_bundle_endpoint_returns_backend_unavailable_when_service_missing() -> None: client = _client(source_planning=FakeSourcePlanningService()) diff --git a/tests/unit/test_plugins.py b/tests/unit/test_plugins.py index 3b009be..f506829 100644 --- a/tests/unit/test_plugins.py +++ b/tests/unit/test_plugins.py @@ -93,11 +93,58 @@ def _embedding_stack(tmp_path: Path, extra: dict | None = None): return stack -def test_discover_plugins_finds_nothing_by_default() -> None: - """No entry points are registered in this environment; the loader is inert.""" +def test_discover_plugins_returns_empty_controlled_registry(monkeypatch) -> None: + """A core-only host has no plugins when the registry boundary is empty.""" + monkeypatch.setattr("groundworkers.plugins.entry_points", lambda **_kwargs: ()) assert discover_plugins() == [] +class FakeEntryPoint: + def __init__(self, name: str, plugin=None, error: Exception | None = None) -> None: + self.name = name + self.plugin = plugin + self.error = error + + def load(self): + if self.error is not None: + raise self.error + return self.plugin + + +def test_discover_plugins_loads_one_and_multiple_plugins(monkeypatch) -> None: + first, second = FakePlugin(), FakeReadinessPlugin() + second.name = "readiness_plugin" + second.config_cls = None + registry = [FakeEntryPoint("first", first)] + monkeypatch.setattr("groundworkers.plugins.entry_points", lambda **_kwargs: registry) + assert discover_plugins() == [first] + + registry.append(FakeEntryPoint("second", second)) + assert discover_plugins() == [first, second] + + +def test_broken_plugin_entry_point_does_not_hide_a_valid_plugin(monkeypatch) -> None: + valid = FakePlugin() + monkeypatch.setattr( + "groundworkers.plugins.entry_points", + lambda **_kwargs: ( + FakeEntryPoint("broken", error=ImportError("plugin import failed")), + FakeEntryPoint("valid", valid), + ), + ) + + assert discover_plugins() == [valid] + + +def test_discovery_rejects_duplicate_plugin_identities(monkeypatch) -> None: + monkeypatch.setattr( + "groundworkers.plugins.entry_points", + lambda **_kwargs: (FakeEntryPoint("one", FakePlugin()), FakeEntryPoint("two", FakePlugin())), + ) + with pytest.raises(ValueError, match="not unique"): + discover_plugins() + + def test_build_application_uses_config_defaults_when_no_section_present( tmp_path: Path, monkeypatch ) -> None: diff --git a/tests/unit/test_required_integration_preflight.py b/tests/unit/test_required_integration_preflight.py new file mode 100644 index 0000000..f026360 --- /dev/null +++ b/tests/unit/test_required_integration_preflight.py @@ -0,0 +1,80 @@ +from __future__ import annotations + +import builtins +from types import SimpleNamespace + +import pytest + +from tests.conftest import _missing_required_tests, _required_test_failures +from tests.integration import test_omop_graph_adapter as graph_integration + + +@pytest.fixture(autouse=True) +def required_integration_config(monkeypatch): + monkeypatch.setenv("GROUNDWORKERS_REQUIRE_INTEGRATION", "1") + monkeypatch.setenv("GROUNDWORKERS_CONFIG_PATH", "/fixture/stack.toml") + + +def test_missing_graph_dependency_fails_required_preflight(monkeypatch): + original_import = builtins.__import__ + + def missing_graph(name, *args, **kwargs): + if name == "omop_graph": + raise ImportError("fixture omitted omop-graph") + return original_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", missing_graph) + + with pytest.raises(RuntimeError, match="omop-graph installed"): + graph_integration._load_graph_adapter() + + +def test_bad_config_fails_required_preflight(monkeypatch): + def invalid_config(**_kwargs): + raise ValueError("invalid fixture config") + + monkeypatch.setattr(graph_integration, "build_app_config", invalid_config) + + with pytest.raises(ValueError, match="invalid fixture config"): + graph_integration._load_graph_adapter() + + +@pytest.mark.parametrize( + ("adapter", "message"), + [ + (None, "adapter was not built"), + (SimpleNamespace(is_available=lambda: False), "backend is not available"), + ], +) +def test_missing_or_unavailable_adapter_fails_required_preflight( + monkeypatch, adapter, message +): + monkeypatch.setattr(graph_integration, "build_app_config", lambda **_kwargs: object()) + monkeypatch.setattr( + graph_integration, + "build_adapters", + lambda _config: SimpleNamespace(omop_graph=adapter), + ) + + with pytest.raises(AssertionError, match=message): + graph_integration._load_graph_adapter() + + +def test_missing_seeded_fixture_concept_fails_required_preflight(): + class IncompleteAdapter: + def get_concept(self, concept_id): + return {"concept_id": concept_id} if concept_id == 201826 else None + + with pytest.raises(AssertionError, match="201837"): + graph_integration._require_seeded_concepts(IncompleteAdapter(), (201826, 201837)) + + +def test_execution_gate_detects_a_disabled_required_test(): + required = {"graph::test_required", "emb::test_required"} + + assert _missing_required_tests(required, {"graph::test_required"}) == [ + "emb::test_required" + ] + assert _required_test_failures(required, {"graph::test_required": "passed"}) == [ + "emb::test_required" + ] diff --git a/tests/unit/test_required_test_gate.py b/tests/unit/test_required_test_gate.py new file mode 100644 index 0000000..9387953 --- /dev/null +++ b/tests/unit/test_required_test_gate.py @@ -0,0 +1,16 @@ +from __future__ import annotations + +from tests.conftest import _missing_required_tests, _required_test_failures + + +def test_required_execution_gate_detects_a_marked_test_removed_from_selection(): + required = {"test_a", "test_b"} + + assert _missing_required_tests(required, {"test_a"}) == ["test_b"] + + +def test_required_execution_gate_rejects_skipped_or_unexecuted_cases(): + required = {"test_a", "test_b", "test_c"} + outcomes = {"test_a": "passed", "test_b": "skipped"} + + assert _required_test_failures(required, outcomes) == ["test_b", "test_c"] diff --git a/tests/unit/test_source_planning_assisted_classifier.py b/tests/unit/test_source_planning_assisted_classifier.py new file mode 100644 index 0000000..631202b --- /dev/null +++ b/tests/unit/test_source_planning_assisted_classifier.py @@ -0,0 +1,200 @@ +from __future__ import annotations + +import asyncio + +import pytest + +from groundworkers.base.errors import GroundworkersError +from groundworkers.services.source_planning.assisted import AssistedColumnRoleClassifier +from groundworkers.services.source_planning.classifier import ColumnRoleClassifier +from groundworkers.services.source_planning.models import ( + AnnotatedTable, + ColumnAnnotation, + ColumnRole, + RawTable, + SourceFormat, +) +from groundworkers.services.source_planning.normalisation import normalise_table +from groundworkers.services.source_planning.service import SourcePlanningService + + +class ResponseModel: + model = "fixture-model" + provider = "fixture" + + def __init__(self, response): + self.response = response + self.sync_calls = 0 + self.async_calls = 0 + self.prompt = "" + + def complete_structured(self, prompt, _schema, **_kwargs): + self.sync_calls += 1 + self.prompt = prompt + return self.response + + async def async_complete_structured(self, prompt, _schema, **_kwargs): + self.async_calls += 1 + self.prompt = prompt + return self.response + + +def _baseline() -> AnnotatedTable: + raw = RawTable( + name="measurements", + headers=["Code", "Label", "Mystery"], + rows=[{"Code": "E11.9", "Label": "Diabetes", "Mystery": "A"}], + sample_rows=[{"Code": "E11.9", "Label": "Diabetes", "Mystery": "A"}], + source_format=SourceFormat.CSV, + row_count=1, + ) + return ColumnRoleClassifier().classify(normalise_table(raw)) + + +def test_real_assisted_classifier_merges_only_candidates_and_recomputes_metadata(): + baseline = _baseline() + model = ResponseModel( + { + "decisions": [ + { + "header": "Mystery", + "role": "values", + "confidence": 0.95, + "packed_value": True, + "notes": "coded response", + }, + {"header": "Code", "role": "annotation", "confidence": 1.0}, + {"header": "Unknown", "role": "label", "confidence": 1.0}, + ] + } + ) + + result = AssistedColumnRoleClassifier(model).classify(baseline=baseline) + + assert model.sync_calls == 1 + assert "Candidate headers to review:\nMystery" in model.prompt + assert result.rows == baseline.rows + assert result.column_annotations["Code"] == baseline.column_annotations["Code"] + assert result.column_annotations["Mystery"].role is ColumnRole.values + assert result.column_annotations["Mystery"].detection_tier == "LLM" + assert result.packed_value_columns == ["Mystery"] + assert result.fallback_columns == ["Mystery"] + assert result.uncertain_columns == [] + assert result.llm_fallback_used is True + assert result.groundable_column_count == 3 + assert result.classification_tier_used == "LLM" + assert result.annotation_notes[-1].startswith("llm-assisted classification") + + +def test_async_classifier_uses_native_async_model_completion(): + baseline = _baseline() + model = ResponseModel({"decisions": [{"header": "Mystery", "role": "codes", "confidence": 0.9}]}) + + result = asyncio.run(AssistedColumnRoleClassifier(model).async_classify(baseline=baseline)) + + assert model.async_calls == 1 + assert model.sync_calls == 0 + assert result.column_annotations["Mystery"].role is ColumnRole.codes + + +def test_classifier_does_not_call_model_when_no_columns_need_review(): + normalised = normalise_table( + RawTable( + name="known", + headers=["Code"], + rows=[{"Code": "E11.9"}], + sample_rows=[{"Code": "E11.9"}], + source_format=SourceFormat.CSV, + row_count=1, + ) + ) + baseline = AnnotatedTable.from_normalised( + normalised, + column_annotations={"Code": ColumnAnnotation(role=ColumnRole.codes, detection_tier="A", confidence=1)}, + ) + model = ResponseModel({"decisions": []}) + + result = AssistedColumnRoleClassifier(model).classify(baseline=baseline) + + assert result is baseline + assert model.sync_calls == 0 + + +@pytest.mark.parametrize( + "response", + [None, {"decisions": [{"header": "Mystery", "role": "codes", "confidence": 2}]}, {"decisions": "bad"}], +) +def test_classifier_rejects_malformed_model_structure(response): + with pytest.raises(GroundworkersError) as excinfo: + AssistedColumnRoleClassifier(ResponseModel(response)).classify(baseline=_baseline()) + assert excinfo.value.code == "QUERY_ERROR" + + +def test_classifier_ignores_unrecognized_model_metadata_and_missing_decisions(): + baseline = _baseline() + model = ResponseModel( + { + "reasoning": "extra provider metadata is tolerated", + "decisions": [ + { + "header": "Mystery", + "role": "codes", + "confidence": 0.9, + "provider_trace": "ignored", + } + ], + } + ) + + result = AssistedColumnRoleClassifier(model).classify(baseline=baseline) + assert result.column_annotations["Mystery"].role is ColumnRole.codes + + empty = AssistedColumnRoleClassifier(ResponseModel({})).classify(baseline=baseline) + assert empty.column_annotations == baseline.column_annotations + + +def test_duplicate_decisions_use_the_last_valid_candidate_decision(): + model = ResponseModel( + { + "decisions": [ + {"header": "Mystery", "role": "label", "confidence": 0.85}, + {"header": "Mystery", "role": "codes", "confidence": 0.9}, + ] + } + ) + + result = AssistedColumnRoleClassifier(model).classify(baseline=_baseline()) + + assert result.column_annotations["Mystery"].role is ColumnRole.codes + + +def test_unknown_headers_and_roles_are_ignored_without_changing_inputs(): + baseline = _baseline() + model = ResponseModel( + { + "decisions": [ + {"header": "Mystery", "role": "not-a-role", "confidence": 0.9}, + {"header": "Not in the table", "role": "label", "confidence": 0.9}, + ] + } + ) + + result = AssistedColumnRoleClassifier(model).classify(baseline=baseline) + + assert result.rows == baseline.rows + assert "Mystery" not in result.column_annotations + assert set(result.column_annotations) == set(baseline.column_annotations) + + +def test_source_planning_service_surfaces_assisted_provider_failure(): + class FailedModel: + def complete_structured(self, *_args, **_kwargs): + raise RuntimeError("fixture provider failure") + + service = SourcePlanningService( + assisted_classifier=AssistedColumnRoleClassifier(FailedModel()), + ) + with pytest.raises(RuntimeError, match="fixture provider failure"): + service.plan_source_assisted( + "Mystery,Other\nA,B\n", filename="input.csv" + ) diff --git a/tests/unit/test_source_planning_document_formats.py b/tests/unit/test_source_planning_document_formats.py new file mode 100644 index 0000000..6572a0a --- /dev/null +++ b/tests/unit/test_source_planning_document_formats.py @@ -0,0 +1,200 @@ +from __future__ import annotations + +import builtins +import io + +import pytest + +from groundworkers.base.errors import GroundworkersError +from groundworkers.services.source_planning.decomposer import TableDecomposer +from groundworkers.services.source_planning.models import SourceFormat +from groundworkers.services.source_planning.service import SourcePlanningService + + +def _xlsx_bytes() -> bytes: + openpyxl = pytest.importorskip("openpyxl") + workbook = openpyxl.Workbook() + sheet = workbook.active + sheet.title = "conditions" + sheet["A1"] = "Example registry export" + sheet["A2"], sheet["B2"], sheet["C2"] = "Code", "Code", "Label" + sheet["A3"], sheet["B3"], sheet["C3"] = "E11.9", "E11.9-alt", "Diabetes" + sheet["A4"], sheet["B4"], sheet["C4"] = None, "F00", "Dementia" + sheet.merge_cells("A3:A4") + output = io.BytesIO() + workbook.save(output) + return output.getvalue() + + +def _docx_bytes(*, with_table: bool = True) -> bytes: + docx = pytest.importorskip("docx") + document = docx.Document() + if with_table: + table = document.add_table(rows=3, cols=2) + for row, values in zip(table.rows, (("Domain", "Code"), ("Condition", "E11.9"), ("", "F00"))): + for cell, value in zip(row.cells, values): + cell.text = value + table.cell(1, 0).merge(table.cell(2, 0)) + else: + document.add_heading("Conditions", level=1) + document.add_paragraph("E11.9 maps to diabetes.") + document.add_paragraph("F00 maps to dementia.") + output = io.BytesIO() + document.save(output) + return output.getvalue() + + +def _minimal_pdf(texts: tuple[str, ...], *, table: bool) -> bytes: + """Build a tiny PDF with stdlib only; table geometry is real PDF drawing data.""" + commands = [] + if table: + for y in (160, 135, 110): + commands.append(f"50 {y} m 250 {y} l S") + for x in (50, 150, 250): + commands.append(f"{x} 160 m {x} 110 l S") + positions = ((55, 143), (155, 143), (55, 118), (155, 118)) + cells = ("code", "label", "E11.9", "Diabetes") + for (x, y), value in zip(positions, cells): + commands.append(f"BT /F1 10 Tf {x} {y} Td ({value}) Tj ET") + else: + commands.extend(f"BT /F1 10 Tf 50 {160 - n * 16} Td ({value}) Tj ET" for n, value in enumerate(texts)) + stream = ("\n".join(commands)).encode("ascii") + objects = [ + b"<< /Type /Catalog /Pages 2 0 R >>", + b"<< /Type /Pages /Kids [3 0 R] /Count 1 >>", + b"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 300 200] /Resources << /Font << /F1 5 0 R >> >> /Contents 4 0 R >>", + b"<< /Length " + str(len(stream)).encode() + b" >>\nstream\n" + stream + b"\nendstream", + b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>", + ] + pdf = bytearray(b"%PDF-1.4\n") + offsets = [0] + for index, obj in enumerate(objects, start=1): + offsets.append(len(pdf)) + pdf.extend(f"{index} 0 obj\n".encode() + obj + b"\nendobj\n") + xref = len(pdf) + pdf.extend(f"xref\n0 {len(offsets)}\n0000000000 65535 f \n".encode()) + for offset in offsets[1:]: + pdf.extend(f"{offset:010d} 00000 n \n".encode()) + pdf.extend(f"trailer\n<< /Size {len(offsets)} /Root 1 0 R >>\nstartxref\n{xref}\n%%EOF\n".encode()) + return bytes(pdf) + + +@pytest.mark.required_format +def test_xlsx_real_workbook_removes_banner_fills_merges_and_deduplicates_headers(): + tables = TableDecomposer().decompose(_xlsx_bytes(), SourceFormat.XLSX, "conditions.xlsx") + + assert len(tables) == 1 + table = tables[0] + assert table.name == "conditions" + assert table.headers == ["Code", "Code_2", "Label"] + assert table.rows == [ + {"Code": "E11.9", "Code_2": "E11.9-alt", "Label": "Diabetes"}, + {"Code": "E11.9", "Code_2": "F00", "Label": "Dementia"}, + ] + assert table.metadata["banner_row_removed"] is True + plan = SourcePlanningService().plan_source(_xlsx_bytes(), filename="conditions.xlsx") + assert plan.plan.format_detected is SourceFormat.XLSX + assert plan.plan.tables[0].headers == table.headers + + +@pytest.mark.required_format +def test_docx_real_table_fills_vertical_merge_and_paragraph_fallback(): + decomposer = TableDecomposer() + table = decomposer.decompose(_docx_bytes(), SourceFormat.DOCX, "conditions.docx")[0] + fallback = decomposer.decompose( + _docx_bytes(with_table=False), SourceFormat.DOCX, "notes.docx" + )[0] + + assert table.headers == ["Domain", "Code"] + assert [row["Domain"] for row in table.rows] == ["Condition", "Condition"] + assert [row["Code"] for row in table.rows] == ["E11.9", "F00"] + assert fallback.metadata == {"docx_source": "paragraphs"} + assert fallback.rows == [ + {"section": "Conditions", "text": "E11.9 maps to diabetes. F00 maps to dementia."} + ] + planned = SourcePlanningService().plan_source(_docx_bytes(), filename="conditions.docx") + assert planned.plan.format_detected is SourceFormat.DOCX + assert planned.plan.tables[0].headers == table.headers + + +@pytest.mark.required_format +def test_pdf_real_table_and_text_fallback(): + pytest.importorskip("pdfplumber") + decomposer = TableDecomposer() + table = decomposer.decompose(_minimal_pdf((), table=True), SourceFormat.PDF, "codes.pdf")[0] + fallback = decomposer.decompose( + _minimal_pdf(("Conditions", "E11.9 is diabetes."), table=False), + SourceFormat.PDF, + "notes.pdf", + )[0] + + assert table.headers == ["code", "label"] + assert table.rows == [{"code": "E11.9", "label": "Diabetes"}] + assert fallback.metadata == {"pdf_source": "text"} + assert "E11.9 is diabetes." in fallback.rows[0]["text"] + planned = SourcePlanningService().plan_source( + _minimal_pdf((), table=True), filename="codes.pdf" + ) + assert planned.plan.format_detected is SourceFormat.PDF + assert planned.plan.tables[0].headers == table.headers + + +@pytest.mark.parametrize( + ("source_format", "content", "filename"), + [ + (SourceFormat.XLSX, b"not an xlsx", "x.xlsx"), + (SourceFormat.DOCX, b"not a docx", "x.docx"), + (SourceFormat.PDF, b"not a pdf", "x.pdf"), + ], +) +def test_corrupt_document_returns_no_tables(source_format, content, filename): + assert TableDecomposer().decompose(content, source_format, filename) == [] + + +@pytest.mark.parametrize( + ("source_format", "content", "filename"), + [ + (SourceFormat.XLSX, lambda: _empty_xlsx(), "empty.xlsx"), + (SourceFormat.DOCX, lambda: _empty_docx(), "empty.docx"), + (SourceFormat.PDF, lambda: _minimal_pdf((), table=False), "empty.pdf"), + ], +) +def test_empty_document_returns_no_tables(source_format, content, filename): + assert TableDecomposer().decompose(content(), source_format, filename) == [] + + +def _empty_xlsx() -> bytes: + openpyxl = pytest.importorskip("openpyxl") + output = io.BytesIO() + openpyxl.Workbook().save(output) + return output.getvalue() + + +def _empty_docx() -> bytes: + docx = pytest.importorskip("docx") + output = io.BytesIO() + docx.Document().save(output) + return output.getvalue() + + +@pytest.mark.parametrize( + ("source_format", "module_name", "hint"), + [ + (SourceFormat.XLSX, "openpyxl", "groundworkers[xlsx]"), + (SourceFormat.DOCX, "docx", "groundworkers[docx]"), + (SourceFormat.PDF, "pdfplumber", "groundworkers[pdf]"), + ], +) +def test_missing_optional_parser_reports_install_hint(monkeypatch, source_format, module_name, hint): + original_import = builtins.__import__ + + def missing_parser(name, *args, **kwargs): + if name == module_name: + raise ImportError(f"blocked {module_name} for contract test") + return original_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", missing_parser) + with pytest.raises(GroundworkersError) as excinfo: + TableDecomposer().decompose(b"x", source_format, "x") + assert excinfo.value.code == "MISSING_DEPENDENCY" + assert hint in excinfo.value.message diff --git a/tests/unit/test_source_planning_tools.py b/tests/unit/test_source_planning_tools.py index b488e9f..718ad1c 100644 --- a/tests/unit/test_source_planning_tools.py +++ b/tests/unit/test_source_planning_tools.py @@ -2,6 +2,7 @@ from groundworkers.base.server import GroundworkersMCPServer from groundworkers.services.source_planning import SourcePlanningService +from groundworkers.services.source_planning.assisted import AssistedColumnRoleClassifier from groundworkers.tools.source_planning_tools import ( register_source_planning_resources, register_source_planning_tools, @@ -69,6 +70,41 @@ def test_source_plan_assisted_returns_backend_unavail_without_llm() -> None: assert result["code"] == "BACKEND_UNAVAIL" +def test_source_plan_assisted_runs_real_service_and_classifier() -> None: + class FixedModel: + def complete_structured(self, _prompt, _schema, **_kwargs): + return { + "decisions": [ + {"header": "unknown_a", "role": "values", "confidence": 0.93} + ] + } + + async def async_complete_structured(self, _prompt, _schema, **_kwargs): + return { + "decisions": [ + {"header": "unknown_a", "role": "values", "confidence": 0.93} + ] + } + + service = SourcePlanningService( + assisted_classifier=AssistedColumnRoleClassifier(FixedModel()) + ) + server = GroundworkersMCPServer("assisted") + register_source_planning_tools(server, service) + + result = server.call( + "source_plan_assisted", + content="unknown_a,unknown_b\nA,B\n", + filename="diagnoses.csv", + include_intermediate=True, + ) + + assert result["llm_tier_used"] is True + annotation = result["annotated_tables"][0]["column_annotations"]["unknown_a"] + assert annotation["role"] == "values" + assert annotation["detection_tier"] == "LLM" + + def test_source_planning_resources_are_registered_and_json_readable() -> None: server = _server() diff --git a/tests/unit/test_tui_app.py b/tests/unit/test_tui_app.py index a63fb66..d3330cc 100644 --- a/tests/unit/test_tui_app.py +++ b/tests/unit/test_tui_app.py @@ -31,8 +31,9 @@ def _snapshot_for(path: str): return load_configuration(config_path=path) -def test_groundworkers_spec_keeps_setup_as_a_registered_page() -> None: +def test_groundworkers_spec_keeps_setup_as_a_registered_page(monkeypatch) -> None: pytest.importorskip("groundskeeping") + monkeypatch.setattr("groundworkers.plugins.discover_plugins", lambda: []) from groundworkers.tui.app import build_groundworkers_tui_spec @@ -86,6 +87,16 @@ def verify_readiness(self, state): assert view.status.value == "warning" assert "prerequisites" in view.message assert tuple(action.label for action in view.actions) == ("Configure", "Verify") + from groundskeeping.app import OperatorApp + + async def run_check() -> None: + app = OperatorApp(spec) + async with app.run_test(size=(120, 40)) as pilot: + await pilot.pause() + tabs = app.query_one("#page-tabs") + assert tabs.styles.display != "none" + + asyncio.run(run_check()) def test_plugin_readiness_renders_generic_table_and_details( @@ -175,13 +186,14 @@ def test_plugin_setup_wizard_exposes_protocol_spec_without_assignment_error( assert controller.start().spec.title == "Example setup" -def test_groundworkers_pages_do_not_cover_workbench(tmp_path: Path) -> None: +def test_groundworkers_pages_do_not_cover_workbench(tmp_path: Path, monkeypatch) -> None: """Layout only. Uses a real config so the page renders its sections. A missing config now opens the location wizard on activate, which is a different journey and is covered by its own tests. """ pytest.importorskip("groundskeeping") + monkeypatch.setattr("groundworkers.plugins.discover_plugins", lambda: []) from groundskeeping.app import OperatorApp From 3c78c4192fcc6bbc08f9f1570e3b26bf1195fcb4 Mon Sep 17 00:00:00 2001 From: Georgie Kennedy Date: Wed, 14 Oct 2026 13:41:29 +1100 Subject: [PATCH 2/3] lazy import of groundskeeping --- src/groundworkers/plugins.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/groundworkers/plugins.py b/src/groundworkers/plugins.py index b9950eb..49d2fc3 100644 --- a/src/groundworkers/plugins.py +++ b/src/groundworkers/plugins.py @@ -20,7 +20,6 @@ from importlib.metadata import entry_points from typing import TYPE_CHECKING, Any, ClassVar, Protocol, runtime_checkable -from groundskeeping.contracts import CommandPlan from oa_configurator import ( # type: ignore[import-untyped] PackageConfigBase, ResolvedDatabase, @@ -36,6 +35,7 @@ ConfigWorkflowSpec, MutationOperation, ) + from groundskeeping.contracts import CommandPlan from groundworkers.services.grounding import ConceptGroundingService from groundworkers.services.mapping import MappingService From 3b141513b8c3b5db102ec7b904c2a6514b4a655b Mon Sep 17 00:00:00 2001 From: Georgie Kennedy Date: Wed, 14 Oct 2026 13:44:14 +1100 Subject: [PATCH 3/3] wheel build test update --- tests/support/wheel_smoke.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tests/support/wheel_smoke.py b/tests/support/wheel_smoke.py index 9c16279..8084da6 100644 --- a/tests/support/wheel_smoke.py +++ b/tests/support/wheel_smoke.py @@ -50,6 +50,9 @@ class FixedModel: def complete_structured(self, _prompt, _schema, **_kwargs): return {"decisions": [{"header": "unknown_a", "role": "values", "confidence": 0.9}]} + async def async_complete_structured(self, _prompt, _schema, **_kwargs): + return self.complete_structured(_prompt, _schema, **_kwargs) + service = SourcePlanningService( assisted_classifier=AssistedColumnRoleClassifier(FixedModel()) )