138 lines
6.1 KiB
Python
138 lines
6.1 KiB
Python
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from django.contrib.auth.models import Permission
|
|
from django.urls import reverse
|
|
|
|
from commerce.integrations.moysklad import MoySkladError
|
|
from commerce.models import IntegrationConnection, IntegrationSyncRun, SourceSystem
|
|
from commerce.services.credentials import decrypt_credentials, encrypt_credentials
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
@pytest.fixture
|
|
def admin_client(client, user_factory):
|
|
client.force_login(user_factory(staff=True, superuser=True))
|
|
return client
|
|
|
|
|
|
@pytest.fixture
|
|
def connection():
|
|
source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
|
|
return IntegrationConnection.objects.create(
|
|
source=source, provider="moysklad", auth_type="token", status="ready",
|
|
encrypted_credentials=encrypt_credentials({"token": "existing-test-token"}),
|
|
)
|
|
|
|
|
|
def connection_data(**overrides):
|
|
return {"provider": "moysklad", "auth_type": "token", "auto_sync": "on",
|
|
"sync_interval_minutes": "15", **overrides}
|
|
|
|
|
|
def action(client, connection, name):
|
|
return client.post(reverse("admin:commerce_integrationconnection_changelist"), {
|
|
"action": name, "_selected_action": [str(connection.pk)],
|
|
})
|
|
|
|
|
|
def test_admin_creates_encrypted_connection_and_preserves_blank_credentials(admin_client):
|
|
response = admin_client.post(
|
|
reverse("admin:commerce_integrationconnection_add"),
|
|
connection_data(token="admin-test-token"),
|
|
)
|
|
assert response.status_code == 302
|
|
connection = IntegrationConnection.objects.get()
|
|
assert connection.source.code == "moysklad"
|
|
assert connection.created_by.is_superuser
|
|
assert "admin-test-token" not in connection.encrypted_credentials
|
|
assert decrypt_credentials(connection) == {"token": "admin-test-token"}
|
|
encrypted = connection.encrypted_credentials
|
|
url = reverse("admin:commerce_integrationconnection_change", args=[connection.pk])
|
|
page = admin_client.get(url)
|
|
assert page.status_code == 200
|
|
assert b"admin-test-token" not in page.content
|
|
assert encrypted.encode() not in page.content
|
|
response = admin_client.post(url, connection_data(sync_interval_minutes="30"))
|
|
assert response.status_code == 302
|
|
connection.refresh_from_db()
|
|
assert connection.encrypted_credentials == encrypted
|
|
assert connection.sync_interval_minutes == 30
|
|
assert connection.next_sync_at is not None
|
|
|
|
|
|
def test_admin_requires_complete_new_auth_and_preserves_password_spaces(admin_client, connection):
|
|
url = reverse("admin:commerce_integrationconnection_change", args=[connection.pk])
|
|
response = admin_client.post(url, connection_data(auth_type="basic"))
|
|
assert response.status_code == 200
|
|
assert {"login", "password"} <= response.context["adminform"].form.errors.keys()
|
|
connection.refresh_from_db()
|
|
assert connection.auth_type == "token"
|
|
response = admin_client.post(url, connection_data(
|
|
auth_type="basic", login="admin@test", password=" password with spaces ",
|
|
))
|
|
assert response.status_code == 302
|
|
connection.refresh_from_db()
|
|
assert decrypt_credentials(connection) == {"login": "admin@test", "password": " password with spaces "}
|
|
assert connection.status == "disconnected"
|
|
|
|
|
|
def test_admin_rejects_basic_auth_for_wildberries(admin_client):
|
|
response = admin_client.post(reverse("admin:commerce_integrationconnection_add"),
|
|
connection_data(provider="wildberries", auth_type="basic", login="test", password="test"))
|
|
assert response.status_code == 200
|
|
assert "auth_type" in response.context["adminform"].form.errors
|
|
assert not IntegrationConnection.objects.exists()
|
|
|
|
|
|
def test_admin_test_action_updates_state_and_reports_failure(admin_client, connection):
|
|
with patch("commerce.admin.test_moysklad_connection", return_value={
|
|
"account_name": "Test company", "account_id": "test-account",
|
|
}):
|
|
assert action(admin_client, connection, "test_connections").status_code == 302
|
|
connection.refresh_from_db()
|
|
assert connection.status == "ready"
|
|
assert connection.account_name == "Test company"
|
|
assert connection.last_tested_at is not None
|
|
assert connection.source.metadata["account_id"] == "test-account"
|
|
with patch("commerce.admin.test_moysklad_connection", side_effect=MoySkladError("Test failure")):
|
|
assert action(admin_client, connection, "test_connections").status_code == 302
|
|
connection.refresh_from_db()
|
|
assert connection.status == "error"
|
|
assert connection.last_error == "Test failure"
|
|
|
|
|
|
def test_admin_sync_enqueues_once_and_blocks_editing_active_connection(
|
|
admin_client, connection, django_capture_on_commit_callbacks,
|
|
):
|
|
with patch("commerce.tasks.sync_connection_run.apply_async") as publish, \
|
|
django_capture_on_commit_callbacks(execute=True):
|
|
assert action(admin_client, connection, "sync_connections").status_code == 302
|
|
assert action(admin_client, connection, "sync_connections").status_code == 302
|
|
publish.assert_called_once()
|
|
assert IntegrationSyncRun.objects.filter(connection=connection).count() == 1
|
|
connection.refresh_from_db()
|
|
assert connection.status == "queued"
|
|
url = reverse("admin:commerce_integrationconnection_change", args=[connection.pk])
|
|
response = admin_client.post(url, connection_data(token="replacement"))
|
|
assert response.status_code == 200
|
|
assert response.context["adminform"].form.non_field_errors()
|
|
connection.refresh_from_db()
|
|
assert decrypt_credentials(connection) == {"token": "existing-test-token"}
|
|
|
|
|
|
def test_view_only_staff_cannot_run_admin_actions(client, user_factory, connection):
|
|
user = user_factory(staff=True)
|
|
user.user_permissions.add(Permission.objects.get(
|
|
content_type__app_label="commerce", codename="view_integrationconnection",
|
|
))
|
|
client.force_login(user)
|
|
assert client.get(reverse(
|
|
"admin:commerce_integrationconnection_change", args=[connection.pk],
|
|
)).status_code == 200
|
|
with patch("commerce.admin.enqueue_connection_sync") as enqueue:
|
|
action(client, connection, "sync_connections")
|
|
enqueue.assert_not_called()
|
|
assert not IntegrationSyncRun.objects.exists()
|