Quentin Mondot
b63b8ba885
Some checks failed
pre-commit / pre-commit (pull_request) Failing after 1m45s
279 lines
11 KiB
Python
279 lines
11 KiB
Python
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
|
|
|
|
import base64
|
|
import json
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
from werkzeug.exceptions import NotFound
|
|
from werkzeug.wrappers import Response
|
|
|
|
from odoo.addons.survey.tests import common
|
|
|
|
|
|
class TestSurveyFileCommon(common.TestSurveyCommon):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.file_content = b"Hello, this is a test file."
|
|
self.file_b64 = base64.b64encode(self.file_content).decode()
|
|
self.file_name = "test_document.pdf"
|
|
|
|
self.question_file = self._add_question(
|
|
self.page_0,
|
|
"Upload your document",
|
|
"file",
|
|
constr_mandatory=False,
|
|
survey_id=self.survey.id,
|
|
)
|
|
self.question_file_required = self._add_question(
|
|
self.page_0,
|
|
"Upload your required document",
|
|
"file",
|
|
constr_mandatory=True,
|
|
survey_id=self.survey.id,
|
|
)
|
|
|
|
def _create_answer_with_file(self):
|
|
answer = self._add_answer(self.survey, False, email="test@example.com")
|
|
line = self.env["survey.user_input.line"].create({
|
|
"user_input_id": answer.id,
|
|
"question_id": self.question_file.id,
|
|
"answer_type": "file",
|
|
"skipped": False,
|
|
"value_file": self.file_b64,
|
|
"value_file_fname": self.file_name,
|
|
})
|
|
return answer, line
|
|
|
|
|
|
class TestSurveyFileSaveLines(TestSurveyFileCommon):
|
|
"""Test the save_lines method for file question type."""
|
|
|
|
def test_save_file_answer(self):
|
|
"""Submitting a file stores base64 data and filename."""
|
|
answer = self._add_answer(self.survey, self.survey_manager.partner_id)
|
|
file_json = json.dumps({"data": self.file_b64, "name": self.file_name})
|
|
|
|
answer.save_lines(self.question_file, file_json)
|
|
|
|
line = answer.user_input_line_ids.filtered(
|
|
lambda l: l.question_id == self.question_file
|
|
)
|
|
self.assertEqual(len(line), 1)
|
|
self.assertEqual(line.answer_type, "file")
|
|
self.assertFalse(line.skipped)
|
|
self.assertEqual(line.value_file, self.file_b64.encode())
|
|
self.assertEqual(line.value_file_fname, self.file_name)
|
|
|
|
def test_save_file_skipped(self):
|
|
"""Submitting empty answer marks the line as skipped."""
|
|
answer = self._add_answer(self.survey, self.survey_manager.partner_id)
|
|
|
|
answer.save_lines(self.question_file, "")
|
|
|
|
line = answer.user_input_line_ids.filtered(
|
|
lambda l: l.question_id == self.question_file
|
|
)
|
|
self.assertEqual(len(line), 1)
|
|
self.assertTrue(line.skipped)
|
|
self.assertFalse(line.answer_type)
|
|
|
|
def test_save_file_update_existing(self):
|
|
"""Submitting a new file updates the existing answer line."""
|
|
answer = self._add_answer(self.survey, self.survey_manager.partner_id)
|
|
file_json_1 = json.dumps({"data": self.file_b64, "name": "first.pdf"})
|
|
answer.save_lines(self.question_file, file_json_1)
|
|
|
|
new_b64 = base64.b64encode(b"Updated content").decode()
|
|
file_json_2 = json.dumps({"data": new_b64, "name": "second.pdf"})
|
|
answer.save_lines(self.question_file, file_json_2)
|
|
|
|
lines = answer.user_input_line_ids.filtered(
|
|
lambda l: l.question_id == self.question_file
|
|
)
|
|
self.assertEqual(len(lines), 1, "Should update, not create a second line")
|
|
self.assertEqual(lines.value_file, new_b64.encode())
|
|
self.assertEqual(lines.value_file_fname, "second.pdf")
|
|
|
|
def test_save_file_then_skip(self):
|
|
"""Uploading a file then submitting empty marks line as skipped."""
|
|
answer = self._add_answer(self.survey, self.survey_manager.partner_id)
|
|
file_json = json.dumps({"data": self.file_b64, "name": self.file_name})
|
|
answer.save_lines(self.question_file, file_json)
|
|
|
|
answer.save_lines(self.question_file, "")
|
|
|
|
line = answer.user_input_line_ids.filtered(
|
|
lambda l: l.question_id == self.question_file
|
|
)
|
|
self.assertEqual(len(line), 1)
|
|
self.assertTrue(line.skipped)
|
|
|
|
|
|
class TestSurveyFileConstraints(TestSurveyFileCommon):
|
|
"""Test _check_file_constraints validation logic."""
|
|
|
|
def test_no_constraints(self):
|
|
"""No max_file_size and no allowed_extensions: any file passes."""
|
|
self.question_file.write({"max_file_size": 0, "allowed_extensions": False})
|
|
answer = self._add_answer(self.survey, self.survey_manager.partner_id)
|
|
file_json = json.dumps({"data": self.file_b64, "name": "anything.exe"})
|
|
answer.save_lines(self.question_file, file_json)
|
|
line = answer.user_input_line_ids.filtered(
|
|
lambda l: l.question_id == self.question_file
|
|
)
|
|
self.assertFalse(line.skipped)
|
|
|
|
def test_file_within_size_limit(self):
|
|
"""File smaller than max_file_size passes."""
|
|
self.question_file.write({"max_file_size": 10})
|
|
answer = self._add_answer(self.survey, self.survey_manager.partner_id)
|
|
file_json = json.dumps({"data": self.file_b64, "name": self.file_name})
|
|
answer.save_lines(self.question_file, file_json)
|
|
line = answer.user_input_line_ids.filtered(
|
|
lambda l: l.question_id == self.question_file
|
|
)
|
|
self.assertFalse(line.skipped)
|
|
|
|
def test_file_exceeds_size_limit(self):
|
|
"""File larger than max_file_size raises ValidationError."""
|
|
self.question_file.write({"max_file_size": 1})
|
|
large_content = b"x" * (1 * 1024 * 1024 + 1)
|
|
large_b64 = base64.b64encode(large_content).decode()
|
|
answer = self._add_answer(self.survey, self.survey_manager.partner_id)
|
|
file_json = json.dumps({"data": large_b64, "name": self.file_name})
|
|
with self.assertRaises(Exception):
|
|
answer.save_lines(self.question_file, file_json)
|
|
|
|
def test_allowed_extension_passes(self):
|
|
"""File with an allowed extension passes."""
|
|
self.question_file.write({"allowed_extensions": ".pdf,.docx"})
|
|
answer = self._add_answer(self.survey, self.survey_manager.partner_id)
|
|
file_json = json.dumps({"data": self.file_b64, "name": "report.docx"})
|
|
answer.save_lines(self.question_file, file_json)
|
|
line = answer.user_input_line_ids.filtered(
|
|
lambda l: l.question_id == self.question_file
|
|
)
|
|
self.assertFalse(line.skipped)
|
|
|
|
def test_disallowed_extension_raises(self):
|
|
"""File with a disallowed extension raises ValidationError."""
|
|
self.question_file.write({"allowed_extensions": ".pdf,.docx"})
|
|
answer = self._add_answer(self.survey, self.survey_manager.partner_id)
|
|
file_json = json.dumps({"data": self.file_b64, "name": "script.exe"})
|
|
with self.assertRaises(Exception):
|
|
answer.save_lines(self.question_file, file_json)
|
|
|
|
def test_both_constraints_valid(self):
|
|
"""File respecting both size and extension constraints passes."""
|
|
self.question_file.write({"max_file_size": 10, "allowed_extensions": ".pdf"})
|
|
answer = self._add_answer(self.survey, self.survey_manager.partner_id)
|
|
file_json = json.dumps({"data": self.file_b64, "name": self.file_name})
|
|
answer.save_lines(self.question_file, file_json)
|
|
line = answer.user_input_line_ids.filtered(
|
|
lambda l: l.question_id == self.question_file
|
|
)
|
|
self.assertFalse(line.skipped)
|
|
|
|
|
|
class TestSurveyFileDisplayName(TestSurveyFileCommon):
|
|
"""Test the display_name computation for file answer lines."""
|
|
|
|
def test_display_name_filename(self):
|
|
answer = self._add_answer(self.survey, self.survey_manager.partner_id)
|
|
line = self.env["survey.user_input.line"].create({
|
|
"user_input_id": answer.id,
|
|
"question_id": self.question_file.id,
|
|
"answer_type": "file",
|
|
"skipped": False,
|
|
"value_file": self.file_b64,
|
|
"value_file_fname": "my_filename.pdf",
|
|
})
|
|
self.assertEqual(line.display_name, "my_filename.pdf")
|
|
|
|
|
|
class TestSurveyFileDownloadController(TestSurveyFileCommon):
|
|
"""Test the file download controller logic with mocked request."""
|
|
|
|
def _call_download(self, survey_token, answer_token, line_id):
|
|
"""Call the controller method with a mocked request context."""
|
|
from odoo.addons.survey_extra_fields.controllers.main import (
|
|
SurveyExtraFieldsController,
|
|
)
|
|
|
|
mock_request = MagicMock()
|
|
mock_request.env = self.env
|
|
mock_request.not_found.return_value = NotFound()
|
|
mock_request.make_response.side_effect = (
|
|
lambda content, headers=None: Response(content, headers=headers)
|
|
)
|
|
|
|
controller = SurveyExtraFieldsController()
|
|
with patch(
|
|
"odoo.addons.survey_extra_fields.controllers.main.request",
|
|
mock_request,
|
|
):
|
|
try:
|
|
return controller.survey_file_download(
|
|
survey_token, answer_token, line_id
|
|
)
|
|
except NotFound:
|
|
return NotFound()
|
|
|
|
def test_download_valid(self):
|
|
"""Valid tokens return the file content."""
|
|
answer, line = self._create_answer_with_file()
|
|
|
|
response = self._call_download(
|
|
self.survey.access_token, answer.access_token, line.id
|
|
)
|
|
|
|
self.assertIsInstance(response, Response)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.data, self.file_content)
|
|
|
|
def test_download_invalid_survey_token(self):
|
|
"""Invalid survey token returns not_found."""
|
|
answer, line = self._create_answer_with_file()
|
|
|
|
response = self._call_download(
|
|
"invalid-token", answer.access_token, line.id
|
|
)
|
|
|
|
self.assertIsInstance(response, NotFound)
|
|
|
|
def test_download_invalid_answer_token(self):
|
|
"""Invalid answer token returns not_found."""
|
|
answer, line = self._create_answer_with_file()
|
|
|
|
response = self._call_download(
|
|
self.survey.access_token, "invalid-token", line.id
|
|
)
|
|
|
|
self.assertIsInstance(response, NotFound)
|
|
|
|
def test_download_line_not_belonging_to_answer(self):
|
|
"""Accessing a line from another answer returns not_found."""
|
|
answer, line = self._create_answer_with_file()
|
|
other_answer = self._add_answer(self.survey, False, email="other@example.com")
|
|
|
|
response = self._call_download(
|
|
self.survey.access_token, other_answer.access_token, line.id
|
|
)
|
|
|
|
self.assertIsInstance(response, NotFound)
|
|
|
|
def test_download_no_file(self):
|
|
"""Accessing a line without file data returns not_found."""
|
|
answer = self._add_answer(self.survey, False, email="test@example.com")
|
|
line = self.env["survey.user_input.line"].create({
|
|
"user_input_id": answer.id,
|
|
"question_id": self.question_file.id,
|
|
"skipped": True,
|
|
})
|
|
|
|
response = self._call_download(
|
|
self.survey.access_token, answer.access_token, line.id
|
|
)
|
|
|
|
self.assertIsInstance(response, NotFound)
|