Skip to content

Commit

Permalink
Merge latest changes (#18)
Browse files Browse the repository at this point in the history 
* Refactor exception printing
* Refactor tests
* Include entry in assertion message
* Replace redundant assertions
* Rename functions with inconsistent prefix
* Update testing_data.py
* Annotate entries
* Add missing encodings
* Refactor is_ipfs.py
* Add missing  ipfs path_pattern check
* Correct wrong assumption about native url validation
* Bump version
  • Loading branch information
@Barabazs
Barabazs authored May 24, 2022
1 parent ae9de23 commit 1c947e2
Show file tree
Hide file tree
Showing 9 changed files with 192 additions and 144 deletions.
64 changes: 33 additions & 31 deletions is_ipfs/is_ipfs.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -48,15 +48,15 @@ def _is_cid(self) -> bool:
elif type(self.input) == bytes:
try:
return cid.is_cid(decode(self.input))
except Exception as error:
print(f"Unexpected {type(error)}, {error}")
except ValueError as error:
print(f"Unexpected ValueError, {error}")
return False
else:
try:
if isinstance(self.input, (cid.CIDv0, cid.CIDv1)):
return cid.is_cid(str(self.input))
except Exception as error:
print(f"Unexpected {type(error)}, {error}")
except ValueError as error:
print(f"Unexpected ValueError, {error}")
return False
return False

Expand All @@ -80,18 +80,19 @@ def _is_integral_ipfs_url(

_hash = match["hash"]

if (
pattern == self.subdomain_gateway_pattern
or pattern == self.native_url_pattern
):
if pattern == self.subdomain_gateway_pattern:
_hash = _hash.lower()
try:
if get_codec(_hash).encoding not in ["base32", "base36"]:
if get_codec(_hash).encoding not in ["base32", "base32hex", "base36"]:
return False
except Exception as error:
print(f"Unexpected {type(error)}, {error}")
except ValueError as error:
print(f"Unexpected ValueError, {error}")
return False
elif pattern == self.path_gateway_pattern:
elif (
(pattern == self.path_gateway_pattern)
or (pattern == self.native_url_pattern)
or (pattern == self.path_pattern)
):
if not str(_hash).startswith("Qm"):
try:
if get_codec(_hash).encoding not in [
Expand All @@ -106,21 +107,21 @@ def _is_integral_ipfs_url(
"base64url",
]:
return False
except Exception as error:
print(f"Unexpected {type(error)}, {error}")
except ValueError as error:
print(f"Unexpected ValueError, {error}")
return False

return Validator(_hash)._is_cid()

def _ipfs_subdomain_url(self) -> bool:
def _is_ipfs_subdomain_url(self) -> bool:
"""
Returns True if the provided url string includes a valid IPFS subdomain (case-insensitive CIDv1) or False otherwise.
"""
return self._is_integral_ipfs_url(
self.subdomain_gateway_pattern,
)

def _ipfs_path_url(self) -> bool:
def _is_ipfs_path_url(self) -> bool:
"""
Returns True if the provided url string is a valid IPFS URL or False otherwise.
"""
Expand All @@ -132,7 +133,7 @@ def _is_ipfs_url(self) -> bool:
"""
Returns True if the provided string is a valid IPFS url or False otherwise.
"""
return self._ipfs_path_url() or self._ipfs_subdomain_url()
return self._is_ipfs_path_url() or self._is_ipfs_subdomain_url()

def _is_ipfs_path(self) -> bool:
"""
Expand Down Expand Up @@ -160,18 +161,15 @@ def _is_integral_ipns_url(

ipns_id = match["hash"]

if (ipns_id) and (
pattern == self.subdomain_gateway_pattern
or pattern == self.native_url_pattern
):
if pattern == self.subdomain_gateway_pattern:
ipns_id = ipns_id.lower()

if Validator(ipns_id)._is_cid():
try:
if get_codec(ipns_id).encoding == "base36":
return True
except Exception as error:
print(f"Unexpected {type(error)}, {error}")
except ValueError as error:
print(f"Unexpected ValueError, {error}")
return False
try:
if ("." not in ipns_id and "-" in ipns_id) and (
Expand All @@ -182,11 +180,15 @@ def _is_integral_ipns_url(
)

return self._id_is_explicit_tld(ipns_id)
except Exception as error:
print(f"Unexpected {type(error)}, {error}")
except ValueError as error:
print(f"Unexpected ValueError, {error}")
return False

elif pattern == self.path_gateway_pattern or pattern == self.path_pattern:
elif (
(pattern == self.path_gateway_pattern)
or (pattern == self.path_pattern)
or (pattern == self.native_url_pattern)
):
if not str(ipns_id).startswith("Qm"):
if self._id_is_explicit_tld(ipns_id):
return True
Expand All @@ -204,21 +206,21 @@ def _is_integral_ipns_url(
"base64url",
]:
return False
except Exception as error:
print(f"Unexpected {type(error)}, {error}")
except ValueError as error:
print(f"Unexpected ValueError, {error}")
return False

return Validator(ipns_id)._is_cid()

def _ipns_subdomain_url(self) -> bool:
def _is_ipns_subdomain_url(self) -> bool:
"""
Returns True if the provided url string includes a valid IPFS subdomain (case-insensitive CIDv1) or False otherwise.
"""
return self._is_integral_ipns_url(
self.subdomain_gateway_pattern,
)

def _ipns_path_url(self) -> bool:
def _is_ipns_path_url(self) -> bool:
"""
Returns True if the provided url string is a valid IPFS URL or False otherwise.
"""
Expand All @@ -230,7 +232,7 @@ def _is_ipns_url(self) -> bool:
"""
Returns True if the provided string is a valid IPFS url or False otherwise.
"""
return self._ipns_path_url() or self._ipns_subdomain_url()
return self._is_ipns_path_url() or self._is_ipns_subdomain_url()

def _is_ipns_path(self) -> bool:
"""
Expand Down
2 changes: 1 addition & 1 deletion setup.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@

setuptools.setup(
name="py-is_ipfs",
version="0.2.0",
version="0.3.0",
description="Python library to identify valid IPFS resources",
long_description=long_description,
long_description_content_type="text/markdown",
Expand Down
40 changes: 20 additions & 20 deletions tests/integration/test_is_ipfs.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,78 +8,78 @@ class TestCase(unittest.TestCase):
def test_all(self):

with self.subTest("Test valid CID entries from fixtures"):
for key, value in testing_data.valid_entries["cid"].items():
for value in testing_data.valid_entries["cid"].values():
for entry in value:
self.assertTrue(Validator(entry).is_ipfs())
self.assertTrue(Validator(entry).is_ipfs(), entry)

with self.subTest("Test invalid CID entries from fixtures"):
for key, value in testing_data.invalid_entries["cid"].items():
for value in testing_data.invalid_entries["cid"].values():
for entry in value:
self.assertFalse(Validator(entry).is_ipfs())
self.assertFalse(Validator(entry).is_ipfs(), entry)

with self.subTest("Test valid IPFS URL entries from fixtures"):
for entry in testing_data.valid_entries["url"]["ipfs"]:
self.assertTrue(Validator(entry).is_ipfs())
self.assertTrue(Validator(entry).is_ipfs(), entry)

with self.subTest("Test invalid IPFS URL entries from fixtures"):
for entry in testing_data.invalid_entries["url"]["ipfs"]:
self.assertFalse(Validator(entry).is_ipfs())
self.assertFalse(Validator(entry).is_ipfs(), entry)

with self.subTest("Test valid IPFS subdomain entries from fixtures"):
for entry in testing_data.valid_entries["subdomain"]["ipfs"]:
self.assertTrue(Validator(entry).is_ipfs())
self.assertTrue(Validator(entry).is_ipfs(), entry)

with self.subTest("Test invalid IPFS subdomain entries from fixtures"):
for entry in testing_data.invalid_entries["subdomain"]["ipfs"]:
self.assertFalse(Validator(entry).is_ipfs())
self.assertFalse(Validator(entry).is_ipfs(), entry)

with self.subTest("Test valid IPFS path entries from fixtures"):
for entry in testing_data.valid_entries["path"]["ipfs"]:
self.assertTrue(Validator(entry).is_ipfs())
self.assertTrue(Validator(entry).is_ipfs(), entry)

with self.subTest("Test invalid IPFS path entries from fixtures"):
for entry in testing_data.invalid_entries["path"]["ipfs"]:
self.assertFalse(Validator(entry).is_ipfs())
self.assertFalse(Validator(entry).is_ipfs(), entry)

with self.subTest("Test valid IPNS URL entries from fixtures"):
for entry in testing_data.valid_entries["url"]["ipns"]:
self.assertTrue(Validator(entry).is_ipfs())
self.assertTrue(Validator(entry).is_ipfs(), entry)

with self.subTest("Test invalid IPNS URL entries from fixtures"):
for entry in testing_data.invalid_entries["url"]["ipns"]:
self.assertFalse(Validator(entry).is_ipfs())
self.assertFalse(Validator(entry).is_ipfs(), entry)

with self.subTest("Test valid IPNS subdomain entries from fixtures"):
for entry in testing_data.valid_entries["subdomain"]["ipns"]:
self.assertTrue(Validator(entry).is_ipfs())
self.assertTrue(Validator(entry).is_ipfs(), entry)

with self.subTest("Test invalid IPNS subdomain entries from fixtures"):
for entry in testing_data.invalid_entries["subdomain"]["ipns"]:
self.assertFalse(Validator(entry).is_ipfs())
self.assertFalse(Validator(entry).is_ipfs(), entry)

with self.subTest("Test valid IPNS path entries from fixtures"):
for entry in testing_data.valid_entries["path"]["ipns"]:
self.assertTrue(Validator(entry).is_ipfs())
self.assertTrue(Validator(entry).is_ipfs(), entry)

with self.subTest("Test invalid IPNS path entries from fixtures"):
for entry in testing_data.invalid_entries["path"]["ipns"]:
self.assertFalse(Validator(entry).is_ipfs())
self.assertFalse(Validator(entry).is_ipfs(), entry)

with self.subTest("Test valid IPFS native URL entries from fixtures"):
for entry in testing_data.valid_entries["native_url"]["ipfs"]:
self.assertTrue(Validator(entry).is_ipfs())
self.assertTrue(Validator(entry).is_ipfs(), entry)

with self.subTest("Test invalid IPFS native URL entries from fixtures"):
for entry in testing_data.invalid_entries["native_url"]["ipfs"]:
self.assertFalse(Validator(entry).is_ipfs())
self.assertFalse(Validator(entry).is_ipfs(), entry)

with self.subTest("Test valid IPNS native URL entries from fixtures"):
for entry in testing_data.valid_entries["native_url"]["ipns"]:
self.assertTrue(Validator(entry).is_ipfs())
self.assertTrue(Validator(entry).is_ipfs(), entry)

with self.subTest("Test invalid IPNS native URL from fixtures"):
for entry in testing_data.invalid_entries["native_url"]["ipns"]:
self.assertFalse(Validator(entry).is_ipfs())
self.assertFalse(Validator(entry).is_ipfs(), entry)


if __name__ == "__main__": # pragma: no cover
Expand Down
Loading

0 comments on commit 1c947e2

Please sign in to comment.