verify_only.py

import requests
from pyiceberg.catalog import load_catalog
from pyiceberg.schema import Schema
from pyiceberg.types import NestedField, StringType, IntegerType

def verify():
    print("Starting verification...")
    
    # 1. Connect to Catalog 1 (default)
    print("\n--- Testing Catalog 1 (default) ---")
    cat1 = load_catalog(
        "cat1",
        **{
            "uri": "http://127.0.0.1:8001",
            "warehouse": "/tmp/warehouse",
            "prefix": "cat1",
        }
    )
    
    ns1 = "ns1"
    try:
        cat1.create_namespace(ns1)
        print(f"✅ Created namespace '{ns1}' in cat1")
    except Exception as e:
        print(f"⚠️ Namespace creation failed in cat1: {e}")

    try:
        cat1.create_table(f"{ns1}.table1", schema=Schema(NestedField(1, "x", IntegerType(), required=True)))
        print(f"✅ Created table '{ns1}.table1' in cat1")
    except Exception as e:
        print(f"⚠️ Table creation failed in cat1: {e}")

    # 4. Catalog Properties (IO Config & Warehouse)
    print("\n--- Testing Catalog Properties ---")
    props_url = "http://127.0.0.1:8001/v1/cat1/config/properties"
    io_config = {
        "s3.endpoint": "http://minio:9000",
        "s3.access-key-id": "minioadmin",
        "s3.secret-access-key": "minioadmin",
        "warehouse": "s3://special-warehouse"
    }
    
    # Set properties
    resp = requests.post(props_url, json=io_config)
    if resp.status_code != 200:
        print(f"❌ Failed to set catalog properties: {resp.text}")
    else:
        print(f"✅ Set catalog properties: {resp.json()}")
        
    # Get properties
    resp = requests.get(props_url)
    if resp.status_code != 200:
        print(f"❌ Failed to get catalog properties: {resp.text}")
    else:
        fetched_props = resp.json()
        assert fetched_props["s3.endpoint"] == "http://minio:9000"
        assert fetched_props["warehouse"] == "s3://special-warehouse"
        print(f"✅ Verified catalog properties persisted")

if __name__ == "__main__":
    verify()