2 Commits

Author SHA1 Message Date
0d521278c7 [FIX] survey_record_generation: partner_id/email on survey.user_input could be wrong
New version : 16.0.1.0.3

_mark_done() only set partner_id/email when it created a *new* res.partner.
If the answer already had a partner_id/email (e.g. inherited from the Odoo
user logged in when the /survey/start link was opened), that value was kept
even though the record creation matched or created a different, correct
contact from the participant's own answers.

Now we always sync partner_id/email to the res.partner actually matched or created
from the participant's answers.

Add a migration to relink partner_id/email on existing done submissions.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-22 18:50:33 +02:00
7a1eacc16c [FIX] survey_record_generation : allow false values in boolean question 2026-09-22 18:50:27 +02:00
13 changed files with 489 additions and 356 deletions

View File

@@ -11,7 +11,7 @@ Allow to create record of any model when sending the form :
* Associate question with fields
* For x2m fields : Associate values to questions
""",
"version": "16.0.1.0.2",
"version": "16.0.1.0.3",
"license": "AGPL-3",
"author": "Elabore",
"website": "https://www.elabore.coop",

View File

@@ -0,0 +1,159 @@
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
"""Relink survey.user_input.partner_id/email to the contact that actually
matches each participant's own answers.
Before this version, survey.user_input.partner_id/email were only updated
when _mark_done() created a *new* res.partner. If the answer already had a
partner_id/email (e.g. inherited from the Odoo user who was logged in when
the /survey/start link was opened, or from some other unidentified cause),
_mark_done() silently kept that value even when the survey_record_creation
config found or created a different, correct contact from the participant's
own answers. This backfill re-runs that resolution for every already-done
answer and fixes partner_id/email accordingly.
It never re-triggers the other survey.user_input._mark_done() overrides
(crm lead / event registration / notification modules, ...): it only calls
the res.partner-matching helpers directly, so it can't create duplicate
leads, registrations, etc. for historical submissions.
A res.partner is expected to already exist for every done submission (it
was necessarily found or created the first time _mark_done() ran), so this
backfill never creates one. It first looks for the survey.generated.record
row _mark_done() logged when it created that partner, and falls back to
re-deriving the match through find_existing_record()/
find_duplicate_if_there_are_fields_with_unicity_check() when there is no
such row (existing partner matched instead of created) or it points to a
partner that has since been deleted (e.g. merged into another contact). If
neither approach finds anything, the record is left untouched and logged
for manual review instead.
"""
import logging
from odoo import SUPERUSER_ID, api
_logger = logging.getLogger(__name__)
def migrate(cr, version):
env = api.Environment(cr, SUPERUSER_ID, {})
user_input_model = env["survey.user_input"]
surveys_with_partner_creation = env["survey.record.creation"].search(
[("model_id.model", "=", "res.partner")]
).survey_id
user_inputs = user_input_model.search(
[
("survey_id", "in", surveys_with_partner_creation.ids),
("state", "=", "done"),
]
)
_logger.info(
"survey_record_generation: relinking partner_id/email on %d done "
"survey.user_input records",
len(user_inputs),
)
fixed_count = 0
not_found_count = 0
for user_input in user_inputs:
record_creations = user_input.survey_id.survey_record_creation_ids.filtered(
lambda rc: rc.model_id.model == "res.partner"
).sorted("sequence")
record_creation = record_creations[:1]
if not record_creation:
continue
# 1) Prefer the res.partner this very submission created, if any
# (the first one, by id, in the rare case there is more than one):
# it's the exact record _mark_done() produced for this answer, no
# guessing involved. This is also the only option for surveys whose
# record creation has neither update_existing_records nor any
# unicity_check field configured, since find_existing_record()/
# find_duplicate...() can then never find anything (nothing to
# search on).
record = False
for generated in user_input.generated_record_ids.sorted("id"):
if (
generated.survey_record_creation_id == record_creation
and generated.created_record_id
and generated.created_record_id._name == "res.partner"
and generated.created_record_id.exists()
):
# The referenced partner may have since been deleted (e.g.
# merged into another contact): in that case it's not usable
# and we fall through to the search-based lookup below.
record = generated.created_record_id
break
# 2) Otherwise, this submission matched an already-existing partner
# instead of creating one (find_existing_record()/find_duplicate...()
# branch of _mark_done()): re-derive it the same way.
if not record:
# Only compute the fields find_existing_record()/find_duplicate...()
# actually read (the search field, and any unicity_check field),
# not every field of the record creation: other fields (e.g. a
# "record" reference to a model defined in a module that depends
# on this one) may not be loadable yet at this point of the
# upgrade, and are useless here anyway since this backfill never
# writes to res.partner.
needed_field_names = set()
if (
record_creation.update_existing_records
and record_creation.field_to_retrieve_existing_records
):
needed_field_names.add(
record_creation.field_to_retrieve_existing_records.name
)
unicity_field_values = record_creation.field_values_ids.filtered(
lambda field_value: field_value.unicity_check
)
needed_field_names.update(unicity_field_values.mapped("field_id.name"))
vals = {}
for field_value in record_creation.field_values_ids:
if field_value.field_id.name not in needed_field_names:
continue
value, __ = user_input_model.get_value_based_on_value_origin(
field_value=field_value,
user_input=user_input,
created_records={},
model="res.partner",
other_record_fields_to_update=[],
)
vals[field_value.field_id.name] = value
existing_record = user_input_model.find_existing_record(
record_creation, vals
)
duplicate = (
user_input_model.find_duplicate_if_there_are_fields_with_unicity_check(
"res.partner", record_creation, vals
)
)
record = duplicate or existing_record
if not record:
_logger.warning(
"survey_record_generation: no existing res.partner found for "
"survey.user_input %s while backfilling partner_id/email, "
"leaving it untouched",
user_input.id,
)
not_found_count += 1
continue
if user_input.partner_id != record or user_input.email != record.email:
user_input.partner_id = record.id
user_input.email = record.email
fixed_count += 1
_logger.info(
"survey_record_generation: fixed %d survey.user_input records "
"(%d without a matching res.partner)",
fixed_count,
not_found_count,
)

View File

@@ -39,6 +39,7 @@ class SurveyUserInput(models.Model):
for user_input in self:
created_records = {}
other_record_fields_to_update: list[SurveyRecordCreationFieldValues] = []
partner_linked_in_this_run = False
record_creation: SurveyRecordCreation
for (
@@ -85,8 +86,6 @@ class SurveyUserInput(models.Model):
try:
with self.env.cr.savepoint():
record = self.env[model].create(vals)
if model == "res.partner" and not self.partner_id:
self.partner_id = record.id
except Exception:
# This a broad exception because it could be IntegrityError,
# EmptyNamesError in case partner_firstname is installed etc...
@@ -103,6 +102,17 @@ class SurveyUserInput(models.Model):
}
)
if model == "res.partner" and not partner_linked_in_this_run:
# Always reflect the partner actually matched/created from
# this participant's own answers, even if partner_id/email
# were already set on the answer (e.g. inherited from the
# logged-in user when the survey link was opened). Only the
# first res.partner record creation of this run wins, in
# case several are configured on the same survey.
user_input.partner_id = record.id
user_input.email = record.email
partner_linked_in_this_run = True
created_records[record_creation.id] = record
# update linked record
@@ -315,28 +325,44 @@ class SurveyUserInput(models.Model):
}
)
@staticmethod
def get_boolean_value(answer_value_char: str, question_title: str) -> bool:
# Below code is a trick to be able to use "simple_choice" question
# with values 'yes' and 'no' and transform it to boolean.
if boolean_value := answer_value_char in [
"1",
"True",
"true",
"Oui",
"oui",
"Yes",
"yes",
]:
return boolean_value
else:
raise UserError(
_(
"[Survey record generation] The boolean value %s(value)s "
"is not supported (for question %(question)s)."
)
% {
"value": answer_value_char,
"question": question_title,
}
def _get_boolean_true_values(self):
"""Tokens interpreted as true, normalized to lowercase.
Instance method so that a third-party module or a localization can
extend the list without rewriting get_boolean_value.
"""
return {"1", "true", "vrai", "yes", "y", "oui", "o", "on", "x"}
def _get_boolean_false_values(self):
"""Tokens interpreted as false, normalized to lowercase."""
return {"0", "false", "faux", "no", "n", "non", "off", ""}
def get_boolean_value(self, answer_value_char: str, question_title: str) -> bool:
"""Convert the technical value of an answer into a boolean.
Allows a boolean field to be filled from a "simple_choice" question
whose suggested answers carry yes/no values.
"""
if not answer_value_char:
# Empty answer: an unset boolean is false, this is not a
# configuration error.
return False
token = str(answer_value_char).strip().casefold()
if token in self._get_boolean_true_values():
return True
if token in self._get_boolean_false_values():
return False
raise UserError(
_(
"[Survey record generation] The boolean value %(value)s is not "
"supported (for question %(question)s)"
)
% {
"value": answer_value_char,
"question": question_title,
}
)

View File

@@ -1 +1,2 @@
from . import test_survey_record_creation
from . import test_end_migration_16_0_1_0_3

View File

@@ -0,0 +1,156 @@
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
import importlib.util
import os
from odoo.addons.survey.tests.common import SurveyCase
def _load_end_migration():
migration_path = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"migrations",
"16.0.1.0.3",
"end-migration.py",
)
spec = importlib.util.spec_from_file_location(
"survey_record_generation_end_migration_16_0_1_0_3", migration_path
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
class TestEndMigration160103(SurveyCase):
"""Covers the use cases the 16.0.1.0.3 end-migration backfill must
handle: relinking survey.user_input.partner_id/email to the contact
that actually matches each participant's own answers, for submissions
whose partner_id/email were left wrong before the fix in this version.
"""
def setUp(self):
super().setUp()
self.migration = _load_end_migration()
self.survey = self.env["survey.survey"].create({"title": "Test Survey"})
self.question_name = self._add_question(
page=None,
name="Name",
qtype="char_box",
survey_id=self.survey.id,
sequence=1,
)
self.res_partner_model = self.env["ir.model"]._get("res.partner")
self.survey_record_creation = self.env["survey.record.creation"].create(
{
"name": "Contact",
"survey_id": self.survey.id,
"model_id": self.res_partner_model.id,
}
)
self.name_field = self.env["ir.model.fields"].search(
[("model", "=", "res.partner"), ("name", "=", "name")]
)
self.env["survey.record.creation.field.values"].create(
{
"survey_record_creation_id": self.survey_record_creation.id,
"survey_id": self.survey.id,
"model_id": self.res_partner_model.id,
"field_id": self.name_field.id,
"value_origin": "question",
"question_id": self.question_name.id,
}
)
def _submit_answer(self, name):
answer = self._add_answer(survey=self.survey, partner=False, email=False)
self._add_answer_line(
question=self.question_name, answer=answer, answer_value=name
)
answer._mark_done()
return answer
def _corrupt_with_wrong_partner(self, answer):
# Simulate a pre-fix record: partner_id/email point to someone
# unrelated to this participant's own answers.
wrong_partner = self.env["res.partner"].create({"name": "Wrong Partner"})
answer.write({"partner_id": wrong_partner.id, "email": "wrong@test.fr"})
return wrong_partner
def test_migrate_uses_generated_record_when_available(self):
# The submission created its own res.partner: the migration must
# relink partner_id/email to that exact contact.
answer = self._submit_answer("Jean")
jean = self.env["res.partner"].search([("name", "=", "Jean")])
self._corrupt_with_wrong_partner(answer)
self.migration.migrate(self.cr, "16.0.1.0.2")
answer.invalidate_recordset()
self.assertEqual(answer.partner_id, jean)
self.assertEqual(answer.email, jean.email)
def test_migrate_falls_back_to_search_when_no_generated_record(self):
# The submission matched an already-existing partner instead of
# creating one: no survey.generated.record row exists for it, so the
# migration must fall back to re-deriving the match through search.
jean = self.env["res.partner"].create(
{"name": "Jean", "email": "jean@test.fr"}
)
self.survey_record_creation.write(
{
"update_existing_records": True,
"field_to_retrieve_existing_records": self.name_field.id,
}
)
answer = self._submit_answer("Jean")
self.assertFalse(answer.generated_record_ids)
self._corrupt_with_wrong_partner(answer)
self.migration.migrate(self.cr, "16.0.1.0.2")
answer.invalidate_recordset()
self.assertEqual(answer.partner_id, jean)
self.assertEqual(answer.email, jean.email)
def test_migrate_falls_back_to_search_when_generated_partner_was_deleted(self):
# The res.partner the submission created has since been deleted
# (e.g. merged into another contact): the migration must not crash
# on the stale reference and must fall back to search instead.
self.survey_record_creation.write(
{
"update_existing_records": True,
"field_to_retrieve_existing_records": self.name_field.id,
}
)
answer = self._submit_answer("Jean")
first_jean = self.env["res.partner"].search([("name", "=", "Jean")])
first_jean.unlink()
# A new contact with the same name takes over.
second_jean = self.env["res.partner"].create(
{"name": "Jean", "email": "jean2@test.fr"}
)
self._corrupt_with_wrong_partner(answer)
self.migration.migrate(self.cr, "16.0.1.0.2")
answer.invalidate_recordset()
self.assertEqual(answer.partner_id, second_jean)
self.assertEqual(answer.email, second_jean.email)
def test_migrate_leaves_answer_untouched_when_no_partner_can_be_found(self):
# No survey.generated.record row (e.g. historical data predating
# that log) and no way to search (neither update_existing_records
# nor unicity_check configured): the migration must leave
# partner_id/email as they are instead of guessing or crashing.
answer = self._submit_answer("Jean")
answer.generated_record_ids.unlink()
wrong_partner = self._corrupt_with_wrong_partner(answer)
self.migration.migrate(self.cr, "16.0.1.0.2")
answer.invalidate_recordset()
self.assertEqual(answer.partner_id, wrong_partner)
self.assertEqual(answer.email, "wrong@test.fr")

View File

@@ -930,6 +930,71 @@ class TestSurveyRecordCreation(SurveyCase):
partner = self.env["res.partner"].search([("name", "=", "Jean")])
self.assertEqual(self.answer.partner_id, partner)
def test_partner_id_and_email_are_corrected_when_creating_a_new_partner(self):
# A wrong partner_id/email is already set on the answer (e.g.
# inherited from the Odoo user who was logged in when the
# /survey/start link was opened). The record creation must still
# fill partner_id/email with the contact actually created from the
# participant's own answers, overriding that wrong value.
wrong_partner = self.env["res.partner"].create({"name": "Wrong Partner"})
self.answer = self._add_answer(
survey=self.survey, partner=wrong_partner, email="wrong@test.fr"
)
self._add_answer_line(
question=self.question_name, answer=self.answer, answer_value="Jean"
)
self.answer._mark_done()
partner = self.env["res.partner"].search([("name", "=", "Jean")])
self.assertEqual(self.answer.partner_id, partner)
self.assertNotEqual(self.answer.partner_id, wrong_partner)
self.assertEqual(self.answer.email, partner.email)
def test_partner_id_and_email_are_corrected_when_matching_an_existing_partner(self):
# Same as above, but this time the record creation matches an
# already-existing partner (via update_existing_records) instead of
# creating a new one: partner_id/email must still be corrected to
# that matched partner, not left as the wrong pre-existing value.
jean = self.env["res.partner"].create({"name": "Jean", "email": "jean@test.fr"})
wrong_partner = self.env["res.partner"].create({"name": "Wrong Partner"})
self.survey_record_creation.write(
{
"update_existing_records": True,
"field_to_retrieve_existing_records": self.name_field.id,
}
)
self.answer = self._add_answer(
survey=self.survey, partner=wrong_partner, email="wrong@test.fr"
)
self._add_answer_line(
question=self.question_name, answer=self.answer, answer_value="Jean"
)
self.answer._mark_done()
self.assertEqual(self.answer.partner_id, jean)
self.assertEqual(self.answer.email, jean.email)
def test_partner_id_and_email_are_corrected_when_matching_a_duplicate(self):
# Same as above, but through the unicity_check/find_duplicate branch
# instead of update_existing_records.
self.name_survey_record_creation_field_values.unicity_check = True
jean = self.env["res.partner"].create({"name": "Jean", "email": "jean@test.fr"})
wrong_partner = self.env["res.partner"].create({"name": "Wrong Partner"})
self.answer = self._add_answer(
survey=self.survey, partner=wrong_partner, email="wrong@test.fr"
)
self._add_answer_line(
question=self.question_name, answer=self.answer, answer_value="Jean"
)
self.answer._mark_done()
self.assertEqual(self.answer.partner_id, jean)
self.assertEqual(self.answer.email, jean.email)
def test_partner_id_in_survey_input_is_filled_up_by_first_contact_record_creation(self):
# In this test, we verify that when creating several contacts with the same survey,
# the 1st created contact is used to fill up survey_input.partner_id
@@ -976,3 +1041,58 @@ class TestSurveyRecordCreation(SurveyCase):
partner = self.env["res.partner"].search([("name", "=", "Jean")])
self.assertEqual(self.answer.partner_id, partner)
def test_partner_id_is_not_overridden_by_second_contact_record_creation(self):
# When several res.partner record creations run on the same survey,
# partner_id/email must be corrected to match the first contact
# created, and the second contact created must not override that
# choice, even though the answer started with a wrong
# partner_id/email.
self.second_question_name = self._add_question(
page=None,
name="Name of second person",
qtype="char_box",
survey_id=self.survey.id,
sequence=1,
)
self.second_contact_creation = self.env["survey.record.creation"].create(
{
"name": "Contact 2",
"survey_id": self.survey.id,
"model_id": self.res_partner_model.id,
}
)
self.env["survey.record.creation.field.values"].create(
{
"survey_record_creation_id": self.second_contact_creation.id,
"survey_id": self.survey.id,
"model_id": self.res_partner_model.id,
"field_id": self.name_field.id,
"value_origin": "question",
"question_id": self.second_question_name.id,
}
)
wrong_partner = self.env["res.partner"].create({"name": "Wrong Partner"})
self.answer = self._add_answer(
survey=self.survey, partner=wrong_partner, email="wrong@test.fr"
)
self._add_answer_line(
question=self.question_name,
answer=self.answer,
answer_value="Jean",
)
self._add_answer_line(
question=self.second_question_name,
answer=self.answer,
answer_value="Jeanne",
)
self.answer._mark_done()
jean = self.env["res.partner"].search([("name", "=", "Jean")])
jeanne = self.env["res.partner"].search([("name", "=", "Jeanne")])
self.assertEqual(self.answer.partner_id, jean)
self.assertNotEqual(self.answer.partner_id, jeanne)
self.assertEqual(self.answer.email, jean.email)

View File

@@ -1 +0,0 @@
from . import models

View File

@@ -1,20 +0,0 @@
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
{
"name": "Survey record generation extra fields",
"summary": "Allow to save files in any model when submitting the survey",
"description": "",
"version": "16.0.1.0.0",
"license": "AGPL-3",
"author": "Elabore",
"website": "https://elabore.coop",
"maintainer": "Elabore",
"category": "Survey",
"depends": ["survey_record_generation", "survey_extra_fields"],
"data": [
],
"installable": True,
# Install this module automatically if all dependency have been previously
# and independently installed. Used for synergetic or glue modules.
"auto_install": True,
}

View File

@@ -1,2 +0,0 @@
from . import survey_record_creation_field_values
from . import survey_user_input

View File

@@ -1,65 +0,0 @@
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
from odoo import _, api, fields, models
from odoo.exceptions import ValidationError
BINARY_QUESTION_TYPES = ["file"]
class SurveyRecordCreationFieldValues(models.Model):
_inherit = "survey.record.creation.field.values"
# add file type to the domain
field_id = fields.Many2one(
domain="[('model_id','=',model_id),"
"('ttype','in',['char','selection','text','html','integer','float',"
"'date','datetime','many2one','many2many','boolean','binary'])]",
)
fixed_value_binary = fields.Binary("Value")
fixed_value_binary_fname = fields.Char("Filename")
@api.depends("survey_id")
def _compute_allowed_question_ids(self):
super()._compute_allowed_question_ids()
for record in self.filtered(lambda r: r.field_id.ttype == "binary"):
if not record.survey_id:
record.allowed_question_ids = False
continue
record.allowed_question_ids = self.env["survey.question"].search([
("survey_id", "=", record.survey_id.id),
("question_type", "in", BINARY_QUESTION_TYPES),
])
@api.onchange("fixed_value_binary", "fixed_value_binary_fname")
def _compute_displayed_value(self):
# display filename instead of file content
binary_values = self.filtered(
lambda r: r.field_id.ttype == "binary" and r.value_origin == "fixed"
)
super(
SurveyRecordCreationFieldValues, self - binary_values
)._compute_displayed_value()
for record in binary_values:
if record.fixed_value_binary_fname:
record.displayed_value = record.fixed_value_binary_fname
elif record.fixed_value_binary:
record.displayed_value = _("File")
else:
record.displayed_value = None
def clean_values(self):
res = super().clean_values()
self.fixed_value_binary = None
self.fixed_value_binary_fname = None
return res
@api.constrains("field_id", "unicity_check")
def _check_binary_unicity_check(self):
for record in self:
if record.field_id.ttype == "binary" and record.unicity_check:
raise ValidationError(_(
"The unicity constraint cannot be used on the binary field "
"%s: binary fields are stored as attachments and cannot be "
"searched.", record.field_id.display_name
))

View File

@@ -1,26 +0,0 @@
from odoo import models
FILE_QUESTION_TYPES = ["file"]
class SurveyUserInput(models.Model):
_inherit = "survey.user_input"
def get_value_from_user_answer(self, field_value, user_input):
# we check for file type before parent checks
if field_value.question_id.question_type not in FILE_QUESTION_TYPES:
return super().get_value_from_user_answer(field_value, user_input)
line = self._get_file_answer_line(field_value, user_input)
if not line:
return None
return line.with_context(bin_size=False).value_file or None
def _get_file_answer_line(self, field_value, user_input):
# return possible lines for this question, otherwise empty recordset
lines = user_input.user_input_line_ids.filtered(
lambda line: line.question_id == field_value.question_id
)
if not lines or lines[0].skipped:
return self.env["survey.user_input.line"]
return lines[0]

View File

@@ -1 +0,0 @@
from . import test_record_generation_extra_fields

View File

@@ -1,214 +0,0 @@
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
import base64
from odoo.tests import tagged
from odoo.addons.survey.tests.common import SurveyCase
@tagged("post_install", "-at_install")
class TestSurveyRecordCreationFileCommon(SurveyCase):
"""Socle commun : un questionnaire avec une question "nom" et une question
"fichier", mappées vers res.partner.name et vers un champ binaire
personnalisé res.partner.x_document.
"""
# Champ binaire cible, créé dynamiquement (voir _setup_target_field).
# Le préfixe x_ est imposé par la contrainte sur state="manual".
FILE_FIELD = "x_document"
def setUp(self):
super().setUp()
# `_add_answer_line` s'appuie sur cette table pour déduire le couple
# (answer_type, champ de stockage) à partir du type de question.
# Le type "file" vient de survey_extra_fields, il n'y figure pas.
self._type_match["file"] = ("file", "value_file")
self.res_partner_model = self.env["ir.model"]._get("res.partner")
self._setup_target_field()
self._setup_survey()
self._setup_record_creation()
self.file_content = b"%PDF-1.4 contenu de test"
self.file_b64 = base64.b64encode(self.file_content)
self.file_name = "document.pdf"
# ------------------------------------------------------------------
# Setup helpers
# ------------------------------------------------------------------
def _setup_target_field(self):
"""Crée un champ binaire personnalisé sur res.partner.
Le module pont doit fonctionner avec n'importe quel champ binaire de
n'importe quel modèle : on teste donc avec un champ personnalisé
plutôt qu'avec un champ existant comme image_1920, dont le type Image
applique un redimensionnement et refuserait un PDF.
"""
self.env["ir.model.fields"].create({
"name": self.FILE_FIELD,
"model_id": self.res_partner_model.id,
"field_description": "Document justificatif",
"ttype": "binary",
"state": "manual",
})
# create() a déjà appelé setup_models() + init_models(), mais ne lève
# pas ce drapeau (seul unlink() le fait). Sans lui, le reset_changes()
# enregistré en addClassCleanup par TransactionCase ne resynchronise
# pas le registry et le champ survit au rollback.
self.registry.registry_invalidated = True
self.file_field = self.env["ir.model.fields"]._get(
"res.partner", self.FILE_FIELD
)
self.name_field = self.env["ir.model.fields"]._get("res.partner", "name")
def _setup_survey(self):
self.survey = self.env["survey.survey"].create({"title": "Test Survey"})
# res.partner.name est obligatoire : sans cette question, la création
# échoue en IntegrityError avant même d'atteindre le fichier.
self.question_name = self._add_question(
page=None,
name="Name",
qtype="char_box",
survey_id=self.survey.id,
sequence=1,
)
self.question_document = self._add_question(
page=None,
name="Document",
qtype="file",
constr_mandatory=False,
survey_id=self.survey.id,
sequence=2,
)
def _setup_record_creation(self):
self.survey_record_creation = self.env["survey.record.creation"].create({
"name": "Contact",
"survey_id": self.survey.id,
"model_id": self.res_partner_model.id,
})
self.name_field_values = self._map_question(
self.name_field, self.question_name
)
self.file_field_values = self._map_question(
self.file_field, self.question_document
)
def _map_question(self, field, question, **kwargs):
values = {
"survey_record_creation_id": self.survey_record_creation.id,
"survey_id": self.survey.id,
"model_id": self.res_partner_model.id,
"field_id": field.id,
"value_origin": "question",
"question_id": question.id,
}
values.update(kwargs)
return self.env["survey.record.creation.field.values"].create(values)
# ------------------------------------------------------------------
# Answer helpers
# ------------------------------------------------------------------
def _answer_survey(self, name="Jean", file_b64=None, file_name=None):
"""Répond au questionnaire et le valide. Retourne le user_input."""
answer = self._add_answer(
survey=self.survey, partner=False, email="jean@test.fr"
)
if name is not None:
self._add_answer_line(
question=self.question_name, answer=answer, answer_value=name
)
if file_b64 is not None:
self._add_answer_line(
question=self.question_document,
answer=answer,
answer_value=file_b64,
value_file_fname=file_name or self.file_name,
)
answer._mark_done()
return answer
def _read_binary(self, record, fname=None):
"""Relit un champ binaire depuis la base, hors cache et hors bin_size."""
fname = fname or self.FILE_FIELD
record.invalidate_recordset([fname])
return record.with_context(bin_size=False)[fname]
class TestSurveyRecordCreationFile(TestSurveyRecordCreationFileCommon):
def test_file_is_copied_to_binary_field(self):
"""Le fichier répondu se retrouve dans le champ binaire cible."""
self._answer_survey(file_b64=self.file_b64)
partner = self.env["res.partner"].search([("name", "=", "Jean")])
self.assertEqual(len(partner), 1)
stored = self._read_binary(partner)
self.assertTrue(stored, "Le champ binaire devrait être renseigné")
self.assertEqual(base64.b64decode(stored), self.file_content)
def test_file_stored_as_attachment(self):
"""Un champ binaire manuel est stocké en pièce jointe, pas en colonne.
Le domaine doit mentionner res_field, sinon ir.attachment._search
injecte ('res_field', '=', False) et masque la pièce jointe.
"""
self._answer_survey(file_b64=self.file_b64)
partner = self.env["res.partner"].search([("name", "=", "Jean")])
attachment = self.env["ir.attachment"].sudo().search([
("res_model", "=", "res.partner"),
("res_id", "=", partner.id),
("res_field", "=", self.FILE_FIELD),
])
self.assertEqual(len(attachment), 1)
self.assertEqual(attachment.raw, self.file_content)
def test_no_file_answered(self):
"""Sans réponse fichier, l'enregistrement est créé, champ binaire vide."""
self._answer_survey(file_b64=None)
partner = self.env["res.partner"].search([("name", "=", "Jean")])
self.assertEqual(len(partner), 1)
self.assertFalse(self._read_binary(partner))
def test_skipped_file_answer(self):
"""Une ligne de réponse explicitement passée ne remplit pas le champ."""
answer = self._add_answer(
survey=self.survey, partner=False, email="jean@test.fr"
)
self._add_answer_line(
question=self.question_name, answer=answer, answer_value="Jean"
)
self.env["survey.user_input.line"].create({
"user_input_id": answer.id,
"question_id": self.question_document.id,
"skipped": True,
})
answer._mark_done()
partner = self.env["res.partner"].search([("name", "=", "Jean")])
self.assertEqual(len(partner), 1)
self.assertFalse(self._read_binary(partner))
def test_file_updates_existing_record(self):
"""En mode mise à jour, le fichier est écrit sur le partner existant."""
partner = self.env["res.partner"].create({"name": "Jean"})
self.survey_record_creation.write({
"update_existing_records": True,
"field_to_retrieve_existing_records": self.name_field.id,
})
self._answer_survey(file_b64=self.file_b64)
self.assertEqual(
self.env["res.partner"].search_count([("name", "=", "Jean")]), 1
)
self.assertEqual(
base64.b64decode(self._read_binary(partner)), self.file_content
)