# -*- coding: utf-8 -*- """Integration tests for the POS sign-or-block guard (require a running Odoo 17). odoo -d -i nacef_pos --test-enable --test-tags /nacef_pos --stop-after-init """ from odoo.exceptions import UserError from odoo.tests import tagged from odoo.addons.point_of_sale.tests.common import TestPoSCommon @tagged("post_install", "-at_install") class TestNacefPosGuard(TestPoSCommon): def setUp(self): super().setUp() self.config = self.basic_config self.product = self.create_product("NACEF Product", self.categ_basic, 10.0) # complete, schema-valid fiscal identity (E0803 requires these) self.company.write({ "nacef_matricule_fiscal": "1234567A", "nacef_accreditation_reference": "ACC-2026-0001", "nacef_nom_commercial": "Cafe Resto", "street": "12 Rue de l'Independance", "city": "Tunis", }) self.config.nacef_ce_serial = "CR12345678" def _seed_state(self, imdf, status="SYNCHRONIZED", offline=5): state = self.env["nacef.smdf.state"].get_or_create(imdf, self.company) state.write({ "status": status, "connection_state": "Online", "cert_request_status": "CERTIFICATE_GENERATED", "available_offline_tickets": offline, }) return state def _make_order(self): data = self.create_ui_order_data([(self.product, 1)]) self.env["pos.order"].create_from_ui([data]) return self.env["pos.order"].search([], order="id desc", limit=1) def test_synchronized_sale_is_signed(self): self.config.nacef_imdf = "IMDFPOS1000000" self._seed_state("IMDFPOS1000000", "SYNCHRONIZED") self.open_new_session() order = self._make_order() self.assertTrue(order.nacef_signed) self.assertTrue(order.nacef_ticket_identifier) self.assertTrue(order.nacef_fiscal_reference) self.assertEqual(order.nacef_imdf, "IMDFPOS1000000") self.assertTrue(order.nacef_online) def test_fiscal_reference_is_gapless(self): self.config.nacef_imdf = "IMDFPOS2000000" self._seed_state("IMDFPOS2000000", "SYNCHRONIZED") self.open_new_session() o1 = self._make_order() o2 = self._make_order() self.assertEqual(o1.nacef_fiscal_reference, "00000001") self.assertEqual(o2.nacef_fiscal_reference, "00000002") def test_sale_appends_audit_entry(self): self.config.nacef_imdf = "IMDFPOS3000000" self._seed_state("IMDFPOS3000000", "SYNCHRONIZED") self.open_new_session() self._make_order() logs = self.env["nacef.audit.log"].search( [("imdf", "=", "IMDFPOS3000000"), ("operation_code", "=", "CASHING")]) self.assertTrue(logs) def test_sale_blocked_when_factory(self): self.config.nacef_imdf = "IMDFBLOCK00000" self._seed_state("IMDFBLOCK00000", "FACTORY") self.open_new_session() with self.assertRaises(UserError): self._make_order() def test_sale_blocked_when_suspended(self): self.config.nacef_imdf = "IMDFSUSP000000" self._seed_state("IMDFSUSP000000", "SUSPENDED") self.open_new_session() with self.assertRaises(UserError): self._make_order() def test_no_imdf_means_no_fiscal_enforcement(self): # A POS with no IMDF configured is not under fiscal enforcement. self.config.nacef_imdf = False self.company.nacef_imdf = False self.open_new_session() order = self._make_order() self.assertFalse(order.nacef_signed) # -- inalterability (E1101/E1102) ---------------------------------- # def test_orders_are_sealed_and_chained(self): self.config.nacef_imdf = "IMDFSEAL000000" self._seed_state("IMDFSEAL000000", "SYNCHRONIZED") self.open_new_session() o1 = self._make_order() o2 = self._make_order() self.assertTrue(o1.nacef_hash) self.assertEqual(o1.nacef_secure_sequence, 1) self.assertEqual(o1.nacef_prev_hash, "GENESIS") self.assertEqual(o2.nacef_secure_sequence, 2) self.assertEqual(o2.nacef_prev_hash, o1.nacef_hash) ok, broken = self.env["pos.order"].nacef_verify_inalterability("IMDFSEAL000000") self.assertTrue(ok) self.assertIsNone(broken) def test_sealed_order_cannot_change_protected_field(self): self.config.nacef_imdf = "IMDFSEAL200000" self._seed_state("IMDFSEAL200000", "SYNCHRONIZED") self.open_new_session() order = self._make_order() with self.assertRaises(UserError): order.write({"amount_total": 999.0}) def test_tampering_is_detected(self): self.config.nacef_imdf = "IMDFTAMP000000" self._seed_state("IMDFTAMP000000", "SYNCHRONIZED") self.open_new_session() order = self._make_order() # bypass ORM guards to simulate a storage-layer edit self.env.cr.execute( "UPDATE pos_order SET amount_total = amount_total + 5 WHERE id = %s", (order.id,)) order.invalidate_recordset(["amount_total"]) ok, broken = self.env["pos.order"].nacef_verify_inalterability("IMDFTAMP000000") self.assertFalse(ok) self.assertEqual(broken, order.id) # -- closing turnover (E1204) + closed-period guard (E1202) --------- # def test_daily_closing_aggregates_turnover(self): self.config.nacef_imdf = "IMDFTURN000000" self._seed_state("IMDFTURN000000", "SYNCHRONIZED") self.open_new_session() self._make_order() self._make_order() closing = self.env["nacef.fiscal.closing"].open_period( "IMDFTURN000000", "daily") closing.action_close() self.assertEqual(closing.order_count, 2) self.assertEqual(closing.period_total, 20.0) self.assertEqual(closing.perpetual_total, 20.0) def test_sale_blocked_in_closed_period(self): self.config.nacef_imdf = "IMDFCLOSED0000" self._seed_state("IMDFCLOSED0000", "SYNCHRONIZED") self.open_new_session() # close today's period first, then a new sale must be refused (E1202) self.env["nacef.fiscal.closing"].open_period( "IMDFCLOSED0000", "daily").action_close() with self.assertRaises(UserError): self._make_order() # -- reprint lockdown / Ticket copie (E0505) ----------------------- # def test_duplicate_is_a_recorded_copy(self): self.config.nacef_imdf = "IMDFDUP0000000" self._seed_state("IMDFDUP0000000", "SYNCHRONIZED") self.open_new_session() order = self._make_order() order.action_nacef_create_duplicate() dup = self.env["pos.order"].search([ ("nacef_origin_reference", "=", order.nacef_fiscal_reference)]) self.assertNotEqual(dup.id, order.id) self.assertEqual(dup.nacef_operation_type, "DUPLICATE") self.assertEqual(dup.nacef_origin_reference, order.nacef_fiscal_reference) self.assertTrue(dup.nacef_signed) self.assertNotEqual(dup.nacef_fiscal_reference, order.nacef_fiscal_reference) # the copy is a fully re-issued, printable ticket (E0505): it keeps # the original's completed state and reuses its payment for display. self.assertEqual(dup.state, order.state) self.assertEqual(dup._nacef_display_payments(), order.payment_ids) # E0501.iv: "un numéro différent doit être affecté au ticket" — the # copy's own bookkeeping references must not collide with the # original's (distinct from the fiscal reference, already asserted # above via nacef_fiscal_reference). self.assertNotEqual(dup.pos_reference, order.pos_reference) self.assertNotEqual(dup.name, "/") # E0506(e): must show the actual cashier, not whoever triggered the # backend duplicate action. self.assertEqual(dup._nacef_cashier_name(), order._nacef_cashier_name()) # -- archiving / fiscal-admin export (E1301-E1304 / E1702) --------- # def test_full_archive_includes_signed_orders(self): from datetime import datetime, timedelta self.config.nacef_imdf = "IMDFARCH000000" self._seed_state("IMDFARCH000000", "SYNCHRONIZED") self.open_new_session() self._make_order() self._make_order() now = datetime.now() arch = self.env["nacef.fiscal.archive"].create({ "imdf": "IMDFARCH000000", "archive_type": "full", "date_start": now - timedelta(hours=1), "date_stop": now + timedelta(hours=1)}) arch.action_generate() self.assertEqual(arch.encaissement_count, 2) self.assertTrue(arch.verify_integrity()) self.assertTrue(arch.filename.endswith(".json")) # -- purge (E1401/E1402) ------------------------------------------- # def test_purge_archives_then_deletes_preserving_counters(self): from datetime import datetime, timedelta self.config.nacef_imdf = "IMDFPURG000000" self._seed_state("IMDFPURG000000", "SYNCHRONIZED") self.open_new_session() self._make_order() self._make_order() closing = self.env["nacef.fiscal.closing"].open_period( "IMDFPURG000000", "daily") closing.action_close() self.assertEqual(closing.perpetual_total, 20.0) purge = self.env["nacef.fiscal.purge"].create({ "imdf": "IMDFPURG000000", "cutoff_date": datetime.now() + timedelta(hours=1), "reason": "test"}) purge.action_purge() self.assertEqual(purge.state, "done") self.assertEqual(purge.purged_count, 2) self.assertEqual(purge.purged_total, 20.0) # E1401: archive was generated before deletion and holds the orders self.assertEqual(purge.archive_id.encaissement_count, 2) self.assertTrue(purge.archive_id.verify_integrity()) # encaissement removed self.assertEqual(self.env["pos.order"].search_count( [("nacef_imdf", "=", "IMDFPURG000000"), ("nacef_signed", "=", True)]), 0) # E1402: cumulative counters + traceability preserved self.assertTrue(closing.exists()) self.assertEqual(closing.perpetual_total, 20.0) self.assertEqual(purge.perpetual_before, purge.perpetual_after) self.assertTrue(self.env["nacef.audit.log"].search_count( [("imdf", "=", "IMDFPURG000000"), ("operation_code", "=", "PURGE")])) # -- ticket mentions (E0503/E0504/E0505) --------------------------- # def test_ticket_mentions(self): Order = self.env["pos.order"] self.assertEqual( Order.new({"nacef_transaction_type": "TRAINING"})._nacef_mention(), "TICKET FORMATION") self.assertEqual( Order.new({"nacef_operation_type": "REFUND"})._nacef_mention(), "TICKET REMBOURSEMENT") self.assertEqual( Order.new({"nacef_operation_type": "DUPLICATE"})._nacef_mention(), "TICKET COPIE") self.assertEqual(Order.new({})._nacef_mention(), "") def test_duplicate_carries_ticket_copie_mention(self): self.config.nacef_imdf = "IMDFMENT000000" self._seed_state("IMDFMENT000000", "SYNCHRONIZED") self.open_new_session() order = self._make_order() action = order.action_nacef_create_duplicate() dup = self.env["pos.order"].search([ ("nacef_origin_reference", "=", order.nacef_fiscal_reference)]) self.assertEqual(dup._nacef_mention(), "TICKET COPIE") # clicking the button must actually print: an ir.actions.report, # not a silent no-op (E0505: "génère un ticket de caisse"). self.assertEqual(action.get("type"), "ir.actions.report") self.assertEqual(action.get("report_name"), "nacef_pos.report_ticket_copy_document") def test_sync_response_exposes_mention_and_types(self): self.config.nacef_imdf = "IMDFMENT200000" self._seed_state("IMDFMENT200000", "SYNCHRONIZED") self.open_new_session() res = self.env["pos.order"].create_from_ui( [self.create_ui_order_data([(self.product, 1)])]) self.assertIn("nacef_mention", res[0]) self.assertIn("nacef_operation_type", res[0]) self.assertIn("nacef_transaction_type", res[0]) # -- receipt QR (E0506 v) ------------------------------------------ # def test_receipt_qr_is_real_png(self): import base64 self.config.nacef_imdf = "IMDFQR00000000" self._seed_state("IMDFQR00000000", "SYNCHRONIZED") self.open_new_session() order = self._make_order() self.assertTrue(order.nacef_qr_code) # a genuine, scannable PNG (not the raw mock text) self.assertEqual(base64.b64decode(order.nacef_qr_code)[:4], b"\x89PNG") # and it is exposed to the POS frontend for the receipt exported = self.env["pos.order"]._export_for_ui(order) self.assertEqual(exported["nacef_qr_code"], order.nacef_qr_code) self.assertEqual(exported["nacef_ticket_identifier"], order.nacef_ticket_identifier) # -- A3 ticket JSON (developers.nacef.tn "Ticket" schema) ---------- # def test_a3_ticket_structure(self): self.config.nacef_imdf = "IMDFA300000000" self._seed_state("IMDFA300000000", "SYNCHRONIZED") self.company.nacef_matricule_fiscal = "1234567L" self.company.nacef_nom_commercial = "Café Resto" tax = self.env["account.tax"].create({ "name": "TVA 7", "amount": 7.0, "amount_type": "percent", "type_tax_use": "sale", "price_include": True, "company_id": self.company.id}) prod = self.create_product("Café", self.categ_basic, 3.0, tax_ids=tax.ids) self.open_new_session() self.env["pos.order"].create_from_ui( [self.create_ui_order_data([(prod, 2)])]) order = self.env["pos.order"].search([], order="id desc", limit=1) self.assertTrue(order.nacef_signed) t = order._nacef_build_ticket_dict(order.nacef_fiscal_reference) for key in ("data_type", "version", "transaction", "merchant_identity", "customer_identity", "sale_details", "tax_summary", "general_discount", "additional_tax", "payment_details", "sale_summary", "delivery_details"): self.assertIn(key, t) # schema-fixed consts self.assertEqual(t["data_type"], "ncf.cashier.operation") self.assertEqual(t["version"], "1.1.4") # transaction / enums self.assertEqual(t["transaction"]["id"], order.nacef_fiscal_reference) self.assertEqual(t["transaction"]["operation"]["op_type"], "TICKET") self.assertEqual(t["transaction"]["operation"]["context"], "SALE") self.assertEqual(t["merchant_identity"]["id"], "1234567L") self.assertEqual(t["merchant_identity"]["id_type"], "MF") self.assertRegex(t["merchant_identity"]["taxpayer_establishment"] ["reference"], r"^\d{3}$") self.assertEqual(t["customer_identity"]["id_type"], "NP") # sale line: qty int, price integer millimes, A5 tax code 10 for 7% sd = t["sale_details"][0] self.assertEqual(sd["quantity"], 2) self.assertIsInstance(sd["product"]["price_pre_tax"], int) self.assertEqual(sd["taxation"][0]["type"], "percent") self.assertEqual(sd["taxation"][0]["tax_code"], "10") # payment method within the schema enum self.assertIn( t["payment_details"]["collection_details"][0]["method"], {"cash", "check", "bank_card", "restaurant_ticket", "mobile_payment", "contre_bon", "wire_transfer"}) # sale_summary: integers, and excl + tax == incl ss = t["sale_summary"] self.assertIsInstance(ss["total_incl_tax"], int) self.assertEqual(ss["total_excl_tax"] + ss["total_tax"], ss["total_incl_tax"]) self.assertTrue(any(x["tax_code"] == "10" for x in t["tax_summary"])) # the whole ticket must be JSON-serialisable and ASCII (E0803) import json json.dumps(t, ensure_ascii=True) # -- E1003: training mode requires an admin profile ---------------- # def _make_order_data(self, training=False): data = self.create_ui_order_data([(self.product, 1)]) if training: data["data"]["nacef_training"] = True return data def test_training_allowed_for_manager(self): # the test user is a POS manager -> training is honoured, not signed self.config.nacef_imdf = "IMDFTRN1000000" self._seed_state("IMDFTRN1000000", "SYNCHRONIZED") self.open_new_session() self.env["pos.order"].create_from_ui([self._make_order_data(training=True)]) order = self.env["pos.order"].search([], order="id desc", limit=1) self.assertEqual(order.nacef_transaction_type, "TRAINING") self.assertFalse(order.nacef_signed) def test_training_denied_for_non_manager(self): # a cashier (no pos manager group) cannot activate training: the flag is # refused, the sale is a real signed transaction, and it is audited. self.config.nacef_imdf = "IMDFTRN2000000" self._seed_state("IMDFTRN2000000", "SYNCHRONIZED") cashier = self.env["res.users"].create({ "name": "Cashier Only", "login": "nacef_cashier_only", "groups_id": [(6, 0, [ self.env.ref("base.group_user").id, self.env.ref("point_of_sale.group_pos_user").id])], }) self.open_new_session() self.env["pos.order"].with_user(cashier).create_from_ui( [self._make_order_data(training=True)]) order = self.env["pos.order"].search([], order="id desc", limit=1) self.assertNotEqual(order.nacef_transaction_type, "TRAINING") self.assertTrue(order.nacef_signed) self.assertTrue(self.env["nacef.audit.log"].search_count( [("imdf", "=", "IMDFTRN2000000"), ("operation_code", "=", "TRAINING"), ("ptype", "=", "ERREUR")])) # -- E0404/E0505: reprint lockdown is persisted -------------------- # def test_mark_printed_persists_and_audits(self): self.config.nacef_imdf = "IMDFPRN0000000" self._seed_state("IMDFPRN0000000", "SYNCHRONIZED") self.open_new_session() order = self._make_order() self.assertFalse(order.nacef_printed) marked = self.env["pos.order"].nacef_mark_printed([order.id]) self.assertEqual(marked, 1) self.assertTrue(order.nacef_printed) # exposed to the UI so a reopened order stays locked self.assertTrue(self.env["pos.order"]._export_for_ui(order)["nacef_printed"]) # a second call is a no-op (already printed) self.assertEqual(self.env["pos.order"].nacef_mark_printed([order.id]), 0) self.assertTrue(self.env["nacef.audit.log"].search_count( [("imdf", "=", "IMDFPRN0000000"), ("operation_code", "=", "PRINTER")])) # -- offline-ticket budget is persisted back to the S-MDF state ---- # def test_offline_budget_decrements_and_persists(self): self.config.nacef_imdf = "IMDFOFF0000000" state = self._seed_state("IMDFOFF0000000", "SYNCHRONIZED", offline=3) state.connection_state = "Offline" self.open_new_session() self._make_order() self.assertEqual(state.available_offline_tickets, 2) self._make_order() self.assertEqual(state.available_offline_tickets, 1) # -- E0506(l): SA/AA fiscal-advantage indicator -------------------- # def test_fiscal_advantage_sa_then_aa(self): self.config.nacef_imdf = "IMDFADV0000000" self._seed_state("IMDFADV0000000", "SYNCHRONIZED") self.open_new_session() # default: no attestation -> SA, fiscal_advantages omitted o1 = self._make_order() self.assertEqual(o1._nacef_fiscal_advantage(), "SA") self.assertNotIn("fiscal_advantages", o1._nacef_build_ticket_dict( o1.nacef_fiscal_reference)["customer_identity"]) # attestation reference supplied at the POS -> AA data = self.create_ui_order_data([(self.product, 1)]) data["data"]["nacef_fiscal_advantage_ref"] = "ATT-2026-0001" self.env["pos.order"].create_from_ui([data]) o2 = self.env["pos.order"].search([], order="id desc", limit=1) self.assertEqual(o2._nacef_fiscal_advantage(), "AA") self.assertEqual(o2._nacef_build_ticket_dict(o2.nacef_fiscal_reference) ["customer_identity"]["fiscal_advantages"], "ATT-2026-0001") self.assertEqual(self.env["pos.order"]._export_for_ui(o2) ["nacef_fiscal_advantage"], "AA") # -- E0803: A3 ticket is JSON-schema validated before signing ------ # def test_ticket_is_schema_validated(self): from odoo.addons.nacef_smdf_client.api import ( validate, validate_ticket, SchemaError) # a real order builds a schema-conformant ticket (no violations) self.config.nacef_imdf = "IMDFVAL0000000" self._seed_state("IMDFVAL0000000", "SYNCHRONIZED") self.open_new_session() order = self._make_order() self.assertEqual( validate(order._nacef_build_ticket_dict(order.nacef_fiscal_reference)), []) # a malformed ticket is rejected with self.assertRaises(SchemaError): validate_ticket({"data_type": "x", "version": "0.0.0"}) # -- E0802: family_code (A4) is schema-valid (2..8 chars) ---------- # def test_family_code_within_schema_length(self): self.config.nacef_imdf = "IMDFFAM0000001" self._seed_state("IMDFFAM0000001", "SYNCHRONIZED") poscat = self.env["pos.category"].create( {"name": "Sandwicherie Gourmande Deluxe"}) # long name prod = self.create_product("Sandwich", self.categ_basic, 5.0) prod.pos_categ_ids = [(6, 0, poscat.ids)] self.open_new_session() self.env["pos.order"].create_from_ui( [self.create_ui_order_data([(prod, 1)])]) order = self.env["pos.order"].search([], order="id desc", limit=1) self.assertTrue(order.nacef_signed) # would raise E0803 if > 8 chars fc = order._nacef_build_ticket_dict( order.nacef_fiscal_reference)["sale_details"][0]["product"]["family_code"] self.assertTrue(2 <= len(fc) <= 8, fc) # an explicit A4 code (when configured) is used verbatim poscat.nacef_family_code = "F001" self.assertEqual(order._nacef_family_code(prod), "F001") # -- [1203]: automatic daily closing at midnight ------------------- # def test_cron_auto_close_daily(self): self.config.nacef_imdf = "IMDFCRON000000" self._seed_state("IMDFCRON000000", "SYNCHRONIZED") self.open_new_session() self._make_order() self.env["nacef.fiscal.closing"].cron_auto_close_daily() closed = self.env["nacef.fiscal.closing"].search([ ("imdf", "=", "IMDFCRON000000"), ("closing_type", "=", "daily"), ("state", "=", "closed")]) self.assertTrue(closed, "auto daily closing did not close the period") # -- E0501/E0502: PROFORMA ticket is creatable --------------------- # def test_proforma_ticket(self): self.config.nacef_imdf = "IMDFPROF000000" self._seed_state("IMDFPROF000000", "SYNCHRONIZED") self.open_new_session() data = self.create_ui_order_data([(self.product, 1)]) data["data"]["nacef_proforma"] = True self.env["pos.order"].create_from_ui([data]) order = self.env["pos.order"].search([], order="id desc", limit=1) self.assertEqual(order.nacef_operation_type, "PROFORMA") self.assertEqual(order._nacef_mention(), "PRO-FORMA") self.assertTrue(order.nacef_signed) # a pro-forma is a quote, not revenue -> excluded from turnover closing = self.env["nacef.fiscal.closing"].open_period( "IMDFPROF000000", "daily") closing.action_close() self.assertEqual(closing.order_count, 0)