<commit_before>import os
import pytest

from polyglotdb.corpus import CorpusContext

def test_basic_query(graph_db):
    with CorpusContext(corpus_name = 'untimed', **graph_db) as g:
        q = g.query_graph(g.word).filter(g.word.label == 'are')
        print(q.cypher())
        assert(all(x.label == 'are' for x in q.all()))

def test_order_by(graph_db):
    with CorpusContext(corpus_name = 'timed', **graph_db) as g:
        q = g.query_graph(g.word).filter(g.word.label == 'are').order_by(g.word.begin.column_name('begin')).times('begin','end')
        prev = 0
        print(q.cypher())
        for x in q.all():
            assert(x.begin > prev)
            prev = x.begin

def test_query_previous(graph_db):
    with CorpusContext(corpus_name = 'untimed', **graph_db) as g:
        q = g.query_graph(g.word).filter(g.word.label == 'are')
        q = q.filter(g.word.previous.label == 'cats')
        print(q.cypher())
        assert(len(list(q.all())) == 1)

def test_query_following(graph_db):
    with CorpusContext(corpus_name = 'untimed', **graph_db) as g:
        q = g.query_graph(g.word).filter(g.word.label == 'are')
        q = q.filter(g.word.following.label == 'too')
        print(q.cypher())
        print(list(q.all()))
        assert(len(list(q.all())) == 1)

def test_query_time(graph_db):
    with CorpusContext(corpus_name = 'timed', **graph_db) as g:
        q = g.query_graph(g.word).filter(g.word.label == 'are')
        q = q.filter(g.word.begin > 2)
        print(q.cypher())
        assert(len(list(q.all())) == 1)
    with CorpusContext(corpus_name = 'timed', **graph_db) as g:
        q = g.query_graph(g.word).filter(g.word.label == 'are')
        q = q.filter(g.word.begin < 2)
        print(q.cypher())
        assert(len(list(q.all())) == 1)

def test_query_contains(graph_db):
    with CorpusContext(corpus_name = 'untimed', **graph_db) as g:
        q = g.query_graph(g.word).filter_contains(g.phone.label == 'aa')
        print(q.cypher())
        assert(len(list(q.all())) == 3)

def test_query_contained_by(graph_db):
    with CorpusContext(corpus_name = 'untimed', **graph_db) as g:
        q = g.query_graph(g.phone).filter(g.phone.label == 'aa')
        q = q.filter_contained_by(g.word.label == 'dogs')
        print(q.cypher())
        assert(len(list(q.all())) == 1)

def test_query_left_aligned_line(graph_db):
    with CorpusContext(corpus_name = 'untimed', **graph_db) as g:
        q = g.query_graph(g.phone).filter(g.phone.label == 'k')
        q = q.filter_left_aligned(g.line)
        print(q.cypher())
        assert(len(list(q.all())) == 1)

def test_query_previous_left_aligned_line(graph_db):
    with CorpusContext(corpus_name = 'untimed', **graph_db) as g:
        q = g.query_graph(g.phone).filter(g.phone.label == 'ae')
        q = q.filter(g.phone.previous.begin == g.line.begin)
        print(q.cypher())
        assert(q.count() == 1)
    with CorpusContext(corpus_name = 'timed', **graph_db) as g:
        q = g.query_graph(g.phone).filter(g.phone.label == 'ae')
        q = q.filter(g.phone.previous.begin == g.line.begin)
        print(q.cypher())
        assert(q.count() == 1)

def test_query_phone_in_line_initial_word(graph_db):
    with CorpusContext(corpus_name = 'untimed', **graph_db) as g:
        word_q = g.query_graph(g.word).filter_left_aligned(g.line)
        assert(len(list(word_q.all())) == 3)
        q = g.query_graph(g.phone).filter(g.phone.label == 'aa')
        q = q.filter_contained_by(g.word.id.in_(word_q))
        print(q.cypher())
        assert(len(list(q.all())) == 1)

