tests.py

# coding: utf8
from __future__ import unicode_literals

import pytest
import tempfile
import shutil
from pathlib import Path
from contextlib import contextmanager
from prodigy.components.db import connect
from prodigy.util import write_jsonl, INPUT_HASH_ATTR, TASK_HASH_ATTR
from prodigy.models.ner import merge_spans

from spacy.language import Language
from spacy.lang.en import English

from ner.ner_teach import ner_teach
from ner.ner_match import ner_match
from ner.ner_manual import ner_manual
from ner.ner_correct import ner_correct
from ner.ner_silver_to_gold import ner_silver_to_gold
from ner.ner_eval_ab import ner_eval_ab
from textcat.textcat_teach import textcat_teach
from textcat.textcat_custom_model import textcat_custom_model
from textcat.textcat_manual import textcat_manual
from textcat.textcat_correct import textcat_correct
from terms.terms_teach import terms_teach
from image.image_manual import image_manual
from other.mark import mark
from other.choice import choice



@pytest.fixture()
def dataset():
    return False


@pytest.fixture
def spacy_model():
    return 'en_core_web_sm'


@pytest.fixture
@Language.component("dummy_textcat")
def dummy_textcat_pipe(doc):
    if doc == 'This is a text about David Bowie':
        doc.cats = {"PERSON": 1.0, "ORG": 0.0}
    elif doc =='Apple makes iPhones':
        doc.cats = {"PERSON": 0.0, "ORG": 1.0}
    else:
        doc.cats = {"PERSON": 0.0, "ORG": 0.0}
    return doc


@pytest.fixture(scope="session")
def nlp():
    return English()


@pytest.fixture
def vectors():
    return 'en_core_web_md'


@pytest.fixture
def labels():
    return ['PERSON', 'ORG']

@pytest.fixture()
def source():
    texts = ['This is a text about David Bowie', 'Apple makes iPhones']
    examples = [{'text': text} for text in texts]
    _, tmp_file = tempfile.mkstemp()
    write_jsonl(tmp_file, examples)
    return tmp_file


@pytest.fixture()
def patterns():
    examples = [{'label': 'PERSON', 'pattern': 'David Bowie'},
                {'label': 'ORG', 'pattern': [{'lower': 'apple'}]}]
    _, tmp_file = tempfile.mkstemp()
    write_jsonl(tmp_file, examples)
    return tmp_file


@contextmanager
def tmp_dataset(name, examples=[]):
    DB = connect()
    DB.add_dataset(name)
    DB.add_examples(examples, datasets=[name])
    yield examples
    DB.drop_dataset(name)


@contextmanager
def make_tmpdir():
    d = Path(tempfile.mkdtemp())
    try:
        yield d
    finally:
        shutil.rmtree(d)


def test_ner_teach(dataset, spacy_model, source, labels, patterns):
    recipe = ner_teach(dataset, spacy_model, source, labels, patterns)
    stream = list(recipe['stream'])
    assert recipe['view_id'] == 'ner'
    assert recipe['dataset'] == dataset
    assert len(stream) == 5
    assert 'spans' in stream[0]
    assert 'tokens' in stream[0]
    assert 'meta' in stream[0]
    assert 'score' in stream[0]['meta']


def test_ner_match(dataset, spacy_model, source, patterns):
    recipe = ner_match(dataset, spacy_model, source, patterns)
    stream = list(recipe['stream'])
    assert recipe['view_id'] == 'ner'
    assert recipe['dataset'] == dataset
    assert len(stream) == 2
    assert 'spans' in stream[0]
    assert len(stream[0]['spans']) == 1
    assert stream[0]['spans'][0]['label'] == 'PERSON'
    assert 'spans' in stream[1]
    assert len(stream[1]['spans']) == 1
    assert stream[1]['spans'][0]['label'] == 'ORG'


def test_ner_manual(dataset, spacy_model, source, labels):
    recipe = ner_manual(dataset, spacy_model, source, labels)
    stream = list(recipe['stream'])
    assert recipe['view_id'] == 'ner_manual'
    assert recipe['dataset'] == dataset
    assert len(stream) == 2
    assert 'tokens' in stream[0]
    assert 'tokens' in stream[1]


def test_ner_correct(dataset, spacy_model, source, labels):
    recipe = ner_correct(dataset, spacy_model, source, labels)
    stream = list(recipe['stream'])
    assert recipe['view_id'] == 'ner_manual'
    assert recipe['dataset'] == dataset
    assert len(stream) == 2
    assert 'spans' in stream[0]
    assert 'tokens' in stream[0]


