313 lines
7.9 KiB
Python
313 lines
7.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
# type: ignore
|
|
from __future__ import annotations
|
|
|
|
import http
|
|
from unittest import mock
|
|
import uuid
|
|
|
|
from django.test import Client
|
|
from django.urls import reverse
|
|
import pytest
|
|
import pytest_mock
|
|
|
|
from hotpocket_backend_testing.services.saves import (
|
|
AssociationsTestingService,
|
|
SaveProcessorTestingService,
|
|
SavesTestingService,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_saves_process_save_task_apply_async(mocker: pytest_mock.MockerFixture,
|
|
async_result,
|
|
) -> mock.Mock:
|
|
return SaveProcessorTestingService().mock_process_save_task_apply_async(
|
|
mocker=mocker, async_result=async_result,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def call(rpc_call_factory):
|
|
return rpc_call_factory(
|
|
'saves.create',
|
|
['https://www.ziomek.dog/'],
|
|
)
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_ok(authenticated_client: Client,
|
|
call,
|
|
account,
|
|
mock_saves_process_save_task_apply_async: mock.Mock,
|
|
):
|
|
# When
|
|
result = authenticated_client.post(
|
|
reverse('ui.rpc'),
|
|
data=call,
|
|
content_type='application/json',
|
|
)
|
|
|
|
# Then
|
|
assert result.status_code == http.HTTPStatus.OK
|
|
|
|
call_result = result.json()
|
|
assert 'error' not in call_result
|
|
|
|
save_pk = uuid.UUID(call_result['result']['target_uuid'])
|
|
association_pk = uuid.UUID(call_result['result']['id'])
|
|
|
|
AssociationsTestingService().assert_created(
|
|
pk=association_pk,
|
|
account_uuid=account.pk,
|
|
target_uuid=save_pk,
|
|
)
|
|
|
|
SavesTestingService().assert_created(
|
|
pk=save_pk,
|
|
account_uuid=account.pk,
|
|
url=call['params'][0],
|
|
is_netloc_banned=False,
|
|
)
|
|
|
|
mock_saves_process_save_task_apply_async.assert_called_once_with(
|
|
kwargs={
|
|
'pk': save_pk,
|
|
},
|
|
)
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_ok_netloc_banned(authenticated_client: Client,
|
|
call,
|
|
account,
|
|
mock_saves_process_save_task_apply_async: mock.Mock,
|
|
):
|
|
# Given
|
|
call['params'][0] = 'https://youtube.com/'
|
|
|
|
# When
|
|
result = authenticated_client.post(
|
|
reverse('ui.rpc'),
|
|
data=call,
|
|
content_type='application/json',
|
|
)
|
|
|
|
# Then
|
|
assert result.status_code == http.HTTPStatus.OK
|
|
|
|
call_result = result.json()
|
|
assert 'error' not in call_result
|
|
|
|
save_pk = uuid.UUID(call_result['result']['target_uuid'])
|
|
|
|
SavesTestingService().assert_created(
|
|
pk=save_pk,
|
|
account_uuid=account.pk,
|
|
url=call['params'][0],
|
|
is_netloc_banned=True,
|
|
)
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_ok_reuse_save(save_out,
|
|
authenticated_client: Client,
|
|
call,
|
|
account,
|
|
mock_saves_process_save_task_apply_async: mock.Mock,
|
|
):
|
|
# Given
|
|
call['params'][0] = save_out.url
|
|
|
|
# When
|
|
result = authenticated_client.post(
|
|
reverse('ui.rpc'),
|
|
data=call,
|
|
content_type='application/json',
|
|
)
|
|
|
|
# Then
|
|
assert result.status_code == http.HTTPStatus.OK
|
|
|
|
call_result = result.json()
|
|
assert 'error' not in call_result
|
|
|
|
save_pk = uuid.UUID(call_result['result']['target_uuid'])
|
|
association_pk = uuid.UUID(call_result['result']['id'])
|
|
|
|
AssociationsTestingService().assert_created(
|
|
pk=association_pk,
|
|
account_uuid=account.pk,
|
|
target_uuid=save_pk,
|
|
)
|
|
|
|
SavesTestingService().assert_reused(
|
|
pk=save_pk,
|
|
reference=save_out,
|
|
)
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_ok_reuse_association(association_out,
|
|
save_out,
|
|
authenticated_client: Client,
|
|
call,
|
|
account,
|
|
mock_saves_process_save_task_apply_async: mock.Mock,
|
|
):
|
|
# Given
|
|
call['params'][0] = save_out.url
|
|
|
|
# When
|
|
result = authenticated_client.post(
|
|
reverse('ui.rpc'),
|
|
data=call,
|
|
content_type='application/json',
|
|
)
|
|
|
|
# Then
|
|
assert result.status_code == http.HTTPStatus.OK
|
|
|
|
call_result = result.json()
|
|
assert 'error' not in call_result
|
|
|
|
association_pk = uuid.UUID(call_result['result']['id'])
|
|
|
|
AssociationsTestingService().assert_reused(
|
|
pk=association_pk,
|
|
reference=association_out,
|
|
)
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_ok_reuse_other_account_save(other_account_save_out,
|
|
authenticated_client: Client,
|
|
call,
|
|
account,
|
|
mock_saves_process_save_task_apply_async: mock.Mock,
|
|
):
|
|
# Given
|
|
call['params'][0] = other_account_save_out.url
|
|
|
|
# When
|
|
result = authenticated_client.post(
|
|
reverse('ui.rpc'),
|
|
data=call,
|
|
content_type='application/json',
|
|
)
|
|
|
|
# Then
|
|
assert result.status_code == http.HTTPStatus.OK
|
|
|
|
call_result = result.json()
|
|
assert 'error' not in call_result
|
|
|
|
save_pk = uuid.UUID(call_result['result']['target_uuid'])
|
|
association_pk = uuid.UUID(call_result['result']['id'])
|
|
|
|
AssociationsTestingService().assert_created(
|
|
pk=association_pk,
|
|
account_uuid=account.pk,
|
|
target_uuid=save_pk,
|
|
)
|
|
|
|
SavesTestingService().assert_reused(
|
|
pk=save_pk,
|
|
reference=other_account_save_out,
|
|
)
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_ok_dont_process_reused_processed_save(processed_save_out,
|
|
authenticated_client: Client,
|
|
call,
|
|
account,
|
|
mock_saves_process_save_task_apply_async: mock.Mock,
|
|
):
|
|
# Given
|
|
call['params'][0] = processed_save_out.url
|
|
|
|
# When
|
|
_ = authenticated_client.post(
|
|
reverse('ui.rpc'),
|
|
data=call,
|
|
content_type='application/json',
|
|
)
|
|
|
|
# Then
|
|
mock_saves_process_save_task_apply_async.assert_not_called()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_empty_url(authenticated_client: Client,
|
|
call,
|
|
account,
|
|
mock_saves_process_save_task_apply_async: mock.Mock,
|
|
):
|
|
# Given
|
|
call['params'][0] = ''
|
|
|
|
# When
|
|
result = authenticated_client.post(
|
|
reverse('ui.rpc'),
|
|
data=call,
|
|
content_type='application/json',
|
|
)
|
|
|
|
# Then
|
|
assert result.status_code == http.HTTPStatus.OK
|
|
|
|
call_result = result.json()
|
|
assert 'error' in call_result
|
|
|
|
assert call_result['error']['data']['url'] == ['blank']
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_invalid_url(authenticated_client: Client,
|
|
call,
|
|
account,
|
|
mock_saves_process_save_task_apply_async: mock.Mock,
|
|
):
|
|
# Given
|
|
call['params'][0] = 'thisisntright'
|
|
|
|
# When
|
|
result = authenticated_client.post(
|
|
reverse('ui.rpc'),
|
|
data=call,
|
|
content_type='application/json',
|
|
)
|
|
|
|
# Then
|
|
assert result.status_code == http.HTTPStatus.OK
|
|
|
|
call_result = result.json()
|
|
assert 'error' in call_result
|
|
|
|
assert call_result['error']['data']['url'] == ['invalid']
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_inactive_account(inactive_account_client: Client, call):
|
|
# When
|
|
result = inactive_account_client.post(
|
|
reverse('ui.rpc'),
|
|
data=call,
|
|
)
|
|
|
|
# Then
|
|
assert result.status_code == http.HTTPStatus.FORBIDDEN
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_anonymous(client: Client, call):
|
|
# When
|
|
result = client.post(
|
|
reverse('ui.rpc'),
|
|
data=call,
|
|
)
|
|
|
|
# Then
|
|
assert result.status_code == http.HTTPStatus.FORBIDDEN
|