Skip to content

Commit 2addd0c

Browse files
jeremijeremi
committed
Improve Notary evidence coverage
1 parent 0b1cc0a commit 2addd0c

2 files changed

Lines changed: 242 additions & 3 deletions

File tree

‎spp_notary_evidence/models/data_provider.py‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22
"""Notary extensions for CEL data providers."""
33

44
import logging
5-
from datetime import datetime, timedelta
5+
from datetime import UTC, datetime, timedelta
66

77
from odoo import _, fields, models
88
from odoo.exceptions import UserError
@@ -487,5 +487,11 @@ def _parse_notary_datetime(self, value):
487487
if isinstance(value, datetime):
488488
return value
489489
if isinstance(value, str):
490-
return fields.Datetime.to_datetime(value.replace("Z", "+00:00"))
490+
try:
491+
return fields.Datetime.to_datetime(value)
492+
except ValueError:
493+
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
494+
if parsed.tzinfo:
495+
parsed = parsed.astimezone(UTC).replace(tzinfo=None)
496+
return parsed
491497
return None

‎spp_notary_evidence/tests/test_notary_evidence.py‎

Lines changed: 234 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,8 +13,9 @@
1313
from odoo.tests import TransactionCase, tagged
1414

1515
from odoo.addons.spp_notary_client.services.client import NotaryClient
16-
from odoo.addons.spp_notary_client.services.exceptions import NotarySubjectIdMissing, NotaryTransportError
16+
from odoo.addons.spp_notary_client.services.exceptions import NotaryError, NotarySubjectIdMissing, NotaryTransportError
1717
from odoo.addons.spp_notary_client.services.schemas import CatalogResponse
18+
from odoo.addons.spp_notary_evidence.models.notary_claim import data_value_type_for_cel, normalize_notary_value_type
1819

1920

2021
@tagged("post_install", "-at_install")
@@ -130,6 +131,71 @@ def test_catalog_sync_creates_claim_and_external_variable(self):
130131
self.assertEqual(claim.variable_id.external_provider_id, self.provider)
131132
self.assertEqual(claim.variable_id.notary_claim_id, claim)
132133

134+
def test_fetch_notary_catalog_supports_legacy_client_shapes(self):
135+
get_client = SimpleNamespace(
136+
get_claim_catalog=lambda: [
137+
{
138+
"id": "legacy-get",
139+
"name": "Legacy Get",
140+
"version": "2026-01",
141+
"type": "bool",
142+
}
143+
]
144+
)
145+
fetch_client = SimpleNamespace(
146+
fetch_claim_catalog=lambda path: [
147+
{
148+
"id": f"legacy-fetch-{path.strip('/')}",
149+
"title": "Legacy Fetch",
150+
"formats": ["json"],
151+
}
152+
]
153+
)
154+
155+
with patch.object(type(self.provider), "_notary_client", return_value=get_client):
156+
get_catalog = self.provider._fetch_notary_catalog()
157+
with patch.object(type(self.provider), "_notary_client", return_value=fetch_client):
158+
fetch_catalog = self.provider._fetch_notary_catalog()
159+
160+
self.assertEqual(get_catalog.claims[0].id, "legacy-get")
161+
self.assertEqual(get_catalog.claims[0].title, "Legacy Get")
162+
self.assertEqual(get_catalog.claims[0].value_type, "bool")
163+
self.assertEqual(fetch_catalog.claims[0].id, "legacy-fetch-claims")
164+
self.assertEqual(fetch_catalog.claims[0].supported_formats, ["json"])
165+
with patch.object(type(self.provider), "_notary_client", return_value=SimpleNamespace()):
166+
with self.assertRaises(UserError):
167+
self.provider._fetch_notary_catalog()
168+
169+
def test_catalog_sync_rejects_non_notary_and_wraps_notary_error(self):
170+
generic_provider = self.Provider.create(
171+
{
172+
"name": "Generic Provider",
173+
"code": f"generic_notary_test_{self._test_id}",
174+
"provider_kind": "generic",
175+
"base_url": "https://generic.example",
176+
"auth_type": "none",
177+
}
178+
)
179+
180+
with self.assertRaises(UserError):
181+
generic_provider.action_sync_notary_claim_catalog()
182+
with patch.object(type(self.provider), "_fetch_notary_catalog", side_effect=NotaryError("catalog down")):
183+
with self.assertRaises(UserError):
184+
self.provider.action_sync_notary_claim_catalog()
185+
186+
def test_provider_actions_return_wizard_and_claim_windows(self):
187+
claim = self._create_claim_with_variable("window-claim", value_type="boolean")
188+
189+
wizard_action = self.provider.action_open_notary_catalog_sync_wizard()
190+
claim_action = self.provider.action_view_notary_claims()
191+
192+
self.assertEqual(wizard_action["res_model"], "spp.notary.catalog.sync.wizard")
193+
self.assertEqual(wizard_action["context"]["default_provider_id"], self.provider.id)
194+
self.assertEqual(claim_action["res_model"], "spp.notary.claim")
195+
self.assertIn(("provider_id", "=", self.provider.id), claim_action["domain"])
196+
self.assertEqual(self.provider.notary_claim_count, 1)
197+
self.assertEqual(claim.provider_id, self.provider)
198+
133199
def test_catalog_sync_marks_missing_claims_unavailable(self):
134200
claim = self._create_claim_with_variable("removed-claim", value_type="boolean")
135201
catalog = SimpleNamespace(claims=[])
@@ -528,6 +594,17 @@ def test_null_policy_returns_none_without_cache_write(self):
528594
)
529595
self.assertFalse(cached)
530596