def test_ner_silver_to_gold(dataset, spacy_model):
    silver_dataset = '__test_ner_silver_to_gold__'
    silver_examples = [
        {
            INPUT_HASH_ATTR: 1,
            TASK_HASH_ATTR: 11,
            'text': 'Hello world',
            'answer': 'accept',
            'spans': [{'start': 0, 'end': 5, 'label': 'PERSON'}]
        },
        {
            INPUT_HASH_ATTR: 1,
            TASK_HASH_ATTR: 12,
            'text': 'Hello world',
            'answer': 'reject',
            'spans': [{'start': 6, 'end': 11, 'label': 'PERSON'}]
        },
        {
            INPUT_HASH_ATTR: 2,
            TASK_HASH_ATTR: 21,
            'text': 'This is a test',
            'answer': 'reject',
            'spans': [{'start': 5, 'end': 7, 'label': 'ORG'}]
        }
    ]
    with tmp_dataset(silver_dataset, silver_examples):
        recipe = ner_silver_to_gold(silver_dataset, dataset, spacy_model)
        stream = list(recipe['stream'])
    assert recipe['view_id'] == 'ner_manual'
    assert recipe['dataset'] == dataset
    assert len(stream) == 2
    assert stream[0]['text'] == 'Hello world'
    assert 'tokens' in stream[0]
    assert stream[1]['text'] == 'This is a test'
    assert 'tokens' in stream[1]

def test_ner_eval_ab(dataset, spacy_model, source):
    recipe = ner_eval_ab(dataset, spacy_model, "blank:en", source, ["ORG"])
    stream = list(recipe['stream'])
    print(stream[0])
    assert stream[0]["A"]["spans"][0]["label"] == "ORG"
    assert len(stream[0]["B"]["spans"]) == 0
    assert len(stream[0]["options"]) == 2
    assert hasattr(recipe['on_exit'], '__call__')

def test_textcat_teach(dataset, spacy_model, source, labels, patterns):
    recipe = textcat_teach(dataset, spacy_model, source, labels, patterns)
    stream = list(recipe['stream'])
    assert recipe['view_id'] == 'classification'
    assert recipe['dataset'] == dataset
    assert len(stream) >= 2
    assert 'label' in stream[0]
    assert 'meta' in stream[0]
    assert 'score' in stream[0]['meta']


def test_textcat_custom_model(dataset, source, labels):
    recipe = textcat_custom_model(dataset, source, labels)
    stream = list(recipe['stream'])
    assert recipe['view_id'] == 'classification'
    assert recipe['dataset'] == dataset
    assert len(stream) >= 1
    assert 'label' in stream[0]


def test_textcat_manual(dataset, source, labels):
    recipe = textcat_manual(dataset, source, labels)
    stream = list(recipe['stream'])
    assert recipe['view_id'] == 'choice'
    assert recipe['dataset'] == dataset
    assert len(stream) == 2
    assert 'options' in stream[0]


def test_textcat_correct(dataset, nlp, source, labels):
    component = "dummy_textcat"
    dummy_textcat_component = nlp.add_pipe(component)
    with make_tmpdir() as tempdir:
        nlp.to_disk(tempdir)
        recipe = textcat_correct(dataset, tempdir, source, labels, False, None, 0.5, component)
    stream = list(recipe['stream'])
    assert recipe['view_id'] == 'choice'
    assert recipe['dataset'] == dataset
    assert len(stream) == 2
    assert 'options' in stream[0]
    assert 'options' in stream[1]


def test_terms_teach(dataset, vectors):
    seeds = ['cat', 'dog', 'mouse']
    recipe = terms_teach(dataset, vectors, seeds)
    assert recipe['view_id'] == 'text'
    assert recipe['dataset'] == dataset


def test_image_manual(dataset):
    img_dir = tempfile.mkdtemp()
    img1 = tempfile.NamedTemporaryFile(dir=img_dir, prefix='1', suffix='.jpg')
    img2 = tempfile.NamedTemporaryFile(dir=img_dir, prefix='2', suffix='.png')
    no_img = tempfile.NamedTemporaryFile(dir=img_dir, prefix='3', suffix='.txt')
    recipe = image_manual(dataset, img_dir, ['PERSON', 'DOG', 'CAT'])
    stream = list(recipe['stream'])
    assert recipe['view_id'] == 'image_manual'
    assert recipe['dataset'] == dataset
    assert len(stream) == 2


def test_mark(dataset, source):
    view_id = 'text'
    recipe = mark(dataset, source, view_id)
    stream = list(recipe['stream'])
    assert recipe['view_id'] == view_id
    assert recipe['dataset'] == dataset
    assert len(stream) == 2
    assert hasattr(recipe['update'], '__call__')
    assert hasattr(recipe['on_load'], '__call__')
    assert hasattr(recipe['on_exit'], '__call__')


def test_choice(dataset, source):
    options = ['OPTION_A', 'OPTION_B', 'OPTION_C']
    recipe = choice(dataset, source, options)
    stream = list(recipe['stream'])
    assert recipe['view_id'] == 'choice'
    assert recipe['dataset'] == dataset
    assert len(stream) == 2
    assert 'options' in stream[0]
    assert len(stream[0]['options']) == 3
    assert stream[0]['options'][0]['id'] == 'OPTION_A'
    assert recipe['config']['choice_style'] == 'single'
    assert recipe['config']['choice_auto_accept']