dtf-system/tests/workflow_test.py

"""Customer identity, correction and final-file trust boundaries against local stack."""
from uuid import uuid4
from urllib.request import urlopen
from tests.smoke_test import Client, approved_quote, upload_bytes, item_spec

def run():
    customer=Client();other=Client();customer.call('/session');other.call('/session')
    uid=upload_bytes(customer,b'LOCAL ORIGINAL ONLY')
    item=item_spec('file','1.01',0,uid)
    profile={'cnpj':'11222333000181','zap':'11999999999','mail':'workflow-'+uuid4().hex[:8]+'@example.test'}
    q=customer.call('/quotes',{'request_key':str(uuid4()),'customer':profile,'items':[item],'freight':{'service':'pickup'}})
    approved_quote(customer,q,[item])
    order=customer.call('/orders/dev-paid',{'quote_id':q['id']});oid=order['id']
    before=list(customer.jar)[0].value
    password='local-test-password-'+uuid4().hex
    customer.call('/account/register',{'customer':profile,'password':password})
    assert customer.call('/account/me')['customer']['mail']==profile['mail']
    assert customer.call('/customer/orders')['orders'][0]['id']==oid
    # Email/CNPJ do not grant ownership; only current guest session is migrated.
    other.call('/customer/orders/'+oid,expected=404)
    other.call('/account/login',{'email':profile['mail'],'password':'wrong-password'},expected=401)
    revoked=Client()
    import http.cookiejar
    cookie=http.cookiejar.Cookie(0,'dtf_session',before,None,False,'localhost.local',False,False,'/',True,False,None,True,None,None,{},False)
    revoked.jar.set_cookie(cookie)
    revoked.call('/customer/orders',expected=401)
    account_scope=customer.call('/session')['cart_scope']
    cookie.value=account_scope;revoked.jar.set_cookie(cookie)
    revoked.call('/customer/orders',expected=401)
    other.call('/account/login',{'email':profile['mail'],'password':password})
    assert other.call('/customer/orders/'+oid)['id']==oid
    print('PASS: registration claims only current guest records, cross-session account login, revoked sessions, owner UUID is not a credential')

    def move(state,version):
        return customer.call('/operator/orders/'+oid+'/move',{'state':state,'version':version,'reason':'Please replace the artwork' if state=='cor' else ''},operator=True)['version']
    version=move('tra',0)
    customer.call('/operator/orders/'+oid+'/move',{'state':'fil','version':version},operator=True,expected=409)
    final_id=upload_bytes(customer,b'LOCAL FINAL VERSION ONE',order_id=oid)
    customer.call('/uploads/'+final_id,expected=404)
    body={'version':version,'files':[{'item_index':0,'upload_id':final_id}],'note':'Manually checked final'}
    customer.call('/operator/orders/'+oid+'/final-files',body,expected=401)
    version=customer.call('/operator/orders/'+oid+'/final-files',body,operator=True)['version']
    detail=customer.call('/customer/orders/'+oid)
    final=detail['files'][0]
    link=customer.call('/customer/orders/'+oid+'/files/'+final['id']+'/download')
    assert urlopen(link['url']).read()==b'LOCAL FINAL VERSION ONE'
    guest=Client();guest.call('/session');guest.call('/customer/orders/'+oid+'/files/'+final['id']+'/download',expected=404)
    version=move('fil',version);version=move('imp',version);version=move('cor',version)
    customer.call('/customer/orders/'+oid+'/files/'+final['id']+'/download',expected=404)
    premature_final=upload_bytes(customer,b'FINAL BEFORE CUSTOMER CORRECTION',order_id=oid)
    version=customer.call('/operator/orders/'+oid+'/final-files',
        {'version':version,'files':[{'item_index':0,'upload_id':premature_final}],
         'note':'Prepared before customer sent the new correction'},operator=True)['version']
    correction_id=upload_bytes(customer,b'LOCAL CORRECTED ORIGINAL')
    payload={'version':version,'files':[{'item_index':0,'upload_id':correction_id}],'note':'Replaced the artwork as requested'}
    guest.call('/customer/orders/'+oid+'/corrections',payload,expected=404)
    customer.call('/customer/orders/'+oid+'/corrections',{**payload,'version':0},expected=409)
    version=customer.call('/customer/orders/'+oid+'/corrections',payload)['version']
    files=customer.call('/operator/orders/'+oid+'/files',operator=True)
    assert any(f['kind']=='correction' and f['active'] and f['upload_id']==correction_id for f in files)
    assert not any(f['kind']=='final' and f['active'] for f in files), 'new correction kept a stale final active'
    version=move('tra',version)
    customer.call('/operator/orders/'+oid+'/move',{'state':'fil','version':version},operator=True,expected=409)
    new_final=upload_bytes(customer,b'LOCAL FINAL VERSION TWO',order_id=oid)
    version=customer.call('/operator/orders/'+oid+'/final-files',{'version':version,'files':[{'item_index':0,'upload_id':new_final}],'note':'Checked corrected final'},operator=True)['version']
    for state in ('fil','imp','fin'):version=move(state,version)
    detail=other.call('/customer/orders/'+oid)
    assert detail['state']=='fin'
    assert sum(f['active'] and f['kind']=='final' for f in detail['files'])==1
    assert any(h['reason']=='Please replace the artwork' for h in detail['history'])
    print('PASS: final-file gate, final revisions, secure customer downloads, correction history/uploads, old final invalidation and reapproval')
    old_cookie=list(other.jar)[0].value
    other.call('/account/logout',{})
    cookie.value=old_cookie;revoked.jar.set_cookie(cookie);revoked.call('/customer/orders',expected=401)
    other.call('/session');assert other.call('/customer/orders')['orders']==[]
    print('PASS: logout revokes server session and signed-out visitors cannot see account orders')

if __name__=='__main__':run()