117 lines
7.3 KiB
Python
117 lines
7.3 KiB
Python
"""Scoped editorial assignment and authorized business teaser on the guarded live QA server."""
|
|
import json
|
|
import struct
|
|
import uuid
|
|
import zlib
|
|
from datetime import datetime, timezone
|
|
from portal_profiles_http_session import Session, actors, PRIVATE, BASE, EXPECTED, REASON, events
|
|
|
|
source = json.loads((PRIVATE / 'opportunity-semantic-http-qa.json').read_text())
|
|
assert source['result'] == 'passed' and source['target'] == EXPECTED
|
|
case = next(c for c in source['cases'] if c['category'] == 'business_sale')
|
|
record_id = case['id']
|
|
path = PRIVATE / 'opportunity-semantic-publication-http-qa.json'
|
|
proof = {'target': EXPECTED, 'url': BASE, 'started_at': datetime.now(timezone.utc).isoformat(),
|
|
'id': record_id, 'category': 'business_sale', 'active_writes': 0, 'interface_demonstrated': False}
|
|
|
|
|
|
def page(s):
|
|
return s.call('GET', f'/minhas-oportunidades/{record_id}/view', inertia=True)['props']
|
|
|
|
|
|
def action(s, name, status=200, **extra):
|
|
p = page(s)
|
|
if name == 'publish':
|
|
extra = {'confirm_publication': True, 'preview_hash': p['profileContext']['approval']['hash'], **extra}
|
|
return s.call('POST', f'/minhas-oportunidades/{record_id}/workflow',
|
|
{'etag': p['etag'], 'action': name, 'reason': REASON, **extra}, status=status)
|
|
|
|
|
|
def png():
|
|
def chunk(kind, data):
|
|
return struct.pack('>I', len(data)) + kind + data + struct.pack('>I', zlib.crc32(kind + data) & 0xffffffff)
|
|
color = uuid.uuid4().bytes[:3]
|
|
pixels = b''.join(b'\x00' + color * 512 for _ in range(512))
|
|
return b'\x89PNG\r\n\x1a\n' + chunk(b'IHDR', struct.pack('>IIBBBBB', 512, 512, 8, 2, 0, 0, 0)) + chunk(b'IDAT', zlib.compress(pixels)) + chunk(b'IEND', b'')
|
|
|
|
|
|
try:
|
|
administrator = Session('super_first').login()
|
|
types = [c['category'] for c in source['cases']]
|
|
grants = administrator.call('GET', '/perfis/gestao', inertia=True)['props']['assignments']
|
|
for key, responsibilities in [('publisher', ['publication_coordinator']), ('bootstrap', [])]:
|
|
if not any(g['user_id'] == actors[key]['id'] and g['role'] == 'publisher' and g['revoked_at'] is None
|
|
and set(responsibilities).issubset(json.loads(g['responsibilities']))
|
|
and set(types).issubset(json.loads(g['opportunity_types'])) for g in grants):
|
|
administrator.call('POST', '/perfis/atribuicoes', {'user_id': actors[key]['id'], 'role': 'publisher',
|
|
'opportunity_types': types, 'responsibilities': responsibilities,
|
|
'reason': 'Atribuição editorial estritamente sintética QA; sem responsáveis institucionais reais'}, status=201)
|
|
promoter = Session('promoter').login()
|
|
reviewer = Session('validator').login()
|
|
coordinator = Session('publisher').login()
|
|
assigned = Session('bootstrap').login()
|
|
if page(promoter)['workflowState'] == 'published':
|
|
action(promoter, 'request_revision')
|
|
else:
|
|
assert page(promoter)['workflowState'] == 'needs_information'
|
|
p = page(promoter)['values']['canonical']
|
|
p['visibility'] = 'public'
|
|
p['executive_summary'] += ' MARCADOR-RESUMO-PRIVADO-SEMANTICA-QA.'
|
|
p['demand_statement'] += ' MARCADOR-DADOS-PRIVADOS-SEMANTICA-QA.'
|
|
teaser = 'Resumo sintético QA autorizado para publicação; sem informação privada nem validade comercial.'
|
|
p['business_transaction']['teaser'] = teaser
|
|
existing_resources = promoter.call('GET', f'/minhas-oportunidades/{record_id}/resources', inertia=True)['props']['resources']
|
|
existing = next((i for i in existing_resources['images'] if i['payload'].get('caption') == 'Ilustração sintética QA' and not i.get('deleted_at')), None)
|
|
image = {'id': existing['id']} if existing else promoter.call('POST', f'/minhas-oportunidades/{record_id}/resources/images', {
|
|
'etag': page(promoter)['etag'], 'creation_key': str(uuid.uuid4()), 'reason': REASON,
|
|
'payload': json.dumps({'title': 'Imagem sintética QA', 'caption': 'Ilustração sintética QA',
|
|
'alt_text': 'Imagem sintética colorida', 'is_cover': True, 'rights_status': 'owned', 'visibility': 'public'})
|
|
}, multipart=png(), status=201)
|
|
resources = promoter.call('GET', f'/minhas-oportunidades/{record_id}/resources', inertia=True)['props']['resources']
|
|
row = next(i for i in resources['images'] if i['id'] == image['id'])
|
|
proof['image_id'] = image['id']
|
|
assert row['file']['malware_scan_status'] == 'clean'
|
|
if not row['payload'].get('derivatives'):
|
|
promoter.call('POST', f"/minhas-oportunidades/{record_id}/resources/images/{image['id']}/prepare",
|
|
{'etag': page(promoter)['etag'], 'version': row['version'], 'reason': REASON})
|
|
resources = promoter.call('GET', f'/minhas-oportunidades/{record_id}/resources', inertia=True)['props']['resources']
|
|
row = next(i for i in resources['images'] if i['id'] == image['id'])
|
|
assert len(row['derivativePreviews']) >= 2
|
|
for derivative in row['derivativePreviews']:
|
|
assert promoter.call('GET', derivative['url'].replace(BASE, ''), binary=True)
|
|
p['images'] = [image['id']]
|
|
promoter.call('PUT', f'/minhas-oportunidades/{record_id}/draft', {'etag': page(promoter)['etag'], 'canonical': p, 'reason': REASON})
|
|
action(promoter, 'submit')
|
|
action(reviewer, 'assign_review', assignee_user_id=actors['validator']['id'])
|
|
for gate in page(reviewer)['canonicalOptions']['gates']:
|
|
action(reviewer, 'review_gate', gate=gate, gate_decision='approved', evidence_document_ids=[case['document_id']])
|
|
action(reviewer, 'approve')
|
|
action(coordinator, 'assign_publication', assignee_user_id=actors['bootstrap']['id'])
|
|
queue = assigned.call('GET', '/minhas-oportunidades?queue=assigned', inertia=True)['props']
|
|
assert any(r['id'] == record_id for r in queue['records']['data'])
|
|
action(coordinator, 'publish', status=403)
|
|
preview = page(assigned)['publicationPreview']
|
|
assert preview['summary'] == preview['description'] == teaser
|
|
action(assigned, 'publish')
|
|
recovered = page(Session('promoter').login())
|
|
assert recovered['workflowState'] == 'published'
|
|
assert recovered['values']['canonical']['executive_summary'] == p['executive_summary']
|
|
anonymous = Session('outsider')
|
|
public_url = '/api/v1/opportunities/' + recovered['record']['slug']
|
|
output = anonymous.call('GET', public_url)
|
|
serialized = json.dumps(output, ensure_ascii=False)
|
|
assert teaser in serialized and 'MARCADOR-RESUMO-PRIVADO' not in serialized and 'MARCADOR-DADOS-PRIVADOS' not in serialized
|
|
proof.update({'result': 'passed', 'coordinator_id': actors['publisher']['id'], 'assignee_id': actors['bootstrap']['id'],
|
|
'assigned_queue_persisted': True, 'different_publisher_denied': 403, 'published': 'public',
|
|
'public_route': public_url, 'public_teaser_only': True, 'private_content_preserved': True,
|
|
'recovered_in_new_mfa_session': True, 'original_clamd_clean': True, 'derived_previews_recovered': len(row['derivativePreviews']),
|
|
'publication_task_id': recovered['profileContext']['publication_task_id']})
|
|
except Exception as error:
|
|
proof.update({'result': 'failed', 'failure': str(error)})
|
|
raise
|
|
finally:
|
|
proof['events'] = events
|
|
proof['completed_at'] = datetime.now(timezone.utc).isoformat()
|
|
path.write_text(json.dumps(proof, ensure_ascii=False, indent=2), encoding='utf-8')
|
|
print(json.dumps({k: proof[k] for k in ['result', 'target', 'id', 'public_teaser_only', 'active_writes']}))
|