Fix explicitly requested empty protocol

upath/_protocol.py

@@ -49,11 +49,11 @@ def _fsspec_protocol_equals(p0: str, p1: str) -> bool:
     try:
         o0 = _fsspec_registry_map[p0]
     except KeyError:
-        raise ValueError(f"Protocol not known: {p0}")
+        raise ValueError(f"Protocol not known: {p0!r}")
     try:
         o1 = _fsspec_registry_map[p1]
     except KeyError:
-        raise ValueError(f"Protocol not known: {p1}")
+        raise ValueError(f"Protocol not known: {p1!r}")
 
     return o0 == o1
 
@@ -81,14 +81,22 @@ def get_upath_protocol(
         pth_protocol = _match_protocol(str(pth))
     # if storage_options and not protocol and not pth_protocol:
     #     protocol = "file"
-    if (
+    if protocol is None:
+        return pth_protocol or ""
+    elif (
         protocol
         and pth_protocol
         and not _fsspec_protocol_equals(pth_protocol, protocol)
     ):
         raise ValueError(
             f"requested protocol {protocol!r} incompatible with {pth_protocol!r}"
         )
+    elif protocol == "" and pth_protocol:
+        # explicitly requested empty protocol, but path has non-empty protocol
+        raise ValueError(
+            f"explicitly requested empty protocol {protocol!r}"
+            f" incompatible with {pth_protocol!r}"
+        )
     return protocol or pth_protocol or ""
 
 

upath/core.py

@@ -1219,7 +1219,10 @@ def __get_pydantic_core_schema__(
                         ),
                         "protocol": core_schema.typed_dict_field(
                             core_schema.with_default_schema(
-                                core_schema.str_schema(), default=""
+                                core_schema.nullable_schema(
+                                    core_schema.str_schema(),
+                                ),
+                                default=None,
                             ),
                             required=False,
                         ),

upath/tests/test_core.py

@@ -439,3 +439,38 @@ def test_open_a_local_upath(tmp_path, protocol):
     u = UPath(p, protocol=protocol)
     with open(u, "rb") as f:
         assert f.read() == b"hello world"
+
+
+@pytest.mark.parametrize(
+    "uri,protocol",
+    [
+        ("s3://bucket/folder", "s3"),
+        ("gs://bucket/folder", "gs"),
+        ("bucket/folder", "s3"),
+        ("memory://folder", "memory"),
+        ("file:/tmp/folder", "file"),
+        ("/tmp/folder", "file"),
+        ("/tmp/folder", ""),
+        ("a/b/c", ""),
+    ],
+)
+def test_constructor_compatible_protocol_uri(uri, protocol):
+    p = UPath(uri, protocol=protocol)
+    assert p.protocol == protocol
+
+
+@pytest.mark.parametrize(
+    "uri,protocol",
+    [
+        ("s3://bucket/folder", "gs"),
+        ("gs://bucket/folder", "s3"),
+        ("memory://folder", "s3"),
+        ("file:/tmp/folder", "s3"),
+        ("s3://bucket/folder", ""),
+        ("memory://folder", ""),
+        ("file:/tmp/folder", ""),
+    ],
+)
+def test_constructor_incompatible_protocol_uri(uri, protocol):
+    with pytest.raises(ValueError, match=r".*incompatible with"):
+        UPath(uri, protocol=protocol)