earthsnake/tests/test_document_es4.py

"""Tests for the es.4 document validator"""

from datetime import datetime, timedelta
from typing import Any, Dict

from nacl.signing import SigningKey
import pytest
from pytest_mock.plugin import MockerFixture

from earthsnake.exc import ValidationError
from earthsnake.document.es4 import Es4Document
from earthsnake.identity import Identity
from earthsnake.path import Path
from earthsnake.share import Share


VALID_DOCUMENT: Dict[str, Any] = {
    'format': 'es.4',
    'timestamp': 1651738871668993,
    'deleteAfter': None,
    'author': '@test.bcz76z52y5dlpohtkmpuj3jsdcvfmebzpcgfmtmhu4u7hlexzreya',
    'path': '/test.txt',
    'workspace': '+test.suffix',
    'signature': (
        'bughac3ooy5qfect4bxdke2zkun2wqufhpivt57vj2mzq52mhqzesjvhyywttxm7qqjyhoskmiqd2lw72qy2u766r'
        'fqvnbwab4qp3gbi'
    ),
    'content': 'test',
    'contentHash': 'bt6dnbamijr6wlgrp5kqmkwwqcwr36ty3fmfyelgrlvwblmhqbiea',
}


@pytest.mark.parametrize(
    'missing',
    Es4Document.CORE_SCHEMA['required'],  # type: ignore
)
def test_validate_missing_keys(missing: str) -> None:
    """Test if validation fails when a required key is missing"""

    raw_document = VALID_DOCUMENT.copy()
    del raw_document[missing]

    with pytest.raises(ValidationError) as ctx:
        Es4Document.from_json(raw_document)

    required_field_names = ', '.join(
        repr(x) for x in Es4Document.CORE_SCHEMA['required']  # type: ignore
    )
    assert str(ctx.value) == f"data must contain [{required_field_names}] properties"


def test_validate_future_document() -> None:
    """Test if validation fails when the document is created far in the future"""

    raw_document = VALID_DOCUMENT.copy()
    raw_document['timestamp'] = int(
        (datetime.utcnow() + timedelta(minutes=20)).timestamp() * 1000000
    )

    with pytest.raises(ValidationError) as ctx:
        Es4Document.from_json(raw_document)

    assert str(ctx.value) == 'timestamp too far in the future'


def test_validate_expiry_before_creation() -> None:
    """Test if validation fails when the document is set to expire before it has been created"""

    raw_document = VALID_DOCUMENT.copy()
    raw_document['timestamp'] = int(
        (datetime.utcnow() + timedelta(minutes=5)).timestamp() * 1000000
    )
    raw_document['deleteAfter'] = raw_document['timestamp'] - 1

    with pytest.raises(ValidationError) as ctx:
        Es4Document.from_json(raw_document)

    assert str(ctx.value) == 'ephemeral doc expired before it was created'


def test_validate_deleteafter_in_past() -> None:
    """Test if validation fails if deleteAfter is in the pas"""

    raw_document = VALID_DOCUMENT.copy()
    yesterday = datetime.utcnow() - timedelta(days=1)
    raw_document['deleteAfter'] = int(yesterday.timestamp() * 1000000)

    with pytest.raises(ValidationError) as ctx:
        Es4Document.from_json(raw_document)

    assert str(ctx.value) == 'ephemeral doc has expired'


def test_validate_not_writable() -> None:
    """Test if validation fails when the author is not allowed to write at the given path"""

    raw_document = VALID_DOCUMENT.copy()
    raw_document['path'] = '/test/~@some.' + ('a' * 52)

    with pytest.raises(ValidationError) as ctx:
        Es4Document.from_json(raw_document)

    assert (
        str(ctx.value)
        == f'Author {raw_document["author"]} cannot write to path {raw_document["path"]}'
    )


def test_validate() -> None:
    """Test if validation succeeds on a valid document"""

    document = Es4Document.from_json(VALID_DOCUMENT)

    assert isinstance(document, Es4Document)


def test_validate_ephemeral() -> None:
    """Test if validation succeeds with a valid deleteAfter value"""

    raw_document = VALID_DOCUMENT.copy()
    raw_document['deleteAfter'] = 9007199254740990
    raw_document['signature'] = (
        'bw7zyncx7u2wrk2jz4ucfjbbc6b5t5rm7ma42vkaaius2yczvqdbvfyfbc6mh345flsaeqizw44gncot5huaskdmk'
        'npkty7mgtzbssdy'
    )

    document = Es4Document.from_json(raw_document)
    assert isinstance(document, Es4Document)


def test_signature_check_error(mocker: MockerFixture) -> None:
    """Test if validation fails if something goes wrong during signature checking"""

    mocker.patch(
        'earthsnake.document.es4.Es4Document.generate_hash',
        side_effect=ValueError('test error'),
    )

    with pytest.raises(ValidationError) as ctx:
        Es4Document.from_json(VALID_DOCUMENT)

    assert str(ctx.value) == 'Cannot check signature: test error'