597+
def test_error_policy_helpers_handle_unknown_policy_and_empty_stale_reads(self):
598+
claim = self._create_claim_with_variable("empty-stale-policy", value_type="string")
599+
error = NotaryTransportError(code="source.unavailable", status_code=503)
600+
601+
self.provider.notary_unavailable_policy = "stale_cache_with_audit"
602+
self.assertEqual(self.provider._read_stale_notary_values(claim.variable_id, [], "current"), {})
603+
self.assertEqual(
604+
self.provider._values_for_notary_error(error, claim.variable_id, [self.partner_a.id], "current"),
605+
{},
606+
)
607+
531608
def test_raise_policy_surfaces_notary_error_as_user_error(self):
532609
claim = self._create_claim_with_variable("raise-policy-claim", value_type="string")
533610
self.provider.notary_unavailable_policy = "raise"
@@ -570,6 +647,68 @@ def test_notary_expires_at_is_clamped_to_minimum_cache_ttl(self):
570647
)
571648
self.assertGreaterEqual(cached.expires_at, fields.Datetime.now() + timedelta(seconds=250))
572649

650+
def test_notary_value_helpers_cover_stale_raw_values_and_fallback_shapes(self):
651+
claim = self._create_claim_with_variable("helper-shapes", value_type="boolean")
652+
stale_expires_at = fields.Datetime.now() - timedelta(minutes=30)
653+
self.env["spp.data.value"].upsert_values(
654+
[
655+
{
656+
"variable_name": claim.variable_id.name,
657+
"subject_id": self.partner_a.id,
658+
"period_key": "past",
659+
"provider": self.provider.code,
660+
"value_json": "raw-stale",
661+
"value_type": "string",
662+
"source_type": "external",
663+
"expires_at": stale_expires_at,
664+
}
665+
]
666+
)
667+
batch_response = SimpleNamespace(
668+
items=[
669+
SimpleNamespace(
670+
input_index=None,
671+
status="succeeded",
672+
claim_results=[],
673+
results=[
674+
SimpleNamespace(
675+
claim_id="helper-shapes",
676+
value=None,
677+
satisfied=False,
678+
expires_at="2026-01-02T03:04:05Z",
679+
)
680+
],
681+
),
682+
SimpleNamespace(input_index=2, status="failed", claim_results=[], results=[]),
683+
]
684+
)
685+
686+
stale = self.provider._read_stale_notary_values(claim.variable_id, [self.partner_a.id], "past")
687+
values = self.provider._values_from_batch_response(batch_response, [(self.partner_b.id, {})], claim)
688+
self.provider._write_notary_values(
689+
claim.variable_id,
690+
{self.partner_c.id: {"value": "do-not-write", "stale": True}},
691+
"current",
692+
)
693+
694+
self.assertEqual(stale[self.partner_a.id]["value"], "raw-stale")
695+
self.assertEqual(values[self.partner_b.id]["value"], False)
696+
self.assertTrue(values[self.partner_b.id]["expires_at"])
697+
self.assertIsNone(self.provider._effective_notary_expires_at(None))
698+
self.provider.notary_min_cache_ttl_seconds = 0
699+
upstream_expires_at = fields.Datetime.now() + timedelta(minutes=5)
700+
self.assertEqual(self.provider._effective_notary_expires_at(upstream_expires_at), upstream_expires_at)
701+
self.assertIsNone(self.provider._parse_notary_datetime(object()))
702+
self.assertFalse(
703+
self.env["spp.data.value"].search(
704+
[
705+
("variable_name", "=", claim.variable_id.name),
706+
("subject_id", "=", self.partner_c.id),
707+
("provider", "=", self.provider.code),
708+
]
709+
)
710+
)
711+
573712
def test_catalog_sync_marks_existing_pinned_claim_as_version_drift(self):
574713
claim = self._create_claim_with_variable("drifting-claim", value_type="boolean")
575714
catalog = SimpleNamespace(
@@ -633,6 +772,85 @@ def test_catalog_sync_wizard_previews_before_confirming(self):
633772
self.Claim.search([("provider_id", "=", self.provider.id), ("external_id", "=", "wizard-preview-claim")])
634773
)
635774

