# -*- coding: utf-8 -*- """Standalone unit tests for the S-MDF client library. Pure Python — no Odoo required. Run with:: cd addons/nacef_smdf_client && python3 -m unittest tests.test_api """ import base64 import unittest # Import the library whether run as an Odoo addon or standalone from the module dir. try: # standalone (cwd == module dir) from api import ( AgentRestNacefClient, CertRequestStatus, EquipmentVersionsInfo, MockNacefClient, NacefError, SICCashRegisterInfo, SICCertificateRequest, SICLogEntry, SMDFManifest, SMDFStatus, SMDFSyncRequest, SMDFTicketInfo, SMDFType, SMDFVersionsInfo, ERROR_BY_CODE, ERROR_CATALOG, ) except ImportError: # inside Odoo from odoo.addons.nacef_smdf_client.api import ( # type: ignore AgentRestNacefClient, CertRequestStatus, EquipmentVersionsInfo, MockNacefClient, NacefError, SICCashRegisterInfo, SICCertificateRequest, SICLogEntry, SMDFManifest, SMDFStatus, SMDFSyncRequest, SMDFTicketInfo, SMDFType, SMDFVersionsInfo, ERROR_BY_CODE, ERROR_CATALOG, ) class TestSerialization(unittest.TestCase): def test_manifest_roundtrip_camelcase_keys(self): wire = { "imdf": "IMDF123", "status": "SYNCHRONIZED", "version": "1.0.0", "type": "SERVER", "certificateInfo": { "certRequestStatus": "CERTIFICATE_GENERATED", "issuanceDate": "2026-01-01T00:00:00.000", "expirationDate": "2027-01-01T00:00:00.000", "expired": False, "revoked": False, }, "maintenanceEnabled": False, "state": "Online", "synchronizationRate": 100, "availableOfflineTickets": 5, "versionsInfo": { "smdf": {"os": "Linux", "current": "1.0.0", "last": "1.0.0", "minimal": "1.0.0"}, "agents": {"AA:BB:CC:DD:EE:FF": { "os": "Windows", "current": "1.0.0", "last": "1.0.0", "minimal": "1.0.0"}}, }, } m = SMDFManifest.from_dict(wire) self.assertEqual(m.imdf, "IMDF123") self.assertEqual(m.certificate_info.cert_request_status, "CERTIFICATE_GENERATED") self.assertEqual(m.available_offline_tickets, 5) self.assertIsInstance(m.version_info, SMDFVersionsInfo) self.assertEqual(m.version_info.smdf.os, "Linux") self.assertIn("AA:BB:CC:DD:EE:FF", m.version_info.agents) # to_dict must reproduce the exact camelCase wire keys out = m.to_dict() self.assertEqual(out["certificateInfo"]["certRequestStatus"], "CERTIFICATE_GENERATED") self.assertEqual(out["availableOfflineTickets"], 5) self.assertEqual( out["versionsInfo"]["agents"]["AA:BB:CC:DD:EE:FF"]["os"], "Windows") def test_ticket_info_wire_keys(self): info = SMDFTicketInfo(base64_ticket="eyJ4IjoxfQ==", total_ht=1000, total_tax=70, operation_type="TICKET", transaction_type="SALE") d = info.to_dict() self.assertEqual(set(d), {"base64Ticket", "totalHT", "totalTax", "operationType", "transactionType"}) def test_none_fields_are_omitted(self): d = SICCashRegisterInfo(model="OdooPOS", serial_number="SN1", version="17.0").to_dict() self.assertEqual(d, {"model": "OdooPOS", "serialNumber": "SN1", "version": "17.0"}) class TestErrorCatalog(unittest.TestCase): def test_known_codes_present(self): for code in (509, 517, 518, 519, 523): self.assertIn(code, ERROR_BY_CODE) def test_error_flags(self): err = NacefError(name="SMDF_NOT_REACHABLE") self.assertEqual(err.code, 503) self.assertTrue(err.retryable) self.assertFalse(err.terminal) rev = NacefError(name="SMDF_REVOKED_CERTIFICATE") self.assertEqual(rev.code, 519) self.assertTrue(rev.terminal) def test_lookup_by_code(self): err = NacefError(code=518) self.assertEqual(err.name, "SMDF_EXPIRED_CERTIFICATE") class TestRestEnvelope(unittest.TestCase): def test_unwrap_success(self): obj = AgentRestNacefClient._unwrap(200, {"object": {"imdf": "X"}, "errorCode": None}) self.assertEqual(obj, {"imdf": "X"}) def test_unwrap_error_raises_integer_code(self): # live API returns an integer errorCode (OpenAPI Response schema) with self.assertRaises(NacefError) as ctx: AgentRestNacefClient._unwrap( 400, {"object": None, "errorCode": 509, "message": "not synced"}) self.assertEqual(ctx.exception.code, 509) self.assertEqual(ctx.exception.name, "SMDF_NOT_SYNCHRONIZED") def test_endpoint_paths_match_collection(self): # trailing slashes must match the official Postman collection exactly self.assertEqual(AgentRestNacefClient.PATHS["certificate"], "/sic/external/certificate/request/") self.assertEqual(AgentRestNacefClient.PATHS["sync"], "/sic/external/sync/request/") self.assertEqual(AgentRestNacefClient.PATHS["sign"], "/sic/external/sign/request/") self.assertEqual(AgentRestNacefClient.PATHS["manifest"], "/sic/external/manifest") self.assertEqual(AgentRestNacefClient.PATHS["log"], "/sic/external/log/") self.assertEqual(AgentRestNacefClient().base_url, "http://localhost:10006") class TestProctestScenario(unittest.TestCase): """Drives the NACEF-PROCTEST-02 21-step scenario against MockNacefClient.""" def setUp(self): self.c = MockNacefClient(imdf="IMDF00000000042", smdf_type=SMDFType.SERVER, offline_capacity=5) self.cert_req = SICCertificateRequest( cash_register_info=SICCashRegisterInfo( model="Odoo NACEF POS", serial_number="TILL-001", version="17.0.1.0.0")) def _sale(self): return self.c.sign_ticket(SMDFTicketInfo( base64_ticket=base64.b64encode(b'{"ref":"x"}').decode(), total_ht=1000, total_tax=70, operation_type="TICKET", transaction_type="SALE")) def test_step3_initial_state_is_factory(self): self.assertEqual(self.c.get_manifest().status, SMDFStatus.FACTORY.value) def test_step4_sale_blocked_before_certificate(self): with self.assertRaises(NacefError) as ctx: self._sale() self.assertEqual(ctx.exception.code, 509) # SMDF_NOT_SYNCHRONIZED def test_step5_6_certificate_request(self): m = self.c.request_certificate(self.cert_req) self.assertEqual(m.status, SMDFStatus.CERT_REQUESTED.value) # step 6 def test_step8_sale_blocked_before_sync(self): self.c.request_certificate(self.cert_req) self.c.simulate_certificate_generation() # registration unit with self.assertRaises(NacefError): self._sale() # must sync first def test_step9_first_sync_returns_ticket_zero(self): self.c.request_certificate(self.cert_req) self.c.simulate_certificate_generation() m = self.c.sync(SMDFSyncRequest(request_pin_update=True)) self.assertEqual(m.status, SMDFStatus.SYNCHRONIZED.value) self.assertIsNotNone(m.ticket_zero_qrcode) self.assertIn(b"NACEF SMDF", m.ticket_zero_bytes) def _bring_to_synchronized(self): self.c.request_certificate(self.cert_req) self.c.simulate_certificate_generation() self.c.sync(SMDFSyncRequest(request_pin_update=True)) def test_step10_online_sales_signed(self): self._bring_to_synchronized() s1, s2 = self._sale(), self._sale() self.assertTrue(s1.ticket_identifier) self.assertNotEqual(s1.ticket_identifier, s2.ticket_identifier) self.assertTrue(base64.b64decode(s1.qrcode_image).startswith(b"ONLINE")) def test_step11_offline_sales_and_counter(self): self._bring_to_synchronized() self.c.set_offline() before = self.c.get_manifest().available_offline_tickets s = self._sale() self.assertTrue(base64.b64decode(s.qrcode_image).startswith(b"OFFLINE")) self.assertEqual(self.c.get_manifest().available_offline_tickets, before - 1) def test_step12_back_online_manifest(self): self._bring_to_synchronized() self.c.set_offline(); self._sale(); self._sale() self.c.set_online() m = self.c.get_manifest() self.assertEqual(m.status, SMDFStatus.SYNCHRONIZED.value) self.assertEqual(m.state, "Online") def test_step14_15_offline_exhaustion_forces_resync(self): self._bring_to_synchronized() self.c.set_offline() for _ in range(5): # capacity == 5 self._sale() with self.assertRaises(NacefError): self._sale() # 6th offline sale blocked self.assertEqual(self.c.get_manifest().status, SMDFStatus.NOT_SYNCHRONIZED.value) def test_step18_19_suspended_blocks_sale(self): self._bring_to_synchronized() self.c.simulate_suspend() with self.assertRaises(NacefError): self._sale() def test_step20_21_revoked_blocks_and_requires_new_request(self): self._bring_to_synchronized() self.c.simulate_revoke() with self.assertRaises(NacefError) as ctx: self._sale() self.assertEqual(ctx.exception.code, 519) # SMDF_REVOKED_CERTIFICATE # a fresh certificate request must be possible afterwards (step 21) m = self.c.request_certificate(self.cert_req) self.assertEqual(m.status, SMDFStatus.CERT_REQUESTED.value) def test_log_is_recorded(self): self.c.log(SICLogEntry(module="lc.agent", operation="SIGN_REQUEST", level="INFO", message="ticket 1")) self.assertEqual(len(self.c.logs), 1) if __name__ == "__main__": unittest.main()