Skip to content
Snippets Groups Projects
Commit 25bfe1c6 authored by Eva Bardou's avatar Eva Bardou Committed by Bastien Abadie
Browse files

Add Unit Tests to Base Worker repo to avoid regressions

parent 35147e42
No related branches found
No related tags found
1 merge request!5Add Unit Tests to Base Worker repo to avoid regressions
Pipeline #77874 passed
*.egg-info
*.pyc
.tox/
......@@ -21,6 +21,27 @@ lint:
script:
- pre-commit run -a
test:
image: python:3
stage: test
cache:
paths:
- .cache/pip
variables:
PIP_CACHE_DIR: "$CI_PROJECT_DIR/.cache/pip"
ARKINDEX_API_SCHEMA_URL: schema.yml
before_script:
- pip install tox
# Download OpenAPI schema from last backend build
- curl https://assets.teklia.com/arkindex/openapi.yml > schema.yml
script:
- tox
test-cookiecutter:
image: python:3
......
......@@ -8,4 +8,4 @@ line_length = 88
default_section=FIRSTPARTY
known_first_party = arkindex,arkindex_common
known_third_party =PIL,apistar,requests,setuptools,tenacity
known_third_party =PIL,apistar,pytest,requests,setuptools,tenacity
......@@ -88,6 +88,9 @@ class ElementsWorker(BaseWorker):
)
def list_elements(self):
assert not (
self.args.elements_list and self.args.element
), "elements-list and element CLI args shouldn't be both set"
out = []
# Process elements from JSON file
......@@ -96,9 +99,8 @@ class ElementsWorker(BaseWorker):
assert isinstance(data, list), "Elements list must be a list"
assert len(data), "No elements in elements list"
out += list(filter(None, [element.get("id") for element in data]))
# Add any extra element from CLI
if self.args.element:
elif self.args.element:
out += self.args.element
return out
......
pytest==5.4.3
pytest-mock==3.2.0
pytest-responses==0.4.0
# -*- coding: utf-8 -*-
import os
import pytest
@pytest.fixture(autouse=True)
def pass_schema(responses):
schema_url = os.environ.get("ARKINDEX_API_SCHEMA_URL")
responses.add_passthru(schema_url)
tests/data/test_image.jpg

14.6 KiB

tests/data/tiled_image.jpg

184 KiB

