(texera) 01/01: test(pyamber): add unit tests for vfs_uri_factory (#6641)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6641-37c68872fa42afdf3646e35e7d9f7b620de05c2f
in repository https://gitbox.apache.org/repos/asf/texera.git

commit 656bc084b4d0c704c7dffe621e5fd26b47da2984
Author: Matthew B. <mg...@uci.edu>
AuthorDate: Mon Jul 20 15:23:37 2026 -0700

    test(pyamber): add unit tests for vfs_uri_factory (#6641)
    
    ### What changes were proposed in this PR?
    - Add `amber/src/test/python/core/storage/test_vfs_uri_factory.py`, a
    new pytest spec for the VFS URI factory, which previously had no
    dedicated unit tests.
    - Cover base-URI construction and result_uri/state_uri segment
    appending.
    - Cover build-to-decode round-trips for result and state URIs (wid/eid,
    full GlobalPortIdentity fields, resource type) and the absent-port None
    branch.
    - Cover all decode error paths: bad scheme, missing query key, and
    unknown resource type.
    ### Any related issues, documentation, discussions?
    Closes: #6640
    ### How was this PR tested?
    - Run: `cd amber && python -m pytest
    src/test/python/core/storage/test_vfs_uri_factory.py`, expect all 10
    tests passing.
    - Test-only change; no production code is modified.
    ### Was this PR authored or co-authored using generative AI tooling?
    Co-authored with Claude Opus 4.8 in compliance with ASF
---
 .../python/core/storage/test_vfs_uri_factory.py    | 140 +++++++++++++++++++++
 1 file changed, 140 insertions(+)

diff --git a/amber/src/test/python/core/storage/test_vfs_uri_factory.py b/amber/src/test/python/core/storage/test_vfs_uri_factory.py
new file mode 100644
index 0000000000..90ea44a52d
--- /dev/null
+++ b/amber/src/test/python/core/storage/test_vfs_uri_factory.py
@@ -0,0 +1,140 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import pytest
+
+from core.storage.vfs_uri_factory import VFSResourceType, VFSURIFactory
+from proto.org.apache.texera.amber.core import (
+    ExecutionIdentity,
+    GlobalPortIdentity,
+    OperatorIdentity,
+    PhysicalOpIdentity,
+    PortIdentity,
+    WorkflowIdentity,
+)
+
+
+def _gpi(
+    op_id: str = "myOp",
+    layer: str = "main",
+    port: int = 0,
+    internal: bool = False,
+    is_input: bool = True,
+) -> GlobalPortIdentity:
+    return GlobalPortIdentity(
+        op_id=PhysicalOpIdentity(
+            logical_op_id=OperatorIdentity(id=op_id), layer_name=layer
+        ),
+        port_id=PortIdentity(id=port, internal=internal),
+        input=is_input,
+    )
+
+
+class TestCreatePortBaseUri:
+    def test_base_uri_encodes_scheme_ids_and_serialized_port(self):
+        uri = VFSURIFactory.create_port_base_uri(
+            WorkflowIdentity(id=7), ExecutionIdentity(id=3), _gpi()
+        )
+        # The base URI stitches together the vfs scheme, the wid/eid segments,
+        # and the serialized global port identity as a single trailing segment.
+        assert uri == (
+            "vfs:///wid/7/eid/3/globalportid/"
+            "(logicalOpId=myOp,layerName=main,portId=0,"
+            "isInternal=false,isInput=true)"
+        )
+
+    def test_result_uri_appends_result_segment(self):
+        base = VFSURIFactory.create_port_base_uri(
+            WorkflowIdentity(id=1), ExecutionIdentity(id=1), _gpi()
+        )
+        assert VFSURIFactory.result_uri(base) == f"{base}/result"
+
+    def test_state_uri_appends_state_segment(self):
+        base = VFSURIFactory.create_port_base_uri(
+            WorkflowIdentity(id=1), ExecutionIdentity(id=1), _gpi()
+        )
+        assert VFSURIFactory.state_uri(base) == f"{base}/state"
+
+
+class TestDecodeUriRoundTrip:
+    def test_result_uri_round_trips_through_decode(self):
+        wid, eid, gpi = (
+            WorkflowIdentity(id=42),
+            ExecutionIdentity(id=9),
+            _gpi(op_id="opA", layer="main", port=2, internal=True, is_input=False),
+        )
+        base = VFSURIFactory.create_port_base_uri(wid, eid, gpi)
+        uri = VFSURIFactory.result_uri(base)
+
+        decoded_wid, decoded_eid, decoded_gpi, resource_type = VFSURIFactory.decode_uri(
+            uri
+        )
+
+        assert decoded_wid.id == 42
+        assert decoded_eid.id == 9
+        assert resource_type == VFSResourceType.RESULT
+        assert decoded_gpi is not None
+        assert decoded_gpi.op_id.logical_op_id.id == "opA"
+        assert decoded_gpi.op_id.layer_name == "main"
+        assert decoded_gpi.port_id.id == 2
+        assert decoded_gpi.port_id.internal is True
+        assert decoded_gpi.input is False
+
+    def test_state_uri_round_trips_through_decode(self):
+        wid, eid, gpi = WorkflowIdentity(id=5), ExecutionIdentity(id=6), _gpi()
+        uri = VFSURIFactory.state_uri(VFSURIFactory.create_port_base_uri(wid, eid, gpi))
+
+        decoded_wid, decoded_eid, decoded_gpi, resource_type = VFSURIFactory.decode_uri(
+            uri
+        )
+
+        assert decoded_wid.id == 5
+        assert decoded_eid.id == 6
+        assert resource_type == VFSResourceType.STATE
+        assert decoded_gpi.op_id.logical_op_id.id == "myOp"
+
+    def test_decode_returns_none_port_when_globalportid_absent(self):
+        # A URI without a globalportid segment yields a None port identity,
+        # exercising the optional branch in decode_uri.
+        wid, eid, port, resource_type = VFSURIFactory.decode_uri(
+            "vfs:///wid/11/eid/22/result"
+        )
+        assert wid.id == 11
+        assert eid.id == 22
+        assert port is None
+        assert resource_type == VFSResourceType.RESULT
+
+
+class TestDecodeUriErrorPaths:
+    def test_rejects_non_vfs_scheme(self):
+        with pytest.raises(ValueError, match="Invalid URI scheme"):
+            VFSURIFactory.decode_uri("http:///wid/1/eid/1/result")
+
+    def test_rejects_missing_key_when_wid_absent(self):
+        # 'wid' is never present, so extract_value fails on the index lookup.
+        with pytest.raises(ValueError, match="Missing value for key: wid"):
+            VFSURIFactory.decode_uri("vfs:///eid/1/result")
+
+    def test_rejects_key_present_but_value_missing(self):
+        # 'eid' is the final segment with no following value; the IndexError
+        # branch of extract_value must surface as a Missing-value error.
+        with pytest.raises(ValueError, match="Missing value for key: eid"):
+            VFSURIFactory.decode_uri("vfs:///wid/1/eid")
+
+    def test_rejects_unknown_resource_type(self):
+        with pytest.raises(ValueError, match="Unknown resource type: bogus"):
+            VFSURIFactory.decode_uri("vfs:///wid/1/eid/1/bogus")