def test_signature_check_bad_signature() -> None:
    """Test if validation fails if the document’s signature doesn’t match the document itself"""

    raw_document = VALID_DOCUMENT.copy()
    raw_document['signature'] = 'b' * 104

    with pytest.raises(ValidationError) as ctx:
        Es4Document.from_json(raw_document)

    assert str(ctx.value) == 'signature is invalid'


def test_remove_extra_fields_no_underscore() -> None:
    """Test if remove_extra_fields() fails on non-specced fields not starting with an underscore"""

    raw_document = VALID_DOCUMENT.copy()
    raw_document.update({'test': 'yes'})

    with pytest.raises(ValidationError) as ctx:
        Es4Document.remove_extra_fields(raw_document)

    assert str(ctx.value) == 'extra document fields must have names starting with an underscore'


def test_remove_extra_fields_non_object() -> None:
    """Test if remove_extra_fields() fails if the validated document is not a JSON object"""

    with pytest.raises(ValidationError) as ctx:
        Es4Document.remove_extra_fields([])  # type: ignore

    assert str(ctx.value) == 'Document is not a plain JSON object'


def test_remove_extra_fields() -> None:
    """Test if remove_extra_fields() correctly extracts field names starting with an underscore"""

    raw_document = VALID_DOCUMENT.copy()
    raw_document.update({'_test': True})

    cleaned_doc, removed = Es4Document.remove_extra_fields(raw_document)

    assert cleaned_doc == VALID_DOCUMENT
    assert removed == {'_test': True}


def test_check_content_hash() -> None:
    """Test if validation fails if contentHash is not the hash of content"""

    raw_document = VALID_DOCUMENT.copy()
    raw_document['content'] = 'other'

    with pytest.raises(ValidationError) as ctx:
        Es4Document.from_json(raw_document)

    assert str(ctx.value) == 'content does not match contentHash'


def test_sign_different_author(identity: Identity) -> None:
    """Test if sign() fails if the signing identity is not the same as the document author"""

    document = Es4Document.from_json(VALID_DOCUMENT)

    with pytest.raises(ValidationError) as ctx:
        document.sign(identity=identity)

    assert str(ctx.value) == 'when signing a document, keypair address must match document author'


def test_sign_new_identity() -> None:
    """Test if signing a document with a different entity also sets the author"""

    other_key_seed = (
        b'`_\x8dm\x18\xeem\xe3\\\xeb_\x1aw)\xcd\xb7\xd8\xd9\xdd\xad\x86\x9a#wQ"F\x95\xa1\x178r'
    )
    identity = Identity('name', sign_key=SigningKey(other_key_seed))
    assert str(identity) == '@name.bu5seaewd4p7cx7ot4ue3m6wpigfa5hmowxgeophe2so72roao5wq'

    document = Es4Document.from_json(VALID_DOCUMENT)
    document.author = identity
    document.sign()

    assert (
        document.signature
        == 'bk76oxkhbydy3itajeeqtryyj7ej5y7hqjffae6ilf4kpyklh2w32hx3ndg6cb3zvlphj46zmuxbetk4cj2fh6'
        '5bhxdtzflvf7oamcbi'
    )
    assert document.author == identity


@pytest.mark.id_name('test')
def test_sign(identity: Identity) -> None:
    """Test if the sign() method updates the document with a correct signature"""

    document = Es4Document.from_json(VALID_DOCUMENT)
    document.sign(identity=identity)

    assert document.signature == (
        'bughac3ooy5qfect4bxdke2zkun2wqufhpivt57vj2mzq52mhqzesjvhyywttxm7qqjyhoskmiqd2lw72qy2u766r'
        'fqvnbwab4qp3gbi'
    )


def test_content_setter() -> None:
    """Check if the content setter recalculates content_hash, too"""

    document = Es4Document.from_json(VALID_DOCUMENT)
    document.content = 'new content'
    assert document.content_hash != VALID_DOCUMENT['contentHash']
    assert document.content_hash == 'b7yzgbde66w3m67r7srsiajj765xsj5hmaz4phuhqp6mejs77syaq'


def test_content_hash_getter() -> None:
    """Test if the content_hash getter recalculates the content hash if it’s not already set"""

    document = Es4Document.from_json(VALID_DOCUMENT)
    document._content = 'new content'  # pylint: disable=protected-access
    document._content_hash = None  # pylint: disable=protected-access

    assert document.content_hash == 'b7yzgbde66w3m67r7srsiajj765xsj5hmaz4phuhqp6mejs77syaq'


def test_validate_unsigned_document(identity: Identity) -> None:
    """Test if signature validation fails if there is no signature yet"""

    document = Es4Document(identity, Share.from_address('+test.share'), Path('/test'))

    with pytest.raises(ValidationError) as ctx:
        document.validate_signature()

    assert str(ctx.value) == 'document has no signature assigned'


def test_content_length() -> None:
    """Test the content_length property"""

    document = Es4Document.from_json(VALID_DOCUMENT)

    assert document.content_length == 4