Python implementation of [Earthstar](https://earthstar-project.org/)
You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
435 lines
14 KiB
435 lines
14 KiB
"""Tests for document querying""" |
|
|
|
from datetime import datetime, timedelta, timezone |
|
|
|
from typing import Literal |
|
|
|
import pytest |
|
|
|
from earthsnake.document.es4 import Es4Document |
|
from earthsnake.exc import ValidationError |
|
from earthsnake.identity import Identity |
|
from earthsnake.path import Path |
|
from earthsnake.query import ( |
|
Cmp, |
|
HistoryMode, |
|
OrderByField, |
|
OrderDirection, |
|
Query, |
|
QueryFilter, |
|
QueryVerdict, |
|
) |
|
|
|
|
|
def test_init_negative_limit() -> None: |
|
"""Test if initialization fails with a negative limit""" |
|
|
|
with pytest.raises(ValidationError) as ctx: |
|
Query(limit=-1) |
|
|
|
assert str(ctx.value) == 'query limit must be 0 or greater' |
|
|
|
|
|
def test_init_mutual_exclusive_start_after() -> None: |
|
"""Test if start_after_local_index and start_after_path are mutually exclusive""" |
|
|
|
with pytest.raises(ValidationError) as ctx: |
|
Query(start_after_local_index=1, start_after_path=Path('/test.txt')) |
|
|
|
assert ( |
|
str(ctx.value) |
|
== 'query start position should be either a local_index or a path, but not both' |
|
) |
|
|
|
|
|
def test_init_order_start_after_mismatch_path() -> None: |
|
"""Test if init fails if ordering by path but starting at a local index""" |
|
|
|
with pytest.raises(ValidationError) as ctx: |
|
Query(start_after_local_index=1, order_by_field=OrderByField.PATH) |
|
|
|
assert str(ctx.value) == 'query would order by path, but instructed to start at a local index' |
|
|
|
|
|
def test_init_order_start_after_mismatch_local_index() -> None: |
|
"""Test if init fails if ordering by local_index but starting at a path""" |
|
|
|
with pytest.raises(ValidationError) as ctx: |
|
Query(start_after_path=Path('/test.txt'), order_by_field=OrderByField.LOCAL_INDEX) |
|
|
|
assert str(ctx.value) == 'query would order by local index, but instructed to start at a path' |
|
|
|
|
|
def test_init_path_startswith_mismatch() -> None: |
|
"""Test if init fails if we match an exact path and a path prefix that doesn’t match path""" |
|
|
|
with pytest.raises(ValidationError) as ctx: |
|
Query(flt=QueryFilter(path=Path('/test.txt'), path_prefix='/other.md')) |
|
|
|
assert ( |
|
str(ctx.value) |
|
== 'we were asked to match an exact path and a path prefix but they don’t match each other' |
|
) |
|
|
|
|
|
def test_init_path_endswith_mismatch() -> None: |
|
"""Test if init fails if we match an exact path and a path suffix that doesn’t match path""" |
|
|
|
with pytest.raises(ValidationError) as ctx: |
|
Query(flt=QueryFilter(path=Path('/test.txt'), path_suffix='.md')) |
|
|
|
assert ( |
|
str(ctx.value) |
|
== 'we were asked to match an exact path and a path suffix but they don’t match each other' |
|
) |
|
|
|
|
|
def test_init_empty_prediction() -> None: |
|
"""Test if verdict is NOTHING if limit is zero""" |
|
|
|
query = Query(limit=0) |
|
|
|
assert query.prediction == QueryVerdict.NOTHING |
|
|
|
|
|
def test_init_filter_prediction() -> None: |
|
"""Test if verdict is SOME if we have a filter set""" |
|
|
|
query = Query(flt=QueryFilter(path=Path('/test.txt'))) |
|
|
|
assert query.prediction == QueryVerdict.SOME |
|
|
|
|
|
@pytest.mark.parametrize( |
|
'history_mode,prediction', |
|
( |
|
pytest.param(HistoryMode.ALL, QueryVerdict.ALL), |
|
pytest.param(HistoryMode.LATEST, QueryVerdict.ALL_LATEST), |
|
), |
|
) |
|
def test_init_all_prediction(history_mode: HistoryMode, prediction: QueryVerdict) -> None: |
|
"""Test if verdict is ALL/ALL_LATEST if we don’t have any filtering parameters set""" |
|
|
|
query = Query(history_mode=history_mode) |
|
|
|
assert query.prediction == prediction |
|
|
|
|
|
def test_match_doc_no_filter(es4_document: Es4Document) -> None: |
|
"""Test if the empty filter works corretly""" |
|
|
|
query = Query() |
|
|
|
assert query.match_document(es4_document) is True |
|
|
|
|
|
def test_match_doc_path(es4_document: Es4Document) -> None: |
|
"""Test if the path works correctly""" |
|
query = Query(flt=QueryFilter(path=Path('/test.txt'))) |
|
|
|
es4_document.path = Path('/test.txt') |
|
assert query.match_document(es4_document) |
|
|
|
es4_document.path = Path('/test.md') |
|
assert not query.match_document(es4_document) |
|
|
|
|
|
def test_match_doc_path_prefix(es4_document: Es4Document) -> None: |
|
"""Test if the path prefix filter works correctly""" |
|
query = Query(flt=QueryFilter(path_prefix='/test')) |
|
|
|
es4_document.path = Path('/test.txt') |
|
assert query.match_document(es4_document) |
|
|
|
es4_document.path = Path('/other.txt') |
|
assert not query.match_document(es4_document) |
|
|
|
|
|
def test_match_doc_path_suffix(es4_document: Es4Document) -> None: |
|
"""Test if the path suffix filter works correctly""" |
|
|
|
query = Query(flt=QueryFilter(path_suffix='.txt')) |
|
|
|
es4_document.path = Path('/test.txt') |
|
assert query.match_document(es4_document) |
|
|
|
es4_document.path = Path('/test.md') |
|
assert not query.match_document(es4_document) |
|
|
|
|
|
def test_match_doc_author(es4_document: Es4Document, identity: Identity) -> None: |
|
"""Test if the author filter works correctly""" |
|
|
|
query = Query(flt=QueryFilter(author=identity)) |
|
|
|
es4_document.author = identity |
|
assert query.match_document(es4_document) |
|
|
|
es4_document.author = Identity.generate('name') |
|
assert not query.match_document(es4_document) |
|
|
|
|
|
@pytest.mark.parametrize('cmp_usage', ('explicit', 'implicit')) |
|
def test_match_timestamp_eq( |
|
cmp_usage: Literal['explicit', 'implicit'], |
|
es4_document: Es4Document, |
|
) -> None: |
|
"""Test if the timestamp filter works correctly with the EQ comparator""" |
|
|
|
flt = QueryFilter(timestamp=datetime(2022, 5, 5, 8, 21, 11, 668993, tzinfo=timezone.utc)) |
|
|
|
if cmp_usage == 'explicit': |
|
flt['timestamp_comp'] = Cmp.EQ |
|
|
|
query = Query(flt=flt) |
|
|
|
assert query.match_document(es4_document) |
|
|
|
es4_document.timestamp += timedelta(days=1) |
|
assert not query.match_document(es4_document) |
|
|
|
|
|
def test_match_timestamp_gt(es4_document: Es4Document) -> None: |
|
"""Test if the timestamp filter works correctly with the GT comparator""" |
|
|
|
flt = QueryFilter( |
|
timestamp=datetime(2022, 5, 5, 8, 21, 11, 668992, tzinfo=timezone.utc), |
|
timestamp_comp=Cmp.GT, |
|
) |
|
|
|
query = Query(flt=flt) |
|
|
|
assert query.match_document(es4_document) |
|
|
|
es4_document.timestamp -= timedelta(microseconds=1) |
|
assert not query.match_document(es4_document) |
|
|
|
|
|
def test_match_timestamp_lt(es4_document: Es4Document) -> None: |
|
"""Test if the timestamp filter works correctly with the LT comparator""" |
|
|
|
flt = QueryFilter( |
|
timestamp=datetime(2022, 5, 5, 8, 21, 11, 668994, tzinfo=timezone.utc), |
|
timestamp_comp=Cmp.LT, |
|
) |
|
|
|
query = Query(flt=flt) |
|
|
|
assert query.match_document(es4_document) |
|
|
|
es4_document.timestamp += timedelta(microseconds=1) |
|
assert not query.match_document(es4_document) |
|
|
|
|
|
@pytest.mark.parametrize('cmp_usage', ('explicit', 'implicit')) |
|
def test_match_content_length_eq( |
|
cmp_usage: Literal['explicit', 'implicit'], |
|
es4_document: Es4Document, |
|
) -> None: |
|
"""Test if the content_length filter works correctly with the EQ comparator""" |
|
|
|
flt = QueryFilter(content_length=4) |
|
|
|
if cmp_usage == 'explicit': |
|
flt['content_length_comp'] = Cmp.EQ |
|
|
|
query = Query(flt=flt) |
|
|
|
assert query.match_document(es4_document) |
|
|
|
es4_document.content = 'other' |
|
assert not query.match_document(es4_document) |
|
|
|
|
|
def test_match_content_length_gt(es4_document: Es4Document) -> None: |
|
"""Test if the content_length filter works correctly with the GT comparator""" |
|
|
|
flt = QueryFilter(content_length=4, content_length_comp=Cmp.GT) |
|
|
|
query = Query(flt=flt) |
|
|
|
assert not query.match_document(es4_document) |
|
|
|
es4_document.content = 'other' |
|
assert query.match_document(es4_document) |
|
|
|
|
|
def test_match_content_length_lt(es4_document: Es4Document) -> None: |
|
"""Test if the content_length filter works correctly with the LT comparator""" |
|
|
|
flt = QueryFilter(content_length=4, content_length_comp=Cmp.LT) |
|
|
|
query = Query(flt=flt) |
|
|
|
assert not query.match_document(es4_document) |
|
|
|
es4_document.content = 'see' |
|
assert query.match_document(es4_document) |
|
|
|
|
|
def test_query_filter_nothing_prediction(es4_document: Es4Document) -> None: |
|
"""Test if the query predicted NOTHING, it returns an empty list""" |
|
|
|
query = Query(limit=0) |
|
|
|
assert not query({es4_document: {}}) |
|
|
|
|
|
@pytest.mark.parametrize('cmp_usage', ('implicit', 'explicit')) |
|
def test_query_filter_path_sort( |
|
cmp_usage: Literal['implicit', 'explicit'], |
|
es4_document: Es4Document, |
|
identity: Identity, |
|
) -> None: |
|
"""Test if ordering by path works""" |
|
|
|
other_document = Es4Document(identity, es4_document.share, Path('/much.txt')) |
|
|
|
if cmp_usage == 'implicit': |
|
query = Query(order_by_field=OrderByField.PATH) |
|
else: |
|
query = Query(order_by_field=OrderByField.PATH, order_direction=OrderDirection.ASC) |
|
|
|
assert query({es4_document: {}, other_document: {}}) == [other_document, es4_document] |
|
|
|
|
|
def test_query_filter_path_sort_desc(es4_document: Es4Document, identity: Identity) -> None: |
|
"""Test if ordering by path in descending order works""" |
|
|
|
other_document = Es4Document(identity, es4_document.share, Path('/much.txt')) |
|
query = Query(order_by_field=OrderByField.PATH, order_direction=OrderDirection.DESC) |
|
|
|
assert query({other_document: {}, es4_document: {}}) == [es4_document, other_document] |
|
|
|
|
|
@pytest.mark.parametrize('cmp_usage', ('implicit', 'explicit')) |
|
def test_query_filter_local_index_sort( |
|
cmp_usage: Literal['implicit', 'explicit'], |
|
es4_document: Es4Document, |
|
identity: Identity, |
|
) -> None: |
|
"""Test if ordering by path works""" |
|
|
|
other_document = Es4Document(identity, es4_document.share, Path('/much.txt')) |
|
|
|
if cmp_usage == 'implicit': |
|
query = Query(order_by_field=OrderByField.LOCAL_INDEX) |
|
else: |
|
query = Query(order_by_field=OrderByField.LOCAL_INDEX, order_direction=OrderDirection.ASC) |
|
|
|
assert query({es4_document: {'local_index': 2}, other_document: {'local_index': 1}}) == [ |
|
other_document, |
|
es4_document, |
|
] |
|
|
|
|
|
def test_query_filter_local_index_sort_desc(es4_document: Es4Document, identity: Identity) -> None: |
|
"""Test if ordering by path in descending order works""" |
|
|
|
other_document = Es4Document(identity, es4_document.share, Path('/much.txt')) |
|
query = Query(order_by_field=OrderByField.LOCAL_INDEX, order_direction=OrderDirection.DESC) |
|
|
|
assert query({other_document: {'local_index': 1}, es4_document: {'local_index': 2}}) == [ |
|
es4_document, |
|
other_document, |
|
] |
|
|
|
|
|
def test_start_after_path_asc(es4_document: Es4Document, identity: Identity) -> None: |
|
"""Test if start_after_path works with ascending sort""" |
|
|
|
other_document = Es4Document(identity, es4_document.share, Path('/much.txt')) |
|
query = Query(order_by_field=OrderByField.PATH, start_after_path=Path('/much.txt')) |
|
|
|
assert query({other_document: {}, es4_document: {}}) == [es4_document] |
|
|
|
|
|
def test_start_after_path_desc(es4_document: Es4Document, identity: Identity) -> None: |
|
"""Test if start_after_path works with descending sort""" |
|
|
|
other_document = Es4Document(identity, es4_document.share, Path('/much.txt')) |
|
query = Query( |
|
order_by_field=OrderByField.PATH, |
|
order_direction=OrderDirection.DESC, |
|
start_after_path=Path('/test.txt'), |
|
) |
|
|
|
assert query({other_document: {}, es4_document: {}}) == [other_document] |
|
|
|
|
|
def test_start_after_local_index_asc(es4_document: Es4Document, identity: Identity) -> None: |
|
"""Test if start_after_local_index works with ascending sort""" |
|
|
|
other_document = Es4Document(identity, es4_document.share, Path('/much.txt')) |
|
query = Query(order_by_field=OrderByField.LOCAL_INDEX, start_after_local_index=1) |
|
|
|
assert query({other_document: {'local_index': 2}, es4_document: {'local_index': 1}}) == [ |
|
other_document |
|
] |
|
|
|
|
|
def test_start_after_local_index_desc(es4_document: Es4Document, identity: Identity) -> None: |
|
"""Test if start_after_path works with descending sort""" |
|
|
|
other_document = Es4Document(identity, es4_document.share, Path('/much.txt')) |
|
query = Query( |
|
order_by_field=OrderByField.LOCAL_INDEX, |
|
order_direction=OrderDirection.DESC, |
|
start_after_local_index=2, |
|
) |
|
|
|
assert query({other_document: {'local_index': 1}, es4_document: {'local_index': 2}}) == [ |
|
other_document |
|
] |
|
|
|
|
|
def test_start_after_with_empty_output() -> None: |
|
"""Test if start_after does not fail with an empty document list""" |
|
|
|
query = Query(start_after_path=Path('/test.txt')) |
|
|
|
assert query({}) == [] |
|
|
|
|
|
@pytest.mark.parametrize('use_start_after', (True, False), ids=('with_start_after', 'without_start_after')) |
|
def test_limit(use_start_after, es4_document: Es4Document, identity: Identity) -> None: |
|
"""Test if document count limiting works""" |
|
|
|
other_document = Es4Document(identity, es4_document.share, Path('/much.txt')) |
|
|
|
if use_start_after: |
|
query = Query(start_after_path=Path('/much.txt'), limit=1) |
|
expected = es4_document |
|
else: |
|
query = Query(limit=1) |
|
expected = other_document |
|
|
|
assert query({es4_document: {}, other_document: {}}) == [expected] |
|
|
|
|
|
def test_byte_limiting_not_reached(es4_document: Es4Document, identity: Identity) -> None: |
|
"""Test if byte limiting doesn’t do anything if the sum of document sizes is not reached""" |
|
|
|
other_document = Es4Document(identity, es4_document.share, Path('/much.txt'), content='much') |
|
query = Query(limit_bytes=15) |
|
|
|
assert query({es4_document: {}, other_document: {}}) == [other_document, es4_document] |
|
|
|
|
|
def test_byte_limiting(es4_document: Es4Document, identity: Identity) -> None: |
|
"""Test if byte limiting works as expected""" |
|
|
|
other_document = Es4Document(identity, es4_document.share, Path('/much.txt'), content='much') |
|
query = Query(limit_bytes=5) |
|
|
|
assert query({es4_document: {}, other_document: {}}) == [other_document] |
|
|
|
|
|
def test_byte_limiting_with_start_after(es4_document: Es4Document, identity: Identity) -> None: |
|
"""Test if byte limiting works as expected when not starting from the beginning""" |
|
|
|
other_document1 = Es4Document(identity, es4_document.share, Path('/much1.txt'), content='much1') |
|
other_document2 = Es4Document(identity, es4_document.share, Path('/much2.txt'), content='much2') |
|
query = Query(limit_bytes=5, start_after_path='/much1.txt') |
|
|
|
assert query({es4_document: {}, other_document1: {}, other_document2: {}}) == [other_document2]
|
|
|