<commit_before>from hiku import query as q
from hiku.graph import Graph, Edge, Field, Link, Option, Root, Many
from hiku.types import Integer
from hiku.validate.query import QueryValidator


def _():
    return 1/0


# TODO: refactor
GRAPH = Graph([
    Edge('e1', [
        Field('f2', Integer, _,
              options=[Option('f2-op1', Integer),
                       Option('f2-op2', Integer, default=1)]),
        Link('l1', Many, _, edge='e2', requires='f2',
             options=[Option('l1-op1', Integer),
                      Option('l1-op2', Integer, default=1)]),
    ]),
    Edge('e2', [
        Field('f3', _),
    ]),
    Root([
        Field('f1', Integer, _),
        Edge('e1', [
            Field('f2', Integer, _,
                  options=[Option('f2-op1', Integer),
                           Option('f2-op2', Integer, default=1)]),
            Link('l1', Many, _, edge='e2', requires='f2',
                 options=[Option('l1-op1', Integer),
                          Option('l1-op2', Integer, default=1)]),
        ]),
        Edge('e2', [
            Field('f3', _),
        ]),
        Link('l2', Many, _, edge='e2', requires=None),
    ]),
])


def check_errors(query, errors):
    validator = QueryValidator(GRAPH)
    validator.visit(query)
    assert validator.errors.list == errors


def test_field():
    # field in the root edge
    check_errors(q.Edge([q.Field('invalid')]), [
        'Field "invalid" is not implemented in the "root" edge',
    ])
    # field in the global edge
    check_errors(q.Edge([q.Link('e1', q.Edge([q.Field('invalid')]))]), [
        'Field "invalid" is not implemented in the "e1" edge',
    ])
    # field in the linked edge
    check_errors(q.Edge([q.Link('l2', q.Edge([q.Field('invalid')]))]), [
        'Field "invalid" is not implemented in the "e2" edge',
    ])


def test_non_field():
    check_errors(q.Edge([q.Field('l2')]), [
        'Trying to query "root.l2" link as it was a field',
    ])
    check_errors(q.Edge([q.Field('e2')]), [
        'Trying to query "e2" edge as it was a field',
    ])


def test_field_options():
    def mk(**kwargs):
        return q.Edge([q.Link('e1', q.Edge([q.Field('f2', **kwargs)]))])

    check_errors(mk(), [
        'Required option "e1.f2:f2-op1" is not specified',
    ])
    check_errors(mk(options={'f2-op1': 1, 'invalid': 2}), [
        'Unknown options for "e1.f2": invalid',
    ])
    check_errors(mk(options={'f2-op1': '1'}), [
        'Invalid type "str" for option "e1.f2:f2-op1" provided',
    ])


def test_link():
    l = q.Link('invalid', q.Edge([]))
    # link in the root edge
    check_errors(q.Edge([l]), [
        'Link "invalid" is not implemented in the "root" edge',
    ])
    # link in the global edge
    check_errors(q.Edge([q.Link('e1', q.Edge([l]))]), [
        'Link "invalid" is not implemented in the "e1" edge',
    ])
    # link in the linked edge
    check_errors(q.Edge([q.Link('l2', q.Edge([l]))]), [
        'Link "invalid" is not implemented in the "e2" edge',
    ])


def test_non_link():
    check_errors(q.Edge([q.Link('f1', q.Edge([]))]), [
        'Trying to query "root.f1" simple field as edge',
    ])