775+
def test_catalog_sync_wizard_default_get_guards_and_preview_errors(self):
776+
generic_provider = self.Provider.create(
777+
{
778+
"name": "Generic Wizard Provider",
779+
"code": f"generic_wizard_notary_{self._test_id}",
780+
"provider_kind": "generic",
781+
"base_url": "https://generic-wizard.example",
782+
"auth_type": "none",
783+
}
784+
)
785+
values = (
786+
self.env["spp.notary.catalog.sync.wizard"]
787+
.with_context(active_model="spp.data.provider", active_id=self.provider.id)
788+
.default_get(["provider_id"])
789+
)
790+
wizard = self.env["spp.notary.catalog.sync.wizard"].create({"provider_id": self.provider.id})
791+
generic_wizard = self.env["spp.notary.catalog.sync.wizard"].create({"provider_id": generic_provider.id})
792+
793+
self.assertEqual(values["provider_id"], self.provider.id)
794+
with self.assertRaises(UserError):
795+
wizard.action_sync_catalog()
796+
with self.assertRaises(UserError):
797+
generic_wizard.action_load_preview()
798+
with patch.object(type(self.provider), "_fetch_notary_catalog", side_effect=NotaryError("preview down")):
799+
with self.assertRaises(UserError):
800+
wizard.action_load_preview()
801+
802+
def test_catalog_sync_wizard_previews_update_no_change_drift_and_unavailable(self):
803+
no_change = self._create_claim_with_variable("wizard-no-change", value_type="boolean")
804+
update = self._create_claim_with_variable("wizard-update", value_type="boolean")
805+
drift = self._create_claim_with_variable("wizard-drift", value_type="number")
806+
unavailable = self._create_claim_with_variable("wizard-unavailable", value_type="string")
807+
catalog = CatalogResponse.model_validate(
808+
{
809+
"claims": [
810+
{
811+
"id": no_change.external_id,
812+
"title": no_change.name,
813+
"description": "",
814+
"version": "2026-01",
815+
"subject_type": "individual",
816+
"disclosure": ["predicate"],
817+
"value_type": "boolean",
818+
},
819+
{
820+
"id": update.external_id,
821+
"title": "Wizard Update Changed",
822+
"description": "Changed description",
823+
"version": "2026-01",
824+
"subject_type": "individual",
825+
"disclosure": ["predicate"],
826+
"value_type": "boolean",
827+
},
828+
{
829+
"id": drift.external_id,
830+
"title": drift.name,
831+
"description": "",
832+
"version": "2026-02",
833+
"subject_type": "individual",
834+
"disclosure": ["predicate"],
835+
"value_type": "number",
836+
},
837+
]
838+
}
839+
)
840+
wizard = self.env["spp.notary.catalog.sync.wizard"].create({"provider_id": self.provider.id})
841+
842+
line_values = wizard._preview_line_values(catalog)
843+
by_claim = {values["claim_id"]: values for values in line_values}
844+
summary = wizard._summary_from_lines(line_values)
845+
846+
self.assertEqual(by_claim[no_change.external_id]["action"], "no_change")
847+
self.assertEqual(by_claim[update.external_id]["action"], "update")
848+
self.assertEqual(by_claim[drift.external_id]["action"], "version_drift")
849+
self.assertEqual(by_claim[drift.external_id]["state_after"], "version_drift")
850+
self.assertEqual(by_claim[unavailable.external_id]["action"], "unavailable")
851+
self.assertIn("Update: 1", summary)
852+
self.assertIn("Version drift: 1", summary)
853+
636854
def test_catalog_sync_wizard_blocks_accessor_collision(self):
637855
catalog = CatalogResponse.model_validate(
638856
{
@@ -667,6 +885,21 @@ def test_catalog_sync_wizard_blocks_accessor_collision(self):
667885
with self.assertRaises(UserError):
668886
wizard.action_sync_catalog()
669887

888+
def test_claim_helpers_normalize_values_and_update_existing_variable(self):
889+
self.assertEqual(self.Claim._build_variable_name("123 Provider", "Claim!!!"), "notary_123_provider_claim")
890+
self.assertEqual(normalize_notary_value_type("decimal"), "number")
891+
self.assertEqual(normalize_notary_value_type("unsupported"), "string")
892+
self.assertEqual(data_value_type_for_cel("date"), "string")
893+
self.assertEqual(data_value_type_for_cel("list"), "json")
894+
895+
claim = self._create_claim_with_variable("helper-variable", value_type="money")
896+
claim.write({"state": "deprecated", "active": False, "subject_type": "group"})
897+
898+
self.assertFalse(claim.variable_id.active)
899+
self.assertEqual(claim.variable_id.state, "inactive")
900+
self.assertEqual(claim.variable_id.applies_to, "group")
901+
self.assertEqual(claim.variable_id.value_type, "money")
902+
670903
def test_real_client_is_created_with_outgoing_log_context(self):
671904
provider = self.provider.with_context()
672905
provider.notary_subject_log_secret = "test-secret"

0 commit comments

Comments
 (0)