def test_query_word_in(graph_db):
    with CorpusContext(corpus_name = 'untimed', **graph_db) as g:
        q = g.query_graph(g.phone).filter(g.phone.label == 'k')
        q = q.filter_contained_by(g.word.label.in_(['cats','dogs','cute']))
        print(q.cypher())
        assert(len(list(q.all())) == 2)

def test_query_coda_phone(graph_db):
    with CorpusContext(corpus_name = 'syllable_morpheme', **graph_db) as g:
        q = g.query_graph(g.phone).filter(g.phone.label == 'k')
        q = q.filter_right_aligned(g.syllable)
        print(q.cypher())
        assert(len(list(q.all())) == 1)

        q = g.query_graph(g.phone).filter(g.phone.label == 'k')
        q = q.filter_not_right_aligned(g.syllable)
        print(q.cypher())
        assert(len(list(q.all())) == 1)

def test_query_onset_phone(graph_db):
    with CorpusContext(corpus_name = 'syllable_morpheme', **graph_db) as g:
        q = g.query_graph(g.phone).filter(g.phone.label == 'k')
        q = q.filter_left_aligned(g.syllable)
        print(q.cypher())
        assert(len(list(q.all())) == 1)

        q = g.query_graph(g.phone).filter(g.phone.label == 'k')
        q = q.filter_not_left_aligned(g.syllable)
        print(q.cypher())<commit_msg>Some tests for alignment<commit_after>import os
import pytest

from polyglotdb.corpus import CorpusContext

def test_basic_query(graph_db):
    with CorpusContext(corpus_name = 'untimed', **graph_db) as g:
        q = g.query_graph(g.word).filter(g.word.label == 'are')
        print(q.cypher())
        assert(all(x.label == 'are' for x in q.all()))

def test_order_by(graph_db):
    with CorpusContext(corpus_name = 'timed', **graph_db) as g:
        q = g.query_graph(g.word).filter(g.word.label == 'are').order_by(g.word.begin.column_name('begin')).times('begin','end')
        prev = 0
        print(q.cypher())
        for x in q.all():
            assert(x.begin > prev)
            prev = x.begin

def test_query_previous(graph_db):
    with CorpusContext(corpus_name = 'untimed', **graph_db) as g:
        q = g.query_graph(g.word).filter(g.word.label == 'are')
        q = q.filter(g.word.previous.label == 'cats')
        print(q.cypher())
        assert(len(list(q.all())) == 1)

def test_query_following(graph_db):
    with CorpusContext(corpus_name = 'untimed', **graph_db) as g:
        q = g.query_graph(g.word).filter(g.word.label == 'are')
        q = q.filter(g.word.following.label == 'too')
        print(q.cypher())
        print(list(q.all()))
        assert(len(list(q.all())) == 1)

def test_query_time(graph_db):
    with CorpusContext(corpus_name = 'timed', **graph_db) as g:
        q = g.query_graph(g.word).filter(g.word.label == 'are')
        q = q.filter(g.word.begin > 2)
        print(q.cypher())
        assert(len(list(q.all())) == 1)
    with CorpusContext(corpus_name = 'timed', **graph_db) as g:
        q = g.query_graph(g.word).filter(g.word.label == 'are')
        q = q.filter(g.word.begin < 2)
        print(q.cypher())
        assert(len(list(q.all())) == 1)

def test_query_contains(graph_db):
    with CorpusContext(corpus_name = 'untimed', **graph_db) as g:
        q = g.query_graph(g.word).filter_contains(g.phone.label == 'aa')
        print(q.cypher())
        assert(len(list(q.all())) == 3)

def test_query_contained_by(graph_db):
    with CorpusContext(corpus_name = 'untimed', **graph_db) as g:
        q = g.query_graph(g.phone).filter(g.phone.label == 'aa')
        q = q.filter_contained_by(g.word.label == 'dogs')
        print(q.cypher())
        assert(len(list(q.all())) == 1)

