add tests for snappy
MODIFY
.travis.yml
.travis.yml
language: python
sudo: required
dist: xenial
+before_install:
+ - sudo apt-get install -y libsnappy-dev
+ - pip install python-snappy
install:
- pip install -r developer_requirements.txt
script:

MODIFY
azure-pipelines.yml
azure-pipelines.yml
@@ -34,6 +34,10 @@ jobs:
versionSpec: '$(python.version)'
architecture: 'x64'
+ - script: |
+ brew install snappy
+ CPPFLAGS="-I/usr/local/include -L/usr/local/lib" python -m pip install python-snappy
+
- script: python -m pip install --upgrade -r developer_requirements.txt
displayName: 'Install dependencies'

ADD
tests/test_compression.py
+import os
+
+import pytest
+
+from fastavro.six import MemoryIO
+import fastavro
+
+
+@pytest.mark.skipif(os.name == "nt", reason="A pain to set up on windows")
+def test_snappy():
+ schema = {
+ "doc": "A weather reading.",
+ "name": "Weather",
+ "namespace": "test",
+ "type": "record",
+ "fields": [
+ {"name": "station", "type": "string"},
+ {"name": "time", "type": "long"},
+ {"name": "temp", "type": "int"},
+ ],
+ }
+
+ records = [
+ {"station": "011990-99999", "temp": 0, "time": 1433269388},
+ {"station": "011990-99999", "temp": 22, "time": 1433270389},
+ {"station": "011990-99999", "temp": -11, "time": 1433273379},
+ {"station": "012650-99999", "temp": 111, "time": 1433275478},
+ ]
+
+ file = MemoryIO()
+ fastavro.writer(file, schema, records, codec="snappy")
+
+ file.seek(0)
+ out_records = list(fastavro.reader(file))
+ assert records == out_records

MODIFY
tests/test_main_cli.py
tests/test_main_cli.py
@@ -114,14 +114,15 @@ def test_cli_arg_schema():
def test_cli_arg_codecs():
# given,
given_cmd_args = [sys.executable, main_py, '--codecs']
- expected_output = sorted(['deflate', 'null'])
+ default_codecs = ("deflate", "null")
# exercise,
result_output = subprocess.check_output(given_cmd_args).decode()
+ result_codecs = [
+ base_text.strip(line)
+ for line in result_output.splitlines()
+ if base_text.strip(line)
+ ]
- # verify
- data = sorted(
- # translate to a sorted list of codecs,
- # strip newlines, remove empty lines
- filter(None, map(base_text.strip, result_output.splitlines())))
- assert data == expected_output
+ for codec in default_codecs:
+ assert codec in result_codecs

add tests for snappy