diff --git a/.github/workflows/iac-tests.yml b/.github/workflows/iac-tests.yml index 81215c4a..1e9e2c27 100644 --- a/.github/workflows/iac-tests.yml +++ b/.github/workflows/iac-tests.yml @@ -214,6 +214,7 @@ jobs: tests/iac/test_iac_databricks_fake_e2e.py tests/iac/test_iac_databricks_jobs_fake_e2e.py tests/iac/test_iac_databricks_genie_fake_e2e.py + tests/iac/test_iac_databricks_genie_estate_fake_e2e.py tests/iac/test_iac_databricks_metrics_fake_e2e.py pytest-args: "-p no:randomly" junit: databricks-fake.xml @@ -221,6 +222,7 @@ jobs: Databricks fake tier=tests/iac/test_iac_databricks_fake_e2e.py Databricks jobs fake tier=tests/iac/test_iac_databricks_jobs_fake_e2e.py Databricks Genie fake tier=tests/iac/test_iac_databricks_genie_fake_e2e.py + Databricks Genie estate fake tier=tests/iac/test_iac_databricks_genie_estate_fake_e2e.py Databricks metric views fake tier=tests/iac/test_iac_databricks_metrics_fake_e2e.py steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 diff --git a/.secrets.baseline b/.secrets.baseline index 5275b5d0..e91b3130 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -139,7 +139,7 @@ "filename": ".github/workflows/iac-tests.yml", "hashed_secret": "a94a8fe5ccb19ba61c4c0873d391e987982fbbd3", "is_verified": false, - "line_number": 350 + "line_number": 352 } ], ".github/workflows/integration.yml": [ @@ -157,7 +157,7 @@ "filename": "AGENTS.md", "hashed_secret": "9d4e1e23bd5b727046a9e3b4b7db57bd8d6ee684", "is_verified": false, - "line_number": 654 + "line_number": 675 } ], "Jenkinsfile": [ @@ -2317,5 +2317,5 @@ } ] }, - "generated_at": "2026-11-12T23:34:40Z" + "generated_at": "2026-11-12T23:57:29Z" } diff --git a/AGENTS.md b/AGENTS.md index d5663feb..9abd0b20 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -381,6 +381,27 @@ changing `parentPath`, leaves the space in the old folder with a note naming it. `databricks_genie_space` action, and `fluid generate iac` notes that its module holds none. Code: `fluid_build/iac/providers/databricks_genie.py`. +### One Genie space over several products (an estate) + +```bash +fluid genie apply tower-management/ --estate tower-management --title "Tower Management" +fluid genie remove --estate tower-management +``` + +`fluid genie apply` takes contract files, globs or folders (searched for `*.fluid.yaml`), and +`--env` and `--dry-run` as `fluid apply` does; each contract passes the same schema and +version gate. Apply the products first: the run checks every table exists and names the +product of one that does not. Genie adds a join of its own for each Unity Catalog foreign key +that touches a table entering a space; the run reads each write back and takes those out, so +a second run changes nothing. The space holds each product's tables +as a product space does, and a join for every foreign key the contracts declare between them +(a column tagged `foreign-key` whose `labels.references` is `.`), the same +keys `fluid apply` declares in Unity Catalog (`databricks_keys.py`). `CAN RUN` goes only to the +principals that read every table of the estate, because the space describes them all. The space +is found again by `[fluid-genie-estate:]` at the end of its description, in the deploying +identity's own folder, and converges to exactly the contracts given. Code: +`fluid_build/cli/genie.py`, `fluid_build/iac/providers/databricks_genie_estate.py`. + --- ## For AI-Assisted Project Creation (Forge) diff --git a/CHANGELOG.md b/CHANGELOG.md index ac26f76e..4c415d5a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,32 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **`fluid genie apply`: one Databricks Genie space over several data products (an estate).** + `fluid genie apply --estate ` creates or updates one AI/BI + Genie space over the Databricks tables of every contract given, so a question can join the + products: `fluid genie apply tower-management/ --estate tower-management`. Each table carries + what a product's own space says of it (column descriptions, allowed values, synonyms, example + questions, one instruction section from its semantics, metrics and `agentPolicy`), and the space + gets a join for every foreign key the contracts declare between its tables: a column tagged + `foreign-key` whose `labels.references` is `.`, resolved as `fluid apply` + resolves it for Unity Catalog. A join is MANY_TO_ONE from the referencing table, ONE_TO_ONE + when the foreign key is that table's own primary key, a composite key is one join, and a key + to its own table joins a second alias. A reference to a product the estate does not hold is + said and left out. `CAN RUN` goes only to the principals that read every table of the estate, + since the space describes them all; the others are named. The products are applied first: the + run checks that every table exists before it writes and names the product of one that does + not. Like a product's space it is found again by a marker, `[fluid-genie-estate:]`, in the + deploying identity's own folder only, changes nothing on a second run, and converges to + exactly the contracts given. `--title`, `--env` (contract overlays) and `--dry-run`; + `fluid genie remove --estate ` trashes it. Each contract passes the schema and version + gate `fluid apply` runs, and an unreadable one is `contract_load_failed`, naming the file. A + product whose `agentPolicy` rules out question answering refuses the estate, and so do more + than 50 tables, Genie's cap (measured). Whenever a table enters a space, Genie adds a join of + its own for each Unity Catalog foreign key that touches it (measured): a run reads each + write back, takes those duplicates out, and says so, so the next run finds no change. This + holds for a product's own space too. A product space's update now names a table that + leaves the space as a table, not as a metric view. + - **A Databricks AI/BI Genie space for a data product.** With `extensions.databricksGenie: {}` (or `{title: ...}`) in a contract with Databricks table exposes, `fluid apply` creates the product's Genie space once its tables, masks and row filters are in place. The space holds @@ -28,7 +54,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 with `parentPath` (below): a space another identity made in its own folder is left alone. `fluid apply --mode dry-run` prints what would change. The space is made through the Genie REST API (the `[databricks]` extra), since the databricks/databricks provider has no Genie - space resource. One space per product; a space over several products is a follow-up. + space resource. One space per product; `fluid genie apply` makes one over several products. - **One Genie space for a product deployed from more than one identity.** `extensions.databricksGenie.parentPath` names a workspace folder, such as `/genie/`. diff --git a/HONESTLY_TESTED.md b/HONESTLY_TESTED.md index 40c314c7..c3f29fa2 100644 --- a/HONESTLY_TESTED.md +++ b/HONESTLY_TESTED.md @@ -102,6 +102,9 @@ Markers declared in `pyproject.toml`. | Stage 1 real tofu, stand-in workspace: Genie space | `tests/iac/test_iac_databricks_genie_fake_e2e.py` | ✅ 4. The real `fluid` CLI and `tofu`: validate, dry-run, apply (the space after the tables, shared with the readers), a second apply that changes nothing, the block removed (trashed); a Genie failure fails the apply after the tables with its fix and the next apply converges; an invalid block refused before `tofu`; `parentPath`: a missing folder stops the apply before `tofu` writes a table, then the space made in the folder and a second identity's apply changing nothing | The stand-in is modelled on the measured API, not the service itself | | Manual live, Free Edition, Genie space, 2026-11-12 | the real `fluid` CLI, no test file: branch `feat/databricks-genie-space`, databricks-sdk 0.162.0, a copy of industry-models tower-management `dp019_site_360` in the scratch catalog `fluid_iactest_genie` (no dbt build, no MENA sovereignty block: the workspace is in us-west-2), 6 synthetic rows | ✅ dry-run (would create, would share), apply `+73` then the space created and shared `CAN_RUN` with two persona service principals, a second apply `+0 ~0 -0` and "no change", the question "Which on-air sites in Amman are colocation-ready with hybrid power and no standing risk?" answered through the Genie Conversation API with the 2 expected sites (its SQL took the latest `snapshot_date` and the inverse of the `at_risk_site_count` filter from the instruction), a metric view created by hand picked up ("updated: metric views", then "no change"), the block removed (trashed). Two defects found and fixed in the run: the metric-view read failed on a schema not created yet, and the service keeps a metric view among the tables. A re-check after the security review (the plugin's own hook over `samples.nyctaxi.trips`, nothing created in Unity Catalog) found the list shows a new space only ~9 s later (a run straight after made a second space) and that a taken title is suffixed on create and 409 on PATCH; fixed, then create, an immediate second run ("no change", same space), and the block removed (trashed, via the folder check) passed live. Swept: the space trashed, `DROP CATALOG fluid_iactest_genie CASCADE`, 0 left | A reader without `SELECT` asking (no persona credential here; the access model is Genie's documented one); a service principal as the deploying identity (its folder is `/Users/`, the one Databricks Asset Bundles build from `current_user.userName`; a create that lands anywhere else is trashed and stops the apply); a space over several products | | Manual live, Free Edition, Genie `parentPath`, 2026-11-12 | the real `fluid` CLI, no test file: branch `feat/databricks-genie-shared-folder` on main 97811add, databricks-sdk 0.162.0, a one-table product in the scratch catalog `fluid_iactest_genie_pp`, its folder `/fluid_iactest_genie_pp` at the workspace root | ✅ A probe measured the API first. A create accepts `/Users/`, `/Workspace/Users/` (accepted this time; an earlier probe that day had a 400), `/Shared/x`, `/Workspace/Shared/x`, a trailing `/` and a root-level folder, and a GET returns the bare form. A missing folder is 404, a relative path 400, and `/shared/x` 404 (paths are case-sensitive). `/Shared` grants `users` `CAN_MANAGE` directly, and every folder and space under it inherits that; the root grants only admins. A space's permissions all come from its folder, and `PATCH {parent_path}` moves it. Then through the CLI: validate refuses `/Workspace/Shared/...`; an apply with the folder missing stops after `tofu plan`, before `tofu apply` (no schema made); a space made without `parentPath` in the user's folder, a dry run says it would move, the apply moves it (same id), and a second apply changes nothing; `users` with `CAN_MANAGE` on the folder is refused before `tofu apply`, and `users` with `CAN_RUN` is a note; `fluid plan` lists `genie_space_`, `generate iac` prints the note, and `diff --exit-on-drift` exits 0 with the space among `added`; with the block removed, the folder's space is left with a note. Swept: the space trashed, the folder deleted, the catalog dropped, 0 left | A second identity, live (no service-principal secret here; the stand-in e2e covers it); a folder a non-admin made inside someone's home | +| Stage 1 unit: Databricks Genie estate (`fluid genie apply`) | `tests/iac/test_iac_databricks_genie_estate.py` (the compile, `converge_estate` and `remove_estate` against the stand-in, `fluid genie` through the real entry point) | ✅ 40. Every product's table in one space, sorted; a join per declared foreign key between them (MANY_TO_ONE from the referencing table, ONE_TO_ONE when it is that table's primary key, one join for a composite key, a second alias for a key to its own table, every column qualified, 32-hex ids sorted); the questions once each; one instruction with a section per product; no measure snippets; CAN_RUN only for the readers of every table, the others named; a reference outside the estate, to a keyless product or to no single primary key said, and a nullable key column or a garbled reference said as `fluid apply` says them; the refusals (an id the marker cannot carry, no contract, no id, a product twice, no Databricks table, a table in two products, an `agentPolicy` against Genie, more than 50 tables, a text past 25,000 characters, a refusal every contract shares said once); create and share, then no change; the joins Genie adds from Unity Catalog keys on a create and on an update that adds a table taken out at once; a product left out updates the space; a table not applied yet named with its product before any write; dry run; remove trashes only the estate's space and a failed one names `fluid genie remove`; a product space and an estate space never touch; two marker copies stop the run; a planted copy left alone; the estate's own command in its messages and the product's wording kept; the stand-in refusing an unknown relationship tag and a 51st table; contract globs (one level) and folders once each, past no symlink or ignored folder; a schema-invalid, pre-0.7 or unreadable contract a typed error naming its file | Genie's answers (only the live row asks one) | +| Stage 1 real tofu, stand-in workspace: Genie estate | `tests/iac/test_iac_databricks_genie_estate_fake_e2e.py` | ✅ 1. Two products applied with the real `fluid apply` and `tofu` (their tables and the Unity Catalog key between them); `fluid genie apply` before the second exists refused, naming it; dry run; the space with the join from the declared foreign key, shared with the reader of both tables only; a second run changes nothing; `fluid genie remove` | The stand-in is modelled on the measured API, not the service itself | +| Manual live, Free Edition, Genie estate, 2026-11-12/13 | the real `fluid` CLI, no test file: branch `feat/databricks-genie-estate`, databricks-sdk 0.162.0, the 19 tower-management products as fluid-demo-lab-run's `databricks_only_forge` track derives them (overlays to us-west-2 and the persona principals; builds left out; the analysts taken off 3 restricted products), each applied with `fluid apply` into the scratch catalog `fluid_iactest_estate` (19 runs, rc 0, the Unity Catalog keys declared), 7 synthetic rows | ✅ dry run (19 tables, 57 sample questions, 24 joins; CAN_RUN for the steward only, the analyst named, 16 of 19), create and share, a question through the Genie Conversation API ("Which site name has the most unbilled kWh?": Basra Port, 1500, by a JOIN of `tenant_billing` to `site_master` on `site_id`), the next run "no change", a product left out (updated: description, sample questions, tables, instructions, joins) and put back, remove, and remove again (nothing to trash). Found and fixed in the run: whenever a table enters a space Genie adds a join of its own for each Unity Catalog foreign key that touches it (24 sent, 48 held; 1 more when the product came back), so the second run updated; each write is now read back and set back. Probes, each space trashed at once: join_specs kept as sent by an update with the same tables, a self-join under two aliases and ONE_TO_ONE taken, an unknown relationship tag refused, 51 tables refused ("A space can have at most 50 tables."), 60 sample questions and a 60,000-character instruction taken. Swept: `DROP CATALOG fluid_iactest_estate CASCADE`; 0 `fluid_iactest_*` catalogs and 0 such spaces left | A reader without SELECT asking (no persona credential here); a service principal as the deploying identity; Genie's answers over more questions; an estate of more than 19 tables | | Stage 2 emulator (nightly, non-gating) | `tests/iac/test_iac_databricks_minilake.py` (job `nightly-databricks-minilake`) | ⚠️ 1 passed, 4 strict xfails (manual dispatch, run 42267640334, 2026-11-06, minilake v1.9.0). Schema, table and grant creates work, the replace gate holds, adding a column plans no replace, and a grant revoke is classified not gated. The four tests about those resources apply the module without its tag assignments (the `untagged` fixture). An xfail is satisfied only by its own gap's failure (`MinilakeGap`), and a pass fails the run, so an image that closes a gap turns the job red until the mark comes off. Earlier: 1/4 from the first run (2026-10-30) to the 2026-11-04 nightly, then 0/4 after #831, every apply stopping at the first tag. | A second plan is not empty and no in-place table update applies: minilake drops a CREATE TABLE's `TBLPROPERTIES` and `COMMENT`, and DuckDB cannot parse the `ALTER TABLE ... SET TBLPROPERTIES` the provider sends on every update of a table that sets properties. No tag is applied: minilake has no entity-tag-assignments API (it answers 501). v1.9.3 (2026-11-05) has both gaps, so the pin stays; an upstream issue for each is drafted, not filed. Jobs and workspace files are not in this flow yet. | | Stage 1 unit: `fluid test` on Databricks, results table, dashboard | `tests/providers/test_databricks_validation.py`, `tests/providers/test_databricks_statements.py`, `tests/providers/test_quality_engine_column_rules.py`, `tests/cli/test_contract_validation_schema_rules.py`, `tests/cli/test_fluid_test_results_table.py`, `tests/cli/test_fluid_test_dashboard.py` | ✅ Against a stubbed `api_client.do`: the Unity Catalog columns and the contract's types through the emitter's type table (`VARCHAR` is `string`), a missing table (404 → `missing_resource`, and every rule reported as not run with no statement), the row count, the warehouse check; the statements over `databricks_enforce.SqlRunner` with `:name` parameters, polling, timeout cancel, chunks, type conversion and scrubbed errors; every rule kind in the Databricks dialect (backticked identifiers, `\'` literals, `timestampdiff` freshness, `RLIKE`), comma-listed columns, values from the column's enum, regex or range rule, references resolved by the `fluid_product_id` tag (same catalog first, ambiguity refused, cached); `schema` rules from the column comparison; the result rows (one per JUnit case, `rule_type` `structure` for the structural ones, expose version, UTC run time), the batched parameterised `INSERT`, the refusals (bad name, a product's own schema, a table without the columns), the fix-its; the dashboard definition's structure and the create, unchanged, re-publish, update-with-etag and create-race paths. | What Databricks accepts: the manual live row below. | | Manual live, Free Edition (founder-approved), 2026-11-12: `fluid test` results | the real `fluid` CLI, no test file: `fluid test --env databricks --output junit --results-table fluid_iactest_dqr.dq.fluid_test_results --results-dashboard '/Workspace/Shared/fluid_iactest_dqr/Data quality'` on the 19 tower-management products in `northwind_towers` (read only: `SELECT` and metadata reads), contracts derived from industry-models main | ✅ 19 runs, 17-34 s each, 18 exit 0; `dp019_site_360` exits 1 because its table is not deployed in the demo workspace (`missing_resource`; its rules are reported as not run, and no statement is sent). 168 rule cases: the 155 on deployed tables all ran (146 pass, 9 freshness rules fail at `warn`: the bronze data is hours old against a 15-minute window), every rule kind the pack uses among them: enum, regex and range values (#4's range rule over five columns included), 23 references (3 of them bronze references outside `consumes[]`, resolved by the `fluid_product_id` tag), `schema` rules and comma-listed completeness rules. 396 rows appended, one per JUnit case; the dashboard was created by the first run, found unchanged by every later one, and updated (PATCH, re-publish) once when its definition changed; in a browser it shows the counters (88.1% of rules passing, 19 products, 20 failing rules at that point), the per-product bars by layer, the trend and the failing-checks table. Afterwards the catalog `fluid_iactest_dqr` was dropped CASCADE and the dashboard deleted. | Lakeview validates widget specs only when the dashboard renders, so a widget shape Databricks changes would show as "Invalid widget definition" without failing the command (the table widget did, until it carried every column key). | diff --git a/fluid_build/cli/bootstrap.py b/fluid_build/cli/bootstrap.py index 986511ec..0230928a 100644 --- a/fluid_build/cli/bootstrap.py +++ b/fluid_build/cli/bootstrap.py @@ -647,6 +647,11 @@ def _cmd(name: str, help_text: Any, *aliases: str) -> Tuple[str, Any, Tuple[str, CoreCommand( "publish", "publish", (_cmd("publish", "Publish data products to enterprise catalogs"),) ), + CoreCommand( + "genie", + "genie", + (_cmd("genie", "Databricks Genie space over several data products (apply · remove)"),), + ), CoreCommand( "bundle", "bundle", diff --git a/fluid_build/cli/genie.py b/fluid_build/cli/genie.py new file mode 100644 index 00000000..4348b9a6 --- /dev/null +++ b/fluid_build/cli/genie.py @@ -0,0 +1,310 @@ +# Copyright 2025 Agentics Transformation Limited +# SPDX-License-Identifier: Apache-2.0 + +"""``fluid genie {apply,remove}``: one Databricks AI/BI Genie space over several data products. + +``fluid genie apply --estate `` creates or updates the +estate's space, over the Databricks tables of every contract given, with joins from the +foreign keys they declare (``iac/providers/databricks_genie_estate.py``). ``fluid genie remove +--estate `` trashes it. A product's own space comes from ``fluid apply`` and +``extensions.databricksGenie`` instead (``databricks_genie.py``). + +Light at import: only the parser. The provider modules load when a subcommand runs, so the +``fluid --help`` cold path never imports them. +""" + +from __future__ import annotations + +import argparse +import logging +from pathlib import Path +from typing import Any, Dict, List, Mapping, Sequence, Tuple + +COMMAND = "genie" +#: The file names a folder argument is searched for. +CONTRACT_SUFFIXES = (".fluid.yaml", ".fluid.yml", ".fluid.json") +_GLOB_CHARS = frozenset("*?[") + +_DESCRIPTION = """\ +One Databricks AI/BI Genie space over several data products (an estate). + +The space holds the Databricks tables of every contract given, with what each contract says +of its columns, its example questions and one instruction, and a join for every foreign key +the contracts declare between those tables (a column tagged foreign-key whose +labels.references is .). Apply the products first (fluid apply): the +space opens on tables that exist. Only the principals that read every table of the estate +are given CAN RUN on it. +""" + + +def register(subparsers: argparse._SubParsersAction) -> None: + """Register ``fluid genie {apply,remove}``.""" + p = subparsers.add_parser( + COMMAND, + help="Databricks Genie space over several data products (apply · remove)", + description=_DESCRIPTION, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + sub = p.add_subparsers(dest="genie_command", required=True) + + apply = sub.add_parser( + "apply", + help="Create or update the estate's Genie space", + description=_DESCRIPTION + + "\nExample:\n fluid genie apply tower-management/ --estate tower-management " + '--title "Tower Management"\n', + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + apply.add_argument( + "contracts", + nargs="+", + metavar="CONTRACT", + help="contract files, glob patterns, or folders (searched for *.fluid.yaml at any depth)", + ) + _estate_argument(apply) + apply.add_argument("--title", default="", help="the space's title (default: the estate id)") + apply.add_argument( + "--env", + default=None, + help="the overlay environment to read each contract with, as for fluid apply --env", + ) + apply.add_argument( + "--dry-run", + action="store_true", + help="read the workspace and print what would change; write nothing", + ) + apply.set_defaults(func=run_apply) + + remove = sub.add_parser( + "remove", + help="Trash the estate's Genie space", + description="Trash the estate's Genie space: the one an apply with this --estate made " + "from this identity.", + ) + _estate_argument(remove) + remove.add_argument( + "--dry-run", action="store_true", help="print what would be trashed; trash nothing" + ) + remove.set_defaults(func=run_remove) + + +def _estate_argument(parser: argparse.ArgumentParser) -> None: + parser.add_argument( + "--estate", + required=True, + metavar="ID", + help="the estate's id (letters, digits, '.', '_', '-'): the next run finds the space " + "by it", + ) + + +def run_apply(args: argparse.Namespace, logger: logging.Logger) -> int: + """``fluid genie apply``: compile the estate from the contracts, then converge its space.""" + from fluid_build.cli._logging import info + from fluid_build.cli.console import cprint_literal + from fluid_build.iac.base import UnsupportedBindingError + from fluid_build.iac.providers.databricks import genie_estate + from fluid_build.iac.providers.databricks_genie_estate import converge_estate + + # The loader says "overlay_applied" for each contract it reads (``fluid.loader``), which for + # a pack is one line per product; the header names the overlay once instead. A missing + # overlay still warns, and --log-level DEBUG shows every line. + loader_log = logging.getLogger("fluid.loader") + level = loader_log.level + if logger.getEffectiveLevel() > logging.DEBUG: + loader_log.setLevel(logging.WARNING) + try: + contracts: List[Tuple[str, Mapping[str, Any]]] = [ + (label, _contract(label, path, args.env, logger)) + for label, path in contract_paths(args.contracts) + ] + finally: + loader_log.setLevel(level) + estate = genie_estate(args.estate, contracts, title=args.title) + say = cprint_literal + header = f"Genie estate {args.estate}: {len(contracts)} contract(s)" + if args.env: + header += f" (--env {args.env})" + if estate.space is not None: + header += f", {len(estate.owners)} table(s), {estate.joins} join(s)" + say(header) + for warning in estate.warnings: + say(f" note: {warning}") + for refusal in estate.refusals[1:]: + say(f" also refused: {refusal} {' '.join(refusal.remediation)}".rstrip()) + try: + converge_estate(estate, say, dry_run=args.dry_run) + except UnsupportedBindingError as exc: + raise _cli_error(exc) from exc + info( + logger, + "genie_estate_dry_run" if args.dry_run else "genie_estate_applied", + estate=args.estate, + products=len(contracts), + tables=len(estate.owners), + joins=estate.joins, + ) + return 0 + + +def run_remove(args: argparse.Namespace, logger: logging.Logger) -> int: + """``fluid genie remove``: trash the estate's space.""" + from fluid_build.cli._logging import info + from fluid_build.cli.console import cprint_literal + from fluid_build.iac.base import UnsupportedBindingError + from fluid_build.iac.providers.databricks_genie_estate import remove_estate + + try: + trashed = remove_estate(args.estate, cprint_literal, dry_run=args.dry_run) + except UnsupportedBindingError as exc: + raise _cli_error(exc) from exc + info( + logger, + "genie_estate_remove_dry_run" if args.dry_run else "genie_estate_removed", + estate=args.estate, + spaces=trashed, + ) + return 0 + + +def _contract(label: str, path: Path, env: Any, logger: logging.Logger) -> Mapping[str, Any]: + """One contract, read with its overlay and gated as ``fluid apply`` gates one (a pre-0.7 + ``fluidVersion`` or a schema-invalid document stops the run); a failure names the file.""" + from fluid_build._errors import FluidUserError + from fluid_build.cli._common import CLIError, load_contract_with_overlay + from fluid_build.cli._common import resolve_env_templates_in_contract as resolve + from fluid_build.cli.plan import _gate_contract_for_plan_or_apply + + try: + contract = resolve(load_contract_with_overlay(str(path), env, logger)) + except (CLIError, FluidUserError): + raise + except Exception as exc: # noqa: BLE001 — a bad file is a typed error, as in fluid apply + raise CLIError( + 1, "contract_load_failed", {"path": label, "env": env, "error": str(exc)} + ) from exc + try: + _gate_contract_for_plan_or_apply(contract, logger, command="genie") + except CLIError as exc: + exc.context.setdefault("path", label) + raise + return contract + + +def contract_paths(items: Sequence[str]) -> List[Tuple[str, Path]]: + """``(label, path)`` per contract: each item a file, a glob pattern (expanded as ``fluid + publish`` expands one, ``**`` a single level) or a folder searched for + :data:`CONTRACT_SUFFIXES` at any depth, in that order, each file once. A folder or + a glob reaches no symlink and no ignored or build-output folder (``.git``, ``node_modules``, + ``.venv``, ``target``, ...) below where it starts; a file named outright is read as named. + Every path passes the CLI's path validator (``cli/security.py``) before it is read.""" + from fluid_build.cli._common import CLIError + from fluid_build.cli.security import validate_cli_path + from fluid_build.util.upstream_discovery import BUILD_OUTPUT_DIRS, IGNORED_DIRS + + skip = IGNORED_DIRS | BUILD_OUTPUT_DIRS + found: List[Path] = [] + for item in items: + for match in _matches(item, skip): + found += _folder_contracts(match, skip) if Path(match).is_dir() else [Path(match)] + missing = [str(p) for p in found if not p.is_file()] + if missing: + raise CLIError(1, "contract_file_not_found", {"path": ", ".join(missing)}) + unique: Dict[Path, Path] = {} + for path in found: + unique.setdefault(path.resolve(), path) + return [ + (str(given), validate_cli_path(given, mode="read", file_type="contract")) + for given in unique.values() + ] + + +def _matches(item: str, skip: frozenset) -> List[str]: + """What one argument names: itself, or what its glob matches that a walk would reach.""" + from glob import glob + + from fluid_build.cli._common import CLIError + + if _GLOB_CHARS & set(item): + base = _glob_base(item) + # As ``fluid publish`` expands one: ``**`` is one folder level, never a walk that + # follows symlinked folders (two symlink loops made a recursive glob run forever). + # A folder argument is the recursive search. + matches = [m for m in sorted(glob(item)) if _reachable(Path(m), base, skip)] + else: + matches = [item] + if not matches: + raise CLIError(1, "contract_file_not_found", {"path": item}) + return matches + + +def _folder_contracts(folder: str, skip: frozenset) -> List[Path]: + """The contract files a folder argument holds; none is an error that names it.""" + from fluid_build.cli._common import CLIError + + inside = _contracts_in(Path(folder), skip) + if not inside: + raise CLIError( + 1, + "genie_estate_no_contracts", + {"path": folder, "error": f"{folder} holds no {' / '.join(CONTRACT_SUFFIXES)} file"}, + ) + return inside + + +def _glob_base(pattern: str) -> Path: + """The folder a glob pattern starts from: its parts before the first with a wildcard.""" + plain: List[str] = [] + for part in Path(pattern).parts: + if _GLOB_CHARS & set(part): + break + plain.append(part) + return Path(*plain) if plain else Path(".") + + +def _reachable(path: Path, base: Path, skip: frozenset) -> bool: + """Whether a glob's match is one a folder walk from ``base`` reaches: no symlink on the way + and no ignored folder below ``base``.""" + try: + below = path.relative_to(base).parts + except ValueError: + below = path.parts + current = base + for part in below: + current = current / part + if current.is_symlink(): + return False + folders = below if path.is_dir() else below[:-1] + return not any(part in skip for part in folders) + + +def _contracts_in(folder: Path, skip: frozenset) -> List[Path]: + """The contract files under ``folder``, in name order, past no symlink and into no + ignored or build-output folder.""" + out: List[Path] = [] + try: + entries = sorted(folder.iterdir()) + except OSError: + return out + for entry in entries: + if entry.is_symlink(): + continue + if entry.is_dir(): + if entry.name not in skip: + out += _contracts_in(entry, skip) + elif entry.is_file() and entry.name.endswith(CONTRACT_SUFFIXES): + out.append(entry) + return out + + +def _cli_error(exc: Any) -> Any: + """A refusal or a Genie API failure as the CLI's typed error, with what to do.""" + from fluid_build.cli._common import CLIError + from fluid_build.iac.providers.databricks_genie import GenieError + + event = "genie_estate_failed" if isinstance(exc, GenieError) else "genie_estate_refused" + return CLIError( + 1, + event, + {"kind": exc.kind, "error": str(exc), "remediation": list(exc.remediation)}, + ) diff --git a/fluid_build/cli/help_formatter.py b/fluid_build/cli/help_formatter.py index d5fc2798..8fac75ac 100644 --- a/fluid_build/cli/help_formatter.py +++ b/fluid_build/cli/help_formatter.py @@ -223,6 +223,10 @@ def _section(icon: str, title: str, color: str, rows: list[tuple[str, str]]) -> [ ("publish", "Publish to enterprise data catalogs"), ("market", "Browse & discover data products [dim]--blueprints[/dim]"), + ( + "genie", + "One Databricks Genie space over several products [dim]apply · remove[/dim]", + ), ("import", "Import existing dbt/Terraform/SQL projects"), ("mcp", "Serve forge tools over Model Context Protocol [dim]serve[/dim]"), ], diff --git a/fluid_build/iac/providers/databricks.py b/fluid_build/iac/providers/databricks.py index 9f559c06..e5781155 100644 --- a/fluid_build/iac/providers/databricks.py +++ b/fluid_build/iac/providers/databricks.py @@ -130,6 +130,8 @@ workspace folder ``extensions.databricksGenie.parentPath`` names, which a product deployed from more than one identity shares. ``fluid plan`` lists the space as a ``databricks_genie_space`` action, and ``fluid generate iac`` notes that its module holds none. + ``fluid genie apply`` makes one space over several products, joined on their foreign keys + (:func:`genie_estate`, ``databricks_genie_estate.py``). * ``binding.location.region`` is the region the workspace is in, declared so a contract's ``sovereignty`` policy can check it (``policy/sovereignty.py`` @@ -2942,6 +2944,38 @@ def _genie(contract: Mapping[str, Any], compiled: _Compiled, env: Mapping[str, s return compile_space(contract, compiled.tables, compiled.grants, env) +def genie_estate( + estate_id: str, + contracts: Sequence[Tuple[str, Mapping[str, Any]]], + env: Optional[Mapping[str, str]] = None, + *, + title: str = "", +) -> Any: + """The Genie space over ``contracts`` (``(source, contract)`` pairs) that ``fluid genie + apply`` asks for (``databricks_genie_estate.compile_estate``). Each contract is compiled as + ``fluid apply`` compiles it, so the space holds the tables and grants that apply made, and + its ``sovereignty`` policy is held as :meth:`DatabricksIacPlugin.emit` holds it. Imported + here, the Light CLI's ``--help`` never needs it.""" + from .databricks_genie_estate import Member, compile_estate + + env = os.environ if env is None else env + members = [] + for source, contract in contracts: + compiled = _compile(contract, env) + if not compiled.refusals: + _enforce_sovereignty(contract, compiled.tables) + members.append( + Member( + source, + contract, + tuple(compiled.tables), + tuple(compiled.grants), + tuple(compiled.refusals), + ) + ) + return compile_estate(estate_id, members, env, title=title) + + @dataclass(frozen=True) class ManagedTable: """A table the module manages, as ``fluid verify`` checks it.""" diff --git a/fluid_build/iac/providers/databricks_genie.py b/fluid_build/iac/providers/databricks_genie.py index 024f5960..fdddd5f4 100644 --- a/fluid_build/iac/providers/databricks_genie.py +++ b/fluid_build/iac/providers/databricks_genie.py @@ -68,6 +68,10 @@ after would make a second); and a title another space in the folder already has comes back suffixed with a timestamp (kept, since a ``PATCH`` to a taken title is 409). +One space over several products (an estate, ``fluid genie apply``) reuses this module's +compile helpers, identity rule and converge (``databricks_genie_estate.py``); its messages name +its own command through :class:`Voice`. + The client is the ``databricks.sdk.WorkspaceClient`` the masks and row filters use (``databricks_enforce.default_client``), through ``api_client.do``. The paths are the SDK's own (databricks-sdk-py ``service/dashboards.py`` and ``service/iam.py``). No message carries a @@ -81,6 +85,8 @@ import os import re import time +from contextlib import contextmanager +from contextvars import ContextVar from dataclasses import dataclass, field from difflib import get_close_matches from typing import ( @@ -89,6 +95,7 @@ Collection, Dict, Iterable, + Iterator, List, Mapping, Optional, @@ -174,6 +181,41 @@ def marker(product_id: str) -> str: return f"[fluid-genie:{product_id}]" +@dataclass(frozen=True) +class Voice: + """How a space's messages name what keeps it in step: a product's space (``fluid apply`` + and the contract's block, the defaults) or an estate's (``databricks_genie_estate.py``).""" + + #: The command that converges the space again. + command: str = "fluid apply" + #: One run of it, as in "the next apply". + run: str = "apply" + #: What the space is the one space of. + scope: str = "product" + #: Where the space's title is set. + title_setting: str = f"extensions.{EXTENSION_KEY}.title" + #: Where the space's folder is set, ``""`` for a space that has no setting for it. + folder_setting: str = f"extensions.{EXTENSION_KEY}.parentPath" + #: How to ask for no space instead, at the start of a sentence. + opt_out: str = f"Remove extensions.{EXTENSION_KEY}" + #: The line that heads what a converge prints. + heading: str = "Databricks Genie space (Genie API, not OpenTofu):" + + +PRODUCT_VOICE = Voice() +_VOICE: ContextVar[Voice] = ContextVar("databricks_genie_voice", default=PRODUCT_VOICE) + + +@contextmanager +def speaking(voice: Voice) -> Iterator[None]: + """The messages said inside the block name what ``voice`` names.""" + token = _VOICE.set(voice) + try: + yield + finally: + _VOICE.reset(token) + + @dataclass(frozen=True) class Space: """What the contract asks the product's Genie space to be.""" @@ -188,6 +230,8 @@ class Space: readers: Tuple[str, ...] #: ``serialized_space``, without the metric views (those are read at apply time). serialized: Dict[str, Any] + #: The marker its description ends with, when it is not the product's (an estate's). + token: str = "" #: The workspace folder the space lives in (``parentPath``, as :func:`workspace_folder` #: writes it), or ``""``: the deploying identity's own folder, where the API puts a space #: when the create names none. @@ -195,7 +239,7 @@ class Space: @property def marker(self) -> str: - return marker(self.product_id) + return self.token or marker(self.product_id) @property def folder(self) -> str: @@ -212,6 +256,8 @@ class Compiled: warnings: List[str] = field(default_factory=list) #: The contract's id: what a space it asked for earlier is found by, block or no block. product_id: str = "" + #: How the converge's messages name what keeps the space. + voice: Voice = PRODUCT_VOICE # ---------------------------------------------------------------------- compile (pure) @@ -437,13 +483,14 @@ def _policy_refusals(table: Any) -> List[UnsupportedBindingError]: where = f"{table.where} policy.agentPolicy" allowed = _string_list(policy.get("allowedUseCases")) denied = _string_list(policy.get("deniedUseCases")) + opt_out = _VOICE.get().opt_out if isinstance(policy.get("allowedModels"), list) and not policy["allowedModels"]: out.append( _refusal( "policy", f"{where}.allowedModels is [], which means no AI access, and a Genie space answers " "questions with a model.", - f"Remove extensions.{EXTENSION_KEY}, or allow AI access in agentPolicy.", + f"{opt_out}, or allow AI access in agentPolicy.", ) ) refused = [case for case in _GENIE_USE_CASES if case in denied] @@ -458,7 +505,7 @@ def _policy_refusals(table: Any) -> List[UnsupportedBindingError]: else "allows neither 'qa' nor 'analysis'" ) + " (question answering over the data), which is what a Genie space is.", - f"Remove extensions.{EXTENSION_KEY}, or allow 'qa' in agentPolicy.allowedUseCases " + f"{opt_out}, or allow 'qa' in agentPolicy.allowedUseCases " "(and take 'qa' and 'analysis' out of deniedUseCases).", ) ) @@ -652,14 +699,18 @@ def _instruction_lines( else: lines += section lines += _lineage_lines(contract) - lines.append( - "How access works: Genie runs each query as the person asking, under that person's own " - "Unity Catalog grants, so column masks and row filters apply. This space grants no " - "access to data: a table the person cannot SELECT returns no rows." - ) + lines.append(ACCESS_LINE) return lines +#: The instruction's last line, in every space. +ACCESS_LINE = ( + "How access works: Genie runs each query as the person asking, under that person's own " + "Unity Catalog grants, so column masks and row filters apply. This space grants no " + "access to data: a table the person cannot SELECT returns no rows." +) + + def _table_section(contract: Mapping[str, Any], table: Any, warnings: List[str]) -> List[str]: """What the space is told about one table: its semantics, metrics and ``agentPolicy``.""" exposure = table.exposure @@ -916,21 +967,34 @@ def converge( Runs after ``tofu apply`` and the masks and row filters, so the tables exist and are protected first. With ``dry_run`` it reads and prints, and writes nothing. Raises - :class:`GenieError` (or a refusal) for what it cannot do. + :class:`GenieError` (or a refusal) for what it cannot do. Its messages name what + ``compiled.voice`` names. """ + with speaking(compiled.voice): + _converge(compiled, say, client_factory, dry_run) + + +def require_warehouse(space: Space) -> None: + """Stop when no SQL warehouse is named: a Genie space runs its queries on one.""" + if not space.warehouse_id: + raise GenieError( + f"{WAREHOUSE_ENV} is not set, and a Genie space runs its queries on that warehouse.", + (f"Set {WAREHOUSE_ENV} to the id of a SQL warehouse in the workspace.",), + ) + + +def _converge( + compiled: Compiled, say: Say, client_factory: Optional[ClientFactory], dry_run: bool +) -> None: if compiled.refusals: raise compiled.refusals[0] space = compiled.space if space is None: _retire(compiled.product_id, say, client_factory, dry_run) return - if not space.warehouse_id: - raise GenieError( - f"{WAREHOUSE_ENV} is not set, and a Genie space runs its queries on that warehouse.", - (f"Set {WAREHOUSE_ENV} to the id of a SQL warehouse in the workspace.",), - ) + require_warehouse(space) client = _client(client_factory) - say("\n Databricks Genie space (Genie API, not OpenTofu):") + say(f"\n {_VOICE.get().heading}") if space.parent_path: _check_folder(client, space.parent_path, say) found = _call(client, "list the Genie spaces", lambda: _find(client, space.marker)) @@ -975,11 +1039,13 @@ def _in(space: Space) -> str: def _at_most_one(spaces: Sequence[Mapping[str, Any]], space: Space) -> None: + voice = _VOICE.get() if len(spaces) > 1: raise GenieError( f"{len(spaces)} Genie spaces carry {space.marker} in their description: " + ", ".join(str(s.get("space_id")) for s in spaces) - + ". fluid apply keeps one space per product and cannot tell which is the product's.", + + f". {voice.command} keeps one space per {voice.scope} and cannot tell which is the " + f"{voice.scope}'s.", ("Trash the extra spaces (or remove the marker from their description), then re-run.",), ) @@ -1206,7 +1272,9 @@ def _update_found( ) -> None: live = _call(client, "read the Genie space", lambda: _get(client, space_id)) title = _title_kept(live.get("title"), space, say) - changes = differences(live, space, serialized) + changes = differences( + live, space, serialized, _tables_of_space(client, live, space, serialized) + ) if not changes: say(f" {space.title!r}: no change ({_url(space_id)})") elif dry_run: @@ -1218,6 +1286,7 @@ def _update_found( lambda: _update(client, space_id, space, serialized, live.get("etag"), title), ) say(f" updated {space.title!r}: {', '.join(changes)} ({_url(space_id)})") + _keep_as_sent(client, space_id, space, serialized, say, made="updated") def _create_new(client: Any, space: Space, serialized: Mapping[str, Any], say: Say) -> str: @@ -1232,9 +1301,50 @@ def _create_new(client: Any, space: Space, serialized: Mapping[str, Any], say: S _title_kept(created.get("title"), space, say) _wait_listed(client, space, space_id, say) _found_again(client, space, space_id) + _keep_as_sent(client, space_id, space, serialized, say, made="made") return space_id +def _keep_as_sent( + client: Any, + space_id: str, + space: Space, + serialized: Mapping[str, Any], + say: Say, + *, + made: str, +) -> None: + """Set the space back to what was sent when the service added to it as it wrote it. + + Measured on a live workspace (2026-11-13): whenever a table enters a space (a create, or + an update that adds one) Genie adds a join for each Unity Catalog foreign key between two + of the space's tables that touches it (its own id, the parent aliased ``_2``), + beside the joins it was sent. An update with the same tables keeps exactly what it is + sent, so one more makes the space what the contracts ask. Left so, the next run would + update the space again. + """ + live = _call(client, "read the Genie space back", lambda: _get(client, space_id)) + changes = differences(live, space, serialized) + if not changes: + return + title = _title_kept(live.get("title"), space, lambda _line: None) + _call( + client, + "set the Genie space back", + lambda: _update(client, space_id, space, serialized, live.get("etag"), title), + ) + path = ("instructions", "join_specs") + held, sent = (len(_dig(doc, path) or ()) for doc in (_held(live), serialized)) + if changes == ["joins"] and held > sent: + say( + f" note: Genie added {held - sent} join(s) of its own as it {made} the space, " + "from the tables' Unity Catalog foreign keys; they were taken out, and the space " + f"holds the {sent} the contracts declare." + ) + else: + say(f" note: the space differed from what was sent ({', '.join(changes)}); set back.") + + def _found_again(client: Any, space: Space, space_id: str) -> None: """Stop, and trash the new space, when it is not where the next apply looks for it: a space every apply could not find would be made again on each one.""" @@ -1250,6 +1360,7 @@ def _found_again(client: Any, space: Space, space_id: str) -> None: ) if landed: return + run = _VOICE.get().run try: _call( client, @@ -1260,13 +1371,13 @@ def _found_again(client: Any, space: Space, space_id: str) -> None: raise GenieError( f"the Genie API put the new space {space_id} in " f"{parent or 'a folder this identity cannot read'}, not in {where}, where the next " - f"apply looks for it, and trashing it failed: {exc}", - (f"Trash space {space_id} by hand before the next apply, which would make another.",), + f"{run} looks for it, and trashing it failed: {exc}", + (f"Trash space {space_id} by hand before the next {run}, which would make another.",), ) from exc raise GenieError( f"the Genie API put the new space in {parent or 'a folder this identity cannot read'}, " - f"not in {where}, where the next apply looks for it, so it was trashed rather than made " - "again on every apply.", + f"not in {where}, where the next {run} looks for it, so it was trashed rather than made " + f"again on every {run}.", ( ( ( @@ -1294,10 +1405,12 @@ def _wait_listed(client: Any, space: Space, space_id: str, say: Say) -> None: except Exception: # noqa: BLE001 — the space exists; only the wait is cut short pass if time.monotonic() >= deadline: + run = _VOICE.get().run + article = "an" if run[:1] in "aeiou" else "a" say( - " note: the Genie list does not show the new space yet; an apply in the next " - "minute could make a second one, which the apply after refuses until one is " - "trashed." + f" note: the Genie list does not show the new space yet; {article} {run} in the " + f"next minute could make a second one, which the {run} after refuses until one " + "is trashed." ) return time.sleep(_POLL_SECONDS) @@ -1313,7 +1426,7 @@ def _title_kept(live: Any, space: Space, say: Say) -> str: say( f" note: the space is titled {live!r}: another space in {space.folder} is " f"already called {wanted!r} and Genie keeps titles unique there. Give this one a " - f"title of its own with extensions.{EXTENSION_KEY}.title." + f"title of its own with {_VOICE.get().title_setting}." ) return live return wanted @@ -1424,6 +1537,8 @@ def _placed( if not found: return [], [] home = _home(client) + voice = _VOICE.get() + scope, setting, run = voice.scope, voice.folder_setting, voice.run product: List[Dict[str, Any]] = [] own: List[Dict[str, Any]] = [] for space in found: @@ -1438,10 +1553,14 @@ def _placed( f" note: left alone: space {space_id} ({space.get('title')!r}) carries " f"{_marker_of(space)} but is in {parent or 'a folder this identity cannot read'}, " "not in " + (f"{folder} or " if folder else "") + "this identity's own folder, " - "so it is not taken for the product's. Trash it, or take the marker out of its " - "description. If it is this product's own space, left in a folder the contract no " - f"longer names as {_WHERE}.parentPath, trash it there by hand: the apply trusts no " - "other folder." + f"so it is not taken for the {scope}'s. Trash it, or take the marker out of its " + "description." + + ( + f" If it is this {scope}'s own space, left in a folder the contract no longer " + f"names as {setting}, trash it there by hand: the {run} trusts no other folder." + if setting + else "" + ) ) return product, own @@ -1468,10 +1587,11 @@ def _in_folder(parent: str, home: str) -> bool: def _marker_of(space: Mapping[str, Any]) -> str: - """The marker a listed space's description ends with (for a message).""" + """The marker a listed space's description ends with (for a message): a product's + ``[fluid-genie:...]`` or an estate's ``[fluid-genie-estate:...]``.""" text = str(space.get("description") or "").rstrip() - start = text.rfind("[fluid-genie:") - return text[start:] if start >= 0 else "a product's marker" + start = text.rfind("[fluid-genie") + return text[start:] if start >= 0 else "a fluid-genie marker" def _get(client: Any, space_id: str) -> Dict[str, Any]: @@ -1512,20 +1632,21 @@ def _update( try: client.api_client.do("PATCH", f"{SPACES_PATH}/{quote(space_id)}", body=body) except Exception as exc: # noqa: BLE001 — a 409 is a taken title or someone else's edit + voice = _VOICE.get() if _status(exc) == 409 and _taken_title(exc): raise GenieError( f"another Genie space in {space.folder} is already called " f"{body['title']!r}, and Genie keeps titles unique there.", ( - f"Give this space a title of its own with extensions.{EXTENSION_KEY}.title, " + f"Give this space a title of its own with {voice.title_setting}, " "or rename the other space.", ), ) from exc if _status(exc) == 409: raise GenieError( - "the Genie space changed while this apply was updating it (its etag no longer " - "matched), so nothing was overwritten.", - ("Re-run fluid apply.",), + f"the Genie space changed while this {voice.run} was updating it (its etag no " + "longer matched), so nothing was overwritten.", + (f"Re-run {voice.command}.",), ) from exc raise @@ -1539,39 +1660,87 @@ def _taken_title(exc: BaseException) -> bool: ) -def differences(live: Mapping[str, Any], space: Space, serialized: Mapping[str, Any]) -> List[str]: +def _tables_of_space( + client: Any, live: Mapping[str, Any], space: Space, serialized: Mapping[str, Any] +) -> set: + """The space's tables, as :func:`differences` names a change: the ones it asks for, and + each data source leaving the space that is still a table in Unity Catalog (a product, or an + expose, left out). One that is a metric view, or is gone, is said as a metric view.""" + path = ("data_sources", "tables") + held = _held(live) + wanted = { + str(t.get("identifier")) for t in _dig(serialized, path) or () if isinstance(t, Mapping) + } + leaving = sorted( + str(t.get("identifier")) + for t in _dig(held, path) or () + if isinstance(t, Mapping) and str(t.get("identifier")) not in wanted + ) + tables = set(space.tables) + for name in leaving: + try: + info = _read_table(client, name) + except Exception: # noqa: BLE001 — only the word for the change depends on it + info = None + if info and str(info.get("table_type") or "").upper() != "METRIC_VIEW": + tables.add(name) + return tables + + +def differences( + live: Mapping[str, Any], + space: Space, + serialized: Mapping[str, Any], + tables: Optional[Collection[str]] = None, +) -> List[str]: """What differs between the live space and the one the contract asks for, by name. A title - the service suffixed because another space has it counts as the same title.""" - changes = [] - if not _same_title(live.get("title"), space.title): - changes.append("title") - if (live.get("description") or "") != space.description: - changes.append("description") - if live.get("warehouse_id") != space.warehouse_id: - changes.append("warehouse") - try: - held = json.loads(live.get("serialized_space") or "{}") - except (TypeError, ValueError): - held = {} + the service suffixed because another space has it counts as the same title. ``tables`` are + the data sources to name as tables (any other as a metric view); by default the space's.""" + changes = [ + name + for name, same in ( + ("title", _same_title(live.get("title"), space.title)), + ("description", (live.get("description") or "") == space.description), + ("warehouse", live.get("warehouse_id") == space.warehouse_id), + ) + if not same + ] + held = _held(live) if held != serialized: - named = [] - if _dig(held, ("config", "sample_questions")) != _dig( - serialized, ("config", "sample_questions") - ): - named.append("sample questions") - named += _source_changes(held, serialized, set(space.tables)) - if _dig(held, ("instructions", "text_instructions")) != _dig( - serialized, ("instructions", "text_instructions") - ): - named.append("instructions") - if _dig(held, ("instructions", "sql_snippets")) != _dig( - serialized, ("instructions", "sql_snippets") - ): - named.append("SQL snippets") + own = set(space.tables if tables is None else tables) + named = ( + _changed_parts(held, serialized, _CONFIG_PARTS) + + _source_changes(held, serialized, own) + + _changed_parts(held, serialized, _INSTRUCTION_PARTS) + ) changes += named or ["other settings"] return changes +#: The parts of ``serialized_space`` an update names a change of, as :func:`differences` +#: names them: the sample questions, then the data sources, then these instructions. +_CONFIG_PARTS = ((("config", "sample_questions"), "sample questions"),) +_INSTRUCTION_PARTS = ( + (("instructions", "text_instructions"), "instructions"), + (("instructions", "join_specs"), "joins"), + (("instructions", "sql_snippets"), "SQL snippets"), +) + + +def _changed_parts( + held: Any, wanted: Any, parts: Sequence[Tuple[Tuple[str, ...], str]] +) -> List[str]: + return [name for path, name in parts if _dig(held, path) != _dig(wanted, path)] + + +def _held(live: Mapping[str, Any]) -> Any: + """The live space's ``serialized_space``, parsed; ``{}`` when it holds no JSON.""" + try: + return json.loads(live.get("serialized_space") or "{}") + except (TypeError, ValueError): + return {} + + def _source_changes(held: Any, wanted: Any, own: set) -> List[str]: """``tables`` and/or ``metric views``: which of the space's data sources differ. Its own tables are the product's; any other entry in the list is a metric view over them.""" @@ -1694,7 +1863,8 @@ def _share(client: Any, space_id: str, readers: Sequence[str], say: Say, dry_run if kept: say( f" note: {', '.join(kept)} also hold CAN_EDIT or CAN_MANAGE on the space, set on it " - "directly; fluid apply keeps them (CAN_MANAGE can read other people's conversations)." + f"directly; {_VOICE.get().command} keeps them (CAN_MANAGE can read other people's " + "conversations)." ) for principal in readers: key = acl_key(principal) @@ -1811,14 +1981,15 @@ def _call(client: Any, what: str, fn: Callable[[], Any]) -> Any: def _remediation(status: Optional[int], what: str) -> str: """What to do about a failed call, by its HTTP status.""" + command = _VOICE.get().command if status in (403, 404): genie = "Genie (AI/BI) is enabled in the workspace and that " if "Genie" in what else "" - return f"Check that {genie}this identity may {what}, then re-run fluid apply." + return f"Check that {genie}this identity may {what}, then re-run {command}." if status == 400: - return "Fix what the message names, then re-run fluid apply." + return f"Fix what the message names, then re-run {command}." if status == 429 or (status or 0) >= 500: - return "The workspace is busy or unavailable: re-run fluid apply in a minute." - return "Check DATABRICKS_HOST and the credential, then re-run fluid apply." + return f"The workspace is busy or unavailable: re-run {command} in a minute." + return f"Check DATABRICKS_HOST and the credential, then re-run {command}." #: The HTTP status each databricks-sdk error class stands for (``databricks/sdk/errors/ @@ -1865,11 +2036,14 @@ def _summary(serialized: Mapping[str, Any], own: Sequence[str]) -> str: tables = sum(1 for name in sources if name in own) views = len(sources) - tables questions = len(_dig(serialized, ("config", "sample_questions")) or ()) + joins = len(_dig(serialized, ("instructions", "join_specs")) or ()) measures = len(_dig(serialized, ("instructions", "sql_snippets", "measures")) or ()) parts = [f"{tables} table(s)"] if views: parts.append(f"{views} metric view(s)") parts.append(f"{questions} sample question(s)") + if joins: + parts.append(f"{joins} join(s)") if measures: parts.append(f"{measures} measure snippet(s)") return ", ".join(parts) diff --git a/fluid_build/iac/providers/databricks_genie_estate.py b/fluid_build/iac/providers/databricks_genie_estate.py new file mode 100644 index 00000000..77ba157b --- /dev/null +++ b/fluid_build/iac/providers/databricks_genie_estate.py @@ -0,0 +1,662 @@ +# Copyright 2025 Agentics Transformation Limited +# SPDX-License-Identifier: Apache-2.0 + +"""One Databricks AI/BI Genie space over a set of data products, an estate: +``fluid genie apply --estate ``. + +A product's own space (``databricks_genie.py``, which ``extensions.databricksGenie`` asks for) +holds that product's tables. An estate's space holds the Databricks tables of every contract it +is given, so one question can span and join them: ``fluid genie apply tower-management/ +--estate tower-management`` makes one space over the pack's products. Each table carries what a +product space says of it (its description, each column's description, allowed values and +synonyms, the product's example questions as sample questions, and an instruction section with +its semantics, metrics and ``agentPolicy``). The space adds: + +* ``instructions.join_specs``: one join per foreign key the contracts declare between the + estate's tables. They are the keys ``fluid apply`` declares in Unity Catalog + (``databricks_keys.py``): a column tagged ``foreign-key`` whose ``labels.references`` is + ``.``, resolved by :func:`databricks_keys.foreign_keys` to the estate table + whose primary key that column is (a composite key from all of its columns). A join is + ``MANY_TO_ONE`` from the referencing table, or ``ONE_TO_ONE`` when the foreign key is that + table's own primary key. Each column in its condition is qualified with its table's alias, as + Genie requires (``databricks-genie-agents`` ``create-genie-agent.md``); a foreign key from a + table to itself gets a second alias for its parent row. A reference to a product the estate + does not hold, or to a column that is not the primary key of exactly one of that product's + tables, is said and left out. +* one text instruction (Genie takes one): a section per product and table, then how access + works. +* no measure snippets: a measure in a space of several tables needs its columns qualified, + which this module does not do. Each metric is in the instruction, with its SQL. + +Who may ask (founder decision, 2026-11-12): ``CAN_RUN`` goes only to the principals that read +every table of the estate (``SELECT`` or ``ALL_PRIVILEGES``). The space's instructions and +column descriptions describe every product, so a principal that reads only some of the tables +would still see the descriptions of the others; each such principal is named in a note instead. +Genie runs every query as the person asking, under that person's own Unity Catalog grants, as in +a product space. + +Identity, as a product space's, without state: the description ends with +``[fluid-genie-estate:]``, a run adopts the space carrying it in the deploying identity's own +folder only (``databricks_genie._placed``), and two such spaces stop the run. The space converges +to exactly the contracts given, so a contract left out of a run leaves the space. +``fluid genie remove --estate `` trashes it. The products are applied first (``fluid +apply``): each run checks that every table exists before it writes, and names the product of +each one that does not. + +Measured on a live workspace (Free Edition, 2026-11-12): the service keeps ``join_specs`` as sent +(so a second run compares equal); it takes a join of a table to itself under two aliases, and +``ONE_TO_ONE``; it refuses a relationship tag it does not know ("Failed to parse export proto"), +and a space of 51 tables ("A space can have at most 50 tables": the 50 of +:data:`databricks_genie.MAX_TABLES`, not the 30 the skill reference gives). It takes 60 sample +questions and an instruction of 400 lines (60,000 characters). It does not check, when a space is +made, that a join's tables are among the space's or that its columns are qualified: Genie fails +on those only when it composes a query, so this module never sends either. +""" + +from __future__ import annotations + +import hashlib +import re +from collections import Counter +from dataclasses import dataclass, field, replace +from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple +from urllib.parse import quote + +from ...providers._sql_safety import validate_ident +from ..base import UnsupportedBindingError +from . import databricks_genie as genie +from .databricks_keys import Candidate, TableKeys, compile_keys, foreign_keys + +#: The marker's tag: a space's description ends with ``[fluid-genie-estate:]``. A product's +#: ends with ``[fluid-genie:]``; neither can end the other, each starting with its own tag. +ESTATE_TAG = "fluid-genie-estate" +#: An estate id: what the marker carries, so no bracket, space or control character. +_ESTATE_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}") +#: The second element of a join's ``sql``: its relationship, left to right. +_RELATIONSHIP = "--rt=FROM_RELATIONSHIP_TYPE_{}--" +_COMMAND = "fluid genie apply" + + +def estate_marker(estate_id: str) -> str: + """The token an estate space's description ends with.""" + return f"[{ESTATE_TAG}:{estate_id}]" + + +def voice(estate_id: str, command: str = _COMMAND) -> genie.Voice: + """How an estate space's messages name what keeps it in step (``command``: the one the + run is, so a failed ``fluid genie remove`` says to re-run that).""" + return genie.Voice( + command=command, + run="run", + scope="estate", + title_setting="--title", + folder_setting="", + opt_out="Leave that product out of the estate", + heading=f"Databricks Genie space for the estate {estate_id} (Genie API):", + ) + + +@dataclass(frozen=True) +class Member: + """One contract of the estate: where it was read, and the Databricks tables, grants and + refusals the emitter resolved for it (``databricks._compile``, as ``fluid apply`` does).""" + + source: str + contract: Mapping[str, Any] + tables: Tuple[Any, ...] = () + #: ``(securable type, full name, principal, privileges)`` per grant. + grants: Tuple[Tuple[str, str, str, Sequence[str]], ...] = () + refusals: Tuple[UnsupportedBindingError, ...] = () + + @property + def product_id(self) -> str: + return genie._text(self.contract.get("id")) + + @property + def name(self) -> str: + return genie._text(self.contract.get("name")) or self.product_id + + +@dataclass +class Estate(genie.Compiled): + """The estate's space (``space``), every refusal, and what a run names it by.""" + + #: Each table's product id, for a message about that table. + owners: Dict[str, str] = field(default_factory=dict) + #: How many contracts the estate holds. + products: int = 0 + + @property + def joins(self) -> int: + """How many joins its space holds (0 without a space).""" + if self.space is None: + return 0 + return len(genie._dig(self.space.serialized, ("instructions", "join_specs")) or ()) + + +# ---------------------------------------------------------------------- compile (pure) + + +def compile_estate( + estate_id: str, + members: Sequence[Member], + env: Mapping[str, str], + *, + title: str = "", +) -> Estate: + """The space over ``members``, and every refusal. Credential-free and offline.""" + estate_id = str(estate_id or "").strip() + out = Estate(product_id=estate_id, voice=voice(estate_id), products=len(members)) + with genie.speaking(out.voice): + out.refusals += _refusals(estate_id, members) + if out.refusals: + return out + key = f"estate:{estate_id}" + ordered = sorted((t for m in members for t in m.tables), key=lambda t: t.full_name) + out.owners = {t.full_name: m.product_id for m in members for t in m.tables} + serialized = _serialized_space(key, members, ordered, out.warnings) + too_long = next( + (text for text in genie._strings_in(serialized) if len(text) > genie.MAX_STRING), + None, + ) + if too_long is not None: + out.refusals.append( + genie._refusal( + "limit", + f"the estate's space would hold a text of {len(too_long)} characters, and " + f"Genie holds at most {genie.MAX_STRING:,} in one: {too_long[:80]!r}...", + "Shorten that text in its contract.", + ) + ) + return out + out.space = genie.Space( + product_id=estate_id, + title=_one_line(title) or estate_id, + description=_description(estate_id, members), + warehouse_id=str(env.get(genie.WAREHOUSE_ENV) or "").strip(), + tables=tuple(t.full_name for t in ordered), + readers=_readers(ordered, [g for m in members for g in m.grants], out.warnings), + serialized=serialized, + token=estate_marker(estate_id), + ) + return out + + +def _refusals(estate_id: str, members: Sequence[Member]) -> List[UnsupportedBindingError]: + """What stops the estate: a bad id, no contract, any contract's own refusal, one without an + id or a table, a product or table given twice, a policy that rules Genie out, too many + tables.""" + if not _ESTATE_ID.fullmatch(estate_id): + return [invalid_estate_id(estate_id)] + if not members: + return [ + genie._refusal( + "no-table", + "no contract was given, and an estate space holds the tables of its contracts.", + "Name the contracts, their folders or a glob: fluid genie apply " + f"--estate {estate_id}.", + ) + ] + out = _contract_refusals(members) + products: Dict[str, str] = {} + tables: Dict[str, str] = {} + for member in members: + out += _member_refusals(member, products, tables) + if len(tables) > genie.MAX_TABLES: + out.append( + genie._refusal( + "limit", + f"the estate has {len(tables)} Databricks tables, and a Genie space holds at most " + f"{genie.MAX_TABLES} tables, views or metric views.", + "Split the estate into smaller ones (one --estate each).", + ) + ) + return out + + +def _member_refusals( + member: Member, products: Dict[str, str], tables: Dict[str, str] +) -> List[UnsupportedBindingError]: + """What one contract makes the estate refuse: no id, a product given twice, no Databricks + table, a table another product has, a policy that rules Genie out. Records its product in + ``products`` and its tables in ``tables``.""" + product = member.product_id + if not product: + return [ + genie._refusal( + "invalid", + f"{member.source} has no id, and the estate names each product by its id (the " + "id its foreign keys reference).", + "Give the contract an id.", + ) + ] + if product in products: + return [ + genie._refusal( + "invalid", + f"{member.source} and {products[product]} are both {product}, and an estate " + "holds each product once.", + "Name one of them.", + ) + ] + products[product] = member.source + out: List[UnsupportedBindingError] = [] + if not member.tables and not member.refusals: + out.append( + genie._refusal( + "no-table", + f"{member.source} ({product}) binds no table expose to databricks, so it has " + "no table for the estate's space.", + "Leave it out of the estate, or bind an expose to databricks " + "(binding.platform: databricks).", + ) + ) + for table in member.tables: + other = tables.setdefault(table.full_name, product) + if other != product: + out.append( + genie._refusal( + "invalid", + f"{table.full_name} is a table of both {other} and {product}.", + "Give each product its own table, or leave one of them out.", + ) + ) + out += [ + UnsupportedBindingError(r.kind, f"{product} {r}", r.remediation) + for r in genie._policy_refusals(table) + ] + return out + + +def _contract_refusals(members: Sequence[Member]) -> List[UnsupportedBindingError]: + """Each contract's own refusals, the ones several contracts share (an unset warehouse, say) + said once with the contracts named.""" + shared: Dict[Tuple[str, str, Tuple[str, ...]], List[str]] = {} + for member in members: + for refusal in member.refusals: + what = (refusal.kind, str(refusal), tuple(refusal.remediation)) + shared.setdefault(what, []).append(member.source) + out = [] + for (kind, message, remediation), sources in shared.items(): + at = ( + ", ".join(sources) + if len(sources) <= 3 + else f"{len(sources)} contracts ({sources[0]}, ...)" + ) + out.append(UnsupportedBindingError(kind, f"{at}: {message}", remediation)) + return out + + +def invalid_estate_id(estate_id: str) -> UnsupportedBindingError: + return genie._refusal( + "invalid", + f"--estate {estate_id!r} is not an estate id: 1 to 128 letters, digits, '.', '_' or '-', " + "starting with a letter or a digit (the space's description ends with " + f"{estate_marker('')}, which is how the next run finds it).", + "Name the estate like a product id: --estate tower-management.", + ) + + +def _serialized_space( + key: str, members: Sequence[Member], tables: Sequence[Any], warnings: List[str] +) -> Dict[str, Any]: + space: Dict[str, Any] = {"version": 2} + questions = _sample_questions(key, members) + if questions: + space["config"] = {"sample_questions": questions} + else: + warnings.append( + "no product of the estate has semantics.labels.ai-example-questions, so its Genie " + "space shows no sample questions." + ) + space["data_sources"] = {"tables": [genie._table_source(t) for t in tables]} + instructions: Dict[str, Any] = { + "text_instructions": [ + { + "id": genie._id(key, "instructions"), + "content": [line + "\n" for line in _instruction_lines(members, warnings)], + } + ] + } + warnings += key_warnings(members) + joins = join_specs(key, members, warnings) + if joins: + instructions["join_specs"] = joins + space["instructions"] = instructions + return space + + +def _by_product(members: Sequence[Member]) -> List[Member]: + return sorted(members, key=lambda m: m.product_id) + + +def _tables_of(member: Member) -> List[Any]: + return sorted(member.tables, key=lambda t: t.full_name) + + +def _sample_questions(key: str, members: Sequence[Member]) -> List[Dict[str, Any]]: + """Every product's ``ai-example-questions``, once each (compared without case).""" + questions: Dict[str, str] = {} + for member in _by_product(members): + for table in _tables_of(member): + labels = genie._labels(genie._semantics(table.exposure).get("labels")) + for question in genie._split(labels.get("ai-example-questions"), "|"): + questions.setdefault(question.casefold(), question) + return genie._by_id( + {"id": genie._id(key, "question", q), "question": [q]} for q in questions.values() + ) + + +def _instruction_lines(members: Sequence[Member], warnings: List[str]) -> List[str]: + """A section per product (and per table, for a product of several), then the access line.""" + lines: List[str] = [] + for member in _by_product(members): + tables = _tables_of(member) + lineage = genie._lineage_lines(member.contract) + if len(tables) == 1: + (table,) = tables + lines.append(f"About {table.full_name} ({_label(member, table)}):") + lines += [ + f" {line}" for line in genie._table_section(member.contract, table, warnings) + ] + else: + lines.append(f"About the data product {member.name} ({member.product_id}):") + for table in tables: + section = genie._table_section(member.contract, table, warnings) + lines.append(f" {table.full_name} ({_expose_title(table)}):") + lines += [f" {line}" for line in section] + lines += [f" {line}" for line in lineage] + lines.append(genie.ACCESS_LINE) + return lines + + +def _expose_title(table: Any) -> str: + return genie._text(table.exposure.get("title")) or genie._text(table.exposure.get("exposeId")) + + +def _label(member: Member, table: Any) -> str: + """``, data product ``, with the expose's title when it is another.""" + title = _expose_title(table) + named = member.name if not title or title == member.name else f"{member.name}: {title}" + return f"{named}, data product {member.product_id}" + + +# ---------------------------------------------------------------------- joins + + +def key_warnings(members: Sequence[Member]) -> List[str]: + """What each table's keys lose before any join is made (``databricks_keys.compile_keys``): + a primary key on a column that is not required, a ``labels.references`` that names no + product and column. ``fluid apply`` says the same of its Unity Catalog keys.""" + out: List[str] = [] + for member in _by_product(members): + for table in _tables_of(member): + notes: List[str] = [] + compile_keys(table.where, table.full_name, table.exposure, table.columns, notes) + out += [f"{member.product_id} {note}" for note in notes] + return out + + +def join_specs(key: str, members: Sequence[Member], warnings: List[str]) -> List[Dict[str, Any]]: + """One ``join_spec`` per foreign key between the estate's tables, sorted by id. + + The parents are the estate's own tables with a primary key, so a foreign key resolves here + as ``fluid apply`` resolves it in Unity Catalog, among the estate's products only. + """ + candidates: Dict[str, List[Candidate]] = { + m.product_id: [ + Candidate(t.full_name, t.keys.primary) + for t in m.tables + if t.keys is not None and t.keys.primary + ] + for m in members + } + aliases = table_aliases(t.full_name for m in members for t in m.tables) + specs: List[Dict[str, Any]] = [] + for member in _by_product(members): + for table in _tables_of(member): + keys: Optional[TableKeys] = table.keys + if keys is None or not keys.references: + continue + at = f"{member.product_id} {table.where}" + joinable = _with_a_parent(at, keys, candidates, warnings) + if not joinable: + continue + found = foreign_keys( + replace(keys, references=tuple(joinable)), + candidates, + [], + own_id=member.product_id, + ) + _say_unjoined(at, joinable, found, warnings) + specs += [_join_spec(key, keys, fk, aliases) for fk in found] + return genie._by_id(specs) + + +def _with_a_parent( + at: str, + keys: TableKeys, + candidates: Mapping[str, Sequence[Candidate]], + warnings: List[str], +) -> List[Any]: + """The references whose product the estate holds with a primary key; each other one said.""" + joinable = [] + for ref in keys.references: + if ref.product not in candidates: + warnings.append( + f"{at} column {ref.column} references {ref.product}, which is not in the " + "estate: the space has no join for it." + ) + elif not candidates[ref.product]: + warnings.append( + f"{at} column {ref.column} references {ref.product}, which declares no " + "primary key (a column tagged primary-key and required: true): the space " + "has no join for it." + ) + else: + joinable.append(ref) + return joinable + + +def _say_unjoined( + at: str, joinable: Sequence[Any], found: Sequence[Any], warnings: List[str] +) -> None: + """Each reference the resolution made no foreign key of.""" + joined = {column for fk in found for column in fk.columns} + for ref in joinable: + if ref.column not in joined: + warnings.append( + f"{at} column {ref.column} references {ref.product}.{ref.parent_column}, " + f"which is not the primary key of exactly one table of {ref.product}: " + "the space has no join for it." + ) + + +def _join_spec(key: str, keys: TableKeys, fk: Any, aliases: Mapping[str, str]) -> Dict[str, Any]: + """The ``join_spec`` of one foreign key. One from a table to itself (a parent row of the + same table) names its right side ``_``, an alias of its own.""" + left, right = aliases[keys.full_name], aliases[fk.parent] + if fk.parent == keys.full_name: + right = f"{left}_{'_'.join(fk.columns)}" + if right in aliases.values() or not _plain(right): + right = f"{left}_{hashlib.sha256(right.encode('utf-8')).hexdigest()[:8]}" + condition = " AND ".join( + f"`{left}`.`{column}` = `{right}`.`{parent}`" + for column, parent in zip(fk.columns, fk.parent_columns, strict=True) + ) + one = set(fk.columns) == set(keys.primary) + return { + "id": genie._id(key, "join", keys.full_name, *fk.columns, fk.parent), + "left": {"identifier": keys.full_name, "alias": left}, + "right": {"identifier": fk.parent, "alias": right}, + "sql": [condition, _RELATIONSHIP.format("ONE_TO_ONE" if one else "MANY_TO_ONE")], + } + + +def table_aliases(full_names: Iterable[str]) -> Dict[str, str]: + """An alias per table, unique in the space: its name (Genie's implicit alias), else its + schema and name, else its catalog, schema and name, else its name and a digest.""" + names = sorted(set(full_names)) + out: Dict[str, str] = {} + for depth in (1, 2, 3): + pending = [n for n in names if n not in out] + labels = {n: "_".join(n.split(".")[-depth:]) for n in pending} + counts = Counter(labels.values()) + taken = set(out.values()) + for name in pending: + label = labels[name] + if counts[label] == 1 and label not in taken and _plain(label): + out[name] = label + for name in names: + if name not in out: + digest = hashlib.sha256(name.encode("utf-8")).hexdigest()[:8] + out[name] = f"{name.split('.')[-1]}_{digest}" + return out + + +def _plain(alias: str) -> bool: + try: + validate_ident(alias) + except ValueError: + return False + return True + + +# ---------------------------------------------------------------------- who may ask + + +def _readers( + tables: Sequence[Any], + grants: Sequence[Tuple[str, str, str, Sequence[str]]], + warnings: List[str], +) -> Tuple[str, ...]: + """The principals that read every table (``SELECT`` or ``ALL_PRIVILEGES``), sorted; each + principal that reads only some of them is named in ``warnings``.""" + spelled: Dict[Tuple[str, str], str] = {} + per_table: List[set] = [] + for table in tables: + readers = set() + for kind, full_name, principal, privileges in grants: + if ( + kind == "table" + and full_name == table.full_name + and genie._READ_PRIVILEGES & set(privileges) + ): + folded = genie._folded_key(genie.acl_key(principal)) + spelled.setdefault(folded, principal) + readers.add(folded) + per_table.append(readers) + everywhere = set.intersection(*per_table) if per_table else set() + partial = sorted(set().union(*per_table) - everywhere, key=lambda k: spelled[k]) + if partial: + counted = ", ".join( + f"{spelled[k]} ({sum(1 for r in per_table if k in r)} of {len(per_table)})" + for k in partial + ) + warnings.append( + "CAN_RUN goes only to the principals that read every table of the estate, since the " + f"space describes them all; not to {counted}." + ) + if not everywhere: + warnings.append( + "no principal reads every table of the estate, so the space is shared with no one: " + "only the identity that makes it (and workspace admins) can ask it." + ) + return tuple(sorted(spelled[k] for k in everywhere)) + + +def _description(estate_id: str, members: Sequence[Member]) -> str: + names = ", ".join(sorted((m.name for m in members), key=str.casefold)) + about = f"A Genie space over the data products {names}." + footer = ( + f"Managed by {_COMMAND} --estate {estate_id} from {len(members)} data product " + f"contracts: changes made here are replaced on the next run. {estate_marker(estate_id)}" + ) + return f"{about}\n\n{footer}" + + +def _one_line(text: Any) -> str: + return " ".join(str(text or "").split()) + + +# ---------------------------------------------------------------------- converge (live) + + +def converge_estate( + estate: Estate, + say: genie.Say, + *, + client_factory: Optional[genie.ClientFactory] = None, + dry_run: bool = False, +) -> None: + """Create or update the estate's space so it is what the contracts ask, after checking + that each of its tables exists. With ``dry_run`` it reads and prints, and writes nothing. + Raises :class:`databricks_genie.GenieError` (or a refusal) for what it cannot do.""" + if estate.refusals: + raise estate.refusals[0] + space = estate.space + if space is None: # a compile without a refusal always has its space + raise genie.GenieError("the estate compiled to no Genie space.") + with genie.speaking(estate.voice): + genie.require_warehouse(space) + client = genie._client(client_factory) + missing = genie._call( + client, + "read the estate's tables in Unity Catalog", + lambda: [t for t in space.tables if genie._read_table(client, t) is None], + ) + if missing: + products = sorted({estate.owners.get(t, "?") for t in missing}) + raise genie.GenieError( + f"{len(missing)} of the estate's tables do not exist, or this identity cannot " + f"read them: {', '.join(missing)}.", + ( + f"Apply {', '.join(products)} first (fluid apply ), then re-run " + f"{_COMMAND}.", + ), + ) + genie.converge(estate, say, client_factory=lambda: client, dry_run=dry_run) + + +def remove_estate( + estate_id: str, + say: genie.Say, + *, + client_factory: Optional[genie.ClientFactory] = None, + dry_run: bool = False, +) -> int: + """Trash the estate's space, found as a run finds it (the marker, in this identity's own + folder); return how many spaces were (or, with ``dry_run``, would be) trashed.""" + estate_id = str(estate_id or "").strip() + if not _ESTATE_ID.fullmatch(estate_id): + raise invalid_estate_id(estate_id) + token = estate_marker(estate_id) + speaking = voice(estate_id, command="fluid genie remove") + with genie.speaking(speaking): + client = genie._client(client_factory) + say(f"\n {speaking.heading}") + found = genie._call(client, "list the Genie spaces", lambda: genie._find(client, token)) + found = genie._call( + client, + "check whose Genie spaces those are", + lambda: genie._placed(client, found, say)[0], + ) + if not found: + say(f" no Genie space in this identity's folder carries {token}: nothing to trash.") + return 0 + for doc in found: + space_id, title = str(doc.get("space_id")), doc.get("title") + if dry_run: + say(f" would trash {title!r} ({space_id})") + continue + genie._call( + client, + "trash the Genie space", + lambda sid=space_id: client.api_client.do( + "DELETE", f"{genie.SPACES_PATH}/{quote(sid)}" + ), + ) + say(f" trashed {title!r} ({space_id})") + if dry_run: + say(" dry-run: no Genie space was trashed.") + return len(found) diff --git a/tests/iac/_databricks_cli.py b/tests/iac/_databricks_cli.py index 8ff7a569..79caef0e 100644 --- a/tests/iac/_databricks_cli.py +++ b/tests/iac/_databricks_cli.py @@ -110,8 +110,10 @@ def make_contract( return contract -def write_contract(root: Path, document: Dict[str, Any]) -> str: - path = root / "contract.fluid.yaml" +def write_contract(root: Path, document: Dict[str, Any], name: str = "contract.fluid.yaml") -> str: + """Write ``document`` to ``root / name`` (its folders made), returning the path.""" + path = root / name + path.parent.mkdir(parents=True, exist_ok=True) path.write_text(yaml.safe_dump(document, sort_keys=False), encoding="utf-8") return str(path) diff --git a/tests/iac/_fake_databricks.py b/tests/iac/_fake_databricks.py index c12e9517..e2b52c4a 100644 --- a/tests/iac/_fake_databricks.py +++ b/tests/iac/_fake_databricks.py @@ -69,6 +69,7 @@ import base64 import copy +import hashlib import json import os import re @@ -1127,6 +1128,8 @@ def __init__(self, endpoint: str): # ``create-genie-agent.md``: ids of 32 lowercase hex characters, id-keyed lists sorted by id, # tables by identifier, text fields arrays of strings). _GENIE_TEXT_FIELDS = ("question", "content", "description", "sql", "synonyms") +#: The relationship types a join's ``--rt=FROM_RELATIONSHIP_TYPE_--`` tag may name. +_GENIE_RELATIONSHIPS = ("ONE_TO_ONE", "ONE_TO_MANY", "MANY_TO_ONE", "MANY_TO_MANY") def _genie_order_problem(items: List[Any], path: str) -> Optional[str]: @@ -1165,6 +1168,32 @@ def _genie_problem(node: Any, path: str, ids: List[str]) -> Optional[str]: return problem +def _genie_limit_problem(doc: Dict[str, Any]) -> Optional[str]: + """The service's limits on a whole ``serialized_space``: one text instruction, a known + relationship tag on each join, 50 tables. The last two measured on a live workspace + (2026-11-12), with these messages; a join's tables and its unqualified columns are not + checked when a space is made.""" + instructions = doc.get("instructions") or {} + if len(instructions.get("text_instructions") or []) > 1: + return "At most 1 text instruction is allowed" + for spec in instructions.get("join_specs") or []: + for part in spec.get("sql") or []: + tag = re.fullmatch(r"--rt=FROM_RELATIONSHIP_TYPE_([A-Za-z_]+)--", str(part)) + if tag and tag.group(1) not in _GENIE_RELATIONSHIPS: + return ( + f"Failed to parse export proto: FROM_RELATIONSHIP_TYPE_{tag.group(1)} " + "(of class java.lang.String)" + ) + if len((doc.get("data_sources") or {}).get("tables") or []) > 50: + return "A space can have at most 50 tables." + return None + + +def writes_since(fake: "FakeDatabricks", start: int) -> List[Tuple[str, str]]: + """The writes (POST, PATCH, PUT, DELETE) ``fake`` served after its ``start``-th request.""" + return [r for r in fake.requests[start:] if r[0] in ("POST", "PATCH", "PUT", "DELETE")] + + # ---------------------------------------------------------------------- the server class FakeDatabricks: def __init__( @@ -1882,13 +1911,51 @@ def _genie_check(self, raw: Any) -> Optional[str]: return problem if len(ids) != len(set(ids)): return "ids must be unique" - if len((doc.get("instructions") or {}).get("text_instructions") or []) > 1: - return "At most 1 text instruction is allowed" + problem = _genie_limit_problem(doc) + if problem: + return problem for source in (doc.get("data_sources") or {}).get("tables") or []: if source.get("identifier") not in {**self.tables, **self.metric_views}: return f"Table {source.get('identifier')!r} does not exist" return None + def _with_uc_joins(self, stored: str, before: Iterable[str]) -> str: + """A written space as the live service keeps it: for each table that entered it (every + one, on a create), a join added for each Unity Catalog FOREIGN KEY between two of its + tables that touches that table, beside the ones it was sent (the child under its own + name, the parent as ``_2``, MANY_TO_ONE; measured 2026-11-13). ``before``: the + tables the space held.""" + doc = json.loads(stored) + names = {t.get("identifier") for t in (doc.get("data_sources") or {}).get("tables") or []} + new = names - set(before) + added = [] + for child in sorted(n for n in names if n in self.tables): + for c in self.tables[child].get("constraints") or (): + if c["type"] != "FOREIGN KEY" or c["parent"] not in names: + continue + if child not in new and c["parent"] not in new: + continue + left, right = child.rsplit(".", 1)[1], c["parent"].rsplit(".", 1)[1] + "_2" + on = " AND ".join( + f"`{left}`.`{a}` = `{right}`.`{b}`" + for a, b in zip(c["columns"], c["parent_columns"], strict=True) + ) + digest = hashlib.sha256(f"{child}\0{c['name']}".encode()).hexdigest() + added.append( + { + "id": "01f1" + digest[:28], + "left": {"identifier": child, "alias": left}, + "right": {"identifier": c["parent"], "alias": right}, + "sql": [on, "--rt=FROM_RELATIONSHIP_TYPE_MANY_TO_ONE--"], + } + ) + if not added: + return stored + instructions = doc.setdefault("instructions", {}) + joins = [*(instructions.get("join_specs") or []), *added] + instructions["join_specs"] = sorted(joins, key=lambda j: j["id"]) + return json.dumps(doc) + def _genie_title_taken(self, title: str, parent: str, other_than: str = "") -> bool: return any( s["title"] == title and s["parent_path"] == parent and sid != other_than @@ -1950,7 +2017,7 @@ def _genie_create(self, m: Any, q: Any, b: Any) -> Tuple[int, Any]: "description": b.get("description") or "", "warehouse_id": b["warehouse_id"], "parent_path": parent, - "serialized_space": self._as_stored(b["serialized_space"]), + "serialized_space": self._with_uc_joins(self._as_stored(b["serialized_space"]), ()), "etag": self._genie_etag(), } self.genie_acl[space_id] = {} @@ -1981,7 +2048,11 @@ def _genie_patch(self, m: Any, q: Any, b: Any) -> Tuple[int, Any]: problem = self._genie_check(b["serialized_space"]) if problem: return _err(400, "INVALID_PARAMETER_VALUE", problem) - space["serialized_space"] = self._as_stored(b["serialized_space"]) + was = json.loads(space["serialized_space"]).get("data_sources") or {} + space["serialized_space"] = self._with_uc_joins( + self._as_stored(b["serialized_space"]), + [t.get("identifier") for t in was.get("tables") or []], + ) if "title" in b and self._genie_title_taken( b["title"], space["parent_path"], other_than=space_id ): diff --git a/tests/iac/test_iac_databricks_genie.py b/tests/iac/test_iac_databricks_genie.py index 09db07fc..990b4db3 100644 --- a/tests/iac/test_iac_databricks_genie.py +++ b/tests/iac/test_iac_databricks_genie.py @@ -30,7 +30,7 @@ from fluid_build.iac.providers import databricks as dbx from fluid_build.iac.providers import databricks_genie as genie -from ._fake_databricks import FakeDatabricks +from ._fake_databricks import FakeDatabricks, writes_since pytestmark = [pytest.mark.unit, pytest.mark.databricks] @@ -517,11 +517,7 @@ def text(self) -> str: return "\n".join(self.said) def writes(self) -> List[Any]: - return [ - r - for r in self.fake.requests[self.start :] - if r[0] in ("POST", "PATCH", "PUT", "DELETE") - ] + return writes_since(self.fake, self.start) def only_space(fake: FakeDatabricks) -> Dict[str, Any]: @@ -668,6 +664,19 @@ def test_a_metric_view_over_the_table_becomes_a_data_source_once_it_exists(fake) assert "updated 'Site 360': metric views (" in Run(fake)(site_contract()).text +def test_an_expose_taken_out_of_the_product_is_said_as_a_table_not_a_metric_view(fake): + doc = site_contract() + second = copy.deepcopy(doc["exposes"][0]) + second["exposeId"] = "site_history" + second["binding"]["location"]["table"] = "site_history" + doc["exposes"].append(second) + fake.create_table("main.asset.site_history", [("site_id", "string", False)]) + Run(fake)(doc) + run = Run(fake)(site_contract()) # the history expose is gone; its table stays in UC + (line,) = [s for s in run.said if "updated 'Site 360'" in s] + assert "tables" in line and "metric views" not in line + + def test_a_metric_view_sent_where_the_service_does_not_keep_it_would_never_converge(fake): """The stand-in keeps a space as the live service does: a metric view sent under ``data_sources.metric_views`` comes back among the tables (measured 2026-11-12).""" @@ -998,6 +1007,11 @@ def test_a_list_that_never_shows_the_new_space_is_said_not_waited_on_forever( fake.genie_list_lag = 1000 run = Run(fake)(site_contract()) assert "created 'Site 360'" in run.text and "does not show the new space yet" in run.text + # The product's own words, as before the estate's Voice. + assert ( + "an apply in the next minute could make a second one, which the apply after refuses " + "until one is trashed." in run.text + ) def test_a_title_another_space_has_is_kept_as_the_service_suffixed_it_and_said(fake): diff --git a/tests/iac/test_iac_databricks_genie_estate.py b/tests/iac/test_iac_databricks_genie_estate.py new file mode 100644 index 00000000..29aa635d --- /dev/null +++ b/tests/iac/test_iac_databricks_genie_estate.py @@ -0,0 +1,832 @@ +# Copyright 2025 Agentics Transformation Limited +# SPDX-License-Identifier: Apache-2.0 + +"""An estate's Databricks Genie space: one space over several products, joined on their keys. + +No ``tofu``: :func:`databricks.genie_estate` over three tower-shaped products (a site master, a +tenancy that references it, a billing ledger that references both), then +:func:`databricks_genie_estate.converge_estate` and ``remove_estate`` against +``_fake_databricks.FakeDatabricks``, and ``fluid genie`` through the real entry point. Pinned: +what the space holds (every table, the joins from the declared foreign keys, the questions once +each, one instruction, no measure snippets), who gets ``CAN_RUN`` (only the readers of every +table), each reference that makes no join said, the refusals, a first run that creates and +shares and a second that changes nothing, a dropped product that updates the space, a missing +table named with its product before anything is written, the estate's own marker and messages, +and ``fluid genie remove``. +""" + +from __future__ import annotations + +import copy +import json +import re +from pathlib import Path +from typing import Any, Dict, Iterator, List, Optional, Sequence + +import pytest + +from fluid_build.cli import genie as genie_cli +from fluid_build.cli._common import CLIError +from fluid_build.iac.providers import databricks as dbx +from fluid_build.iac.providers import databricks_enforce +from fluid_build.iac.providers import databricks_genie as genie +from fluid_build.iac.providers import databricks_genie_estate as estate_mod + +from ._databricks_cli import isolate_workspace, run_fluid, write_contract +from ._fake_databricks import FakeDatabricks, writes_since + +pytestmark = [pytest.mark.unit, pytest.mark.databricks] + +ENV = {"DATABRICKS_WAREHOUSE_ID": "wh-1"} +ESTATE = "towers" +MARKER = "[fluid-genie-estate:towers]" +SITES = "bronze.asset.site_master_v1" +TENANCY = "silver.commercial.active_tenancy_v1" +BILLING = "silver.finance.tenant_billing_v1" +SITES_T = "main.bronze_site_master.site_master" +TENANCY_T = "main.silver_active_tenancy.active_tenancy" +BILLING_T = "main.silver_tenant_billing.tenant_billing" +STEWARDS = "group:stewards" +ANALYSTS = "group:analysts" +FINANCE = "group:finance" + + +def column( + name: str, + type_: str = "STRING", + *, + required: bool = False, + pk: bool = False, + fk: Optional[str] = None, + **extra: Any, +) -> Dict[str, Any]: + out: Dict[str, Any] = {"name": name, "type": type_, **extra} + if required or pk: + out["required"] = True + tags = (["primary-key"] if pk else []) + (["foreign-key"] if fk else []) + if tags: + out["tags"] = tags + if fk: + out["labels"] = {**out.get("labels", {}), "references": fk} + return out + + +def product( + product_id: str, + name: str, + schema: str, + table: str, + columns: List[Dict[str, Any]], + *, + readers: Sequence[str] = (STEWARDS, ANALYSTS), + questions: str = "", + agent_policy: Optional[Dict[str, Any]] = None, + catalog: str = "main", +) -> Dict[str, Any]: + policy: Dict[str, Any] = {"authz": {"readers": list(readers)}} + if agent_policy is not None: + policy["agentPolicy"] = agent_policy + exposure: Dict[str, Any] = { + "exposeId": table, + "kind": "table", + "title": name, + "description": f"{name}: one row per record.", + "binding": { + "platform": "databricks", + "format": "delta_table", + "location": {"database": catalog, "schema": schema, "table": table}, + }, + "policy": policy, + "contract": {"schema": columns}, + } + if questions: + exposure["semantics"] = { + "labels": {"ai-example-questions": questions, "ai-instructions": f"About {name}."} + } + return { + "fluidVersion": "0.7.6", + "kind": "DataProduct", + "id": product_id, + "name": name, + "description": f"The {name} product.", + "metadata": {"owner": {"team": "towers"}, "layer": "Gold"}, + "exposes": [exposure], + } + + +def sites(**kw: Any) -> Dict[str, Any]: + return product( + SITES, + "Site Master", + "bronze_site_master", + "site_master", + [column("site_id", pk=True), column("site_name"), column("country")], + questions="How many sites per country? | Which sites are on air?", + **kw, + ) + + +def tenancy(**kw: Any) -> Dict[str, Any]: + return product( + TENANCY, + "Active Tenancy", + "silver_active_tenancy", + "active_tenancy", + [ + column("tenancy_id", pk=True), + column("site_id", required=True, fk=f"{SITES}.site_id"), + column("operator"), + ], + # The second repeats a site question in another case: the space asks it once. + questions="Which operators hold the most tenancies? | how many sites per country?", + **kw, + ) + + +def billing(**kw: Any) -> Dict[str, Any]: + kw.setdefault("readers", (STEWARDS, FINANCE)) + return product( + BILLING, + "Tenant Billing", + "silver_tenant_billing", + "tenant_billing", + [ + column("invoice_line_id", pk=True), + column("tenancy_id", fk=f"{TENANCY}.tenancy_id"), + column("site_id", fk=f"{SITES}.site_id"), + column("amount", "DECIMAL(14,2)"), + ], + questions="What is the unbilled energy per site?", + **kw, + ) + + +def three() -> List[Dict[str, Any]]: + return [sites(), tenancy(), billing()] + + +def estate( + *docs: Dict[str, Any], env: Optional[Dict[str, str]] = None, **kw: Any +) -> estate_mod.Estate: + return dbx.genie_estate( + kw.pop("estate_id", ESTATE), + [(f"{d.get('id') or 'unnamed'}.fluid.yaml", d) for d in docs], + ENV if env is None else env, + **kw, + ) + + +def space_of(*docs: Dict[str, Any], **kw: Any) -> genie.Space: + result = estate(*docs, **kw) + assert result.refusals == [], result.refusals + assert result.space is not None + return result.space + + +def joins_of(space: genie.Space) -> List[Dict[str, Any]]: + return space.serialized["instructions"].get("join_specs") or [] + + +def text_of(space: genie.Space) -> str: + return "".join(space.serialized["instructions"]["text_instructions"][0]["content"]) + + +def refusal_text(result: estate_mod.Estate) -> str: + return "\n".join(f"{r.kind}: {r} {' '.join(r.remediation)}" for r in result.refusals) + + +# ---------------------------------------------------------------------- what the space holds + + +def test_the_space_holds_every_products_table_joined_on_their_foreign_keys(): + space = space_of(*three()) + + assert space.tables == (SITES_T, TENANCY_T, BILLING_T) + assert [t["identifier"] for t in space.serialized["data_sources"]["tables"]] == sorted( + space.tables + ) + assert space.title == ESTATE + assert space.marker == MARKER and space.description.endswith(MARKER) + assert space.description.startswith( + "A Genie space over the data products Active Tenancy, Site Master, Tenant Billing." + ) + + joins = joins_of(space) + assert [j["id"] for j in joins] == sorted(j["id"] for j in joins) + assert all(re.fullmatch(r"[0-9a-f]{32}", j["id"]) for j in joins) + by_pair = {(j["left"]["identifier"], j["right"]["identifier"]): j for j in joins} + assert set(by_pair) == {(TENANCY_T, SITES_T), (BILLING_T, TENANCY_T), (BILLING_T, SITES_T)} + # Every column qualified by its table's alias, the referencing table on the left. + assert by_pair[(BILLING_T, TENANCY_T)]["left"] == { + "identifier": BILLING_T, + "alias": "tenant_billing", + } + assert by_pair[(BILLING_T, TENANCY_T)]["right"] == { + "identifier": TENANCY_T, + "alias": "active_tenancy", + } + assert by_pair[(BILLING_T, TENANCY_T)]["sql"] == [ + "`tenant_billing`.`tenancy_id` = `active_tenancy`.`tenancy_id`", + "--rt=FROM_RELATIONSHIP_TYPE_MANY_TO_ONE--", + ] + assert by_pair[(TENANCY_T, SITES_T)]["sql"][0] == ( + "`active_tenancy`.`site_id` = `site_master`.`site_id`" + ) + # A space of several tables carries no measure snippet (its columns would need qualifying). + assert "sql_snippets" not in space.serialized["instructions"] + + +def test_each_question_is_asked_once_and_the_instruction_has_a_section_per_product(): + space = space_of(*three()) + questions = space.serialized["config"]["sample_questions"] + assert sorted(q["question"][0] for q in questions) == [ + "How many sites per country?", + "What is the unbilled energy per site?", + "Which operators hold the most tenancies?", + "Which sites are on air?", + ] + assert [q["id"] for q in questions] == sorted(q["id"] for q in questions) + + (instruction,) = space.serialized["instructions"]["text_instructions"] + assert re.fullmatch(r"[0-9a-f]{32}", instruction["id"]) + text = text_of(space) + for table, label in ( + (SITES_T, "Site Master, data product bronze.asset.site_master_v1"), + (TENANCY_T, "Active Tenancy, data product silver.commercial.active_tenancy_v1"), + (BILLING_T, "Tenant Billing, data product silver.finance.tenant_billing_v1"), + ): + assert f"About {table} ({label}):\n" in text + assert " About Site Master.\n" in text and " Primary key:" not in text + assert text.endswith(genie.ACCESS_LINE + "\n") + + +def test_only_the_principals_that_read_every_table_get_can_run(): + result = estate(*three()) + # Stewards read all three; analysts the sites and tenancies; finance only the billing. + assert result.space.readers == ("stewards",) + (note,) = [w for w in result.warnings if "CAN_RUN" in w] + assert note.endswith("not to analysts (2 of 3), finance (1 of 3).") + + +def test_no_principal_reading_every_table_shares_with_no_one_and_says_so(): + result = estate(sites(readers=(ANALYSTS,)), billing(readers=(FINANCE,))) + assert result.space.readers == () + assert any("shared with no one" in w for w in result.warnings) + + +def test_a_title_of_its_own_on_one_line(): + space = space_of(*three(), title=" Tower Management ") + assert space.title == "Tower Management" + assert space.description.endswith(MARKER) + + +def test_a_foreign_key_that_is_the_tables_own_primary_key_is_one_to_one(): + site_360 = product( + "gold.asset.site_360_v1", + "Site 360", + "gold_site_360", + "site_360", + [column("site_id", pk=True, fk=f"{SITES}.site_id"), column("tenancy_count", "INTEGER")], + ) + (join,) = joins_of(space_of(sites(), site_360)) + assert join["sql"] == [ + "`site_360`.`site_id` = `site_master`.`site_id`", + "--rt=FROM_RELATIONSHIP_TYPE_ONE_TO_ONE--", + ] + + +def test_a_composite_foreign_key_is_one_join_on_all_its_columns(): + meters = product( + "bronze.iot.meters_v1", + "Meters", + "bronze_meters", + "meters", + [column("site_id", pk=True), column("meter_no", pk=True), column("kind")], + ) + readings = product( + "bronze.iot.readings_v1", + "Readings", + "bronze_readings", + "readings", + [ + column("reading_id", pk=True), + column("site_id", fk="bronze.iot.meters_v1.site_id"), + column("meter_no", fk="bronze.iot.meters_v1.meter_no"), + ], + ) + (join,) = joins_of(space_of(meters, readings)) + assert join["sql"] == [ + "`readings`.`site_id` = `meters`.`site_id` AND `readings`.`meter_no` = `meters`.`meter_no`", + "--rt=FROM_RELATIONSHIP_TYPE_MANY_TO_ONE--", + ] + + +def test_a_foreign_key_to_its_own_table_joins_a_second_alias(): + hierarchy = sites() + hierarchy["exposes"][0]["contract"]["schema"].append( + column("parent_site_id", fk=f"{SITES}.site_id") + ) + (join,) = joins_of(space_of(hierarchy)) + assert join["left"] == {"identifier": SITES_T, "alias": "site_master"} + assert join["right"] == {"identifier": SITES_T, "alias": "site_master_parent_site_id"} + assert ( + join["sql"][0] == "`site_master`.`parent_site_id` = `site_master_parent_site_id`.`site_id`" + ) + + +def test_a_reference_that_makes_no_join_is_said(): + outside = estate(tenancy()) # its site_id references a product the estate does not hold + assert outside.refusals == [] and joins_of(outside.space) == [] + assert any(f"references {SITES}, which is not in the estate" in w for w in outside.warnings) + + keyless = sites() + keyless["exposes"][0]["contract"]["schema"][0].pop("tags") + result = estate(keyless, tenancy()) + assert joins_of(result.space) == [] + assert any(f"references {SITES}, which declares no primary key" in w for w in result.warnings) + + by_country = tenancy() + by_country["exposes"][0]["contract"]["schema"].append(column("country", fk=f"{SITES}.country")) + result = estate(sites(), by_country) + assert len(joins_of(result.space)) == 1 # site_id still joins + assert any( + f"column country references {SITES}.country, which is not the primary key of exactly " + f"one table of {SITES}" in w + for w in result.warnings + ) + + +def test_what_a_tables_keys_lose_is_said_as_fluid_apply_says_it(): + nullable = sites() + nullable["exposes"][0]["contract"]["schema"][0].pop("required") + result = estate(nullable, tenancy()) + assert joins_of(result.space) == [] + assert any( + w.startswith(f"{SITES} exposes[0] (site_master) tags site_id primary-key without required") + for w in result.warnings + ) + + garbled = tenancy() + garbled["exposes"][0]["contract"]["schema"][1]["labels"]["references"] = "site_id" + result = estate(sites(), garbled) + assert joins_of(result.space) == [] + assert any( + "labels.references is 'site_id', not ." in w for w in result.warnings + ) + + +def test_tables_of_one_name_get_aliases_of_their_own(): + a = product( + "a.orders_v1", "A", "sales", "orders", [column("order_id", pk=True), column("item")] + ) + b = product( + "b.orders_v1", + "B", + "ops", + "orders", + [column("order_id", pk=True), column("sales_order", fk="a.orders_v1.order_id")], + ) + (join,) = joins_of(space_of(a, b)) + assert join["left"]["alias"] == "ops_orders" and join["right"]["alias"] == "sales_orders" + assert join["sql"][0] == "`ops_orders`.`sales_order` = `sales_orders`.`order_id`" + + +def test_table_aliases_stay_unique_however_the_names_collide(): + names = ["x.a_b.c", "x.a.b_c", "y.a.b_c", "z.q.c"] + aliases = estate_mod.table_aliases(names) + assert len(set(aliases.values())) == len(names) + assert aliases["z.q.c"] == "q_c" + + +# ---------------------------------------------------------------------- refusals + + +@pytest.mark.parametrize("bad", ["", "tower management", "[x]", "-towers", "a" * 129]) +def test_an_estate_id_the_marker_cannot_carry_is_refused(bad): + result = estate(*three(), estate_id=bad) + assert result.space is None + (refusal,) = result.refusals + assert refusal.kind == "databricks-genie-invalid" and "is not an estate id" in str(refusal) + + +def test_what_the_estate_cannot_hold_is_refused(): + assert "no contract was given" in refusal_text(estate()) + + nameless = sites() + nameless.pop("id") + assert "has no id" in refusal_text(estate(nameless, tenancy())) + + assert f"are both {SITES}" in refusal_text(estate(sites(), sites())) + + local = sites() + local["exposes"][0]["binding"] = {"platform": "local", "format": "csv", "location": {}} + text = refusal_text(estate(local, tenancy())) + assert "binds no table expose to databricks" in text and "Leave it out of the estate" in text + + twin = tenancy() + twin["exposes"][0]["binding"]["location"] = { + "database": "main", + "schema": "bronze_site_master", + "table": "site_master", + } + assert f"{SITES_T} is a table of both" in refusal_text(estate(sites(), twin)) + + +def test_a_product_whose_agent_policy_rules_genie_out_refuses_the_estate(): + result = estate(sites(), tenancy(agent_policy={"deniedUseCases": ["qa"]})) + assert result.space is None + (refusal,) = result.refusals + assert refusal.kind == "databricks-genie-policy" + assert str(refusal).startswith(f"{TENANCY} exposes[0] (active_tenancy) policy.agentPolicy") + assert refusal.remediation[0].startswith("Leave that product out of the estate, or allow") + + +def test_more_tables_than_a_space_holds_are_refused(monkeypatch): + monkeypatch.setattr(genie, "MAX_TABLES", 2) + text = refusal_text(estate(*three())) + assert "the estate has 3 Databricks tables" in text and "at most 2" in text + + +def test_a_refusal_every_contract_shares_is_said_once(): + result = estate(*three(), env={}) + (refusal,) = [r for r in result.refusals if "DATABRICKS_WAREHOUSE_ID" in str(r)] + assert str(refusal).startswith( + f"{SITES}.fluid.yaml, {TENANCY}.fluid.yaml, {BILLING}.fluid.yaml: " + ) + + +def test_a_text_longer_than_genie_holds_is_refused(): + wordy = tenancy() + wordy["exposes"][0]["semantics"]["labels"]["ai-instructions"] = "x" * (genie.MAX_STRING + 1) + result = estate(sites(), wordy) + assert result.space is None + text = refusal_text(result) + assert "databricks-genie-limit" in text and "at most 25,000" in text + + +# ---------------------------------------------------------------------- against the stand-in + + +@pytest.fixture +def fake(monkeypatch) -> Iterator[FakeDatabricks]: + server = FakeDatabricks(catalogs=("main",)).start() + monkeypatch.setenv(dbx.WAREHOUSE_ENV, server.warehouse_id) + monkeypatch.setenv( + "DATABRICKS_TOKEN", "dapi-genie-estate-test-token" + ) # pragma: allowlist secret + for table, columns in ( + (SITES_T, [("site_id", "string", False), ("site_name", "string", True)]), + (TENANCY_T, [("tenancy_id", "string", False), ("site_id", "string", False)]), + (BILLING_T, [("invoice_line_id", "string", False), ("tenancy_id", "string", True)]), + ): + server.create_table(table, columns + [("extra", "string", True)]) + try: + yield server + finally: + server.stop() + + +class Run: + def __init__(self, fake: FakeDatabricks): + self.fake = fake + self.said: List[str] = [] + self.start = len(fake.requests) + + def __call__(self, *docs: Dict[str, Any], **kw: Any) -> "Run": + compiled = dbx.genie_estate( + ESTATE, [(d["id"], d) for d in docs], {dbx.WAREHOUSE_ENV: self.fake.warehouse_id} + ) + estate_mod.converge_estate( + compiled, self.said.append, client_factory=self.fake.client, **kw + ) + return self + + def remove(self, **kw: Any) -> int: + return estate_mod.remove_estate( + ESTATE, self.said.append, client_factory=self.fake.client, **kw + ) + + @property + def text(self) -> str: + return "\n".join(self.said) + + def writes(self) -> List[Any]: + return writes_since(self.fake, self.start) + + +def only_space(fake: FakeDatabricks) -> Dict[str, Any]: + (space,) = fake.genie_spaces.values() + return space + + +def test_a_first_run_creates_and_shares_and_a_second_changes_nothing(fake): + first = Run(fake)(*three()) + space = only_space(fake) + assert space["title"] == ESTATE and space["description"].endswith(MARKER) + assert "Databricks Genie space for the estate towers (Genie API):" in first.text + assert "created 'towers': 3 table(s), 4 sample question(s), 3 join(s)" in first.text + assert fake.genie_acl[space["space_id"]] == {("group_name", "stewards"): "CAN_RUN"} + stored = json.loads(space["serialized_space"]) + assert len(stored["instructions"]["join_specs"]) == 3 + assert [w[0] for w in first.writes()] == ["POST", "PUT"] + + second = Run(fake)(*three()) + assert second.writes() == [] and "'towers': no change" in second.text + assert fake.unknown == [] + + +def test_joins_genie_adds_from_unity_catalog_keys_are_taken_out_at_once(fake): + # Whenever a table enters a space, Genie adds a join for each Unity Catalog foreign key that + # touches it, beside the ones it is sent (measured 2026-11-13); left so, every next run would + # update the space. + for child, column, parent in ( + (TENANCY_T, "site_id", SITES_T), + (BILLING_T, "tenancy_id", TENANCY_T), + ): + fake.tables[child]["constraints"].append( + { + "type": "FOREIGN KEY", + "name": f"fluid_fk_{child.rsplit('.', 1)[1]}__{column}", + "columns": [column], + "parent": parent, + "parent_columns": [column], + } + ) + ours = space_of(*three()).serialized["instructions"]["join_specs"] + + first = Run(fake)(*three()) + assert [w[0] for w in first.writes()] == ["POST", "PATCH", "PUT"] + assert "Genie added 2 join(s) of its own as it made the space" in first.text + assert "the space holds the 3 the contracts declare" in first.text + assert json.loads(only_space(fake)["serialized_space"])["instructions"]["join_specs"] == ours + rerun = Run(fake)(*three()) + assert rerun.writes() == [] + + Run(fake)(sites(), tenancy()) # billing leaves the space + back = Run(fake)(*three()) # and comes back: Genie joins it again + assert [w[0] for w in back.writes()] == ["PATCH", "PATCH", "PUT"] + assert "Genie added 1 join(s) of its own as it updated the space" in back.text + assert json.loads(only_space(fake)["serialized_space"])["instructions"]["join_specs"] == ours + again = Run(fake)(*three()) + assert again.writes() == [] and "'towers': no change" in again.text + + +def test_a_product_left_out_of_a_run_leaves_the_space(fake): + Run(fake)(*three()) + run = Run(fake)(sites(), tenancy()) + assert [w[0] for w in run.writes()] == ["PATCH", "PUT"] + (line,) = [s for s in run.said if "updated 'towers'" in s] + assert "tables" in line and "joins" in line and "instructions" in line + stored = json.loads(only_space(fake)["serialized_space"]) + assert [t["identifier"] for t in stored["data_sources"]["tables"]] == [SITES_T, TENANCY_T] + # Analysts read both tables now, so they hold CAN_RUN too. + assert fake.genie_acl[only_space(fake)["space_id"]] == { + ("group_name", "analysts"): "CAN_RUN", + ("group_name", "stewards"): "CAN_RUN", + } + rerun = Run(fake)(sites(), tenancy()) + assert rerun.writes() == [] + + +def test_a_table_not_applied_yet_is_named_with_its_product_before_anything_is_written(fake): + del fake.tables[BILLING_T] + run = Run(fake) + with pytest.raises(genie.GenieError) as exc: + run(*three()) + assert ( + f"1 of the estate's tables do not exist, or this identity cannot read them: {BILLING_T}" + in (str(exc.value)) + ) + assert exc.value.remediation == ( + f"Apply {BILLING} first (fluid apply ), then re-run fluid genie apply.", + ) + assert run.writes() == [] and fake.genie_spaces == {} + + +def test_a_dry_run_reads_and_writes_nothing(fake): + run = Run(fake)(*three(), dry_run=True) + assert run.writes() == [] and fake.genie_spaces == {} + assert "would create 'towers': 3 table(s), 4 sample question(s), 3 join(s)" in run.text + assert "would share it (CAN_RUN) with stewards" in run.text + + +def test_remove_trashes_the_estates_space_and_nothing_else(fake): + product_space = fake.add_genie_space("Site Master", f"Its own. [fluid-genie:{SITES}]") + other_estate = fake.add_genie_space("Other", "Another. [fluid-genie-estate:other]") + Run(fake)(*three()) + estate_id = next( + sid for sid, s in fake.genie_spaces.items() if s["description"].endswith(MARKER) + ) + + dry = Run(fake) + removed = dry.remove(dry_run=True) + assert removed == 1 and dry.writes() == [] + assert f"would trash 'towers' ({estate_id})" in dry.text + + run = Run(fake) + removed = run.remove() + assert removed == 1 + assert run.writes() == [("DELETE", f"/api/2.0/genie/spaces/{estate_id}")] + assert set(fake.genie_spaces) == {product_space, other_estate} + + again = Run(fake) + removed = again.remove() + assert removed == 0 and again.writes() == [] + assert f"no Genie space in this identity's folder carries {MARKER}: nothing to trash." in ( + again.text + ) + + +def test_a_product_space_and_an_estate_space_never_touch_each_other(fake): + Run(fake)(*three()) + estate_space = only_space(fake)["space_id"] + product_doc = copy.deepcopy(sites()) + product_doc["extensions"] = {genie.EXTENSION_KEY: {}} + env = {dbx.WAREHOUSE_ENV: fake.warehouse_id} + genie.converge( + dbx._genie(product_doc, dbx._compile(product_doc, env), env), + lambda _line: None, + client_factory=fake.client, + ) + assert len(fake.genie_spaces) == 2 + Run(fake)(*three()) # the estate's run still finds its own space, and changes nothing + assert len(fake.genie_spaces) == 2 and estate_space in fake.genie_spaces + + +def test_two_spaces_carrying_the_estates_marker_stop_the_run(fake): + fake.add_genie_space("One", f"A copy. {MARKER}") + fake.add_genie_space("Two", f"Another copy. {MARKER}") + with pytest.raises(genie.GenieError) as exc: + Run(fake)(*three()) + assert "fluid genie apply keeps one space per estate and cannot tell which is the estate's" in ( + str(exc.value) + ) + + +def test_a_space_planted_with_the_marker_in_another_folder_is_left_alone(fake): + planted = fake.add_genie_space( + "Planted", f"Fill me. {MARKER}", parent_path="/Users/someone.else@example.com" + ) + run = Run(fake)(*three()) + assert f"left alone: space {planted} ('Planted') carries {MARKER}" in run.text + assert len(fake.genie_spaces) == 2 and fake.genie_acl[planted] == {} + + +def test_a_direct_editor_is_kept_and_said_in_the_estates_words(fake): + Run(fake)(*three()) + space_id = only_space(fake)["space_id"] + fake.genie_acl[space_id][("user_name", "editor@towers.example")] = "CAN_EDIT" + run = Run(fake)(*three()) + assert "editor@towers.example also hold CAN_EDIT or CAN_MANAGE" in run.text + assert "fluid genie apply keeps them" in run.text + assert fake.genie_acl[space_id][("user_name", "editor@towers.example")] == "CAN_EDIT" + + +def test_an_edit_in_between_says_to_re_run_the_estates_command(fake): + Run(fake)(*three()) + fake.fail_route("PATCH", "/api/2.0/genie/spaces", 409, "etag does not match") + with pytest.raises(genie.GenieError) as exc: + Run(fake)(sites(), tenancy()) + assert "changed while this run was updating it" in str(exc.value) + assert exc.value.remediation == ("Re-run fluid genie apply.",) + # A product's space speaks as before once the estate's run is over. + assert genie._remediation(400, "x") == "Fix what the message names, then re-run fluid apply." + + +def test_the_stand_in_refuses_what_the_service_refuses(fake): + space = space_of(*three()) + serialized = copy.deepcopy(space.serialized) + serialized["instructions"]["join_specs"][0]["sql"][1] = "--rt=FROM_RELATIONSHIP_TYPE_SOME--" + client = fake.client() + with pytest.raises(Exception, match="Failed to parse export proto"): + client.api_client.do("POST", genie.SPACES_PATH, body=genie._body(space, serialized)) + many = copy.deepcopy(space.serialized) + many["data_sources"]["tables"] = [{"identifier": f"main.s.t{i:02d}"} for i in range(51)] + with pytest.raises(Exception, match="at most 50 tables"): + client.api_client.do("POST", genie.SPACES_PATH, body=genie._body(space, many)) + + +# ---------------------------------------------------------------------- fluid genie + + +def test_contract_paths_reads_files_globs_and_folders_once_each(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + write_contract(tmp_path, sites(), "pack/bronze/dp001_site_master.fluid.yaml") + write_contract(tmp_path, tenancy(), "pack/silver/dp005_active_tenancy.fluid.yaml") + write_contract(tmp_path, {"not": "a contract"}, "pack/silver/notes.yaml") + write_contract(tmp_path, billing(), "pack/node_modules/x.fluid.yaml") + (tmp_path / "pack/link.fluid.yaml").symlink_to( + tmp_path / "pack/bronze/dp001_site_master.fluid.yaml" + ) + + expected = [ + "pack/bronze/dp001_site_master.fluid.yaml", + "pack/silver/dp005_active_tenancy.fluid.yaml", + ] + for items in (["pack"], ["pack/**/*.fluid.yaml"], ["pack", "pack/**/*.fluid.yaml"]): + assert [label for label, _ in genie_cli.contract_paths(items)] == expected, items + # A file named outright is read as named, once however often it is named. + one = "pack/silver/dp005_active_tenancy.fluid.yaml" + assert [label for label, _ in genie_cli.contract_paths([one, one, "pack/silver"])] == [one] + (tmp_path / "empty").mkdir() + with pytest.raises(CLIError) as exc: + genie_cli.contract_paths(["empty"]) + assert exc.value.event == "genie_estate_no_contracts" + with pytest.raises(CLIError) as exc: + genie_cli.contract_paths(["missing.fluid.yaml"]) + assert exc.value.event == "contract_file_not_found" + + +@pytest.fixture +def workspace(monkeypatch, tmp_path: Path, fake: FakeDatabricks) -> Path: + isolate_workspace( + monkeypatch, + tmp_path, + host=fake.endpoint, + warehouse_id=fake.warehouse_id, + plugin_cache=str(tmp_path / "cache"), + ) + monkeypatch.setattr(databricks_enforce, "default_client", fake.client) + for name, doc in ( + ("pack/bronze/dp001.fluid.yaml", sites()), + ("pack/silver/dp005.fluid.yaml", tenancy()), + ("pack/silver/dp016.fluid.yaml", billing()), + ): + write_contract(tmp_path, doc, name) + return tmp_path + + +def test_fluid_genie_apply_then_again_then_remove(workspace, fake, capfd): + rc, out = run_fluid(capfd, "genie", "apply", "pack", "--estate", ESTATE, "--dry-run") + assert rc == 0, out + assert "Genie estate towers: 3 contract(s), 3 table(s), 3 join(s)" in out + assert "would create 'towers'" in out and fake.genie_spaces == {} + assert "not to analysts (2 of 3), finance (1 of 3)" in out + + rc, out = run_fluid(capfd, "genie", "apply", "pack", "--estate", ESTATE, "--title", "Towers") + assert rc == 0, out + assert "created 'Towers': 3 table(s), 4 sample question(s), 3 join(s)" in out + assert only_space(fake)["description"].endswith(MARKER) + + rc, out = run_fluid(capfd, "genie", "apply", "pack", "--estate", ESTATE, "--title", "Towers") + assert rc == 0 and "'Towers': no change" in out + + rc, out = run_fluid(capfd, "genie", "remove", "--estate", ESTATE) + assert rc == 0 and "trashed 'Towers'" in out + assert fake.genie_spaces == {} and fake.unknown == [] + + +def test_fluid_genie_refusals_and_failures_are_typed_errors_with_their_fix(workspace, fake, capfd): + rc, out = run_fluid(capfd, "genie", "apply", "pack", "--estate", "two words") + assert rc == 1 and "ERR_GENIE_ESTATE_REFUSED" in out and "is not an estate id" in out + + fake.fail_route("POST", "/api/2.0/genie/spaces", 403, "PERMISSION_DENIED: Genie is disabled") + rc, out = run_fluid(capfd, "genie", "apply", "pack", "--estate", ESTATE) + assert rc == 1 and "ERR_GENIE_ESTATE_FAILED" in out + assert "then re-run fluid genie apply" in out and "Genie (AI/BI) is enabled" in out + assert fake.genie_spaces == {} + + +def test_a_glob_reaches_no_ignored_folder_it_names_and_follows_no_symlink_loop( + tmp_path, monkeypatch +): + monkeypatch.chdir(tmp_path) + write_contract(tmp_path, sites(), "ok/a.fluid.yaml") + write_contract(tmp_path, tenancy(), "target/b.fluid.yaml") + assert [label for label, _ in genie_cli.contract_paths(["*"])] == ["ok/a.fluid.yaml"] + # Two symlink loops made a recursive glob run forever; a glob is one level per ``**``. + (tmp_path / "ok" / "loop0").symlink_to(tmp_path) + (tmp_path / "ok" / "loop1").symlink_to(tmp_path) + labels = [label for label, _ in genie_cli.contract_paths(["**/*.fluid.yaml"])] + assert labels == ["ok/a.fluid.yaml"] + + +def test_a_failed_remove_says_to_re_run_remove(fake): + fake.fail_route("GET", "/api/2.0/genie/spaces", 500, "INTERNAL_ERROR: down") + with pytest.raises(genie.GenieError) as exc: + Run(fake).remove() + assert exc.value.remediation == ( + "The workspace is busy or unavailable: re-run fluid genie remove in a minute.", + ) + + +def test_a_contract_fluid_apply_would_refuse_stops_the_run_with_its_file(workspace, fake, capfd): + broken = tenancy() + broken["exposes"][0]["policy"] = "readers: everyone" + write_contract(workspace, broken, "bad/shape/dp005.fluid.yaml") + rc, out = run_fluid(capfd, "genie", "apply", "pack", "bad/shape", "--estate", ESTATE) + assert rc == 1 and "ERR_GENIE_CONTRACT_INVALID" in out + assert "bad/shape/dp005.fluid.yaml" in out + + old = sites() + old["fluidVersion"] = "0.5.7" + write_contract(workspace, old, "bad/old/dp001.fluid.yaml") + rc, out = run_fluid(capfd, "genie", "apply", "bad/old", "--estate", ESTATE) + assert rc == 1 and "ERR_CONTRACT_VERSION_UNSUPPORTED" in out + + garbled = workspace / "bad" / "yaml" / "dp016.fluid.yaml" + garbled.parent.mkdir(parents=True) + garbled.write_text("id: [unclosed\n", encoding="utf-8") + rc, out = run_fluid(capfd, "genie", "apply", "bad/yaml", "--estate", ESTATE) + assert rc == 1 and "ERR_CONTRACT_LOAD_FAILED" in out and "bad/yaml/dp016.fluid.yaml" in out + assert fake.genie_spaces == {} diff --git a/tests/iac/test_iac_databricks_genie_estate_fake_e2e.py b/tests/iac/test_iac_databricks_genie_estate_fake_e2e.py new file mode 100644 index 00000000..2561ef06 --- /dev/null +++ b/tests/iac/test_iac_databricks_genie_estate_fake_e2e.py @@ -0,0 +1,156 @@ +# Copyright 2025 Agentics Transformation Limited +# SPDX-License-Identifier: Apache-2.0 + +"""An estate's Genie space through the real ``fluid`` CLI and a real ``tofu``, against the +in-process stand-in (``_fake_databricks.py``). + +The journey of a platform team: two products applied with ``fluid apply`` (their tables, and +the Unity Catalog keys between them), ``fluid genie apply`` over both before the second exists +(refused, naming it), then a dry run, the space with the join from the declared foreign key, a +second run that changes nothing, and ``fluid genie remove``. The SDK client is the stand-in's +(``databricks_enforce.default_client`` patched, the seam the Genie modules read it through). +""" + +from __future__ import annotations + +import copy +import json +from pathlib import Path +from typing import Any, Dict, Iterator + +import pytest + +from fluid_build.iac import runner +from fluid_build.iac.providers import databricks_enforce + +from ._databricks_cli import ( # noqa: F401 — plugin_cache is a fixture + apply_contract, + isolate_workspace, + make_contract, + plugin_cache, + run_fluid, + write_contract, +) +from ._fake_databricks import FakeDatabricks + +pytestmark = [ + pytest.mark.integration, + pytest.mark.provider, + pytest.mark.databricks, + pytest.mark.skipif(runner.tofu_path() is None, reason="tofu binary not installed"), +] + +SITES, TENANCY = "dbx.fake.sites", "dbx.fake.tenancy" +SITES_T, TENANCY_T = "main.sites.site_master", "main.tenancy.active_tenancy" + + +@pytest.fixture +def fake() -> Iterator[FakeDatabricks]: + server = FakeDatabricks().start() + try: + yield server + finally: + server.stop() + + +@pytest.fixture +def workspace(monkeypatch, tmp_path: Path, fake: FakeDatabricks, plugin_cache: str) -> Path: + isolate_workspace( + monkeypatch, + tmp_path, + host=fake.endpoint, + warehouse_id=fake.warehouse_id, + plugin_cache=plugin_cache, + ) + monkeypatch.setattr(databricks_enforce, "default_client", fake.client) + return tmp_path + + +def product(product_id: str, schema: str, table: str, columns, *, readers, question: str): + doc = make_contract(product_id=product_id, readers=readers) + doc["name"] = table.replace("_", " ").title() + exposure = doc["exposes"][0] + exposure["exposeId"] = table + exposure["binding"]["location"].update(schema=schema, table=table) + exposure["contract"]["schema"] = copy.deepcopy(columns) + exposure["semantics"] = {"labels": {"ai-example-questions": question}} + return doc + + +def sites() -> Dict[str, Any]: + return product( + SITES, + "sites", + "site_master", + [ + {"name": "site_id", "type": "STRING", "required": True, "tags": ["primary-key"]}, + {"name": "country", "type": "STRING"}, + ], + readers=("group:analysts", "group:stewards"), + question="How many sites per country?", + ) + + +def tenancy() -> Dict[str, Any]: + return product( + TENANCY, + "tenancy", + "active_tenancy", + [ + {"name": "tenancy_id", "type": "STRING", "required": True, "tags": ["primary-key"]}, + { + "name": "site_id", + "type": "STRING", + "required": True, + "tags": ["foreign-key"], + "labels": {"references": f"{SITES}.site_id"}, + }, + ], + readers=("group:stewards",), + question="Which operators hold the most tenancies per country?", + ) + + +def test_apply_two_products_then_one_space_over_both(workspace, fake, capfd): + rc, out = apply_contract( + capfd, write_contract(workspace, sites(), "products/sites/contract.fluid.yaml") + ) + assert rc == 0, out + tenancy_path = write_contract(workspace, tenancy(), "products/tenancy/contract.fluid.yaml") + + # The second product is not applied yet: the run says which, and writes nothing. + rc, out = run_fluid(capfd, "genie", "apply", "products", "--estate", "towers") + assert rc == 1, out + assert f"Apply {TENANCY} first" in out and fake.genie_spaces == {} + + rc, out = apply_contract(capfd, tenancy_path) + assert rc == 0, out + assert SITES_T in fake.tables and TENANCY_T in fake.tables + + rc, out = run_fluid(capfd, "genie", "apply", "products", "--estate", "towers", "--dry-run") + assert rc == 0, out + assert "Genie estate towers: 2 contract(s), 2 table(s), 1 join(s)" in out + assert "would create 'towers': 2 table(s), 2 sample question(s), 1 join(s)" in out + assert fake.genie_spaces == {} + + rc, out = run_fluid(capfd, "genie", "apply", "products", "--estate", "towers") + assert rc == 0, out + (space,) = fake.genie_spaces.values() + assert space["description"].endswith("[fluid-genie-estate:towers]") + stored = json.loads(space["serialized_space"]) + (join,) = stored["instructions"]["join_specs"] + assert join["left"]["identifier"] == TENANCY_T and join["right"]["identifier"] == SITES_T + assert join["sql"] == [ + "`active_tenancy`.`site_id` = `site_master`.`site_id`", + "--rt=FROM_RELATIONSHIP_TYPE_MANY_TO_ONE--", + ] + # Stewards read both tables; analysts only the sites, so they are named, not shared. + assert fake.genie_acl[space["space_id"]] == {("group_name", "stewards"): "CAN_RUN"} + assert "not to analysts (1 of 2)" in out + + rc, out = run_fluid(capfd, "genie", "apply", "products", "--estate", "towers") + assert rc == 0 and "'towers': no change" in out, out + + rc, out = run_fluid(capfd, "genie", "remove", "--estate", "towers") + assert rc == 0 and f"trashed 'towers' ({space['space_id']})" in out, out + assert fake.genie_spaces == {} and fake.unknown == [] diff --git a/tests/test_cli_help_style.py b/tests/test_cli_help_style.py index ac521924..46d4550f 100644 --- a/tests/test_cli_help_style.py +++ b/tests/test_cli_help_style.py @@ -49,6 +49,7 @@ "demo", "publish", "market", + "genie", "import", "policy-check", "test",