mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 16:28:15 +00:00
Henry Guo
a6a0e29282
* fix(table-catalog): support Spark REST commits * chore(deps): update s3s SigV4 revision --------- Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
648 lines
31 KiB
Python
648 lines
31 KiB
Python
#!/usr/bin/env python3
|
|
"""Unit tests for the RustFS table catalog engine compatibility helper."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import unittest
|
|
|
|
import engine_compatibility
|
|
|
|
|
|
class EngineCompatibilityTest(unittest.TestCase):
|
|
def test_matrix_records_required_engine_scenarios_without_overclaiming(self) -> None:
|
|
matrix = engine_compatibility.engine_compatibility_matrix()
|
|
by_client = {entry["client"]: entry for entry in matrix}
|
|
|
|
self.assertIn("PyIceberg", by_client)
|
|
self.assertIn("Spark Iceberg REST catalog", by_client)
|
|
self.assertIn("Trino Iceberg REST catalog", by_client)
|
|
self.assertIn("DuckDB Iceberg", by_client)
|
|
self.assertIn("StarRocks Iceberg REST catalog", by_client)
|
|
|
|
pyiceberg = by_client["PyIceberg"]
|
|
self.assertEqual(pyiceberg["status"], "automated-smoke")
|
|
self.assertContainsScenario(pyiceberg, "create-namespace", "automated")
|
|
self.assertContainsScenario(pyiceberg, "create-table", "automated")
|
|
self.assertContainsScenario(pyiceberg, "append", "automated")
|
|
self.assertContainsScenario(pyiceberg, "reload-table", "automated")
|
|
self.assertContainsScenario(pyiceberg, "drop-table", "automated-with-cleanup")
|
|
|
|
spark = by_client["Spark Iceberg REST catalog"]
|
|
self.assertEqual(spark["status"], "manual-live-harness")
|
|
self.assertContainsScenario(spark, "create-namespace", "manual-live-harness")
|
|
self.assertContainsScenario(spark, "create-table", "manual-live-harness")
|
|
self.assertContainsScenario(spark, "append", "manual-live-harness")
|
|
self.assertContainsScenario(spark, "reload-table", "manual-live-harness")
|
|
self.assertContainsScenario(spark, "drop-table", "manual-live-harness")
|
|
self.assertContainsScenario(spark, "commit-conflict", "manual-validation-required")
|
|
|
|
trino = by_client["Trino Iceberg REST catalog"]
|
|
self.assertEqual(trino["status"], "manual-live-read-probe")
|
|
self.assertContainsScenario(trino, "catalog-load", "manual-live-probe")
|
|
|
|
def test_spark_config_uses_rustfs_rest_catalog_and_s3fileio(self) -> None:
|
|
config = engine_compatibility.spark_catalog_config(
|
|
endpoint="http://127.0.0.1:9000",
|
|
warehouse="rustfs-s3table-smoke",
|
|
access_key="rustfsadmin",
|
|
secret_key="rustfsadmin",
|
|
region="us-east-1",
|
|
catalog_name="rustfs",
|
|
rest_path="/iceberg",
|
|
rest_signing_name="s3",
|
|
)
|
|
|
|
self.assertEqual(config["spark.sql.catalog.rustfs"], "org.apache.iceberg.spark.SparkCatalog")
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.type"], "rest")
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.uri"], "http://127.0.0.1:9000/iceberg")
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.warehouse"], "rustfs-s3table-smoke")
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.prefix"], "rustfs-s3table-smoke")
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.io-impl"], "org.apache.iceberg.aws.s3.S3FileIO")
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.s3.endpoint"], "http://127.0.0.1:9000")
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.rest.signing-name"], "s3")
|
|
|
|
def test_spark_vendor_config_formats_aws_s3tables_profile(self) -> None:
|
|
config = engine_compatibility.spark_vendor_catalog_config(
|
|
profile="aws-s3tables",
|
|
endpoint="https://s3tables.us-east-1.amazonaws.com",
|
|
warehouse="ignored",
|
|
access_key="rustfsadmin",
|
|
secret_key="rustfsadmin",
|
|
region="us-east-1",
|
|
catalog_name="rustfs",
|
|
account_id="123456789012",
|
|
table_bucket="analytics",
|
|
catalog_uri=None,
|
|
warehouse_name=None,
|
|
)
|
|
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.uri"], "https://s3tables.us-east-1.amazonaws.com/iceberg")
|
|
self.assertEqual(
|
|
config["spark.sql.catalog.rustfs.warehouse"],
|
|
"arn:aws:s3tables:us-east-1:123456789012:bucket/analytics",
|
|
)
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.rest.signing-name"], "s3tables")
|
|
self.assertNotIn("spark.sql.catalog.rustfs.s3.endpoint", config)
|
|
self.assertNotIn("spark.sql.catalog.rustfs.s3.path-style-access", config)
|
|
self.assertNotIn("spark.sql.catalog.rustfs.s3.access-key-id", config)
|
|
self.assertNotIn("spark.sql.catalog.rustfs.s3.secret-access-key", config)
|
|
|
|
def test_spark_vendor_config_formats_cloudflare_catalog_profile(self) -> None:
|
|
config = engine_compatibility.spark_vendor_catalog_config(
|
|
profile="cloudflare-r2-data-catalog",
|
|
endpoint="https://example.r2.cloudflarestorage.com",
|
|
warehouse="ignored",
|
|
access_key="rustfsadmin",
|
|
secret_key="rustfsadmin",
|
|
region="auto",
|
|
catalog_name="rustfs",
|
|
account_id="000000000000",
|
|
table_bucket="ignored",
|
|
catalog_uri="https://catalog.example.com/iceberg",
|
|
warehouse_name="analytics",
|
|
)
|
|
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.uri"], "https://catalog.example.com/iceberg")
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.warehouse"], "analytics")
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.rest.signing-name"], "s3")
|
|
self.assertNotIn("spark.sql.catalog.rustfs.s3.endpoint", config)
|
|
self.assertNotIn("spark.sql.catalog.rustfs.s3.path-style-access", config)
|
|
self.assertNotIn("spark.sql.catalog.rustfs.s3.access-key-id", config)
|
|
self.assertNotIn("spark.sql.catalog.rustfs.s3.secret-access-key", config)
|
|
|
|
def test_spark_vendor_config_keeps_endpoint_for_s3_compatible_profiles(self) -> None:
|
|
config = engine_compatibility.spark_vendor_catalog_config(
|
|
profile="minio-aistor",
|
|
endpoint="https://minio.example.com",
|
|
warehouse="analytics",
|
|
access_key="rustfsadmin",
|
|
secret_key="rustfsadmin",
|
|
region="us-east-1",
|
|
catalog_name="rustfs",
|
|
account_id="000000000000",
|
|
table_bucket="analytics",
|
|
catalog_uri=None,
|
|
warehouse_name=None,
|
|
)
|
|
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.uri"], "https://minio.example.com/_iceberg")
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.s3.endpoint"], "https://minio.example.com")
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.s3.path-style-access"], "true")
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.s3.access-key-id"], "rustfsadmin")
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.s3.secret-access-key"], "rustfsadmin")
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.rest.signing-name"], "s3tables")
|
|
|
|
def test_spark_sql_smoke_covers_lifecycle_append_reload_and_cleanup(self) -> None:
|
|
sql = engine_compatibility.spark_sql_smoke(
|
|
catalog_name="rustfs",
|
|
namespace="sales",
|
|
table="orders",
|
|
cleanup=True,
|
|
)
|
|
|
|
self.assertIn("CREATE NAMESPACE IF NOT EXISTS rustfs.`sales`", sql)
|
|
self.assertIn("DROP TABLE IF EXISTS rustfs.`sales`.`orders`", sql)
|
|
self.assertIn("CREATE TABLE rustfs.`sales`.`orders`", sql)
|
|
self.assertIn("INSERT INTO rustfs.`sales`.`orders`", sql)
|
|
self.assertIn("REFRESH TABLE rustfs.`sales`.`orders`", sql)
|
|
self.assertIn("SELECT COUNT(*) AS row_count FROM rustfs.`sales`.`orders`", sql)
|
|
self.assertIn("DROP TABLE IF EXISTS rustfs.`sales`.`orders`", sql)
|
|
self.assertIn("DROP NAMESPACE IF EXISTS rustfs.`sales`", sql)
|
|
|
|
def test_spark_sql_rejects_unsafe_identifiers(self) -> None:
|
|
with self.assertRaisesRegex(ValueError, "Spark identifier"):
|
|
engine_compatibility.spark_sql_smoke(
|
|
catalog_name="rustfs",
|
|
namespace="sales`prod",
|
|
table="orders",
|
|
)
|
|
|
|
def test_cli_prints_machine_readable_engine_matrix(self) -> None:
|
|
payload = engine_compatibility.cli_json(["--print-engine-matrix"])
|
|
document = json.loads(payload)
|
|
|
|
self.assertIn("engine_compatibility", document)
|
|
self.assertTrue(document["engine_compatibility"])
|
|
|
|
def test_vendor_compatibility_audit_records_provider_gaps(self) -> None:
|
|
audit = engine_compatibility.vendor_compatibility_audit()
|
|
by_profile = {entry["profile"]: entry for entry in audit["providers"]}
|
|
|
|
self.assertEqual(audit["claim_boundary"], "reference profiles and gap audit")
|
|
self.assertEqual(audit["promotion_policy"], "live evidence required before compatibility claims expand")
|
|
self.assertIn("aws-s3tables", by_profile)
|
|
self.assertIn("minio-aistor", by_profile)
|
|
self.assertIn("cloudflare-r2-data-catalog", by_profile)
|
|
self.assertIn("oss-tables", by_profile)
|
|
|
|
aws = by_profile["aws-s3tables"]
|
|
self.assertEqual(aws["signing_name"], "s3tables")
|
|
self.assertEqual(aws["warehouse_shape"], "arn:aws:s3tables:{region}:{account_id}:bucket/{table_bucket}")
|
|
self.assertIn("iceberg-rest-routes", aws["audit_categories"])
|
|
self.assertIn("error-shapes", aws["audit_categories"])
|
|
self.assertIn("permission-actions", aws["audit_categories"])
|
|
self.assertIn("maintenance-semantics", aws["audit_categories"])
|
|
self.assertIn("credential-vending", aws["audit_categories"])
|
|
self.assertEqual(aws["rustfs_claim"], "profile-generator-only")
|
|
|
|
minio = by_profile["minio-aistor"]
|
|
self.assertEqual(minio["catalog_base_path"], "/_iceberg")
|
|
self.assertEqual(minio["signing_name"], "s3tables")
|
|
self.assertEqual(minio["rustfs_claim"], "alias-smoke-plus-profile-generator")
|
|
self.assertIn("private-extension-parity", minio["not_claimed"])
|
|
|
|
cloudflare = by_profile["cloudflare-r2-data-catalog"]
|
|
self.assertEqual(cloudflare["auth_model"], "api-token")
|
|
self.assertIn("managed-maintenance-parity", cloudflare["not_claimed"])
|
|
self.assertEqual(cloudflare["rustfs_claim"], "profile-generator-only")
|
|
|
|
oss = by_profile["oss-tables"]
|
|
self.assertEqual(oss["signing_name"], "osstables")
|
|
self.assertEqual(oss["warehouse_shape"], "acs:osstables:{region}:{account_id}:bucket/{table_bucket}")
|
|
self.assertIn("provider-error-code-parity", oss["not_claimed"])
|
|
|
|
self.assertIn("full AWS S3 Tables compatibility", audit["global_not_claimed"])
|
|
self.assertIn("full Cloudflare R2 Data Catalog interoperability", audit["global_not_claimed"])
|
|
|
|
def test_vendor_compatibility_audit_has_stable_validation_steps(self) -> None:
|
|
audit = engine_compatibility.vendor_compatibility_audit()
|
|
step_names = {step["name"] for step in audit["required_validation_steps"]}
|
|
|
|
self.assertIn("route-and-field-shape", step_names)
|
|
self.assertIn("error-response-shape", step_names)
|
|
self.assertIn("permission-action-mapping", step_names)
|
|
self.assertIn("maintenance-behavior", step_names)
|
|
self.assertIn("credential-model", step_names)
|
|
self.assertIn("client-live-run", step_names)
|
|
|
|
route_step = next(step for step in audit["required_validation_steps"] if step["name"] == "route-and-field-shape")
|
|
self.assertIn("request path", route_step["evidence"])
|
|
self.assertIn("response fields", route_step["evidence"])
|
|
self.assertEqual(route_step["status"], "manual-audit-required")
|
|
|
|
def test_cli_prints_vendor_compatibility_audit(self) -> None:
|
|
payload = engine_compatibility.cli_json(["--print-vendor-audit"])
|
|
document = json.loads(payload)
|
|
|
|
self.assertIn("vendor_compatibility_audit", document)
|
|
profiles = {entry["profile"] for entry in document["vendor_compatibility_audit"]["providers"]}
|
|
self.assertIn("aws-s3tables", profiles)
|
|
self.assertIn("minio-aistor", profiles)
|
|
self.assertIn("cloudflare-r2-data-catalog", profiles)
|
|
self.assertIn("oss-tables", profiles)
|
|
|
|
def test_cli_prints_vendor_spark_config(self) -> None:
|
|
payload = engine_compatibility.cli_json(
|
|
[
|
|
"--print-spark-config",
|
|
"--profile",
|
|
"aws-s3tables",
|
|
"--region",
|
|
"us-east-1",
|
|
"--account-id",
|
|
"123456789012",
|
|
"--table-bucket",
|
|
"analytics",
|
|
]
|
|
)
|
|
document = json.loads(payload)
|
|
config = document["spark_config"]
|
|
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.uri"], "https://s3tables.us-east-1.amazonaws.com/iceberg")
|
|
self.assertEqual(
|
|
config["spark.sql.catalog.rustfs.warehouse"],
|
|
"arn:aws:s3tables:us-east-1:123456789012:bucket/analytics",
|
|
)
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.rest.signing-name"], "s3tables")
|
|
self.assertNotIn("spark.sql.catalog.rustfs.s3.endpoint", config)
|
|
self.assertNotIn("spark.sql.catalog.rustfs.s3.path-style-access", config)
|
|
self.assertNotIn("spark.sql.catalog.rustfs.s3.access-key-id", config)
|
|
self.assertNotIn("spark.sql.catalog.rustfs.s3.secret-access-key", config)
|
|
|
|
def test_cli_prints_oss_tables_spark_config(self) -> None:
|
|
payload = engine_compatibility.cli_json(
|
|
[
|
|
"--print-spark-config",
|
|
"--profile",
|
|
"oss-tables",
|
|
"--endpoint",
|
|
"https://cn-hangzhou.oss-tables.aliyuncs.com",
|
|
"--region",
|
|
"cn-hangzhou",
|
|
"--account-id",
|
|
"123456789012",
|
|
"--table-bucket",
|
|
"analytics",
|
|
]
|
|
)
|
|
document = json.loads(payload)
|
|
config = document["spark_config"]
|
|
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.uri"], "https://cn-hangzhou.oss-tables.aliyuncs.com/iceberg")
|
|
self.assertEqual(
|
|
config["spark.sql.catalog.rustfs.warehouse"],
|
|
"acs:osstables:cn-hangzhou:123456789012:bucket/analytics",
|
|
)
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.rest.signing-name"], "osstables")
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.s3.endpoint"], "https://oss-cn-hangzhou.aliyuncs.com")
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.s3.access-key-id"], "rustfsadmin")
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.s3.secret-access-key"], "rustfsadmin")
|
|
self.assertNotIn("spark.sql.catalog.rustfs.s3.path-style-access", config)
|
|
|
|
def test_cli_spark_config_keeps_explicit_rest_overrides(self) -> None:
|
|
payload = engine_compatibility.cli_json(
|
|
[
|
|
"--print-spark-config",
|
|
"--endpoint",
|
|
"http://127.0.0.1:9000/",
|
|
"--rest-path",
|
|
"/_iceberg",
|
|
"--rest-signing-name",
|
|
"s3tables",
|
|
]
|
|
)
|
|
document = json.loads(payload)
|
|
config = document["spark_config"]
|
|
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.uri"], "http://127.0.0.1:9000/_iceberg")
|
|
self.assertEqual(config["spark.sql.catalog.rustfs.rest.signing-name"], "s3tables")
|
|
|
|
def test_live_conformance_harness_pins_clients_and_records_commands(self) -> None:
|
|
harness = engine_compatibility.live_conformance_harness(
|
|
endpoint="http://127.0.0.1:9000",
|
|
warehouse="rustfs-s3table-smoke",
|
|
access_key="rustfsadmin",
|
|
secret_key="rustfsadmin",
|
|
region="us-east-1",
|
|
catalog_name="rustfs",
|
|
namespace="smoke",
|
|
table="events",
|
|
rest_path="/iceberg",
|
|
rest_signing_name="s3",
|
|
pyiceberg_version="0.10.0",
|
|
spark_version="3.5.4",
|
|
iceberg_version="1.7.1",
|
|
scala_version="2.12",
|
|
cleanup=True,
|
|
)
|
|
|
|
self.assertEqual(harness["mode"], "manual-or-ci-optional")
|
|
self.assertEqual(harness["ci_gate"], "RUSTFS_TABLE_CATALOG_LIVE_CONFORMANCE=1")
|
|
self.assertEqual(harness["expected_results"]["row_count"], 2)
|
|
self.assertIn("RustFS endpoint is reachable", harness["prerequisites"])
|
|
|
|
by_client = {client["name"]: client for client in harness["clients"]}
|
|
pyiceberg = by_client["PyIceberg"]
|
|
self.assertEqual(pyiceberg["version"], "0.10.0")
|
|
self.assertIn("pyiceberg[pyarrow]==0.10.0", pyiceberg["install"])
|
|
self.assertIn("scripts/table-catalog/pyiceberg_smoke.py", pyiceberg["command"])
|
|
self.assertIn("--replace", pyiceberg["command"])
|
|
self.assertIn("--cleanup", pyiceberg["command"])
|
|
|
|
spark = by_client["Spark Iceberg REST catalog"]
|
|
self.assertEqual(spark["spark_version"], "3.5.4")
|
|
self.assertEqual(spark["iceberg_version"], "1.7.1")
|
|
self.assertIn("iceberg-spark-runtime-3.5_2.12:1.7.1", spark["packages"])
|
|
self.assertIn("iceberg-aws-bundle:1.7.1", spark["packages"])
|
|
self.assertIn("spark-sql", spark["command"])
|
|
self.assertIn("spark.sql.catalog.rustfs.uri=http://127.0.0.1:9000/iceberg", spark["command"])
|
|
self.assertIn("SELECT COUNT(*) AS row_count", spark["sql"])
|
|
|
|
trino = by_client["Trino Iceberg REST catalog"]
|
|
self.assertEqual(trino["status"], "manual-live-read-probe")
|
|
self.assertIn("connector.name=iceberg", trino["catalog_properties"])
|
|
self.assertIn("iceberg.catalog.type=rest", trino["catalog_properties"])
|
|
self.assertIn("iceberg.rest-catalog.uri=http://127.0.0.1:9000/iceberg", trino["catalog_properties"])
|
|
self.assertIn("trino", trino["command"])
|
|
self.assertIn("SELECT COUNT(*) AS row_count", trino["sql"])
|
|
self.assertEqual(trino["write_compatibility"], "not-claimed")
|
|
|
|
duckdb = by_client["DuckDB Iceberg"]
|
|
self.assertEqual(duckdb["status"], "manual-live-read-probe")
|
|
self.assertIn("LOAD httpfs", duckdb["sql"])
|
|
self.assertIn("LOAD iceberg", duckdb["sql"])
|
|
self.assertIn("iceberg_scan", duckdb["sql"])
|
|
self.assertEqual(duckdb["write_compatibility"], "not-claimed")
|
|
|
|
snowflake = by_client["Snowflake Open Catalog / Iceberg integrations"]
|
|
self.assertEqual(snowflake["status"], "manual-reference-probe")
|
|
self.assertIn("CREATE OR REPLACE EXTERNAL VOLUME", snowflake["sql"])
|
|
self.assertEqual(snowflake["live_rustfs_interoperability"], "not-claimed")
|
|
|
|
databend = by_client["Databend"]
|
|
self.assertEqual(databend["status"], "manual-live-s3-stage-probe")
|
|
self.assertIn("CREATE STAGE", databend["sql"])
|
|
self.assertIn("SELECT COUNT(*)", databend["sql"])
|
|
self.assertEqual(databend["iceberg_rest_catalog"], "not-claimed")
|
|
|
|
def test_live_conformance_harness_records_evidence_contract(self) -> None:
|
|
harness = engine_compatibility.live_conformance_harness(
|
|
endpoint="http://127.0.0.1:9000",
|
|
warehouse="rustfs-s3table-smoke",
|
|
access_key="rustfsadmin",
|
|
secret_key="rustfsadmin",
|
|
region="us-east-1",
|
|
catalog_name="rustfs",
|
|
namespace="smoke",
|
|
table="events",
|
|
rest_path="/iceberg",
|
|
rest_signing_name="s3",
|
|
pyiceberg_version="0.10.0",
|
|
spark_version="3.5.4",
|
|
iceberg_version="1.7.1",
|
|
scala_version="2.12",
|
|
)
|
|
|
|
evidence = harness["evidence"]
|
|
self.assertEqual(evidence["result"], "operator-recorded")
|
|
self.assertIn("rustfs_build", evidence["required_run_metadata"])
|
|
self.assertIn("catalog_backing", evidence["required_run_metadata"])
|
|
self.assertIn("client_version", evidence["required_run_metadata"])
|
|
self.assertIn("metadata_location", evidence["required_run_metadata"])
|
|
self.assertIn("expected_status", evidence["required_run_metadata"])
|
|
self.assertIn("observed_status", evidence["required_run_metadata"])
|
|
|
|
table_by_client = {row["client"]: row for row in evidence["result_table_template"]}
|
|
self.assertEqual(table_by_client["PyIceberg"]["claim_after_pass"], "automated-smoke")
|
|
self.assertEqual(table_by_client["Spark Iceberg REST catalog"]["claim_after_pass"], "manual-live-verified")
|
|
self.assertEqual(table_by_client["Trino Iceberg REST catalog"]["write_claim_after_pass"], "not-claimed")
|
|
self.assertEqual(table_by_client["DuckDB Iceberg"]["write_claim_after_pass"], "not-claimed")
|
|
self.assertIn("manual-live", " ".join(evidence["promotion_rules"]))
|
|
self.assertIn("not-claimed", " ".join(evidence["promotion_rules"]))
|
|
|
|
def test_live_conformance_evidence_record_validates_claim_promotion(self) -> None:
|
|
record = engine_compatibility.live_conformance_evidence_record(
|
|
client_name="PyIceberg",
|
|
client_version="0.10.0",
|
|
scenario="create-append-reload-scan-direct-rest-probes",
|
|
rustfs_build="rustfs-test",
|
|
git_sha="abc123",
|
|
catalog_backing="durable-strong",
|
|
endpoint="http://127.0.0.1:9000",
|
|
warehouse="rustfs-s3table-smoke",
|
|
rest_path="/iceberg",
|
|
namespace="smoke",
|
|
table="events",
|
|
metadata_location="s3://rustfs-s3table-smoke/tables/table-id/metadata/v2.metadata.json",
|
|
run_timestamp_utc="2026-07-09T00:00:00Z",
|
|
operator="ci",
|
|
expected_status="pass",
|
|
observed_status="pass",
|
|
row_count=2,
|
|
cleanup_result="dropped-table-and-namespace",
|
|
claim="automated-smoke",
|
|
command="python3 scripts/table-catalog/pyiceberg_smoke.py --cleanup",
|
|
)
|
|
|
|
validation = engine_compatibility.validate_live_conformance_evidence(record)
|
|
|
|
self.assertEqual(record["schema"], "rustfs-s3tables-live-conformance-evidence")
|
|
self.assertEqual(record["schema_version"], 1)
|
|
self.assertEqual(validation["status"], "accepted")
|
|
self.assertEqual(validation["claim"], "automated-smoke")
|
|
self.assertEqual(validation["client_name"], "PyIceberg")
|
|
|
|
def test_live_conformance_evidence_rejects_missing_or_mismatched_results(self) -> None:
|
|
record = engine_compatibility.live_conformance_evidence_record(
|
|
client_name="Spark Iceberg REST catalog",
|
|
client_version="3.5.4/iceberg-1.7.1",
|
|
scenario="create-namespace-create-table-append-refresh-count-cleanup",
|
|
rustfs_build="rustfs-test",
|
|
git_sha="abc123",
|
|
catalog_backing="durable-strong",
|
|
endpoint="http://127.0.0.1:9000",
|
|
warehouse="rustfs-s3table-smoke",
|
|
rest_path="/iceberg",
|
|
namespace="smoke",
|
|
table="events",
|
|
metadata_location="s3://rustfs-s3table-smoke/tables/table-id/metadata/v2.metadata.json",
|
|
run_timestamp_utc="2026-07-09T00:00:00Z",
|
|
operator="ci",
|
|
expected_status="pass",
|
|
observed_status="fail",
|
|
row_count=0,
|
|
cleanup_result="not-run",
|
|
claim="manual-live-verified",
|
|
command="spark-sql -f /tmp/rustfs-s3tables-smoke-events-spark.sql",
|
|
)
|
|
|
|
with self.assertRaisesRegex(ValueError, "observed_status"):
|
|
engine_compatibility.validate_live_conformance_evidence(record)
|
|
|
|
passing = dict(record, observed_status="pass", row_count=2)
|
|
del passing["metadata_location"]
|
|
with self.assertRaisesRegex(ValueError, "metadata_location"):
|
|
engine_compatibility.validate_live_conformance_evidence(passing)
|
|
|
|
overclaim = dict(record, observed_status="pass", row_count=2, client_name="Trino Iceberg REST catalog", claim="manual-live-write-verified")
|
|
with self.assertRaisesRegex(ValueError, "claim"):
|
|
engine_compatibility.validate_live_conformance_evidence(overclaim)
|
|
|
|
unrecorded = dict(record, expected_status="pass", observed_status="pass", row_count=2, rustfs_build="operator-recorded")
|
|
with self.assertRaisesRegex(ValueError, "rustfs_build"):
|
|
engine_compatibility.validate_live_conformance_evidence(unrecorded)
|
|
|
|
failed_claim = dict(record, expected_status="fail", observed_status="fail", row_count=0)
|
|
with self.assertRaisesRegex(ValueError, "expected_status"):
|
|
engine_compatibility.validate_live_conformance_evidence(failed_claim)
|
|
|
|
def test_cli_prints_live_conformance_evidence_schema(self) -> None:
|
|
payload = engine_compatibility.cli_json(["--print-live-evidence-schema"])
|
|
document = json.loads(payload)
|
|
|
|
schema = document["live_conformance_evidence_schema"]
|
|
self.assertEqual(schema["schema"], "rustfs-s3tables-live-conformance-evidence")
|
|
self.assertIn("metadata_location", schema["required_fields"])
|
|
self.assertIn("claim", schema["required_fields"])
|
|
self.assertEqual(schema["claim_promotion"]["Trino Iceberg REST catalog"], ["manual-live-read-verified"])
|
|
|
|
def test_production_operations_guide_covers_release_boundaries(self) -> None:
|
|
guide = engine_compatibility.production_operations_guide(
|
|
endpoint="http://127.0.0.1:9000",
|
|
warehouse="rustfs-s3table-smoke",
|
|
namespace="smoke",
|
|
table="events",
|
|
rest_path="/iceberg",
|
|
)
|
|
|
|
self.assertEqual(guide["claim_boundary"], "Iceberg REST Catalog S3 Tables implementation")
|
|
section_by_name = {section["name"]: section for section in guide["sections"]}
|
|
|
|
self.assertIn("live-client-conformance", section_by_name)
|
|
self.assertIn("catalog-backing-cutover", section_by_name)
|
|
self.assertIn("maintenance-operations", section_by_name)
|
|
self.assertIn("recovery-and-disaster-rehearsal", section_by_name)
|
|
self.assertIn("permissions-and-credentials", section_by_name)
|
|
self.assertIn("unsupported-claim-governance", section_by_name)
|
|
|
|
cutover = section_by_name["catalog-backing-cutover"]
|
|
self.assertIn("GET /iceberg/v1/rustfs-s3table-smoke/catalog/migration", cutover["commands"])
|
|
self.assertIn("blockers is empty", " ".join(cutover["pass_criteria"]))
|
|
|
|
maintenance = section_by_name["maintenance-operations"]
|
|
self.assertIn("quarantine", " ".join(maintenance["fail_closed_signals"]))
|
|
self.assertIn("audit-events", " ".join(maintenance["required_evidence"]))
|
|
|
|
unsupported = section_by_name["unsupported-claim-governance"]
|
|
self.assertIn("full AWS S3 Tables control-plane API parity", unsupported["not_claimed"])
|
|
self.assertNotIn("fully compatible", json.dumps(guide).lower())
|
|
|
|
def test_production_operations_guide_threads_target_args_into_commands(self) -> None:
|
|
guide = engine_compatibility.production_operations_guide(
|
|
endpoint="https://rustfs.example.com",
|
|
warehouse="lakehouse-prod",
|
|
namespace="sales",
|
|
table="orders",
|
|
rest_path="/_iceberg",
|
|
)
|
|
|
|
live = {section["name"]: section for section in guide["sections"]}["live-client-conformance"]
|
|
commands = "\n".join(live["commands"])
|
|
|
|
self.assertIn("--endpoint https://rustfs.example.com", commands)
|
|
self.assertIn("--bucket lakehouse-prod", commands)
|
|
self.assertIn("--warehouse lakehouse-prod", commands)
|
|
self.assertIn("--namespace sales", commands)
|
|
self.assertIn("--table orders", commands)
|
|
self.assertIn("--rest-path /_iceberg", commands)
|
|
self.assertNotIn("rustfs-s3table-smoke", commands)
|
|
|
|
def test_production_operations_guide_encodes_catalog_path_segments(self) -> None:
|
|
guide = engine_compatibility.production_operations_guide(
|
|
endpoint="http://127.0.0.1:9000",
|
|
warehouse="lake bucket",
|
|
namespace="sales/prod",
|
|
table="orders table",
|
|
rest_path="/_iceberg",
|
|
)
|
|
|
|
section_by_name = {section["name"]: section for section in guide["sections"]}
|
|
cutover = section_by_name["catalog-backing-cutover"]
|
|
maintenance = section_by_name["maintenance-operations"]
|
|
recovery = section_by_name["recovery-and-disaster-rehearsal"]
|
|
credentials = section_by_name["permissions-and-credentials"]
|
|
|
|
self.assertIn("GET /_iceberg/v1/lake%20bucket/catalog/migration", cutover["commands"])
|
|
encoded_table_path = "/_iceberg/v1/lake%20bucket/namespaces/sales%2Fprod/tables/orders%20table"
|
|
for commands in [maintenance["commands"], recovery["commands"], credentials["commands"]]:
|
|
rendered = "\n".join(commands)
|
|
self.assertIn(encoded_table_path, rendered)
|
|
self.assertNotIn("orders table", rendered)
|
|
self.assertNotIn("/namespaces/sales/prod/", rendered)
|
|
|
|
def test_cli_prints_live_conformance_harness(self) -> None:
|
|
payload = engine_compatibility.cli_json(
|
|
[
|
|
"--print-live-conformance",
|
|
"--pyiceberg-version",
|
|
"0.10.0",
|
|
"--spark-version",
|
|
"3.5.4",
|
|
"--iceberg-version",
|
|
"1.7.1",
|
|
"--cleanup",
|
|
]
|
|
)
|
|
document = json.loads(payload)
|
|
|
|
self.assertEqual(document["live_conformance"]["mode"], "manual-or-ci-optional")
|
|
self.assertEqual(document["live_conformance"]["expected_results"]["row_count"], 2)
|
|
self.assertIn("evidence", document["live_conformance"])
|
|
|
|
def test_cli_prints_production_operations_guide(self) -> None:
|
|
payload = engine_compatibility.cli_json(
|
|
[
|
|
"--print-operations-guide",
|
|
"--warehouse",
|
|
"rustfs-s3table-smoke",
|
|
"--namespace",
|
|
"smoke",
|
|
"--table",
|
|
"events",
|
|
]
|
|
)
|
|
document = json.loads(payload)
|
|
|
|
self.assertEqual(
|
|
document["production_operations_guide"]["claim_boundary"],
|
|
"Iceberg REST Catalog S3 Tables implementation",
|
|
)
|
|
sections = {section["name"] for section in document["production_operations_guide"]["sections"]}
|
|
self.assertIn("live-client-conformance", sections)
|
|
self.assertIn("unsupported-claim-governance", sections)
|
|
|
|
def test_live_conformance_harness_sanitizes_sql_file_path(self) -> None:
|
|
harness = engine_compatibility.live_conformance_harness(
|
|
endpoint="http://127.0.0.1:9000",
|
|
warehouse="rustfs-s3table-smoke",
|
|
access_key="rustfsadmin",
|
|
secret_key="rustfsadmin",
|
|
region="us-east-1",
|
|
catalog_name="rustfs",
|
|
namespace="sales/prod",
|
|
table="orders 2026",
|
|
rest_path="/iceberg",
|
|
rest_signing_name="s3",
|
|
pyiceberg_version="0.10.0",
|
|
spark_version="3.5.4",
|
|
iceberg_version="1.7.1",
|
|
scala_version="2.12",
|
|
)
|
|
|
|
spark = {client["name"]: client for client in harness["clients"]}["Spark Iceberg REST catalog"]
|
|
self.assertEqual(spark["sql_file"], "/tmp/rustfs-s3tables-sales-prod-orders-2026-spark.sql")
|
|
|
|
def assertContainsScenario(self, entry: dict[str, object], name: str, status: str) -> None:
|
|
scenarios = entry.get("scenarios")
|
|
self.assertIsInstance(scenarios, list)
|
|
for scenario in scenarios:
|
|
if isinstance(scenario, dict) and scenario.get("name") == name:
|
|
self.assertEqual(scenario.get("status"), status)
|
|
return
|
|
self.fail(f"missing scenario {name!r} in {entry.get('client')!r}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|