# -*- coding: utf-8 -*-
import logging
import os
import sys
from pathlib import Path
from arkindex_worker import logger
from arkindex_worker.worker import BaseWorker
def test_init_default_local_share():
worker = BaseWorker()
assert worker.work_dir == os.path.expanduser("~/.local/share/arkindex")
def test_init_default_xdg_data_home(monkeypatch):
path = str(Path(__file__).absolute().parent)
monkeypatch.setenv("XDG_DATA_HOME", path)
worker = BaseWorker()
assert worker.work_dir == f"{path}/arkindex"
def test_init_var_ponos_data_given(monkeypatch):
path = str(Path(__file__).absolute().parent)
monkeypatch.setenv("PONOS_DATA", path)
worker = BaseWorker()
assert worker.work_dir == f"{path}/current"
def test_cli_default(mocker):
worker = BaseWorker()
spy = mocker.spy(worker, "add_arguments")
assert not spy.called
assert logger.level == logging.NOTSET
assert not hasattr(worker, "api_client")
worker.configure()
assert spy.called
assert spy.call_count == 1
assert not worker.args.verbose
assert logger.level == logging.NOTSET
assert worker.api_client
logger.setLevel(logging.NOTSET)
def test_cli_arg_verbose_given(mocker):
worker = BaseWorker()
spy = mocker.spy(worker, "add_arguments")
assert not spy.called
assert logger.level == logging.NOTSET
assert not hasattr(worker, "api_client")
mocker.patch.object(sys, "argv", ["worker", "-v"])
worker.configure()
assert spy.called
assert spy.call_count == 1
assert worker.args.verbose
assert logger.level == logging.DEBUG
assert worker.api_client
logger.setLevel(logging.NOTSET)
# -*- coding: utf-8 -*-
import pytest
from requests import HTTPError
from arkindex_worker.models import Element
def test_no_image_url():
url = Element({"zone": None}).image_url()
assert not url
def test_image_url_iiif():
url = Element({"zone": {"image": {"url": "http://something/"}}}).image_url()
assert url == "http://something/full/full/0/default.jpg"
def test_image_url_iiif_resize():
url = Element({"zone": {"image": {"url": "http://something/"}}}).image_url(500)
assert url == "http://something/full/500/0/default.jpg"
def test_image_url_iiif_append_slash():
url = Element({"zone": {"image": {"url": "http://something"}}}).image_url()
assert url == "http://something/full/full/0/default.jpg"
def test_image_url_s3():
url = Element(
{
"zone": {
"image": {"s3_url": "http://s3/something", "url": "http://something/"}
}
}
).image_url()
assert url == "http://s3/something"
def test_image_url_s3_resize():
url = Element(
{
"zone": {
"image": {"s3_url": "http://s3/something", "url": "http://something/"}
}
}
).image_url(500)
assert url == "http://s3/something"
def test_open_image(mocker):
open_mock = mocker.patch(
"arkindex_worker.models.open_image", return_value="an image!"
)
elt = Element(
{
"zone": {
"image": {
"url": "http://something",
"server": {"max_width": None, "max_height": None},
},
"polygon": [[0, 0], [181, 0], [181, 240], [0, 240], [0, 0]],
}
}
)
assert elt.open_image() == "an image!"
assert open_mock.call_count == 1
assert open_mock.call_args == mocker.call(
"http://something/full/full/0/default.jpg"
)
def test_open_image_resize_portrait(mocker):
open_mock = mocker.patch(
"arkindex_worker.models.open_image", return_value="an image!"
)
elt = Element(
{
"zone": {
"image": {
"url": "http://something",
"width": 400,
"height": 600,
"server": {"max_width": None, "max_height": None},
},
"polygon": [[0, 0], [400, 0], [400, 600], [0, 600], [0, 0]],
}
}
)
# Resize = original size
assert elt.open_image(max_size=600) == "an image!"
assert open_mock.call_count == 1
assert open_mock.call_args == mocker.call(
"http://something/full/full/0/default.jpg"
)
# Resize = smaller height
assert elt.open_image(max_size=400) == "an image!"
assert open_mock.call_count == 2
assert open_mock.call_args == mocker.call(
"http://something/full/266,400/0/default.jpg"
)
# Resize = bigger height
assert elt.open_image(max_size=800) == "an image!"
assert open_mock.call_count == 3
assert open_mock.call_args == mocker.call(
"http://something/full/full/0/default.jpg"
)
def test_open_image_resize_partial_element(mocker):
open_mock = mocker.patch(
"arkindex_worker.models.open_image", return_value="an image!"
)
elt = Element(
{
"zone": {
"image": {
"url": "http://something",
"width": 400,
"height": 600,
"server": {"max_width": None, "max_height": None},
},
"polygon": [[0, 0], [200, 0], [200, 600], [0, 600], [0, 0]],
}
}
)
assert elt.open_image(max_size=400) == "an image!"
assert open_mock.call_count == 1
assert open_mock.call_args == mocker.call(
"http://something/full/full/0/default.jpg"
)
def test_open_image_resize_landscape(mocker):
open_mock = mocker.patch(
"arkindex_worker.models.open_image", return_value="an image!"
)
elt = Element(
{
"zone": {
"image": {
"url": "http://something",
"width": 600,
"height": 400,
"server": {"max_width": None, "max_height": None},
},
"polygon": [[0, 0], [600, 0], [600, 400], [0, 400], [0, 0]],
}
}
)
# Resize = original size
assert elt.open_image(max_size=600) == "an image!"
assert open_mock.call_count == 1
assert open_mock.call_args == mocker.call(
"http://something/full/full/0/default.jpg"
)
# Resize = smaller width
assert elt.open_image(max_size=400) == "an image!"
assert open_mock.call_count == 2
assert open_mock.call_args == mocker.call(
"http://something/full/400,266/0/default.jpg"
)
# Resize = bigger width
assert elt.open_image(max_size=800) == "an image!"
assert open_mock.call_count == 3
assert open_mock.call_args == mocker.call(
"http://something/full/full/0/default.jpg"
)
def test_open_image_resize_square(mocker):
open_mock = mocker.patch(
"arkindex_worker.models.open_image", return_value="an image!"
)
elt = Element(
{
"zone": {
"image": {
"url": "http://something",
"width": 400,
"height": 400,
"server": {"max_width": None, "max_height": None},
},
"polygon": [[0, 0], [400, 0], [400, 400], [0, 400], [0, 0]],
}
}
)
# Resize = original size
assert elt.open_image(max_size=400) == "an image!"
assert open_mock.call_count == 1
assert open_mock.call_args == mocker.call(
"http://something/full/full/0/default.jpg"
)
# Resize = smaller
assert elt.open_image(max_size=200) == "an image!"
assert open_mock.call_count == 2
assert open_mock.call_args == mocker.call(
"http://something/full/200,200/0/default.jpg"
)
# Resize = bigger
assert elt.open_image(max_size=800) == "an image!"
assert open_mock.call_count == 3
assert open_mock.call_args == mocker.call(
"http://something/full/full/0/default.jpg"
)
def test_open_image_resize_tiles(mocker):
mocker.patch("arkindex_worker.models.open_image", return_value="an image!")
elt = Element(
{
"zone": {
"image": {
"url": "http://something",
"server": {"max_width": 600, "max_height": 600},
},
"polygon": [[0, 0], [800, 0], [800, 800], [0, 800], [0, 0]],
}
}
)
with pytest.raises(NotImplementedError):
elt.open_image(max_size=400)
def test_open_image_requires_zone():
with pytest.raises(ValueError) as e:
Element({"id": "42"}).open_image()
assert str(e.value) == "Element 42 has no zone"
def test_open_image_s3(mocker):
open_mock = mocker.patch(
"arkindex_worker.models.open_image", return_value="an image!"
)
elt = Element(
{"zone": {"image": {"url": "http://something", "s3_url": "http://s3url"}}}
)
assert elt.open_image() == "an image!"
assert open_mock.call_count == 1
assert open_mock.call_args == mocker.call("http://s3url")
def test_open_image_s3_retry(mocker):
response_mock = mocker.MagicMock()
response_mock.status_code = 403
mocker.patch(
"arkindex_worker.models.open_image",
return_value="an image!",
side_effect=HTTPError(response=response_mock),
)
elt = Element(
{
"id": "cafe",
"zone": {"image": {"url": "http://something", "s3_url": "http://oldurl"}},
}
)
with pytest.raises(NotImplementedError):
elt.open_image()
def test_open_image_s3_retry_once(mocker):
response_mock = mocker.MagicMock()
response_mock.status_code = 403
mocker.patch(
"arkindex_worker.models.open_image",
side_effect=HTTPError(response=response_mock),
)
elt = Element(
{
"id": "cafe",
"zone": {"image": {"url": "http://something", "s3_url": "http://oldurl"}},
}
)
with pytest.raises(NotImplementedError):
elt.open_image()
# -*- coding: utf-8 -*-
import json
import os
import sys
import tempfile
from argparse import Namespace
from uuid import UUID
import pytest
from apistar.exceptions import ErrorResponse
from arkindex_worker.models import Element
from arkindex_worker.worker import ElementsWorker
def test_cli_default(monkeypatch):
_, path = tempfile.mkstemp()
with open(path, "w") as f:
json.dump(
[
{"id": "volumeid", "type": "volume"},
{"id": "pageid", "type": "page"},
{"id": "actid", "type": "act"},
{"id": "surfaceid", "type": "surface"},
],
f,
)
monkeypatch.setenv("TASK_ELEMENTS", path)
worker = ElementsWorker()
worker.configure()
assert worker.args.elements_list.name == path
assert not worker.args.element
os.unlink(path)
def test_cli_arg_elements_list_given(mocker):
_, path = tempfile.mkstemp()
with open(path, "w") as f:
json.dump(
[
{"id": "volumeid", "type": "volume"},
{"id": "pageid", "type": "page"},
{"id": "actid", "type": "act"},
{"id": "surfaceid", "type": "surface"},
],
f,
)
mocker.patch.object(sys, "argv", ["worker", "--elements-list", path])
worker = ElementsWorker()
worker.configure()
assert worker.args.elements_list.name == path
assert not worker.args.element
os.unlink(path)
def test_cli_arg_element_one_given_not_uuid(mocker):
mocker.patch.object(sys, "argv", ["worker", "--element", "1234"])
worker = ElementsWorker()
with pytest.raises(SystemExit):
worker.configure()
def test_cli_arg_element_one_given(mocker):
mocker.patch.object(
sys, "argv", ["worker", "--element", "12341234-1234-1234-1234-123412341234"]
)
worker = ElementsWorker()
worker.configure()
assert worker.args.element == [UUID("12341234-1234-1234-1234-123412341234")]
# elements_list is None because TASK_ELEMENTS environment variable isn't set
assert not worker.args.elements_list
def test_cli_arg_element_many_given(mocker):
mocker.patch.object(
sys,
"argv",
[
"worker",
"--element",
"12341234-1234-1234-1234-123412341234",
"43214321-4321-4321-4321-432143214321",
],
)
worker = ElementsWorker()
worker.configure()
assert worker.args.element == [
UUID("12341234-1234-1234-1234-123412341234"),
UUID("43214321-4321-4321-4321-432143214321"),
]
# elements_list is None because TASK_ELEMENTS environment variable isn't set
assert not worker.args.elements_list
def test_list_elements_elements_list_arg_wrong_type(monkeypatch):
_, path = tempfile.mkstemp()
with open(path, "w") as f:
json.dump({}, f)
monkeypatch.setenv("TASK_ELEMENTS", path)
worker = ElementsWorker()
worker.configure()
os.unlink(path)
with pytest.raises(AssertionError) as e:
worker.list_elements()
assert str(e.value) == "Elements list must be a list"
def test_list_elements_elements_list_arg_empty_list(monkeypatch):
_, path = tempfile.mkstemp()
with open(path, "w") as f:
json.dump([], f)
monkeypatch.setenv("TASK_ELEMENTS", path)
worker = ElementsWorker()
worker.configure()
os.unlink(path)
with pytest.raises(AssertionError) as e:
worker.list_elements()
assert str(e.value) == "No elements in elements list"
def test_list_elements_elements_list_arg_missing_id(monkeypatch):
_, path = tempfile.mkstemp()
with open(path, "w") as f:
json.dump([{"type": "volume"}], f)
monkeypatch.setenv("TASK_ELEMENTS", path)
worker = ElementsWorker()
worker.configure()
os.unlink(path)
elt_list = worker.list_elements()
assert elt_list == []
def test_list_elements_elements_list_arg(monkeypatch):
_, path = tempfile.mkstemp()
with open(path, "w") as f:
json.dump(
[
{"id": "volumeid", "type": "volume"},
{"id": "pageid", "type": "page"},
{"id": "actid", "type": "act"},
{"id": "surfaceid", "type": "surface"},
],
f,
)
monkeypatch.setenv("TASK_ELEMENTS", path)
worker = ElementsWorker()
worker.configure()
os.unlink(path)
elt_list = worker.list_elements()
assert elt_list == ["volumeid", "pageid", "actid", "surfaceid"]
def test_list_elements_element_arg(mocker):
mocker.patch(
"arkindex_worker.worker.argparse.ArgumentParser.parse_args",
return_value=Namespace(
element=["volumeid", "pageid"], verbose=False, elements_list=None
),
)
worker = ElementsWorker()
worker.configure()
elt_list = worker.list_elements()
assert elt_list == ["volumeid", "pageid"]
def test_list_elements_both_args_error(mocker):
_, path = tempfile.mkstemp()
with open(path, "w") as f:
json.dump(
[
{"id": "volumeid", "type": "volume"},
{"id": "pageid", "type": "page"},
{"id": "actid", "type": "act"},
{"id": "surfaceid", "type": "surface"},
],
f,
)
mocker.patch(
"arkindex_worker.worker.argparse.ArgumentParser.parse_args",
return_value=Namespace(
element=["anotherid", "againanotherid"],
verbose=False,
elements_list=open(path),
),
)
worker = ElementsWorker()
worker.configure()
os.unlink(path)
with pytest.raises(AssertionError) as e:
worker.list_elements()
assert str(e.value) == "elements-list and element CLI args shouldn't be both set"
def test_create_sub_element_wrong_element():
worker = ElementsWorker()
with pytest.raises(AssertionError) as e:
worker.create_sub_element(
element=None,
type="something",
name="0",
polygon=[[1, 1], [2, 2], [2, 1], [1, 2]],
)
assert str(e.value) == "element shouldn't be null and should be of type Element"
with pytest.raises(AssertionError) as e:
worker.create_sub_element(
element="not element type",
type="something",
name="0",
polygon=[[1, 1], [2, 2], [2, 1], [1, 2]],
)
assert str(e.value) == "element shouldn't be null and should be of type Element"
def test_create_sub_element_wrong_type():
worker = ElementsWorker()
elt = Element({"zone": None})
with pytest.raises(AssertionError) as e:
worker.create_sub_element(
element=elt, type=None, name="0", polygon=[[1, 1], [2, 2], [2, 1], [1, 2]],
)
assert str(e.value) == "type shouldn't be null and should be of type str"
with pytest.raises(AssertionError) as e:
worker.create_sub_element(
element=elt, type=1234, name="0", polygon=[[1, 1], [2, 2], [2, 1], [1, 2]],
)
assert str(e.value) == "type shouldn't be null and should be of type str"
def test_create_sub_element_wrong_name():
worker = ElementsWorker()
elt = Element({"zone": None})
with pytest.raises(AssertionError) as e:
worker.create_sub_element(
element=elt,
type="something",
name=None,
polygon=[[1, 1], [2, 2], [2, 1], [1, 2]],
)
assert str(e.value) == "name shouldn't be null and should be of type str"
with pytest.raises(AssertionError) as e:
worker.create_sub_element(
element=elt,
type="something",
name=1234,
polygon=[[1, 1], [2, 2], [2, 1], [1, 2]],
)
assert str(e.value) == "name shouldn't be null and should be of type str"
def test_create_sub_element_wrong_polygon():
worker = ElementsWorker()
elt = Element({"zone": None})
with pytest.raises(AssertionError) as e:
worker.create_sub_element(
element=elt, type="something", name="0", polygon=None,
)
assert str(e.value) == "polygon shouldn't be null and should be of type list"
with pytest.raises(AssertionError) as e:
worker.create_sub_element(
element=elt, type="something", name="O", polygon="not a polygon",
)
assert str(e.value) == "polygon shouldn't be null and should be of type list"
with pytest.raises(AssertionError) as e:
worker.create_sub_element(
element=elt, type="something", name="O", polygon=[[1, 1], [2, 2]],
)
assert str(e.value) == "polygon should have at least three points"
with pytest.raises(AssertionError) as e:
worker.create_sub_element(
element=elt,
type="something",
name="O",
polygon=[[1, 1, 1], [2, 2, 1], [2, 1, 1], [1, 2, 1]],
)
assert str(e.value) == "polygon points should be lists of two items"
with pytest.raises(AssertionError) as e:
worker.create_sub_element(
element=elt, type="something", name="O", polygon=[[1], [2], [2], [1]],
)
assert str(e.value) == "polygon points should be lists of two items"
with pytest.raises(AssertionError) as e:
worker.create_sub_element(
element=elt,
type="something",
name="O",
polygon=[["not a coord", 1], [2, 2], [2, 1], [1, 2]],
)
assert str(e.value) == "polygon points should be lists of two numbers"
def test_create_sub_element_api_error(responses):
worker = ElementsWorker()
worker.configure()
elt = Element(
{
"id": "12341234-1234-1234-1234-123412341234",
"corpus": {"id": "11111111-1111-1111-1111-111111111111"},
"zone": {"image": {"id": "22222222-2222-2222-2222-222222222222"}},
}
)
responses.add(
responses.POST,
"https://arkindex.teklia.com/api/v1/elements/create/",
status=500,
)
with pytest.raises(ErrorResponse):
worker.create_sub_element(
element=elt,
type="something",
name="0",
polygon=[[1, 1], [2, 2], [2, 1], [1, 2]],
)
assert len(responses.calls) == 1
assert (
responses.calls[0].request.url
== "https://arkindex.teklia.com/api/v1/elements/create/"
)
def test_create_sub_element(responses):
worker = ElementsWorker()
worker.configure()
elt = Element(
{
"id": "12341234-1234-1234-1234-123412341234",
"corpus": {"id": "11111111-1111-1111-1111-111111111111"},
"zone": {"image": {"id": "22222222-2222-2222-2222-222222222222"}},
}
)
responses.add(
responses.POST,
"https://arkindex.teklia.com/api/v1/elements/create/",
status=200,
)
worker.create_sub_element(
element=elt,
type="something",
name="0",
polygon=[[1, 1], [2, 2], [2, 1], [1, 2]],
)
assert len(responses.calls) == 1
assert (
responses.calls[0].request.url
== "https://arkindex.teklia.com/api/v1/elements/create/"
)
# -*- coding: utf-8 -*-
import math
from io import BytesIO
from pathlib import Path
import pytest
from PIL import Image, ImageChops, ImageOps
from arkindex_worker.image import download_tiles
FIXTURES = Path(__file__).absolute().parent / "data"
TILE = FIXTURES / "test_image.jpg"
FULL_IMAGE = FIXTURES / "tiled_image.jpg"
def _root_mean_square(img_a, img_b):
"""
Get the root-mean-square difference between two images for fuzzy matching
See https://effbot.org/zone/pil-comparing-images.htm
"""
h = ImageChops.difference(img_a, img_b).histogram()
return math.sqrt(
sum((value * ((idx % 256) ** 2) for idx, value in enumerate(h)))
/ float(img_a.size[0] * img_a.size[1])
)
def test_download_tiles(responses):
expected = Image.open(FULL_IMAGE).convert("RGB")
with TILE.open("rb") as tile:
tile_bytes = tile.read()
responses.add(
responses.GET,
"http://nowhere/info.json",
json={"width": 543, "height": 720, "tiles": [{"width": 181, "height": 240}]},
)
for x in (0, 181, 362):
for y in (0, 240, 480):
responses.add(
responses.GET,
f"http://nowhere/{x},{y},181,240/full/0/default.jpg",
body=tile_bytes,
)
actual = download_tiles("http://nowhere")
assert _root_mean_square(expected, actual) <= 5.0
def test_download_tiles_crop(responses):
"""
Ensure download_tiles does not care about tiles that are slightly bigger than expected
(occasional issue with the Harvard IDS image server where 1024×1024 tiles sometimes are returned as 1024x1025)
"""
expected = Image.open(FULL_IMAGE).convert("RGB")
tile_bytes = BytesIO()
with TILE.open("rb") as tile:
# Add one extra pixel to each tile to return slightly bigger tiles
ImageOps.pad(Image.open(tile), (181, 241)).save(tile_bytes, format="jpeg")
tile_bytes = tile_bytes.getvalue()
responses.add(
responses.GET,
"http://nowhere/info.json",
json={"width": 543, "height": 720, "tiles": [{"width": 181, "height": 240}]},
)
for x in (0, 181, 362):
for y in (0, 240, 480):
responses.add(
responses.GET,
f"http://nowhere/{x},{y},181,240/full/0/default.jpg",
body=tile_bytes,
)
actual = download_tiles("http://nowhere")
assert _root_mean_square(expected, actual) <= 5.0
def test_download_tiles_small(responses):
small_tile = BytesIO()
Image.new("RGB", (1, 1)).save(small_tile, format="jpeg")
small_tile.seek(0)
responses.add(
responses.GET,
"http://nowhere/info.json",
json={"width": 543, "height": 720, "tiles": [{"width": 181, "height": 240}]},
)
responses.add(
responses.GET,
"http://nowhere/0,0,181,240/full/0/default.jpg",
body=small_tile.read(),
)
with pytest.raises(ValueError) as e:
download_tiles("http://nowhere")
assert str(e.value) == "Expected size 181×240 for tile 0,0, but got 1×1"
# -*- coding: utf-8 -*-
import pytest
from apistar.exceptions import ErrorResponse
from arkindex_worker.reporting import Reporter
def test_init():
reporter = Reporter("worker")
assert "started" in reporter.report_data
del reporter.report_data["started"]
assert reporter.report_data == {"slug": "worker", "version": "0.0", "elements": {}}
def test_process():
reporter = Reporter("worker")
reporter.process("myelement")
assert "myelement" in reporter.report_data["elements"]
element_data = reporter.report_data["elements"]["myelement"]
assert "started" in element_data
del element_data["started"]
assert element_data == {
"elements": {},
"transcriptions": {},
"classifications": {},
"errors": [],
}
def test_add_element():
reporter = Reporter("worker")
reporter.add_element("myelement", type="text_line")
assert "myelement" in reporter.report_data["elements"]
element_data = reporter.report_data["elements"]["myelement"]
del element_data["started"]
assert element_data == {
"elements": {"text_line": 1},
"transcriptions": {},
"classifications": {},
"errors": [],
}
def test_add_classification():
reporter = Reporter("worker")
reporter.add_classification("myelement", "three")
assert "myelement" in reporter.report_data["elements"]
element_data = reporter.report_data["elements"]["myelement"]
del element_data["started"]
assert element_data == {
"elements": {},
"transcriptions": {},
"classifications": {"three": 1},
"errors": [],
}
def test_add_classifications():
reporter = Reporter("worker")
with pytest.raises(AssertionError):
reporter.add_classifications("myelement", {"not": "a list"})
reporter.add_classifications(
"myelement", [{"class_name": "three"}, {"class_name": "two"}]
)
reporter.add_classifications(
"myelement",
[
{"class_name": "three"},
{"class_name": "two", "high_confidence": True},
{"class_name": "three", "confidence": 0.42},
],
)
assert "myelement" in reporter.report_data["elements"]
element_data = reporter.report_data["elements"]["myelement"]
del element_data["started"]
assert element_data == {
"elements": {},
"transcriptions": {},
"classifications": {"three": 3, "two": 2},
"errors": [],
}
def test_add_transcription():
reporter = Reporter("worker")
reporter.add_transcription("myelement", "word")
assert "myelement" in reporter.report_data["elements"]
element_data = reporter.report_data["elements"]["myelement"]
del element_data["started"]
assert element_data == {
"elements": {},
"transcriptions": {"word": 1},
"classifications": {},
"errors": [],
}
def test_add_transcriptions():
reporter = Reporter("worker")
with pytest.raises(AssertionError):
reporter.add_transcriptions("myelement", {"not": "a list"})
reporter.add_transcriptions("myelement", [{"type": "word"}, {"type": "line"}])
reporter.add_transcriptions(
"myelement",
[
{"type": "word"},
{"type": "line", "text": "something"},
{"type": "word", "confidence": 0.42},
],
)
assert "myelement" in reporter.report_data["elements"]
element_data = reporter.report_data["elements"]["myelement"]
del element_data["started"]
assert element_data == {
"elements": {},
"transcriptions": {"word": 3, "line": 2},
"classifications": {},
"errors": [],
}
def test_error():
reporter = Reporter("worker")
reporter.error("myelement", ZeroDivisionError("What have you done"))
reporter.error(
"myelement",
ErrorResponse(
title="I'm a teapot",
status_code=418,
content='{"coffee": "Can\'t touch this"}',
),
)
assert reporter.report_data["elements"]["myelement"]["errors"] == [
{"class": "ZeroDivisionError", "message": "What have you done"},
{
"class": "ErrorResponse",
"message": "I'm a teapot",
"status_code": 418,
"content": '{"coffee": "Can\'t touch this"}',
},
]
tox.ini 0 → 100644
[tox]
envlist = arkindex_worker
skipsdist=True
[testenv]
passenv = ARKINDEX_API_SCHEMA_URL
commands =
pytest
deps =
-rrequirements.txt
-rtests-requirements.txt
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment