"""Contracts for the multi-currency storage substrate (issue #552). CurrencyType and DatedConversionRate are modeled after the Salesforce standard objects of the same names (key prefixes 01L / 04w, field shapes taken from a v63 describe of a multi-currency org). DatedConversionRate periods are half-open [StartDate, NextStartDate); the pack enforces the Salesforce "Manage Dated Exchange Rates" invariant — appending a period closes the previous open period — through a before-save default flow plus an after-save close flow. These tests pin the shapes both flows and API consumers rely on. """ import json import unittest from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[1] CRM_PACK = REPO_ROOT / "salesforce-crm-standard" OPEN_ENDED = "9999-12-31" def load_json(path: Path): with path.open(encoding="utf-8") as handle: return json.load(handle) def object_spec(api_name: str): for spec in load_json(CRM_PACK / "objects" / "standard_objects_spec.json"): if spec.get("apiName") == api_name: return spec raise AssertionError(f"{api_name} object spec missing") def field_spec(spec: dict, api_name: str): for field in spec.get("fields", []): if field.get("apiName") == api_name: return field raise AssertionError(f"{spec.get('apiName')}.{api_name} field spec missing") def pack_component(kind: str, business_key: str): for component in load_json(CRM_PACK / "pack.json")["components"]: if component.get("kind") == kind and component.get("businessKey") == business_key: return component raise AssertionError(f"pack.json component {kind}/{business_key} missing") def flow(name: str): return load_json(CRM_PACK / "flows" / f"{name}.flow.json") def filters_by_field(element: dict): return {f["field"]: f for f in element.get("filters", [])} class CurrencyTypeShapeTest(unittest.TestCase): def test_object_identity_matches_salesforce(self): spec = object_spec("CurrencyType") self.assertEqual(spec.get("keyPrefix"), "01L") self.assertEqual(spec.get("tableType"), "standard_object") self.assertEqual(spec.get("sharingModel"), "PublicRead") def test_field_shapes(self): spec = object_spec("CurrencyType") self.assertEqual( [f["apiName"] for f in spec["fields"]], ["IsoCode", "ConversionRate", "DecimalPlaces", "IsActive", "IsCorporate"], ) iso = field_spec(spec, "IsoCode") self.assertTrue(iso.get("required")) self.assertTrue(iso.get("unique")) self.assertEqual(iso.get("writeProtectionPolicy"), "immutable_after_create") self.assertEqual(field_spec(spec, "ConversionRate").get("type"), "double") self.assertEqual(field_spec(spec, "DecimalPlaces").get("type"), "integer") self.assertEqual(field_spec(spec, "DecimalPlaces").get("default"), "2") for flag in ("IsActive", "IsCorporate"): field = field_spec(spec, flag) self.assertEqual(field.get("type"), "boolean") self.assertEqual(field.get("default"), "0") class DatedConversionRateShapeTest(unittest.TestCase): def test_object_identity_matches_salesforce(self): spec = object_spec("DatedConversionRate") self.assertEqual(spec.get("keyPrefix"), "04w") self.assertEqual(spec.get("tableType"), "standard_object") self.assertEqual(spec.get("sharingModel"), "PublicRead") def test_period_fields(self): spec = object_spec("DatedConversionRate") self.assertEqual( [f["apiName"] for f in spec["fields"]], ["IsoCode", "ConversionRate", "StartDate", "NextStartDate"], ) self.assertTrue(field_spec(spec, "IsoCode").get("required")) self.assertTrue(field_spec(spec, "StartDate").get("required")) self.assertEqual(field_spec(spec, "StartDate").get("type"), "date") next_start = field_spec(spec, "NextStartDate") self.assertEqual(next_start.get("type"), "date") self.assertFalse(next_start.get("required", False)) def test_write_protection_mirrors_the_salesforce_describe(self): # Salesforce describe for this object: IsoCode/StartDate are # createable-once, NextStartDate is neither createable nor updateable # (platform-derived), ConversionRate is fully writable. # # NextStartDate MUST stay caller-unwritable. The period chain is a # function of the StartDate set, so a caller-supplied end date is a # redundant input that can only contradict it — every shape of that # contradiction (overrunning, falling short, inverted, # finite-on-the-latest-row) is a gap or an overlap the flows would # otherwise have to special-case. Dropping the input is what makes the # derivation total. # # updateable:false is wrong for all of these fields — that path drops # writes silently instead of rejecting them, and it would silently eat # the close flow's own truncation write. spec = object_spec("DatedConversionRate") for name in ("IsoCode", "StartDate"): self.assertEqual( field_spec(spec, name).get("writeProtectionPolicy"), "immutable_after_create", name, ) # createable:false drops a caller-supplied value before the flows run # (what Salesforce does); the policy still lets the pack flows write the # derived value — trusted_only would NOT, because the create-path write # check runs after before-save flows under the CALLER's context, so a # flow-assigned trusted_only field is rejected as if the caller sent it. next_start = field_spec(spec, "NextStartDate") self.assertIs(next_start.get("createable"), False) self.assertEqual(next_start.get("writeProtectionPolicy"), "immutable_after_create") self.assertNotIn("writeProtectionPolicy", field_spec(spec, "ConversionRate")) def test_key_prefixes_unique_across_spec(self): specs = load_json(CRM_PACK / "objects" / "standard_objects_spec.json") prefixes = [s["keyPrefix"] for s in specs if s.get("keyPrefix")] self.assertEqual(len(prefixes), len(set(prefixes))) # The new prefixes must also be case-insensitively free: Salesforce # distinguishes prefixes by case (00K vs 00k ship in this spec), but # runtime registries may not, so new objects should not lean on case. others = [ s["keyPrefix"].lower() for s in specs if s["apiName"] not in ("CurrencyType", "DatedConversionRate") ] self.assertNotIn("01l", others) self.assertNotIn("04w", others) class DefaultNextStartFlowContractTest(unittest.TestCase): def setUp(self): self.flow = flow("Std_DatedConversionRateDefaultNextStart") def test_trigger_is_before_save_create_only(self): start = self.flow["start"] self.assertEqual(start["object"], "DatedConversionRate") self.assertEqual(start["triggerType"], "RecordBeforeSave") self.assertEqual(start["recordTriggerType"], "Create") self.assertTrue(self.flow.get("runAtomically")) def test_duplicate_period_rejected_before_save(self): # One period per (IsoCode, StartDate) — the Salesforce dated-rate rule. # The platform has no composite-unique constraint, so the guard is a # pre-save lookup + custom error; same-instant racers that slip past it # leave two identical open periods (documented residual), never a # mutual truncation to zero. lookup = next(l for l in self.flow["recordLookups"] if l["name"] == "GetDuplicatePeriod") by_field = filters_by_field(lookup) self.assertEqual( by_field["IsoCode"]["value"]["elementReference"], "$Record.IsoCode" ) self.assertEqual(by_field["StartDate"]["operator"], "EqualTo") self.assertEqual( by_field["StartDate"]["value"]["elementReference"], "$Record.StartDate" ) self.assertEqual( self.flow["start"]["connector"]["targetReference"], "GetDuplicatePeriod" ) decision = next(d for d in self.flow["decisions"] if d["name"] == "DuplicatePeriodExists") [rule] = decision["rules"] self.assertEqual(rule["connector"]["targetReference"], "FailDuplicatePeriod") self.assertEqual(decision["defaultConnector"]["targetReference"], "GetNextPeriod") [custom_error] = self.flow["customErrors"] self.assertEqual(custom_error["name"], "FailDuplicatePeriod") [message] = custom_error["customErrorMessages"] # Authored status code = the Salesforce-native one. Verified at runtime: # the record-triggered before-save path currently collapses every # customError to FIELD_CUSTOM_VALIDATION_EXCEPTION (the authored message # survives), so this pins the declaration, not today's wire code. self.assertEqual(message["statusCode"], "DUPLICATE_VALUE") def test_next_start_date_is_derived_not_supplied(self): # Two outcomes only, and they are total: no later period => open-ended, # otherwise cap at that period's start. There is no "caller supplied a # value" branch because NextStartDate is not caller-writable. decision = next(d for d in self.flow["decisions"] if d["name"] == "RouteNextStart") rules = {r["name"]: r for r in decision["rules"]} self.assertEqual(set(rules), {"NoLaterPeriod"}) [condition] = rules["NoLaterPeriod"]["conditions"] self.assertEqual(condition["leftValueReference"], "GetNextPeriod.StartDate") self.assertEqual(condition["operator"], "IsNull") self.assertTrue(condition["rightValue"]["booleanValue"]) self.assertEqual( rules["NoLaterPeriod"]["connector"]["targetReference"], "AssignOpenEnded" ) self.assertEqual( decision["defaultConnector"]["targetReference"], "AssignCapAtNextPeriod" ) def test_lookup_finds_earliest_later_period_of_same_currency(self): lookup = next(l for l in self.flow["recordLookups"] if l["name"] == "GetNextPeriod") self.assertEqual(lookup["object"], "DatedConversionRate") by_field = filters_by_field(lookup) self.assertEqual( by_field["IsoCode"]["value"]["elementReference"], "$Record.IsoCode" ) self.assertEqual(by_field["StartDate"]["operator"], "GreaterThan") self.assertEqual( by_field["StartDate"]["value"]["elementReference"], "$Record.StartDate" ) self.assertEqual(lookup["sortField"], "StartDate") self.assertEqual(lookup["sortOrder"], "Asc") self.assertTrue(lookup["getFirstRecordOnly"]) # Explicit nulls on no-match keep the found/none decision's IsNull # probe well-defined instead of reading an unset variable. self.assertTrue(lookup["assignNullValuesIfNoRecordsFound"]) def test_assignment_targets_and_open_ended_literal(self): assignments = {a["name"]: a for a in self.flow["assignments"]} [open_item] = assignments["AssignOpenEnded"]["assignmentItems"] self.assertEqual(open_item["assignToReference"], "$Record.NextStartDate") self.assertEqual(open_item["value"]["dateValue"], OPEN_ENDED) [cap_item] = assignments["AssignCapAtNextPeriod"]["assignmentItems"] self.assertEqual(cap_item["assignToReference"], "$Record.NextStartDate") self.assertEqual(cap_item["value"]["elementReference"], "GetNextPeriod.StartDate") class CloseOnAppendFlowContractTest(unittest.TestCase): def setUp(self): self.flow = flow("Std_DatedConversionRateCloseOnAppend") def test_trigger_is_after_save_create_only(self): start = self.flow["start"] self.assertEqual(start["object"], "DatedConversionRate") # Cross-record writes need RecordAfterSave; a before-save flow can only # assign to $Record. self.assertEqual(start["triggerType"], "RecordAfterSave") # Create-only: a rate correction on an existing row must not re-close # neighbours (Salesforce keeps StartDate/IsoCode immutable instead). self.assertEqual(start["recordTriggerType"], "Create") # runAtomically is rejected at install time for RecordAfterSave flows # (the platform supports it only on RecordBeforeSave). self.assertFalse(self.flow.get("runAtomically", False)) def test_filtered_update_closes_exactly_the_covering_periods(self): [update] = self.flow["recordUpdates"] self.assertEqual(update["object"], "DatedConversionRate") self.assertEqual(update["filterLogic"], "and") by_field = filters_by_field(update) self.assertEqual(set(by_field), {"IsoCode", "Id", "StartDate", "NextStartDate"}) self.assertEqual( by_field["IsoCode"]["value"]["elementReference"], "$Record.IsoCode" ) # Self-exclusion: without it the freshly created open-ended row would # match its own StartDate <= S filter and close itself. self.assertEqual(by_field["Id"]["operator"], "NotEqualTo") self.assertEqual(by_field["Id"]["value"]["elementReference"], "$Record.Id") # Strict coverage window [StartDate < S < NextStartDate]. Strict < is # load-bearing: equal-StartDate siblings are rejected pre-save by the # duplicate guard, and closing them here would let two same-instant # creates truncate each other into empty periods (no rate at all). self.assertEqual(by_field["StartDate"]["operator"], "LessThan") self.assertEqual( by_field["StartDate"]["value"]["elementReference"], "$Record.StartDate" ) self.assertEqual(by_field["NextStartDate"]["operator"], "GreaterThan") self.assertEqual( by_field["NextStartDate"]["value"]["elementReference"], "$Record.StartDate" ) [assignment] = update["inputAssignments"] self.assertEqual(assignment["field"], "NextStartDate") self.assertEqual(assignment["value"]["elementReference"], "$Record.StartDate") class PackRegistrationTest(unittest.TestCase): def test_flows_registered_with_object_spec_dependency(self): for business_key in ( "Std_DatedConversionRateDefaultNextStart", "Std_DatedConversionRateCloseOnAppend", ): component = pack_component("standard_flow", business_key) self.assertEqual(component["manageability"], "defaults_only") self.assertEqual( component["path"], f"flows/{business_key}.flow.json" ) self.assertIn( {"kind": "object_spec", "businessKey": "standard_objects_spec:crm"}, component.get("dependsOn", []), ) if __name__ == "__main__": unittest.main()