test_backend_process.py

# Copyright 2020 ACSONE
# @author: Simone Orsi <simahawk@gmail.com>
# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).

import base64

from odoo import fields
from odoo.exceptions import UserError
from odoo.tools import mute_logger

from .common import EDIBackendCommonComponentRegistryTestCase
from .fake_components import FakeInputProcess


class EDIBackendTestCase(EDIBackendCommonComponentRegistryTestCase):
    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        cls._build_components(
            # TODO: test all components lookup
            cls,
            FakeInputProcess,
        )
        vals = {
            "model": cls.partner._name,
            "res_id": cls.partner.id,
            "exchange_file": base64.b64encode(b"1234"),
        }
        cls.record = cls.backend.create_record("test_csv_input", vals)

    def setUp(self):
        super().setUp()
        FakeInputProcess.reset_faked()

    def test_process_record(self):
        self.record.write({"edi_exchange_state": "input_received"})
        now = fields.Datetime.now()
        self.record.action_exchange_process()
        self.assertTrue(FakeInputProcess.check_called_for(self.record))
        self.assertRecordValues(
            self.record, [{"edi_exchange_state": "input_processed"}]
        )
        self.record.refresh()
        self.assertAlmostEqual(
            (self.record.exchanged_on - now).total_seconds(),
            0, places=2)

    def test_process_record_with_error(self):
        self.record.write({"edi_exchange_state": "input_received"})
        self.record._set_file_content("TEST %d" % self.record.id)
        self.record.with_context(
            test_break_process="OOPS! Something went wrong :("
        ).action_exchange_process()
        self.assertTrue(FakeInputProcess.check_called_for(self.record))
        self.assertRecordValues(
            self.record,
            [
                {
                    "edi_exchange_state": "input_processed_error",
                }
            ],
        )
        self.assertIn("OOPS! Something went wrong :(", self.record.exchange_error)

    @mute_logger("odoo.models.unlink")
    def test_process_no_file_record(self):
        self.record.write({"edi_exchange_state": "input_received"})
        self.record.exchange_file = False
        with self.assertRaises(UserError):
            self.record.action_exchange_process()

    def test_process_outbound_record(self):
        vals = {
            "model": self.partner._name,
            "res_id": self.partner.id,
        }
        record = self.backend.create_record("test_csv_output", vals)
        record._set_file_content("TEST %d" % record.id)
        with self.assertRaises(UserError):
            record.action_exchange_process()

    # TODO: test ack file are processed