A bunch of testing for IC1 repos (#1423)

TODO:
- [x] rebase is failing

Author: paraseba

icechunk-python/tests/conftest.py

@@ -7,21 +7,27 @@
 from icechunk import Repository, in_memory_storage, local_filesystem_storage
 
 
-def parse_repo(store: Literal["local", "memory"], path: str) -> Repository:
+def parse_repo(
+    store: Literal["local", "memory"], path: str, spec_version: int | None
+) -> Repository:
     if store == "local":
         return Repository.create(
             storage=local_filesystem_storage(path),
+            spec_version=spec_version,
         )
     if store == "memory":
         return Repository.create(
             storage=in_memory_storage(),
+            spec_version=spec_version,
         )
 
 
 @pytest.fixture(scope="function")
-def repo(request: pytest.FixtureRequest, tmpdir: str) -> tuple[Repository, str]:
+def repo(
+    request: pytest.FixtureRequest, tmpdir: str, any_spec_version: int | None
+) -> tuple[Repository, str]:
     param = request.param
-    repo = parse_repo(param, tmpdir)
+    repo = parse_repo(param, tmpdir, spec_version=any_spec_version)
     return repo, tmpdir
 
 
@@ -54,6 +60,8 @@ def write_chunks_to_minio(
     return etags
 
 
-@pytest.fixture(scope="function", params=[1, 2, None])
+@pytest.fixture(
+    scope="function", params=[1, 2, None], ids=["spec-v1", "spec-v2", "no-spec-version"]
+)
 def any_spec_version(request: pytest.FixtureRequest) -> int | None:
     return request.param

icechunk-python/tests/test_commit_properties.py

@@ -6,9 +6,10 @@
 import zarr
 
 
-def test_property_types() -> None:
+def test_property_types(any_spec_version: int | None) -> None:
     repo = ic.Repository.create(
         storage=ic.in_memory_storage(),
+        spec_version=any_spec_version,
     )
     session = repo.writable_session("main")
     store = session.store
@@ -39,9 +40,10 @@ class NoJson:
     pass
 
 
-def test_invalid_property_types() -> None:
+def test_invalid_property_types(any_spec_version: int | None) -> None:
     repo = ic.Repository.create(
         storage=ic.in_memory_storage(),
+        spec_version=any_spec_version,
     )
     session = repo.writable_session("main")
     store = session.store

icechunk-python/tests/test_concurrency.py

@@ -48,9 +48,10 @@ async def list_store(store: icechunk.IcechunkStore, barrier: asyncio.Barrier) ->
         await asyncio.sleep(0.1)
 
 
-async def test_concurrency() -> None:
+async def test_concurrency(any_spec_version: int | None) -> None:
     repo = icechunk.Repository.open_or_create(
         storage=icechunk.in_memory_storage(),
+        create_version=any_spec_version,
     )
 
     session = repo.writable_session("main")
@@ -93,7 +94,7 @@ async def test_concurrency() -> None:
 
 
 @pytest.mark.filterwarnings("ignore:datetime.datetime.utcnow")
-async def test_thread_concurrency() -> None:
+async def test_thread_concurrency(any_spec_version: int | None) -> None:
     """Run multiple threads doing different type of operations for SECONDS_TO_RUN seconds.
 
     The threads execute 5 types of operations: reads, native writes, virtual writes, deletes and lists.
@@ -150,6 +151,7 @@ async def test_thread_concurrency() -> None:
         storage=storage,
         config=config,
         authorize_virtual_chunk_access=credentials,
+        spec_version=any_spec_version,
     )
 
     session = repo.writable_session("main")

icechunk-python/tests/test_config.py

@@ -10,13 +10,16 @@
 
 
 @pytest.fixture(scope="function")
-def tmp_store(tmpdir: Path) -> Generator[tuple[icechunk.IcechunkStore, str]]:
+def tmp_store(
+    tmpdir: Path, any_spec_version: int | None
+) -> Generator[tuple[icechunk.IcechunkStore, str]]:
     repo_path = f"{tmpdir}"
     config = icechunk.RepositoryConfig.default()
     config.inline_chunk_threshold_bytes = 5
     repo = icechunk.Repository.open_or_create(
         storage=icechunk.local_filesystem_storage(repo_path),
         config=config,
+        create_version=any_spec_version,
     )
 
     session = repo.writable_session("main")
@@ -25,24 +28,26 @@ def tmp_store(tmpdir: Path) -> Generator[tuple[icechunk.IcechunkStore, str]]:
     yield store, repo_path
 
 
-def test_config_fetch() -> None:
+def test_config_fetch(any_spec_version: int | None) -> None:
     config = icechunk.RepositoryConfig.default()
     config.inline_chunk_threshold_bytes = 5
     storage = icechunk.in_memory_storage()
     repo = icechunk.Repository.create(
         storage=storage,
         config=config,
+        spec_version=any_spec_version,
     )
 
     assert repo.config == config
     assert icechunk.Repository.fetch_config(storage) == config
 
 
-def test_config_save() -> None:
+def test_config_save(any_spec_version: int | None) -> None:
     config = icechunk.RepositoryConfig.default()
     storage = icechunk.in_memory_storage()
     repo = icechunk.Repository.create(
         storage=storage,
+        spec_version=any_spec_version,
     )
 
     config.inline_chunk_threshold_bytes = 5
@@ -142,10 +147,11 @@ def test_virtual_chunk_containers() -> None:
     assert config.virtual_chunk_containers["s3://testbucket/"] == container
 
 
-def test_can_change_deep_config_values() -> None:
+def test_can_change_deep_config_values(any_spec_version: int | None) -> None:
     storage = icechunk.in_memory_storage()
     repo = icechunk.Repository.create(
         storage=storage,
+        spec_version=any_spec_version,
     )
     config = icechunk.RepositoryConfig(
         inline_chunk_threshold_bytes=11,
@@ -240,7 +246,7 @@ def test_spec_version() -> None:
     assert icechunk.spec_version() >= 1
 
 
-def test_config_from_store() -> None:
+def test_config_from_store(any_spec_version: int | None) -> None:
     config = icechunk.RepositoryConfig.default()
     config.inline_chunk_threshold_bytes = 5
 
@@ -252,6 +258,7 @@ def test_config_from_store() -> None:
     repo = icechunk.Repository.create(
         storage=storage,
         config=config,
+        spec_version=any_spec_version,
     )
     session = repo.writable_session("main")
     store = session.store

icechunk-python/tests/test_conflicts.py

@@ -9,9 +9,10 @@
 
 
 @pytest.fixture
-def repo(tmpdir: Path) -> icechunk.Repository:
+def repo(tmpdir: Path, any_spec_version: int | None) -> icechunk.Repository:
     repo = icechunk.Repository.create(
-        storage=icechunk.local_filesystem_storage(str(tmpdir))
+        storage=icechunk.local_filesystem_storage(str(tmpdir)),
+        spec_version=any_spec_version,
     )
 
     session = repo.writable_session("main")
@@ -192,10 +193,11 @@ def test_rebase_chunks_with_ours(
     assert array_c[0, 1] == 1
 
 
-async def test_rebase_async() -> None:
+async def test_rebase_async(any_spec_version: int | None) -> None:
     """Test async rebase functionality with conflict detection and resolution."""
     repo = await icechunk.Repository.create_async(
         storage=icechunk.in_memory_storage(),
+        spec_version=any_spec_version,
     )
 
     # Set up initial state

icechunk-python/tests/test_credentials.py

@@ -191,6 +191,7 @@ def test_s3_refreshable_credentials_refresh(
 def test_s3_refreshable_credentials_pickle_with_optimization(
     tmp_path: Path,
     scatter_initial_credentials: bool,
+    any_spec_version: int | None,
 ) -> None:
     """Verifies pickled repos don't need to call get_credentials again if scatter_initial_credentials=True"""
     path = tmp_path / "calls.txt"
@@ -208,7 +209,10 @@ def test_s3_refreshable_credentials_pickle_with_optimization(
         scatter_initial_credentials=scatter_initial_credentials,
     )
     # let's create and use a repo
-    repo = Repository.create(storage=st)
+    repo = Repository.create(
+        storage=st,
+        spec_version=any_spec_version,
+    )
     assert Repository.exists(st)
     assert Repository.exists(st)
 

icechunk-python/tests/test_dask.py

@@ -16,14 +16,17 @@
 from xarray.testing import assert_identical
 
 
-def test_store_dask() -> None:
+def test_store_dask(any_spec_version: int | None) -> None:
     shape = (100, 100)
     dask_chunks = (20, 20)
     dask_array = dask.array.random.random(shape, chunks=dask_chunks)
 
     zarr_chunks = (10, 10)
     with tempfile.TemporaryDirectory() as tmpdir:
-        repo = Repository.create(local_filesystem_storage(tmpdir))
+        repo = Repository.create(
+            local_filesystem_storage(tmpdir),
+            spec_version=any_spec_version,
+        )
         session = repo.writable_session("main")
         group = zarr.group(store=session.store, overwrite=True)
 
@@ -52,27 +55,35 @@ def test_store_dask() -> None:
             store_dask(sources=[dask_array], targets=[zarray])
 
 
-def test_distributed() -> None:
+def test_distributed(any_spec_version: int | None) -> None:
     with distributed.Client():  # type: ignore [no-untyped-call]
         ds = create_test_data().chunk(dim1=3, dim2=4)
-        with roundtrip(ds, commit=True) as actual:
+        with roundtrip(ds, commit=True, spec_version=any_spec_version) as actual:
             assert_identical(actual, ds)
 
 
 @pytest.mark.parametrize("scheduler", ["threads", "processes"])
-def test_dask_schedulers(scheduler) -> None:
+def test_dask_schedulers(scheduler, any_spec_version: int | None) -> None:
     with dask.config.set(scheduler=scheduler):
         ds = create_test_data().chunk(dim1=3, dim2=4)
-        with roundtrip(ds, commit=scheduler == "processes") as actual:
+        with roundtrip(
+            ds,
+            commit=scheduler == "processes",
+            spec_version=any_spec_version,
+        ) as actual:
             assert_identical(actual, ds)
 
 
 @pytest.mark.parametrize("scheduler", ["threads", "processes"])
-def test_xarray_to_icechunk_nested_pickling(scheduler) -> None:
+def test_xarray_to_icechunk_nested_pickling(
+    scheduler, any_spec_version: int | None
+) -> None:
     with dask.config.set(scheduler=scheduler):
         ds = create_test_data(dim_sizes=(2, 3, 4)).chunk(-1)
         with tempfile.TemporaryDirectory() as tmpdir:
-            repo = Repository.create(local_filesystem_storage(tmpdir))
+            repo = Repository.create(
+                local_filesystem_storage(tmpdir), spec_version=any_spec_version
+            )
             session = repo.writable_session("main")
 
             to_icechunk(ds, session=session, mode="w")
@@ -93,11 +104,14 @@ def test_xarray_to_icechunk_nested_pickling(scheduler) -> None:
 
 
 @pytest.mark.parametrize("scheduler", ["threads", "processes"])
-def test_fork_session_deep_copies(scheduler) -> None:
+def test_fork_session_deep_copies(scheduler, any_spec_version: int | None) -> None:
     with dask.config.set(scheduler=scheduler):
         ds = create_test_data(dim_sizes=(2, 3, 4)).drop_encoding().chunk(dim3=1)
         with tempfile.TemporaryDirectory() as tmpdir:
-            repo = Repository.create(local_filesystem_storage(tmpdir))
+            repo = Repository.create(
+                local_filesystem_storage(tmpdir),
+                spec_version=any_spec_version,
+            )
 
             session = repo.writable_session("main")
             ds.to_zarr(session.store, mode="w", compute=False)

icechunk-python/tests/test_distributed_append.py

@@ -21,7 +21,7 @@
 )
 
 
-def do_test(scheduler) -> None:
+def do_test(scheduler, spec_version: int | None) -> None:
     # Writing the initial dataset
     if scheduler in ["processes", "sync"]:
         CHUNKX = 3
@@ -69,7 +69,7 @@ def plot() -> None:
         }
     ).chunk(x=CHUNKX, y=CHUNKY)
 
-    repo = ic.Repository.open_or_create(IC_STORAGE)
+    repo = ic.Repository.open_or_create(IC_STORAGE, create_version=spec_version)
     session = repo.writable_session("main")
     with dask.config.set(scheduler=scheduler):
         to_icechunk(initial, session=session, mode="w", split_every=SPLIT_EVERY)
@@ -124,8 +124,8 @@ def plot() -> None:
 
 
 @pytest.mark.parametrize("scheduler", ["threads", "processes"])
-def test_dask_distributed_appends(scheduler) -> None:
-    do_test(scheduler)
+def test_dask_distributed_appends(scheduler, any_spec_version: int | None) -> None:
+    do_test(scheduler, spec_version=any_spec_version)
 
 
 if __name__ == "__main__":

icechunk-python/tests/test_distributed_writers.py

@@ -22,7 +22,9 @@
 CHUNKS_PER_TASK = 2
 
 
-def mk_repo(use_object_store: bool = False) -> icechunk.Repository:
+def mk_repo(
+    spec_version: int | None, use_object_store: bool = False
+) -> icechunk.Repository:
     if use_object_store:
         storage = s3_object_store_storage(
             endpoint_url="http://localhost:9000",
@@ -50,13 +52,16 @@ def mk_repo(use_object_store: bool = False) -> icechunk.Repository:
     repo = icechunk.Repository.open_or_create(
         storage=storage,
         config=repo_config,
+        create_version=spec_version,
     )
 
     return repo
 
 
 @pytest.mark.parametrize("use_object_store", [False, True])
-async def test_distributed_writers(use_object_store: bool) -> None:
+async def test_distributed_writers(
+    use_object_store: bool, any_spec_version: int | None
+) -> None:
     """Write to an array using uncoordinated writers, distributed via Dask.
 
     We create a big array, and then we split into workers, each worker gets
@@ -65,7 +70,7 @@ async def test_distributed_writers(use_object_store: bool) -> None:
     does a distributed commit. When done, we open the store again and verify
     we can write everything we have written.
     """
-    repo = mk_repo(use_object_store)
+    repo = mk_repo(any_spec_version, use_object_store)
     session = repo.writable_session(branch="main")
     store = session.store
 

icechunk-python/tests/test_error.py

@@ -10,10 +10,15 @@
 import zarr
 
 
-def test_error_message_when_snapshot_deleted(tmpdir: Path) -> None:
+def test_error_message_when_snapshot_deleted(
+    tmpdir: Path, any_spec_version: int | None
+) -> None:
     tmpdir = Path(tmpdir)
     storage = ic.local_filesystem_storage(str(tmpdir))
-    repo = ic.Repository.create(storage=storage)
+    repo = ic.Repository.create(
+        storage=storage,
+        spec_version=any_spec_version,
+    )
 
     rmtree(tmpdir / "snapshots")
 
@@ -27,10 +32,15 @@ def test_error_message_when_snapshot_deleted(tmpdir: Path) -> None:
         zarr.group(store=session.store, overwrite=True)
 
 
-def test_error_message_when_manifest_file_altered(tmpdir: Path) -> None:
+def test_error_message_when_manifest_file_altered(
+    tmpdir: Path, any_spec_version: int | None
+) -> None:
     tmpdir = Path(tmpdir)
     storage = ic.local_filesystem_storage(str(tmpdir))
-    repo = ic.Repository.create(storage=storage)
+    repo = ic.Repository.create(
+        storage=storage,
+        spec_version=any_spec_version,
+    )
 
     session = repo.writable_session("main")
     store = session.store