"""Contrats simplifiés, fictifs : ni SDK ni API ni plug-in Dataverse exécuté.

Le dictionnaire expose remplacement, choix du message et erreur après écriture.
La protection de révision est locale et séquentielle, sans garantie distribuée.
"""
from copy import deepcopy

ROW = '11111111-2222-4333-8444-555555555555'
KEY = ('ENV-RECETTE', ROW, 'PARIS')
BASE = {'receipt_ref':'R104', 'sku':'POMPE-12', 'received_units':8,
        'unit':'PIECE', 'source_revision':7, 'status':'RECEIVED'}
V8 = dict(BASE, received_units=10, source_revision=8)
FIELDS = set(BASE)

class Rejected(ValueError):
    pass

class AfterWriteError(RuntimeError):
    pass

def validate(key, payload):
    if key != KEY:
        raise Rejected('Destination non autorisée')
    if set(payload) != FIELDS:
        raise Rejected('Instantané incomplet ou champ non prévu')
    if (payload['receipt_ref'] != 'R104' or payload['sku'] != 'POMPE-12'
        or payload['unit'] != 'PIECE' or payload['status'] != 'RECEIVED'):
        raise Rejected('Identité métier ou unité incohérente')
    if type(payload['received_units']) is not int or payload['received_units'] < 0:
        raise Rejected('Quantité non entière ou négative')
    if type(payload['source_revision']) is not int or payload['source_revision'] < 1:
        raise Rejected('Révision absente ou invalide')

def raw_write(store, key, payload, kind='elastic', guards=frozenset({'Create','Update'}), fail_after=False):
    # Contraste pédagogique, non reproduction du pipeline Microsoft.
    message = 'Upsert' if kind == 'elastic' else ('Update' if key in store else 'Create')
    if message in guards:
        validate(key, payload)
    if kind == 'elastic':
        store[key] = deepcopy(payload)
    else:
        store[key] = dict(store.get(key, {}), **payload)
    if fail_after:
        raise AfterWriteError('Erreur du traitement après écriture')
    return message

def protected_write(store, key, payload):
    validate(key, payload)  # avant l'effet
    previous = store.get(key)
    if previous:
        old = previous['source_revision']
        new = payload['source_revision']
        if new < old or (new == old and payload != previous):
            raise Rejected('Révision ancienne ou contradictoire')
    store[key] = deepcopy(payload)

def confirmed(observed_key, observed_payload, expected):
    return observed_key == KEY and observed_payload == expected

checks = 0
def check(value):
    global checks
    assert value
    checks += 1

def rejects(payload, key=KEY):
    store = {KEY:deepcopy(BASE)}
    try:
        protected_write(store,key,payload)
    except Rejected:
        return store == {KEY:BASE}
    return False

partial = {'received_units':10,'source_revision':8}
raw = {KEY:deepcopy(BASE)}
check(raw_write(raw,KEY,partial) == 'Upsert')
check(raw[KEY] == partial)
check('receipt_ref' not in raw[KEY])
check(rejects(partial))
negative = dict(V8,received_units=-2)
check(raw_write(raw,KEY,negative) == 'Upsert')
check(raw[KEY]['received_units'] == -2)
check(rejects(negative))
standard = {KEY:deepcopy(BASE)}
try:
    raw_write(standard,KEY,negative,kind='standard')
except Rejected:
    check(standard[KEY] == BASE)
else:
    raise AssertionError('Validation standard non exécutée')
store = {KEY:deepcopy(BASE)}
protected_write(store,KEY,V8)
check(store[KEY] == V8)
check(store[KEY]['receipt_ref'] == 'R104' and store[KEY]['unit'] == 'PIECE')
protected_write(store,KEY,V8)
check(len(store) == 1 and store[KEY]['received_units'] == 10)
check(rejects(dict(V8,received_units=None)))
check(rejects(dict(V8,received_units=True)))
check(rejects(dict(V8,received_units=1.5)))
check(rejects(dict(V8,unit='CARTON')))
check(rejects(dict(V8,receipt_ref='R105')))
check(rejects(dict(V8,sku='POMPE-13')))
check(rejects(dict(V8,status='PENDING')))
check(rejects(dict(V8,source_revision=0)))
check(rejects(dict(V8,source_revision=6)))
check(rejects(dict(BASE,received_units=10)))
check(rejects(dict(V8,unexpected='value')))
check(rejects(V8,('ENV-PROD',ROW,'PARIS')))
check(rejects(V8,('ENV-RECETTE',ROW,'LYON')))
check(rejects(V8,('ENV-RECETTE',ROW,None)))
check(confirmed(KEY,V8,V8))
check(not confirmed(KEY,BASE,V8))
check(not confirmed(('ENV-RECETTE',ROW,'LYON'),V8,V8))
failed = {KEY:deepcopy(BASE)}
try:
    raw_write(failed,KEY,V8,guards=frozenset({'Upsert'}),fail_after=True)
except AfterWriteError:
    check(failed[KEY] == V8)
else:
    raise AssertionError('Panne tardive non simulée')
check(confirmed(KEY,failed[KEY],V8))
wrong_partition = {KEY:deepcopy(BASE)}
raw_write(wrong_partition,('ENV-RECETTE',ROW,'LYON'),V8)
check(len(wrong_partition) == 2 and wrong_partition[KEY] == BASE)
print(f'{checks} contrôles réussis ; 3 défauts illustrés : corps partiel, validation non appelée, erreur après écriture.')
print('R104/POMPE-12 : 8 pièces révision7 →10 pièces révision8 ; rejeu :10, pas20.')
print('Partition LYON incorrecte : seconde identité dans le témoin ; protection locale : refus sans effet.')
print('Aucun Dataverse, SDK, plug-in, jeton de session ou accès réseau exécuté.')