Skip to content

Commit

Permalink
format:fix
Browse files Browse the repository at this point in the history
  • Loading branch information
@muddymudskipper
muddymudskipper committed Jan 12, 2024
1 parent c4ae561 commit 89449ec
Show file tree
Hide file tree
Showing 3 changed files with 40 additions and 71 deletions.
49 changes: 20 additions & 29 deletions cmem_plugin_uuid/transform/__init__.py
View file
Original file line number Diff line number Diff line change
@@ -1,27 +1,28 @@
"""UUID6 transform plugin module"""
from typing import Sequence
import re
import uuid
from collections.abc import Sequence

import uuid6
from cmem_plugin_base.dataintegration.description import Plugin, PluginParameter
from cmem_plugin_base.dataintegration.plugins import TransformPlugin
from cmem_plugin_base.dataintegration.types import BoolParameterType

from cmem_plugin_uuid.utils import (
uuid3_uuid5_namespace_param,
uuid_convert_param_in,
uuid_convert_param_out,
node_to_int,
clock_seq_to_int,
get_namespace_uuid,
namespace_hex,
node_to_int,
uuid3_uuid5_namespace_param,
uuid_convert_param_in,
uuid_convert_param_out,
)


@Plugin(
label="UUID1",
categories=["Value", "Identifier"],
description="Generate a UUIDv1 from a host ID, sequence number, and the "
"current time",
description="Generate a UUIDv1 from a host ID, sequence number, and the " "current time",
documentation="""
UUIDv1 is generated from a host ID, sequence number, and the current
time.
Expand Down Expand Up @@ -65,8 +66,7 @@ def transform(self, inputs: Sequence[Sequence[str]]) -> Sequence[str]:
if len(inputs) != 0:
for collection in inputs:
result += [
str(uuid.uuid1(node=self.node, clock_seq=self.clock_seq))
for _ in collection
str(uuid.uuid1(node=self.node, clock_seq=self.clock_seq)) for _ in collection
]
else:
result += [str(uuid.uuid1(node=self.node, clock_seq=self.clock_seq))]
Expand Down Expand Up @@ -128,7 +128,7 @@ def transform(self, inputs: Sequence[Sequence[str]]) -> Sequence[str]:
else:
result += [str(uuid.uuid3(namespace_uuid, _))]
else:
raise IOError("No input for UUID3")
raise OSError("No input for UUID3")
return result


Expand Down Expand Up @@ -206,15 +206,14 @@ def transform(self, inputs: Sequence[Sequence[str]]) -> Sequence[str]:
else:
result += [str(uuid.uuid5(namespace_uuid, _))]
else:
raise IOError("No input for UUID5")
raise OSError("No input for UUID5")
return result


@Plugin(
label="UUID6",
categories=["Value", "Identifier"],
description="Generate a UUIDv6 from a host ID, sequence number, and the "
"current time",
description="Generate a UUIDv6 from a host ID, sequence number, and the " "current time",
documentation="""
UUIDv6 is generated from a host ID, sequence number, and the current
time.
Expand Down Expand Up @@ -261,8 +260,7 @@ def transform(self, inputs: Sequence[Sequence[str]]) -> Sequence[str]:
if len(inputs) != 0:
for collection in inputs:
result += [
str(uuid6.uuid6(node=self.node, clock_seq=self.clock_seq))
for _ in collection
str(uuid6.uuid6(node=self.node, clock_seq=self.clock_seq)) for _ in collection
]
else:
result += [str(uuid6.uuid6(node=self.node, clock_seq=self.clock_seq))]
Expand Down Expand Up @@ -293,7 +291,7 @@ def transform(self, inputs: Sequence[Sequence[str]]) -> Sequence[str]:
except ValueError as exc:
raise ValueError(f"{_} is not a valid UUIDv1 string") from exc
else:
raise IOError("No input for UUID1 to UUID6")
raise OSError("No input for UUID1 to UUID6")
return result


Expand Down Expand Up @@ -381,13 +379,9 @@ def __init__(self, from_format=None, to_format=None):
self.to_ = to_format

def uuid_validate(self, test_uuid, uuid_string):
"""Warning if UUID string not standard (versions 1 to 8)
"""
Warning if UUID string not standard (versions 1 to 8)
"""

pattern = (
r"^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$"
)
pattern = r"^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$"

if not re.match(pattern, str(test_uuid)):
self.log.warning(
Expand All @@ -396,15 +390,12 @@ def uuid_validate(self, test_uuid, uuid_string):
)

def convert_uuid(self, uuid_string, result=None):
"""convert UUID string"""

"""Convert UUID string"""
if self.from_ == "uuid_hex":
try:
in_uuid = uuid.UUID(uuid_string)
except ValueError as exc:
raise ValueError(
f"{uuid_string} is not a valid 32-bit UUID string"
) from exc
raise ValueError(f"{uuid_string} is not a valid 32-bit UUID string") from exc

elif self.from_ == "int":
try:
Expand Down Expand Up @@ -441,7 +432,7 @@ def transform(self, inputs: Sequence[Sequence[str]]) -> Sequence[str]:
for collection in inputs:
result += [self.convert_uuid(_) for _ in collection]
else:
raise IOError("No input for UUID Convert")
raise OSError("No input for UUID Convert")
return result


Expand All @@ -460,5 +451,5 @@ def transform(self, inputs: Sequence[Sequence[str]]) -> Sequence[str]:
for collection in inputs:
result += [str(uuid.UUID(_).version) for _ in collection]
else:
raise IOError("No input for UUID Version")
raise OSError("No input for UUID Version")
return result
19 changes: 7 additions & 12 deletions cmem_plugin_uuid/utils.py
View file
Original file line number Diff line number Diff line change
@@ -1,11 +1,11 @@
"""Utilities for cmem-plugin-uuid"""

import uuid
from hashlib import md5, sha1
from collections import OrderedDict
from binascii import unhexlify
from cmem_plugin_base.dataintegration.parameter.choice import ChoiceParameterType
from collections import OrderedDict
from hashlib import md5, sha1

from cmem_plugin_base.dataintegration.parameter.choice import ChoiceParameterType

uuid3_uuid5_namespace_param = ChoiceParameterType(
OrderedDict(
Expand Down Expand Up @@ -50,8 +50,7 @@


def node_to_int(node: str):
"""
Convert a string representation of a node byte array to an integer.
"""Convert a string representation of a node byte array to an integer.
E.g. 01:23:45:67:89:AB -> 1250999896491
"""
try:
Expand All @@ -71,8 +70,7 @@ def clock_seq_to_int(clock_seq: str):


def namespace_hex(value, uuid_version):
"""
Return hex string from input value
"""Return hex string from input value
"""
hex_value = None
if uuid_version == 3:
Expand All @@ -87,8 +85,7 @@ def get_namespace_uuid(
namespace=None,
uuid_version=None,
):
"""returns namespace UUID"""

"""Returns namespace UUID"""
namespace_uuid = None

if namespace == "namespace_url":
Expand All @@ -103,8 +100,6 @@ def get_namespace_uuid(
if namespace_as_uuid:
namespace_uuid = uuid.UUID(namespace)
else:
namespace_uuid = uuid.UUID(
hex=namespace_hex(namespace, uuid_version), version=1
)
namespace_uuid = uuid.UUID(hex=namespace_hex(namespace, uuid_version), version=1)

return namespace_uuid
43 changes: 13 additions & 30 deletions tests/test_uuid.py
View file
Original file line number Diff line number Diff line change
@@ -1,21 +1,22 @@
"""Plugin tests."""
import uuid
import uuid6
from hashlib import md5, sha1

import uuid6

from cmem_plugin_uuid.transform import (
UUID1,
UUID3,
UUID4,
UUID5,
UUID6,
UUID1ToUUID6,
UUID7,
UUID8,
UUID1ToUUID6,
UUIDConvert,
UUIDVersion,
)


# Test UUID1


Expand Down Expand Up @@ -64,9 +65,7 @@ def test_uuid3_with_namespace_param_as_uuid():
assert len(result) == 2
for i, item in enumerate(result):
assert uuid.UUID(item).version == 3
assert item == str(
uuid.uuid3(namespace=uuid.UUID(namespace), name=input_values[i])
)
assert item == str(uuid.uuid3(namespace=uuid.UUID(namespace), name=input_values[i]))


def test_uuid3_with_namespace_param_as_string():
Expand Down Expand Up @@ -192,9 +191,7 @@ def test_uuid5_with_namespace_param_as_uuid():
assert len(result) == 2
for i, item in enumerate(result):
assert uuid.UUID(item).version == 5
assert item == str(
uuid.uuid5(namespace=uuid.UUID(namespace), name=input_values[i])
)
assert item == str(uuid.uuid5(namespace=uuid.UUID(namespace), name=input_values[i]))


def test_uuid5_with_namespace_param_as_string():
Expand Down Expand Up @@ -284,9 +281,7 @@ def test_uuid5_with_empty_namespace():
assert result == [
str(
uuid.UUID(
sha1(input_values[0][0].encode(), usedforsecurity=False).hexdigest()[
:32
],
sha1(input_values[0][0].encode(), usedforsecurity=False).hexdigest()[:32],
version=5,
)
)
Expand Down Expand Up @@ -379,54 +374,42 @@ def test_uuid8_with_input():
def test_uuid_convert_to_uuid():
test_uuid = uuid.uuid4()
input_values = [str(test_uuid)]
result = UUIDConvert(from_format="uuid_hex", to_format="uuid").transform(
inputs=[input_values]
)
result = UUIDConvert(from_format="uuid_hex", to_format="uuid").transform(inputs=[input_values])
assert result == [str(test_uuid)]


def test_uuid_convert_to_hex():
test_uuid = uuid.uuid4()
input_values = [str(test_uuid)]
result = UUIDConvert(from_format="uuid_hex", to_format="hex").transform(
inputs=[input_values]
)
result = UUIDConvert(from_format="uuid_hex", to_format="hex").transform(inputs=[input_values])
assert result == [str(test_uuid.hex)]


def test_uuid_convert_to_int():
test_uuid = uuid.uuid4()
input_values = [str(test_uuid)]
result = UUIDConvert(from_format="uuid_hex", to_format="int").transform(
inputs=[input_values]
)
result = UUIDConvert(from_format="uuid_hex", to_format="int").transform(inputs=[input_values])
assert result == [str(test_uuid.int)]


def test_uuid_convert_to_urn():
test_uuid = uuid.uuid4()
input_values = [str(test_uuid)]
result = UUIDConvert(from_format="uuid_hex", to_format="urn").transform(
inputs=[input_values]
)
result = UUIDConvert(from_format="uuid_hex", to_format="urn").transform(inputs=[input_values])
assert result == [str(test_uuid.urn)]


def test_uuid_convert_from_int():
test_uuid = uuid.uuid4()
input_values = [str(test_uuid.int)]
result = UUIDConvert(from_format="int", to_format="uuid").transform(
inputs=[input_values]
)
result = UUIDConvert(from_format="int", to_format="uuid").transform(inputs=[input_values])
assert result == [str(test_uuid)]


def test_uuid_convert_from_urn():
test_uuid = uuid.uuid4()
input_values = [str(test_uuid.urn)]
result = UUIDConvert(from_format="urn", to_format="uuid").transform(
inputs=[input_values]
)
result = UUIDConvert(from_format="urn", to_format="uuid").transform(inputs=[input_values])
assert result == [str(test_uuid)]


Expand Down

0 comments on commit 89449ec

Please sign in to comment.