def test_link_options():
    def mk(**kwargs):
        return q.Edge([q.Link('e1', q.Edge([q.Link('l1', q.Edge([]),
                                                   **kwargs)]))])

    check_errors(mk(), [
        'Required option "e1.l1:l1-op1" is not specified',
    ])
    check_errors(mk(options={'l1-op1': 1, 'invalid': 2}), [
        'Unknown options for "e1.l1": invalid',
    ])
    check_errors(mk(options={'l1-op1': '1'}), [<commit_msg>Added tests to validate query with complex fields<commit_after>import pytest

from hiku import query as q
from hiku.graph import Graph, Edge, Field, Link, Option, Root, Many
from hiku.types import Integer, Record, Sequence, Optional
from hiku.validate.query import QueryValidator


def _():
    return 1/0


# TODO: refactor
GRAPH = Graph([
    Edge('e1', [
        Field('f2', Integer, _,
              options=[Option('f2-op1', Integer),
                       Option('f2-op2', Integer, default=1)]),
        Link('l1', Many, _, edge='e2', requires='f2',
             options=[Option('l1-op1', Integer),
                      Option('l1-op2', Integer, default=1)]),
    ]),
    Edge('e2', [
        Field('f3', _),
    ]),
    Root([
        Field('f1', Integer, _),
        Field('f1-maybe', Optional[Record[{'attr': Integer}]], _),
        Field('f1-one', Record[{'attr': Integer}], _),
        Field('f1-many', Sequence[Record[{'attr': Integer}]], _),
        Edge('e1', [
            Field('f2', Integer, _,
                  options=[Option('f2-op1', Integer),
                           Option('f2-op2', Integer, default=1)]),
            Link('l1', Many, _, edge='e2', requires='f2',
                 options=[Option('l1-op1', Integer),
                          Option('l1-op2', Integer, default=1)]),
        ]),
        Edge('e2', [
            Field('f3', _),
        ]),
        Link('l2', Many, _, edge='e2', requires=None),
    ]),
])


def check_errors(query, errors):
    validator = QueryValidator(GRAPH)
    validator.visit(query)
    assert validator.errors.list == errors


def test_field():
    # field in the root edge
    check_errors(q.Edge([q.Field('invalid')]), [
        'Field "invalid" is not implemented in the "root" edge',
    ])
    # field in the global edge
    check_errors(q.Edge([q.Link('e1', q.Edge([q.Field('invalid')]))]), [
        'Field "invalid" is not implemented in the "e1" edge',
    ])
    # field in the linked edge
    check_errors(q.Edge([q.Link('l2', q.Edge([q.Field('invalid')]))]), [
        'Field "invalid" is not implemented in the "e2" edge',
    ])
    # simple field as edge
    check_errors(q.Edge([q.Link('f1', q.Edge([]))]), [
        'Trying to query "root.f1" simple field as edge',
    ])


@pytest.mark.parametrize('field_name', ['f1-maybe', 'f1-one', 'f1-many'])
def test_field_complex(field_name):
    check_errors(q.Edge([q.Link(field_name, q.Edge([]))]), [])
    check_errors(q.Edge([q.Link(field_name, q.Edge([q.Field('invalid')]))]), [
        'Unknown field name',
    ])
    check_errors(q.Edge([q.Link(field_name, q.Edge([q.Field('attr')]))]), [])


def test_non_field():
    check_errors(q.Edge([q.Field('l2')]), [
        'Trying to query "root.l2" link as it was a field',
    ])
    check_errors(q.Edge([q.Field('e2')]), [
        'Trying to query "e2" edge as it was a field',
    ])


def test_field_options():
    def mk(**kwargs):
        return q.Edge([q.Link('e1', q.Edge([q.Field('f2', **kwargs)]))])

    check_errors(mk(), [
        'Required option "e1.f2:f2-op1" is not specified',
    ])
    check_errors(mk(options={'f2-op1': 1, 'invalid': 2}), [
        'Unknown options for "e1.f2": invalid',
    ])
    check_errors(mk(options={'f2-op1': '1'}), [
        'Invalid type "str" for option "e1.f2:f2-op1" provided',
    ])


def test_link():
    l = q.Link('invalid', q.Edge([]))
    # link in the root edge
    check_errors(q.Edge([l]), [
        'Link "invalid" is not implemented in the "root" edge',
    ])
    # link in the global edge
    check_errors(q.Edge([q.Link('e1', q.Edge([l]))]), [
        'Link "invalid" is not implemented in the "e1" edge',
    ])
    # link in the linked edge
    check_errors(q.Edge([q.Link('l2', q.Edge([l]))]), [
        'Link "invalid" is not implemented in the "e2" edge',
    ])


def test_link_options():
    def mk(**kwargs):
        return q.Edge([q.Link('e1', q.Edge([q.Link('l1', q.Edge([]),
                                                   **kwargs)]))])

    check_errors(mk(), [
        'Required option "e1.l1:l1-op1" is not specified',
    ])
    check_errors(mk(options={'l1-op1': 1, 'invalid': 2}), [
        'Unknown options for "e1.l1": invalid',
    ])
    check_errors(mk(options={'l1-op1': '1'}), [
{'fraction_non_alphanumeric': 0.1900889453621347, 'fraction_numerical': 0.022871664548919948, 'mean_word_length': 2.5028927458834, 'pattern_counts': {'":': 4, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4712253', 'n_tokens_mistral': 2710, 'n_tokens_neox': 2660, 'n_words': 602}