This is an automated email from the ASF dual-hosted git repository.
kojiromike pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/avro.git
The following commit(s) were added to refs/heads/master by this push:
new a3e4572 AVRO-2921: Type Fixes for avro.codecs (#1260)
a3e4572 is described below
commit a3e4572dc6de4097cdf9d1e9fe984ed59ba281ad
Author: Michael A. Smith <[email protected]>
AuthorDate: Wed Jun 16 19:28:10 2021 -0400
AVRO-2921: Type Fixes for avro.codecs (#1260)
* AVRO-2921: Fixup Test Coverage
Ensure test coverage is good before adding more type hints.
* AVRO-2921: Type Fixes for avro.codecs
---
lang/py/avro/codecs.py | 86 ++++++++++++++--------------
lang/py/avro/datafile.py | 2 +-
lang/py/avro/test/gen_interop_data.py | 2 +-
lang/py/avro/test/test_bench.py | 4 +-
lang/py/avro/test/test_datafile.py | 2 +-
lang/py/avro/test/test_init.py | 2 +-
lang/py/avro/test/test_io.py | 2 +-
lang/py/avro/test/test_ipc.py | 2 +-
lang/py/avro/test/test_protocol.py | 19 +++---
lang/py/avro/test/test_schema.py | 30 ++++------
lang/py/avro/test/test_script.py | 2 +-
lang/py/avro/test/test_tether_task.py | 2 +-
lang/py/avro/test/test_tether_task_runner.py | 22 +------
lang/py/avro/test/test_tether_word_count.py | 4 +-
lang/py/avro/tether/tether_task_runner.py | 20 +++++--
15 files changed, 92 insertions(+), 109 deletions(-)
diff --git a/lang/py/avro/codecs.py b/lang/py/avro/codecs.py
index fa89d77..e1655fe 100644
--- a/lang/py/avro/codecs.py
+++ b/lang/py/avro/codecs.py
@@ -30,8 +30,8 @@ import abc
import binascii
import io
import struct
-import sys
import zlib
+from typing import Dict, Tuple, Type
import avro.errors
import avro.io
@@ -42,6 +42,11 @@ import avro.io
STRUCT_CRC32 = struct.Struct(">I") # big-endian unsigned int
+def _check_crc32(bytes_: bytes, checksum: bytes) -> None:
+ if binascii.crc32(bytes_) & 0xFFFFFFFF != STRUCT_CRC32.unpack(checksum)[0]:
+ raise avro.errors.AvroException("Checksum failure")
+
+
try:
import bz2
@@ -65,8 +70,9 @@ except ImportError:
class Codec(abc.ABC):
"""Abstract base class for all Avro codec classes."""
+ @staticmethod
@abc.abstractmethod
- def compress(self, data):
+ def compress(data: bytes) -> Tuple[bytes, int]:
"""Compress the passed data.
:param data: a byte string to be compressed
@@ -76,8 +82,9 @@ class Codec(abc.ABC):
:return: compressed data and its length
"""
+ @staticmethod
@abc.abstractmethod
- def decompress(self, readers_decoder):
+ def decompress(readers_decoder: avro.io.BinaryDecoder) ->
avro.io.BinaryDecoder:
"""Read compressed data via the passed BinaryDecoder and decompress it.
:param readers_decoder: a BinaryDecoder object currently being used for
@@ -91,22 +98,26 @@ class Codec(abc.ABC):
class NullCodec(Codec):
- def compress(self, data):
+ @staticmethod
+ def compress(data: bytes) -> Tuple[bytes, int]:
return data, len(data)
- def decompress(self, readers_decoder):
+ @staticmethod
+ def decompress(readers_decoder: avro.io.BinaryDecoder) ->
avro.io.BinaryDecoder:
readers_decoder.skip_long()
return readers_decoder
class DeflateCodec(Codec):
- def compress(self, data):
+ @staticmethod
+ def compress(data: bytes) -> Tuple[bytes, int]:
# The first two characters and last character are zlib
# wrappers around deflate data.
compressed_data = zlib.compress(data)[2:-1]
return compressed_data, len(compressed_data)
- def decompress(self, readers_decoder):
+ @staticmethod
+ def decompress(readers_decoder: avro.io.BinaryDecoder) ->
avro.io.BinaryDecoder:
# Compressed data is stored as (length, data), which
# corresponds to how the "bytes" type is encoded.
data = readers_decoder.read_bytes()
@@ -119,11 +130,13 @@ class DeflateCodec(Codec):
if has_bzip2:
class BZip2Codec(Codec):
- def compress(self, data):
+ @staticmethod
+ def compress(data: bytes) -> Tuple[bytes, int]:
compressed_data = bz2.compress(data)
return compressed_data, len(compressed_data)
- def decompress(self, readers_decoder):
+ @staticmethod
+ def decompress(readers_decoder: avro.io.BinaryDecoder) ->
avro.io.BinaryDecoder:
length = readers_decoder.read_long()
data = readers_decoder.read(length)
uncompressed = bz2.decompress(data)
@@ -133,35 +146,34 @@ if has_bzip2:
if has_snappy:
class SnappyCodec(Codec):
- def compress(self, data):
+ @staticmethod
+ def compress(data: bytes) -> Tuple[bytes, int]:
compressed_data = snappy.compress(data)
# A 4-byte, big-endian CRC32 checksum
compressed_data += STRUCT_CRC32.pack(binascii.crc32(data) &
0xFFFFFFFF)
return compressed_data, len(compressed_data)
- def decompress(self, readers_decoder):
+ @staticmethod
+ def decompress(readers_decoder: avro.io.BinaryDecoder) ->
avro.io.BinaryDecoder:
# Compressed data includes a 4-byte CRC32 checksum
length = readers_decoder.read_long()
data = readers_decoder.read(length - 4)
uncompressed = snappy.decompress(data)
checksum = readers_decoder.read(4)
- self.check_crc32(uncompressed, checksum)
+ _check_crc32(uncompressed, checksum)
return avro.io.BinaryDecoder(io.BytesIO(uncompressed))
- def check_crc32(self, bytes, checksum):
- checksum = STRUCT_CRC32.unpack(checksum)[0]
- if binascii.crc32(bytes) & 0xFFFFFFFF != checksum:
- raise avro.errors.AvroException("Checksum failure")
-
if has_zstandard:
class ZstandardCodec(Codec):
- def compress(self, data):
+ @staticmethod
+ def compress(data: bytes) -> Tuple[bytes, int]:
compressed_data = zstd.ZstdCompressor().compress(data)
return compressed_data, len(compressed_data)
- def decompress(self, readers_decoder):
+ @staticmethod
+ def decompress(readers_decoder: avro.io.BinaryDecoder) ->
avro.io.BinaryDecoder:
length = readers_decoder.read_long()
data = readers_decoder.read(length)
uncompressed = bytearray()
@@ -175,27 +187,15 @@ if has_zstandard:
return avro.io.BinaryDecoder(io.BytesIO(uncompressed))
-def get_codec(codec_name):
- codec_name = codec_name.lower()
- if codec_name == "null":
- return NullCodec()
- if codec_name == "deflate":
- return DeflateCodec()
- if codec_name == "bzip2" and has_bzip2:
- return BZip2Codec()
- if codec_name == "snappy" and has_snappy:
- return SnappyCodec()
- if codec_name == "zstandard" and has_zstandard:
- return ZstandardCodec()
- raise avro.errors.UnsupportedCodec(f"Unsupported codec: {codec_name}. (Is
it installed?)")
-
-
-def supported_codec_names():
- codec_names = ["null", "deflate"]
- if has_bzip2:
- codec_names.append("bzip2")
- if has_snappy:
- codec_names.append("snappy")
- if has_zstandard:
- codec_names.append("zstandard")
- return codec_names
+KNOWN_CODECS: Dict[str, Type[Codec]] = {
+ name[:-5].lower(): class_
+ for name, class_ in globals().items()
+ if class_ != Codec and name.endswith("Codec") and isinstance(class_, type)
and issubclass(class_, Codec)
+}
+
+
+def get_codec(codec_name: str) -> Type[Codec]:
+ try:
+ return KNOWN_CODECS[codec_name]
+ except KeyError:
+ raise avro.errors.UnsupportedCodec(f"Unsupported codec: {codec_name}.
(Is it installed?)")
diff --git a/lang/py/avro/datafile.py b/lang/py/avro/datafile.py
index 8a660bc..b766854 100644
--- a/lang/py/avro/datafile.py
+++ b/lang/py/avro/datafile.py
@@ -53,7 +53,7 @@ META_SCHEMA = avro.schema.parse(
)
NULL_CODEC = "null"
-VALID_CODECS = avro.codecs.supported_codec_names()
+VALID_CODECS = avro.codecs.KNOWN_CODECS.keys()
VALID_ENCODINGS = ["binary"] # not used yet
CODEC_KEY = "avro.codec"
diff --git a/lang/py/avro/test/gen_interop_data.py
b/lang/py/avro/test/gen_interop_data.py
index 68672ba..73b47ce 100644
--- a/lang/py/avro/test/gen_interop_data.py
+++ b/lang/py/avro/test/gen_interop_data.py
@@ -27,7 +27,7 @@ import avro.io
import avro.schema
NULL_CODEC = "null"
-CODECS_TO_VALIDATE = avro.codecs.supported_codec_names()
+CODECS_TO_VALIDATE = avro.codecs.KNOWN_CODECS.keys()
DATUM = {
"intField": 12,
diff --git a/lang/py/avro/test/test_bench.py b/lang/py/avro/test/test_bench.py
index 3bd1689..34c926b 100644
--- a/lang/py/avro/test/test_bench.py
+++ b/lang/py/avro/test/test_bench.py
@@ -52,9 +52,9 @@ MAX_WRITE_SECONDS = 3 if platform.python_implementation() ==
"PyPy" else 1
MAX_READ_SECONDS = 3 if platform.python_implementation() == "PyPy" else 1
-try:
+try: # pragma: no cover
randbytes = random.randbytes # type: ignore
-except AttributeError:
+except AttributeError: # pragma: no cover
def randbytes(n):
"""Polyfill for random.randbytes in Python < 3.9"""
diff --git a/lang/py/avro/test/test_datafile.py
b/lang/py/avro/test/test_datafile.py
index 5bb0d28..bb98857 100644
--- a/lang/py/avro/test/test_datafile.py
+++ b/lang/py/avro/test/test_datafile.py
@@ -28,7 +28,7 @@ import avro.datafile
import avro.io
import avro.schema
-CODECS_TO_VALIDATE = avro.codecs.supported_codec_names()
+CODECS_TO_VALIDATE = avro.codecs.KNOWN_CODECS.keys()
TEST_PAIRS = tuple(
(avro.schema.parse(schema), datum)
for schema, datum in (
diff --git a/lang/py/avro/test/test_init.py b/lang/py/avro/test/test_init.py
index edd14c8..e37dc19 100644
--- a/lang/py/avro/test/test_init.py
+++ b/lang/py/avro/test/test_init.py
@@ -28,5 +28,5 @@ class TestVersion(unittest.TestCase):
self.assertTrue(hasattr(avro, "__version__"))
-if __name__ == "__main__":
+if __name__ == "__main__": # pragma: no coverage
unittest.main()
diff --git a/lang/py/avro/test/test_io.py b/lang/py/avro/test/test_io.py
index 973f51d..db4d043 100644
--- a/lang/py/avro/test/test_io.py
+++ b/lang/py/avro/test/test_io.py
@@ -561,5 +561,5 @@ def load_tests(loader, default_tests, pattern):
return suite
-if __name__ == "__main__":
+if __name__ == "__main__": # pragma: no coverage
unittest.main()
diff --git a/lang/py/avro/test/test_ipc.py b/lang/py/avro/test/test_ipc.py
index 59006d7..969988a 100644
--- a/lang/py/avro/test/test_ipc.py
+++ b/lang/py/avro/test/test_ipc.py
@@ -39,5 +39,5 @@ class TestIPC(unittest.TestCase):
self.assertEqual("/", client_with_default_path.req_resource)
-if __name__ == "__main__":
+if __name__ == "__main__": # pragma: no coverage
unittest.main()
diff --git a/lang/py/avro/test/test_protocol.py
b/lang/py/avro/test/test_protocol.py
index 373abb9..b3cd7bd 100644
--- a/lang/py/avro/test/test_protocol.py
+++ b/lang/py/avro/test/test_protocol.py
@@ -411,17 +411,16 @@ class ProtocolParseTestCase(unittest.TestCase):
"""Parsing a valid protocol should not error."""
try:
self.test_proto.parse()
- except avro.errors.ProtocolParseException:
+ except avro.errors.ProtocolParseException: # pragma: no coverage
self.fail(f"Valid protocol failed to parse: {self.test_proto!s}")
def parse_invalid(self):
"""Parsing an invalid protocol should error."""
- try:
+ with self.assertRaises(
+ (avro.errors.ProtocolParseException,
avro.errors.SchemaParseException),
+ msg=f"Invalid protocol should not have parsed:
{self.test_proto!s}",
+ ):
self.test_proto.parse()
- except (avro.errors.ProtocolParseException,
avro.errors.SchemaParseException):
- pass
- else:
- self.fail(f"Invalid protocol should not have parsed:
{self.test_proto!s}")
class ErrorProtocolTestCase(unittest.TestCase):
@@ -439,8 +438,9 @@ class ErrorProtocolTestCase(unittest.TestCase):
def check_error_protocol_exists(self):
"""Protocol messages should always have at least a string error
protocol."""
p = self.test_proto.parse()
- for k, m in p.messages.items():
- self.assertIsNotNone(m.errors, f"Message {k} did not have the
expected implicit string error protocol.")
+ if p.messages is not None:
+ for k, m in p.messages.items():
+ self.assertIsNotNone(m.errors, f"Message {k} did not have the
expected implicit string error protocol.")
class RoundTripParseTestCase(unittest.TestCase):
@@ -468,8 +468,9 @@ def load_tests(loader, default_tests, pattern):
suite.addTests(loader.loadTestsFromTestCase(TestMisc))
suite.addTests(ProtocolParseTestCase(ex) for ex in EXAMPLES)
suite.addTests(RoundTripParseTestCase(ex) for ex in VALID_EXAMPLES)
+ suite.addTests(ErrorProtocolTestCase(ex) for ex in VALID_EXAMPLES)
return suite
-if __name__ == "__main__":
+if __name__ == "__main__": # pragma: no coverage
unittest.main()
diff --git a/lang/py/avro/test/test_schema.py b/lang/py/avro/test/test_schema.py
index 52ecca5..1a8e96a 100644
--- a/lang/py/avro/test/test_schema.py
+++ b/lang/py/avro/test/test_schema.py
@@ -691,18 +691,12 @@ class TestMisc(unittest.TestCase):
def test_parse_invalid_symbol(self):
"""Disabling enumschema symbol validation should allow invalid symbols
to pass."""
test_schema_string = json.dumps({"type": "enum", "name": "AVRO2174",
"symbols": ["white space"]})
-
- try:
- case = avro.schema.parse(test_schema_string,
validate_enum_symbols=True)
- except avro.errors.InvalidName:
- pass
- else:
- self.fail("When enum symbol validation is enabled, " "an invalid
symbol should raise InvalidName.")
-
+ with self.assertRaises(avro.errors.InvalidName, msg="When enum symbol
validation is enabled, an invalid symbol should raise InvalidName."):
+ avro.schema.parse(test_schema_string, validate_enum_symbols=True)
try:
- case = avro.schema.parse(test_schema_string,
validate_enum_symbols=False)
- except avro.errors.InvalidName:
- self.fail("When enum symbol validation is disabled, " "an invalid
symbol should not raise InvalidName.")
+ avro.schema.parse(test_schema_string, validate_enum_symbols=False)
+ except avro.errors.InvalidName: # pragma: no coverage
+ self.fail("When enum symbol validation is disabled, an invalid
symbol should not raise InvalidName.")
class SchemaParseTestCase(unittest.TestCase):
@@ -724,7 +718,7 @@ class SchemaParseTestCase(unittest.TestCase):
with warnings.catch_warnings(record=True) as actual_warnings:
try:
self.test_schema.parse()
- except (avro.errors.AvroException,
avro.errors.SchemaParseException):
+ except (avro.errors.AvroException,
avro.errors.SchemaParseException): # pragma: no coverage
self.fail(f"Valid schema failed to parse:
{self.test_schema!s}")
actual_messages = [str(wmsg.message) for wmsg in actual_warnings]
if self.test_schema.warnings:
@@ -735,12 +729,10 @@ class SchemaParseTestCase(unittest.TestCase):
def parse_invalid(self):
"""Parsing an invalid schema should error."""
- try:
+ with self.assertRaises(
+ (avro.errors.AvroException, avro.errors.SchemaParseException),
msg=f"Invalid schema should not have parsed: {self.test_schema!s}"
+ ):
self.test_schema.parse()
- except (avro.errors.AvroException, avro.errors.SchemaParseException):
- pass
- else:
- self.fail(f"Invalid schema should not have parsed:
{self.test_schema!s}")
class RoundTripParseTestCase(unittest.TestCase):
@@ -824,7 +816,7 @@ class OtherAttributesTestCase(unittest.TestCase):
sch = self.test_schema.parse()
try:
self.assertNotEqual(sch, object(), "A schema is never equal to a
non-schema instance.")
- except AttributeError:
+ except AttributeError: # pragma: no coverage
self.fail("Comparing a schema to a non-schema should be False, but
not error.")
round_trip = avro.schema.parse(str(sch))
self.assertEqual(
@@ -1256,5 +1248,5 @@ def load_tests(loader, default_tests, pattern):
return suite
-if __name__ == "__main__":
+if __name__ == "__main__": # pragma: no coverage
unittest.main()
diff --git a/lang/py/avro/test/test_script.py b/lang/py/avro/test/test_script.py
index 32437dc..07fef9b 100644
--- a/lang/py/avro/test/test_script.py
+++ b/lang/py/avro/test/test_script.py
@@ -189,7 +189,7 @@ class TestWrite(unittest.TestCase):
for filename in (self.csv_file, self.json_file, self.schema_file):
try:
os.unlink(filename)
- except OSError:
+ except OSError: # pragma: no coverage
continue
def _run(self, *args, **kw):
diff --git a/lang/py/avro/test/test_tether_task.py
b/lang/py/avro/test/test_tether_task.py
index ca50247..f00251e 100644
--- a/lang/py/avro/test/test_tether_task.py
+++ b/lang/py/avro/test/test_tether_task.py
@@ -111,5 +111,5 @@ class TestTetherTask(unittest.TestCase):
proc.kill()
-if __name__ == "__main__":
+if __name__ == "__main__": # pragma: no coverage
unittest.main()
diff --git a/lang/py/avro/test/test_tether_task_runner.py
b/lang/py/avro/test/test_tether_task_runner.py
index 25b7cd2..7696161 100644
--- a/lang/py/avro/test/test_tether_task_runner.py
+++ b/lang/py/avro/test/test_tether_task_runner.py
@@ -43,8 +43,6 @@ class TestTetherTaskRunner(unittest.TestCase):
proc = None
try:
# launch the server in a separate process
- env = dict()
- env["PYTHONPATH"] = ":".join(sys.path)
parent_port = avro.tether.util.find_port()
pyfile = avro.test.mock_tether_parent.__file__
@@ -59,13 +57,6 @@ class TestTetherTaskRunner(unittest.TestCase):
runner =
avro.tether.tether_task_runner.TaskRunner(avro.test.word_count_task.WordCountTask())
runner.start(outputport=parent_port, join=False)
- for _ in range(12):
- if runner.server is not None:
- break
- time.sleep(1)
- else:
- raise RuntimeError("Server never started")
-
# Test sending various messages to the server and ensuring they
are processed correctly
requestor = avro.tether.tether_task.HTTPRequestor(
"localhost",
@@ -140,8 +131,6 @@ class TestTetherTaskRunner(unittest.TestCase):
# shutdown the logging
logging.shutdown()
- except Exception as e:
- raise
finally:
# close the process
if not (proc is None):
@@ -158,8 +147,6 @@ class TestTetherTaskRunner(unittest.TestCase):
runnerproc = None
try:
# launch the server in a separate process
- env = dict()
- env["PYTHONPATH"] = ":".join(sys.path)
parent_port = avro.tether.util.find_port()
pyfile = avro.test.mock_tether_parent.__file__
@@ -171,16 +158,13 @@ class TestTetherTaskRunner(unittest.TestCase):
time.sleep(1)
# start the tether_task_runner in a separate process
- env = {"AVRO_TETHER_OUTPUT_PORT": f"{parent_port}"}
- env["PYTHONPATH"] = ":".join(sys.path)
-
runnerproc = subprocess.Popen(
[
sys.executable,
avro.tether.tether_task_runner.__file__,
"avro.test.word_count_task.WordCountTask",
],
- env=env,
+ env={"AVRO_TETHER_OUTPUT_PORT": f"{parent_port}",
"PYTHONPATH": ":".join(sys.path)},
)
# possible race condition wait for the process to start
@@ -191,8 +175,6 @@ class TestTetherTaskRunner(unittest.TestCase):
# so we give the subprocess time to start up
time.sleep(1)
- except Exception as e:
- raise
finally:
# close the process
if not (runnerproc is None):
@@ -202,5 +184,5 @@ class TestTetherTaskRunner(unittest.TestCase):
proc.kill()
-if __name__ == ("__main__"):
+if __name__ == ("__main__"): # pragma: no coverage
unittest.main()
diff --git a/lang/py/avro/test/test_tether_word_count.py
b/lang/py/avro/test/test_tether_word_count.py
index abcac02..5c876ee 100644
--- a/lang/py/avro/test/test_tether_word_count.py
+++ b/lang/py/avro/test/test_tether_word_count.py
@@ -70,7 +70,7 @@ _OUT_SCHEMA = """{
_PYTHON_PATH =
os.pathsep.join([os.path.dirname(os.path.dirname(avro.__file__)),
os.path.dirname(__file__)])
-def _has_java():
+def _has_java(): # pragma: no coverage
"""Detect if this system has a usable java installed.
On most systems, this is just checking if `java` is in the PATH.
@@ -177,5 +177,5 @@ class TestTetherWordCount(unittest.TestCase):
self.assertDictEqual(actual_counts, expected_counts)
-if __name__ == "__main__":
+if __name__ == "__main__": # pragma: no coverage
unittest.main()
diff --git a/lang/py/avro/tether/tether_task_runner.py
b/lang/py/avro/tether/tether_task_runner.py
index ccd5a02..fcd2dfb 100644
--- a/lang/py/avro/tether/tether_task_runner.py
+++ b/lang/py/avro/tether/tether_task_runner.py
@@ -134,8 +134,9 @@ class TaskRunner:
implements the logic for the mapper and reducer phases
"""
- server = None
+ _server = None
sthread = None
+ timeout = 12 # number of seconds to wait for server to come up.
def __init__(self, task):
"""
@@ -150,6 +151,14 @@ class TaskRunner:
raise avro.errors.AvroException("task must be an instance of
tether task")
self.task = task
+ @property
+ def server(self):
+ for t in range(self.timeout):
+ if self._server:
+ return self._server
+ time.sleep(1)
+ raise RuntimeError("Server never started")
+
def start(self, outputport=None, join=True):
"""
Start the server
@@ -170,9 +179,9 @@ class TaskRunner:
address = ("localhost", port)
def thread_run(task_runner=None):
- task_runner.server = http.server.HTTPServer(address,
HTTPHandlerGen(task_runner))
- task_runner.server.allow_reuse_address = True
- task_runner.server.serve_forever()
+ task_runner._server = http.server.HTTPServer(address,
HTTPHandlerGen(task_runner))
+ task_runner._server.allow_reuse_address = True
+ task_runner._server.serve_forever()
# create a separate thread for the http server
sthread = threading.Thread(target=thread_run, kwargs={"task_runner":
self})
@@ -185,7 +194,7 @@ class TaskRunner:
# wait for the other thread to finish
if join:
self.task.ready_for_shutdown.wait()
- self.server.shutdown()
+ self._server.shutdown()
# should we do some kind of check to make sure it exits
self.log.info("Shutdown the logger")
@@ -196,7 +205,6 @@ class TaskRunner:
"""
Handler for the close message
"""
-
self.task.close()
