test: add integration tests for load_view

Author: spr0els

tests/conftest.py

@@ -2353,6 +2353,13 @@ def table_list(table_name: str) -> list[str]:
     return [f"{table_name}_{idx}" for idx in range(NUM_TABLES)]
 
 
+@pytest.fixture()
+def view_name() -> str:
+    prefix = "my_iceberg_view-"
+    random_tag = "".join(choice(string.ascii_letters) for _ in range(RANDOM_LENGTH))
+    return (prefix + random_tag).lower()
+
+
 @pytest.fixture()
 def database_name() -> str:
     prefix = "my_iceberg_database-"

tests/integration/test_rest_catalog.py

@@ -16,10 +16,15 @@
 # under the License.
 # pylint:disable=redefined-outer-name
 
+import time
+
 import pytest
 from pytest_lazy_fixtures import lf
 
 from pyiceberg.catalog.rest import RestCatalog
+from pyiceberg.exceptions import NoSuchViewError
+from pyiceberg.schema import Schema
+from pyiceberg.view.metadata import SQLViewRepresentation, ViewVersion
 
 TEST_NAMESPACE_IDENTIFIER = "TEST NS"
 
@@ -62,3 +67,47 @@ def test_create_namespace_if_already_existing(catalog: RestCatalog) -> None:
     catalog.create_namespace_if_not_exists(TEST_NAMESPACE_IDENTIFIER)
 
     assert catalog.namespace_exists(TEST_NAMESPACE_IDENTIFIER)
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("catalog", [lf("session_catalog")])
+def test_load_view(catalog: RestCatalog, table_schema_nested: Schema, database_name: str, view_name: str) -> None:
+    identifier = (database_name, view_name)
+    if not catalog.namespace_exists(database_name):
+        catalog.create_namespace(database_name)
+
+    view_version = ViewVersion(
+        version_id=1,
+        schema_id=1,
+        timestamp_ms=int(time.time() * 1000),
+        summary={},
+        representations=[
+            SQLViewRepresentation(
+                type="sql",
+                sql="SELECT 1 as some_col",
+                dialect="spark",
+            )
+        ],
+        default_namespace=["default"],
+    )
+    view = catalog.create_view(identifier, table_schema_nested, view_version=view_version)
+    loaded_view = catalog.load_view(identifier)
+    assert view.name() == loaded_view.name()
+    assert view.metadata == loaded_view.metadata
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("catalog", [lf("session_catalog")])
+def test_load_view_with_table_ident(
+    catalog: RestCatalog, table_name: str, table_schema_nested: Schema, database_name: str
+) -> None:
+    table_identifier = (database_name, table_name)
+    if not catalog.namespace_exists(database_name):
+        catalog.create_namespace(database_name)
+
+    if not catalog.table_exists(table_identifier):
+        catalog.create_table(table_identifier, table_schema_nested)
+
+    assert catalog.table_exists(table_identifier)
+    with pytest.raises(NoSuchViewError):
+        catalog.load_view(table_identifier)