def test_query_left_aligned_line(graph_db):
    with CorpusContext(corpus_name = 'untimed', **graph_db) as g:
        q = g.query_graph(g.phone).filter(g.phone.label == 'k')
        q = q.filter_left_aligned(g.line)
        print(q.cypher())
        assert(len(list(q.all())) == 1)

def test_query_previous_left_aligned_line(graph_db):
    with CorpusContext(corpus_name = 'untimed', **graph_db) as g:
        q = g.query_graph(g.phone).filter(g.phone.label == 'ae')
        q = q.filter(g.phone.previous.begin == g.line.begin)
        print(q.cypher())
        assert(q.count() == 1)
    with CorpusContext(corpus_name = 'timed', **graph_db) as g:
        q = g.query_graph(g.phone).filter(g.phone.label == 'ae')
        q = q.filter(g.phone.previous.begin == g.line.begin)
        print(q.cypher())
        assert(q.count() == 1)

def test_query_phone_in_line_initial_word(graph_db):
    with CorpusContext(corpus_name = 'untimed', **graph_db) as g:
        word_q = g.query_graph(g.word).filter_left_aligned(g.line)
        assert(len(list(word_q.all())) == 3)
        q = g.query_graph(g.phone).filter(g.phone.label == 'aa')
        q = q.filter_contained_by(g.word.id.in_(word_q))
        print(q.cypher())
        assert(len(list(q.all())) == 1)

def test_query_word_in(graph_db):
    with CorpusContext(corpus_name = 'untimed', **graph_db) as g:
        q = g.query_graph(g.phone).filter(g.phone.label == 'k')
        q = q.filter_contained_by(g.word.label.in_(['cats','dogs','cute']))
        print(q.cypher())
        assert(len(list(q.all())) == 2)

def test_query_coda_phone(graph_db):
    with CorpusContext(corpus_name = 'syllable_morpheme', **graph_db) as g:
        q = g.query_graph(g.phone).filter(g.phone.label == 'k')
        q = q.filter_right_aligned(g.syllable)
        print(q.cypher())
        assert(len(list(q.all())) == 1)

        q = g.query_graph(g.phone).filter(g.phone.label == 'k')
        q = q.filter(g.phone.end == g.syllable.end)
        print(q.cypher())
        assert(len(list(q.all())) == 1)

        q = g.query_graph(g.phone).filter(g.phone.label == 'k')
        q = q.filter_not_right_aligned(g.syllable)
        print(q.cypher())
        assert(len(list(q.all())) == 1)

        q = g.query_graph(g.phone).filter(g.phone.label == 'k')
        q = q.filter(g.phone.end != g.syllable.end)
        print(q.cypher())
        assert(len(list(q.all())) == 1)

def test_query_onset_phone(graph_db):
    with CorpusContext(corpus_name = 'syllable_morpheme', **graph_db) as g:
        q = g.query_graph(g.phone).filter(g.phone.label == 'k')
        q = q.filter_left_aligned(g.syllable)
        print(q.cypher())
        assert(len(list(q.all())) == 1)

        q = g.query_graph(g.phone).filter(g.phone.label == 'k')
        q = q.filter(g.phone.begin == g.syllable.begin)
        print(q.cypher())
        assert(len(list(q.all())) == 1)

        q = g.query_graph(g.phone).filter(g.phone.label == 'k')
        q = q.filter_not_left_aligned(g.syllable)
        print(q.cypher())
        assert(len(list(q.all())) == 1)

        q = g.query_graph(g.phone).filter(g.phone.label == 'k')
        q = q.filter(g.phone.begin != g.syllable.begin)
        print(q.cypher())
{'fraction_non_alphanumeric': 0.17719315214237355, 'fraction_numerical': 0.0038688461166457105, 'mean_word_length': 2.9495798319327733, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4771308', 'n_tokens_mistral': 3980, 'n_tokens_neox': 3875, 'n_words': 590}