mirror of
https://github.com/eitchtee/WYGIWYH.git
synced 2026-04-23 17:18:44 +02:00
feat(api): add endpoints for importing files and getting account balance
This commit is contained in:
Herculino Trotta
404
app/apps/api/tests/test_imports.py
Normal file
404
app/apps/api/tests/test_imports.py
Normal file
@@ -0,0 +1,404 @@
|
||||
from io import BytesIO
|
||||
from unittest.mock import patch
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.core.files.uploadedfile import SimpleUploadedFile
|
||||
from django.test import TestCase, override_settings
|
||||
from rest_framework import status
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from apps.import_app.models import ImportProfile, ImportRun
|
||||
|
||||
|
||||
@override_settings(
|
||||
STORAGES={
|
||||
"default": {"BACKEND": "django.core.files.storage.FileSystemStorage"},
|
||||
"staticfiles": {
|
||||
"BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage"
|
||||
},
|
||||
},
|
||||
WHITENOISE_AUTOREFRESH=True,
|
||||
)
|
||||
class ImportAPITests(TestCase):
|
||||
"""Tests for the Import API endpoint"""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up test data"""
|
||||
User = get_user_model()
|
||||
self.user = User.objects.create_user(
|
||||
email="testuser@test.com", password="testpass123"
|
||||
)
|
||||
self.client = APIClient()
|
||||
self.client.force_authenticate(user=self.user)
|
||||
|
||||
# Create a basic import profile with minimal valid YAML config
|
||||
self.profile = ImportProfile.objects.create(
|
||||
name="Test Profile",
|
||||
version=ImportProfile.Versions.VERSION_1,
|
||||
yaml_config="""
|
||||
file_type: csv
|
||||
date_format: "%Y-%m-%d"
|
||||
column_mapping:
|
||||
date:
|
||||
source: date
|
||||
description:
|
||||
source: description
|
||||
amount:
|
||||
source: amount
|
||||
transaction_type:
|
||||
detection_method: always_expense
|
||||
is_paid:
|
||||
detection_method: always_paid
|
||||
account:
|
||||
source: account
|
||||
match_field: name
|
||||
""",
|
||||
)
|
||||
|
||||
@patch("apps.import_app.tasks.process_import.defer")
|
||||
@patch("django.core.files.storage.FileSystemStorage.save")
|
||||
@patch("django.core.files.storage.FileSystemStorage.path")
|
||||
def test_create_import_success(self, mock_path, mock_save, mock_defer):
|
||||
"""Test successful file upload creates ImportRun and queues task"""
|
||||
mock_save.return_value = "test_file.csv"
|
||||
mock_path.return_value = "/usr/src/app/temp/test_file.csv"
|
||||
|
||||
csv_content = b"date,description,amount,account\n2025-01-01,Test,100,Main"
|
||||
file = SimpleUploadedFile(
|
||||
"test_file.csv", csv_content, content_type="text/csv"
|
||||
)
|
||||
|
||||
response = self.client.post(
|
||||
"/api/import/import/",
|
||||
{"profile_id": self.profile.id, "file": file},
|
||||
format="multipart",
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
|
||||
self.assertIn("import_run_id", response.data)
|
||||
self.assertEqual(response.data["status"], "queued")
|
||||
|
||||
# Verify ImportRun was created
|
||||
import_run = ImportRun.objects.get(id=response.data["import_run_id"])
|
||||
self.assertEqual(import_run.profile, self.profile)
|
||||
self.assertEqual(import_run.file_name, "test_file.csv")
|
||||
|
||||
# Verify task was deferred
|
||||
mock_defer.assert_called_once_with(
|
||||
import_run_id=import_run.id,
|
||||
file_path="/usr/src/app/temp/test_file.csv",
|
||||
user_id=self.user.id,
|
||||
)
|
||||
|
||||
def test_create_import_missing_profile(self):
|
||||
"""Test request without profile_id returns 400"""
|
||||
csv_content = b"date,description,amount\n2025-01-01,Test,100"
|
||||
file = SimpleUploadedFile(
|
||||
"test_file.csv", csv_content, content_type="text/csv"
|
||||
)
|
||||
|
||||
response = self.client.post(
|
||||
"/api/import/import/",
|
||||
{"file": file},
|
||||
format="multipart",
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
self.assertIn("profile_id", response.data)
|
||||
|
||||
def test_create_import_missing_file(self):
|
||||
"""Test request without file returns 400"""
|
||||
response = self.client.post(
|
||||
"/api/import/import/",
|
||||
{"profile_id": self.profile.id},
|
||||
format="multipart",
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
self.assertIn("file", response.data)
|
||||
|
||||
def test_create_import_invalid_profile(self):
|
||||
"""Test request with non-existent profile returns 400"""
|
||||
csv_content = b"date,description,amount\n2025-01-01,Test,100"
|
||||
file = SimpleUploadedFile(
|
||||
"test_file.csv", csv_content, content_type="text/csv"
|
||||
)
|
||||
|
||||
response = self.client.post(
|
||||
"/api/import/import/",
|
||||
{"profile_id": 99999, "file": file},
|
||||
format="multipart",
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
self.assertIn("profile_id", response.data)
|
||||
|
||||
@patch("apps.import_app.tasks.process_import.defer")
|
||||
@patch("django.core.files.storage.FileSystemStorage.save")
|
||||
@patch("django.core.files.storage.FileSystemStorage.path")
|
||||
def test_create_import_xlsx(self, mock_path, mock_save, mock_defer):
|
||||
"""Test successful XLSX file upload"""
|
||||
mock_save.return_value = "test_file.xlsx"
|
||||
mock_path.return_value = "/usr/src/app/temp/test_file.xlsx"
|
||||
|
||||
# Create a simple XLSX-like content (just for the upload test)
|
||||
xlsx_content = BytesIO(b"PK\x03\x04") # XLSX files start with PK header
|
||||
file = SimpleUploadedFile(
|
||||
"test_file.xlsx",
|
||||
xlsx_content.getvalue(),
|
||||
content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||||
)
|
||||
|
||||
response = self.client.post(
|
||||
"/api/import/import/",
|
||||
{"profile_id": self.profile.id, "file": file},
|
||||
format="multipart",
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
|
||||
self.assertIn("import_run_id", response.data)
|
||||
|
||||
def test_unauthenticated_request(self):
|
||||
"""Test unauthenticated request returns 403"""
|
||||
unauthenticated_client = APIClient()
|
||||
|
||||
csv_content = b"date,description,amount\n2025-01-01,Test,100"
|
||||
file = SimpleUploadedFile(
|
||||
"test_file.csv", csv_content, content_type="text/csv"
|
||||
)
|
||||
|
||||
response = unauthenticated_client.post(
|
||||
"/api/import/import/",
|
||||
{"profile_id": self.profile.id, "file": file},
|
||||
format="multipart",
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
||||
|
||||
|
||||
@override_settings(
|
||||
STORAGES={
|
||||
"default": {"BACKEND": "django.core.files.storage.FileSystemStorage"},
|
||||
"staticfiles": {
|
||||
"BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage"
|
||||
},
|
||||
},
|
||||
WHITENOISE_AUTOREFRESH=True,
|
||||
)
|
||||
class ImportProfileAPITests(TestCase):
|
||||
"""Tests for the Import Profile API endpoints"""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up test data"""
|
||||
User = get_user_model()
|
||||
self.user = User.objects.create_user(
|
||||
email="testuser@test.com", password="testpass123"
|
||||
)
|
||||
self.client = APIClient()
|
||||
self.client.force_authenticate(user=self.user)
|
||||
|
||||
self.profile1 = ImportProfile.objects.create(
|
||||
name="Profile 1",
|
||||
version=ImportProfile.Versions.VERSION_1,
|
||||
yaml_config="""
|
||||
file_type: csv
|
||||
date_format: "%Y-%m-%d"
|
||||
column_mapping:
|
||||
date:
|
||||
source: date
|
||||
description:
|
||||
source: description
|
||||
amount:
|
||||
source: amount
|
||||
transaction_type:
|
||||
detection_method: always_expense
|
||||
is_paid:
|
||||
detection_method: always_paid
|
||||
account:
|
||||
source: account
|
||||
match_field: name
|
||||
""",
|
||||
)
|
||||
self.profile2 = ImportProfile.objects.create(
|
||||
name="Profile 2",
|
||||
version=ImportProfile.Versions.VERSION_1,
|
||||
yaml_config="""
|
||||
file_type: csv
|
||||
date_format: "%Y-%m-%d"
|
||||
column_mapping:
|
||||
date:
|
||||
source: date
|
||||
description:
|
||||
source: description
|
||||
amount:
|
||||
source: amount
|
||||
transaction_type:
|
||||
detection_method: always_income
|
||||
is_paid:
|
||||
detection_method: always_unpaid
|
||||
account:
|
||||
source: account
|
||||
match_field: name
|
||||
""",
|
||||
)
|
||||
|
||||
def test_list_profiles(self):
|
||||
"""Test listing all profiles"""
|
||||
response = self.client.get("/api/import/profiles/")
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertEqual(response.data["count"], 2)
|
||||
self.assertEqual(len(response.data["results"]), 2)
|
||||
|
||||
def test_retrieve_profile(self):
|
||||
"""Test retrieving a specific profile"""
|
||||
response = self.client.get(f"/api/import/profiles/{self.profile1.id}/")
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertEqual(response.data["id"], self.profile1.id)
|
||||
self.assertEqual(response.data["name"], "Profile 1")
|
||||
self.assertIn("yaml_config", response.data)
|
||||
|
||||
def test_retrieve_nonexistent_profile(self):
|
||||
"""Test retrieving a non-existent profile returns 404"""
|
||||
response = self.client.get("/api/import/profiles/99999/")
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
|
||||
|
||||
def test_profiles_unauthenticated(self):
|
||||
"""Test unauthenticated request returns 403"""
|
||||
unauthenticated_client = APIClient()
|
||||
response = unauthenticated_client.get("/api/import/profiles/")
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
||||
|
||||
|
||||
@override_settings(
|
||||
STORAGES={
|
||||
"default": {"BACKEND": "django.core.files.storage.FileSystemStorage"},
|
||||
"staticfiles": {
|
||||
"BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage"
|
||||
},
|
||||
},
|
||||
WHITENOISE_AUTOREFRESH=True,
|
||||
)
|
||||
class ImportRunAPITests(TestCase):
|
||||
"""Tests for the Import Run API endpoints"""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up test data"""
|
||||
User = get_user_model()
|
||||
self.user = User.objects.create_user(
|
||||
email="testuser@test.com", password="testpass123"
|
||||
)
|
||||
self.client = APIClient()
|
||||
self.client.force_authenticate(user=self.user)
|
||||
|
||||
self.profile1 = ImportProfile.objects.create(
|
||||
name="Profile 1",
|
||||
version=ImportProfile.Versions.VERSION_1,
|
||||
yaml_config="""
|
||||
file_type: csv
|
||||
date_format: "%Y-%m-%d"
|
||||
column_mapping:
|
||||
date:
|
||||
source: date
|
||||
description:
|
||||
source: description
|
||||
amount:
|
||||
source: amount
|
||||
transaction_type:
|
||||
detection_method: always_expense
|
||||
is_paid:
|
||||
detection_method: always_paid
|
||||
account:
|
||||
source: account
|
||||
match_field: name
|
||||
""",
|
||||
)
|
||||
self.profile2 = ImportProfile.objects.create(
|
||||
name="Profile 2",
|
||||
version=ImportProfile.Versions.VERSION_1,
|
||||
yaml_config="""
|
||||
file_type: csv
|
||||
date_format: "%Y-%m-%d"
|
||||
column_mapping:
|
||||
date:
|
||||
source: date
|
||||
description:
|
||||
source: description
|
||||
amount:
|
||||
source: amount
|
||||
transaction_type:
|
||||
detection_method: always_income
|
||||
is_paid:
|
||||
detection_method: always_unpaid
|
||||
account:
|
||||
source: account
|
||||
match_field: name
|
||||
""",
|
||||
)
|
||||
|
||||
# Create import runs
|
||||
self.run1 = ImportRun.objects.create(
|
||||
profile=self.profile1,
|
||||
file_name="file1.csv",
|
||||
status=ImportRun.Status.FINISHED,
|
||||
)
|
||||
self.run2 = ImportRun.objects.create(
|
||||
profile=self.profile1,
|
||||
file_name="file2.csv",
|
||||
status=ImportRun.Status.QUEUED,
|
||||
)
|
||||
self.run3 = ImportRun.objects.create(
|
||||
profile=self.profile2,
|
||||
file_name="file3.csv",
|
||||
status=ImportRun.Status.FINISHED,
|
||||
)
|
||||
|
||||
def test_list_all_runs(self):
|
||||
"""Test listing all runs"""
|
||||
response = self.client.get("/api/import/runs/")
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertEqual(response.data["count"], 3)
|
||||
self.assertEqual(len(response.data["results"]), 3)
|
||||
|
||||
def test_list_runs_by_profile(self):
|
||||
"""Test filtering runs by profile_id"""
|
||||
response = self.client.get(f"/api/import/runs/?profile_id={self.profile1.id}")
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertEqual(response.data["count"], 2)
|
||||
for run in response.data["results"]:
|
||||
self.assertEqual(run["profile"], self.profile1.id)
|
||||
|
||||
def test_list_runs_by_other_profile(self):
|
||||
"""Test filtering runs by another profile_id"""
|
||||
response = self.client.get(f"/api/import/runs/?profile_id={self.profile2.id}")
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertEqual(response.data["count"], 1)
|
||||
self.assertEqual(response.data["results"][0]["profile"], self.profile2.id)
|
||||
|
||||
def test_retrieve_run(self):
|
||||
"""Test retrieving a specific run"""
|
||||
response = self.client.get(f"/api/import/runs/{self.run1.id}/")
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertEqual(response.data["id"], self.run1.id)
|
||||
self.assertEqual(response.data["file_name"], "file1.csv")
|
||||
self.assertEqual(response.data["status"], "FINISHED")
|
||||
|
||||
def test_retrieve_nonexistent_run(self):
|
||||
"""Test retrieving a non-existent run returns 404"""
|
||||
response = self.client.get("/api/import/runs/99999/")
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
|
||||
|
||||
def test_runs_unauthenticated(self):
|
||||
"""Test unauthenticated request returns 403"""
|
||||
unauthenticated_client = APIClient()
|
||||
response = unauthenticated_client.get("/api/import/runs/")
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
||||
Reference in New Issue
Block a user