167 lines
4.1 KiB
Python
167 lines
4.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
# type: ignore
|
|
from __future__ import annotations
|
|
|
|
import http
|
|
from unittest import mock
|
|
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
from django.test import Client
|
|
from django.urls import reverse
|
|
import pytest
|
|
from pytest_django import asserts
|
|
import pytest_mock
|
|
|
|
from hotpocket_backend_testing.services import UITestingService
|
|
|
|
|
|
@pytest.fixture
|
|
def pocket_csv_file(pocket_csv_content):
|
|
result = SimpleUploadedFile(
|
|
'part_000000.csv',
|
|
pocket_csv_content.encode('utf-8'),
|
|
content_type='application/csv;charset=utf-8',
|
|
)
|
|
|
|
yield result
|
|
|
|
|
|
@pytest.fixture
|
|
def payload(pocket_csv_file):
|
|
return {
|
|
'csv': pocket_csv_file,
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def override_settings_upload_path(settings, tmp_path):
|
|
settings.UPLOADS_PATH = tmp_path
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_ui_import_from_pocket_task_apply_async(mocker: pytest_mock.MockerFixture,
|
|
async_result,
|
|
) -> mock.Mock:
|
|
return UITestingService().mock_import_from_pocket_task_apply_async(
|
|
mocker=mocker, async_result=async_result,
|
|
)
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_ok(override_settings_upload_path,
|
|
authenticated_client: Client,
|
|
payload,
|
|
tmp_path,
|
|
mock_ui_import_from_pocket_task_apply_async,
|
|
account,
|
|
pocket_csv_file,
|
|
pocket_csv_content,
|
|
):
|
|
# When
|
|
result = authenticated_client.post(
|
|
reverse('ui.imports.pocket'),
|
|
data=payload,
|
|
)
|
|
|
|
# Then
|
|
asserts.assertRedirects(
|
|
result,
|
|
reverse('ui.associations.index'),
|
|
fetch_redirect_response=False,
|
|
)
|
|
|
|
uploaded_file_path = tmp_path / f'{account.pk}-{pocket_csv_file.name}'
|
|
assert uploaded_file_path.is_file() is True
|
|
assert uploaded_file_path.read_text() == pocket_csv_content.replace('\r\n', '\n')
|
|
|
|
mock_ui_import_from_pocket_task_apply_async.assert_called_once_with(
|
|
kwargs={
|
|
'account_uuid': account.pk,
|
|
'csv_path': str(uploaded_file_path),
|
|
},
|
|
)
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_invalid_all_empty(override_settings_upload_path,
|
|
authenticated_client: Client,
|
|
payload,
|
|
mock_ui_import_from_pocket_task_apply_async,
|
|
):
|
|
# Given
|
|
effective_payload = {
|
|
key: ''
|
|
for key
|
|
in payload.keys()
|
|
}
|
|
|
|
# When
|
|
result = authenticated_client.post(
|
|
reverse('ui.imports.pocket'),
|
|
data=effective_payload,
|
|
)
|
|
|
|
# Then
|
|
assert result.status_code == http.HTTPStatus.OK
|
|
|
|
assert 'csv' in result.context['form'].errors
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_invalid_all_missing(override_settings_upload_path,
|
|
authenticated_client: Client,
|
|
payload,
|
|
mock_ui_import_from_pocket_task_apply_async,
|
|
):
|
|
# Given
|
|
effective_payload = {}
|
|
|
|
# When
|
|
result = authenticated_client.post(
|
|
reverse('ui.imports.pocket'),
|
|
data=effective_payload,
|
|
)
|
|
|
|
# Then
|
|
assert result.status_code == http.HTTPStatus.OK
|
|
|
|
assert 'csv' in result.context['form'].errors
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_inactive_account(inactive_account_client: Client, payload):
|
|
# When
|
|
result = inactive_account_client.post(
|
|
reverse('ui.imports.pocket'),
|
|
data=payload,
|
|
)
|
|
|
|
# Then
|
|
asserts.assertRedirects(
|
|
result,
|
|
reverse(
|
|
'ui.accounts.login',
|
|
query=[('next', reverse('ui.imports.pocket'))],
|
|
),
|
|
fetch_redirect_response=False,
|
|
)
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_anonymous(client: Client, payload):
|
|
# When
|
|
result = client.post(
|
|
reverse('ui.imports.pocket'),
|
|
data=payload,
|
|
)
|
|
|
|
# Then
|
|
asserts.assertRedirects(
|
|
result,
|
|
reverse(
|
|
'ui.accounts.login',
|
|
query=[('next', reverse('ui.imports.pocket'))],
|
|
),
|
|
fetch_redirect_response=False,
|
|
)
|