Oficina: dados de referência, deltas e ativação

"""Original DR reference-data and RMA observation models. Python 3.13 stdlib.
python3 run.py --output evidence.json
No Swift SDK, API call, credentials, actual directory records or messages.
The delta format, activation window, cache TTL and decision labels are fictional.
"""
import argparse
import copy
import hashlib
import json
import sys
from pathlib import Path

checks=[]
def check(name, actual, expected):
    assert actual==expected,(name,actual,expected)
    checks.append(dict(name=name,actual=actual,expected=expected,passed=True))
def error(fn):
    try:fn()
    except ValueError as e:return str(e)
    raise AssertionError('Expected local model rejection')

def apply_delta(snapshot, delta):
    if delta['base']!=snapshot['version']:raise ValueError('wrong base')
    if delta['version']<=delta['base']:raise ValueError('non-increasing version')
    changed=copy.deepcopy(snapshot)
    seen=set()
    for op in delta['operations']:
        key=op['id']
        if key in seen:raise ValueError('duplicate operation identity')
        seen.add(key)
        if op['kind']=='delete':
            if key not in changed['records']:raise ValueError('delete absent record')
            del changed['records'][key]
        elif op['kind']=='upsert':
            if not op.get('route'):raise ValueError('missing route')
            changed['records'][key]=op['route']
        else:raise ValueError('unknown operation')
    changed['version']=delta['version']
    return changed

def activate(candidate, approved, now, effective_from):
    # Test record activation only, not a Swift schedule or bank instruction.
    return candidate if approved and now>=effective_from else None

def usable_cache(entry, request, now, ttl):
    # Query scope includes direction through ordered issuer/correspondent.
    key_fields=('environment','issuer','correspondent','service','message')
    return (all(entry[k]==request[k] for k in key_fields)
            and 0<=now-entry['observed_at']<ttl)

def classify(http,payload):
    # Original consumer decision labels, not Swift API response fields.
    if http!=200:return 'API_OBSERVATION_FAILED'
    if not isinstance(payload,dict):return 'INCOMPLETE_OBSERVATION'
    state=payload.get('authorization_status')
    if state=='AUTHORIZED':return 'RELATION_REPORTED_AUTHORIZED'
    if state=='NOT-AUTHORIZED':
        return 'RELATION_REPORTED_NOT_AUTHORIZED' if payload.get('reason') in {'REVOKED','REJECTED','EXPIRED','NON-EXISTENT'} else 'INCOMPLETE_OBSERVATION'
    return 'INCOMPLETE_OBSERVATION'

def main():
    base=dict(version=10,records={'PARTY-A':'ROUTE-OLD','PARTY-B':'ROUTE-B'})
    delta=dict(base=10,version=11,operations=[dict(kind='upsert',id='PARTY-A',route='ROUTE-NEW'),dict(kind='delete',id='PARTY-B'),dict(kind='upsert',id='PARTY-C',route='ROUTE-C')])
    next_state=apply_delta(base,delta)
    check('delta updates one record',next_state['records']['PARTY-A'],'ROUTE-NEW')
    check('delta applies deletion','PARTY-B' in next_state['records'],False)
    check('delta inserts record',next_state['records']['PARTY-C'],'ROUTE-C')
    check('base remains unchanged',base,dict(version=10,records={'PARTY-A':'ROUTE-OLD','PARTY-B':'ROUTE-B'}))
    check('version advances only after validation',next_state['version'],11)
    check('wrong baseline rejected',error(lambda:apply_delta(base,{**delta,'base':9})),'wrong base')
    check('same version rejected',error(lambda:apply_delta(base,{**delta,'version':10})),'non-increasing version')
    check('repeat delta rejected',error(lambda:apply_delta(next_state,delta)),'wrong base')
    bad={**delta,'operations':delta['operations']+[dict(kind='upsert',id='PARTY-A',route='OTHER')]}
    check('duplicate change identity rejected',error(lambda:apply_delta(base,bad)),'duplicate operation identity')
    check('failed candidate does not mutate active snapshot',base,dict(version=10,records={'PARTY-A':'ROUTE-OLD','PARTY-B':'ROUTE-B'}))
    check('absent deletion rejected',error(lambda:apply_delta(base,{**delta,'operations':[dict(kind='delete',id='ABSENT')]})),'delete absent record')
    check('missing route rejected',error(lambda:apply_delta(base,{**delta,'operations':[dict(kind='upsert',id='PARTY-A')]})),'missing route')
    check('future effective version held',activate(next_state,True,9,10),None)
    check('effective boundary activates',activate(next_state,True,10,10)['version'],11)
    check('unapproved version held',activate(next_state,False,10,10),None)
    req=dict(environment='LIVE',issuer='PARTY-A',correspondent='PARTY-B',service='DR-SERVICE',message='DR-TYPE-1')
    cached={**req,'observed_at':100,'state':'AUTHORIZED'}
    check('cache within local scope and TTL',usable_cache(cached,req,104,5),True)
    check('cache expires at exact local boundary',usable_cache(cached,req,105,5),False)
    check('future timestamp rejected',usable_cache(cached,req,99,5),False)
    for name,change in [('environment',{'environment':'PILOT'}),('issuer',{'issuer':'PARTY-C'}),('correspondent',{'correspondent':'PARTY-C'}),('service',{'service':'DR-OTHER'}),('message',{'message':'DR-TYPE-2'}),('direction',{'issuer':'PARTY-B','correspondent':'PARTY-A'})]:
        check('cache '+name+' mismatch rejected',usable_cache(cached,{**req,**change},104,5),False)
    check('HTTP success may report not authorized',classify(200,{'authorization_status':'NOT-AUTHORIZED','reason':'REVOKED'}),'RELATION_REPORTED_NOT_AUTHORIZED')
    check('HTTP success reports authorization only',classify(200,{'authorization_status':'AUTHORIZED'}),'RELATION_REPORTED_AUTHORIZED')
    for status in (401,403,429,503):
        check(str(status)+' is failed observation',classify(status,{}),'API_OBSERVATION_FAILED')
    check('missing status is incomplete',classify(200,{}),'INCOMPLETE_OBSERVATION')
    check('unknown status is incomplete',classify(200,{'authorization_status':'PENDING'}),'INCOMPLETE_OBSERVATION')
    check('negative state missing reason is incomplete locally',classify(200,{'authorization_status':'NOT-AUTHORIZED'}),'INCOMPLETE_OBSERVATION')
    check('non object is incomplete',classify(200,[]),'INCOMPLETE_OBSERVATION')
    check('request cache does not model settlement','settlement' in cached,False)
    output=dict(scope='Original in-memory synthetic models; not a SwiftRef file parser, RMA implementation, actual API query, directory validation, message authorization or settlement evidence.',python=sys.version.split()[0],runnerSha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),passed=len(checks),checks=checks)
    parser=argparse.ArgumentParser();parser.add_argument('--output',required=True)
    Path(parser.parse_args().output).write_text(json.dumps(output,indent=2)+'\n');print(json.dumps({'passed':len(checks),'scope':output['scope']}))
if __name__=='__main__':main()

Tópicos relacionados: Identificadores, relações e dados de referência · Estados, reconciliação e recuperação sem duplicar · Continuidade e passagem à operação

Referência: SwiftRef · DR Swift knowledge assessment2026.10; independent professional curriculum