Files
Investidor-V2/tools/independent-documents-http-qa.py
nala 01405cbf8a
tests / ci (push) Canceled after 0s
oportunidades 5
2026-10-10 14:15:28 +02:00

121 lines
8.1 KiB
Python

"""Land originals and explicit room sharing via QA18097, with real MFA and scan."""
import base64
import hashlib
import json
import subprocess
import uuid
from datetime import datetime, timezone
from portal_profiles_http_session import Session, PRIVATE, EXPECTED, REASON, actors, events, PHP, ROOT
actors['legacy_manager'] = json.loads((PRIVATE / 'resource-http-actors-private.json').read_text(encoding='utf-8-sig'))['manager']
path = PRIVATE / 'independent-documents-http-qa.json'
receipt_path = PRIVATE / 'independent-documents-http-receipts-qa.json'
receipts = json.loads(receipt_path.read_text(encoding='utf-8-sig')) if receipt_path.exists() else {}
proof = {'target': EXPECTED, 'started_at': datetime.now(timezone.utc).isoformat(), 'active_writes': 0, 'interface_observed': False, 'land': [], 'room': []}
def save():
receipt_path.write_text(json.dumps(receipts, indent=2), encoding='utf-8')
def originals():
expression = "require 'tests/Support/SyntheticDocument.php';$i=imagecreatetruecolor(16,16);ob_start();imagepng($i);$p=ob_get_clean();ob_start();imagejpeg($i);$j=ob_get_clean();echo json_encode(['pdf'=>base64_encode(Tests\\Support\\SyntheticDocument::pdf()),'png'=>base64_encode($p),'jpg'=>base64_encode($j)]);"
value = subprocess.check_output([PHP, '-d', 'extension=C:/Users/macbook/.config/investmoz/php-8.5.11/ext/php_gd.dll', '-r', expression], cwd=ROOT)
return [(kind, mime, base64.b64decode(data)) for (kind, data), mime in zip(json.loads(value).items(), ['application/pdf', 'image/png', 'image/jpeg'])]
try:
owner = Session('promoter').login()
candidate = Session('colleague').login()
outsider = Session('outsider').login()
manager = Session('legacy_manager').login()
files = originals()
if 'case_id' not in receipts:
result = owner.call('POST', '/investidor/terra-duat', {'payload': {'full_name': 'Synthetic QA document consumer', 'country': 'MZ', 'declared_origin_country_code': 'PT'}}, status=201)
receipts['case_id'] = result['id']; save()
base = '/investidor/terra-duat/' + receipts['case_id']
def case():
response = owner.call('GET', base, inertia=True)
return response.get('props', response)
for kind, mime, data in files:
name = 'qa-consumer-original.' + kind
page = case()
existing = next((d for d in page['documents'] if d['file_name'] == name), None)
if existing is None:
owner.call('POST', base + '/documentos', {'version': page['case']['version'], 'category': 'other'}, multipart=(name, mime, data), status=201)
existing = next(d for d in case()['documents'] if d['file_name'] == name)
url = base + '/documentos/' + existing['id']
original = owner.call('GET', url, binary=True)
assert original == data
manager.call('GET', url, status=404, binary=True)
outsider.call('GET', url, status=404, binary=True)
proof['land'].append({'type': kind, 'attachment_id': existing['id'], 'sha256': hashlib.sha256(data).hexdigest(), 'owner': 21, 'manager_private_route_denied': 404, 'other_owner_denied': 404})
print_page = owner.call('GET', base + '/imprimir', inertia=True)
assert print_page['component'] == 'land-assistance/print' and 'Não constitui' in print_page['props']['disclaimer']
fresh = Session('promoter').login()
response = fresh.call('GET', base, inertia=True)
restored = response.get('props', response)
assert restored['case']['payload']['country'] == 'MZ'
assert restored['case']['payload']['declared_origin_country_code'] == 'PT'
for row, (_, _, data) in zip(proof['land'], files):
assert fresh.call('GET', base + '/documentos/' + row['attachment_id'], binary=True) == data
proof['land_print'] = {'inertia_template_recovered': True, 'pdf_generated_by_browser': False, 'new_mfa_recovery': True, 'country_origin_recovered': ['MZ', 'PT']}
if 'offer_id' not in receipts:
draft = owner.call('POST', '/parcerias/rascunhos', {'client_key': str(uuid.uuid4()), 'title': 'Synthetic QA document sharing only', 'kind': 'seeking', 'partnership_type': 'joint_venture', 'offer': 'Synthetic QA capacity', 'seeking': 'Synthetic QA contribution', 'payload': {'summary': 'Synthetic QA; no actual undertaking', 'consent': True}, 'origins': [], 'locations': []})
receipts['offer_id'] = draft['id']; save()
offer_base = '/parcerias/' + str(receipts['offer_id'])
offer = owner.call('GET', offer_base)['record']
if offer['status'] == 'draft':
owner.call('POST', offer_base + '/estado', {'version': offer['version'], 'action': 'submit', 'reason': REASON})
offer = owner.call('GET', offer_base)['record']
if offer['status'] != 'published':
manager.call('POST', offer_base + '/estado', {'version': offer['version'], 'action': 'publish', 'reason': REASON})
if 'interest_id' not in receipts:
interest = candidate.call('POST', offer_base + '/interesses', {'proposal': 'Synthetic QA sharing request', 'payload': {'consent': True}})
receipts['interest_id'] = interest['id']; save()
interest = next(i for i in owner.call('GET', offer_base)['interests'] if i['id'] == receipts['interest_id'])
for action in ['shortlist', 'connect']:
if interest.get('room_id'):
break
if action == 'shortlist' and interest['status'] not in ['submitted', 'clarification']:
continue
owner.call('POST', '/parcerias/candidaturas/' + str(interest['id']), {'version': interest['version'], 'action': action, 'reason': REASON})
interest = next(i for i in owner.call('GET', offer_base)['interests'] if i['id'] == receipts['interest_id'])
receipts['room_id'] = interest['room_id']; save()
room_base = '/parcerias/salas/' + str(receipts['room_id'])
for kind, mime, data in files:
name = 'qa-room-original.' + kind
current = owner.call('GET', room_base)
row = next((d for d in current['documents'] if d['file_name'] == name), None)
if row is None:
owner.call('POST', room_base + '/documentos', {'shared': 0}, multipart=(name, mime, data))
row = next(d for d in owner.call('GET', room_base)['documents'] if d['file_name'] == name)
url = room_base + '/documentos/' + str(row['id'])
if row['shared']:
owner.call('POST', url + '/partilha', {'shared': False})
assert owner.call('GET', url, binary=True) == data
candidate.call('GET', url, status=404, binary=True)
candidate.call('POST', url + '/partilha', {'shared': True}, status=403)
owner.call('POST', url + '/partilha', {'shared': True})
assert candidate.call('GET', url, binary=True) == data
manager.call('GET', url, status=403, binary=True)
outsider.call('GET', url, status=403, binary=True)
fresh_candidate = Session('colleague').login()
assert fresh_candidate.call('GET', url, binary=True) == data
owner.call('POST', url + '/partilha', {'shared': False})
fresh_candidate.call('GET', url, status=404, binary=True)
assert all(d['id'] != row['id'] for d in fresh_candidate.call('GET', room_base)['documents'])
assert owner.call('GET', url, binary=True) == data
proof['room'].append({'type': kind, 'attachment_id': row['id'], 'sha256': hashlib.sha256(data).hexdigest(), 'private_candidate': 404,
'share_change_by_non_author': 403, 'explicit_share_recovered': True, 'new_mfa_session': True, 'share_revoked': 404,
'manager_non_participant': 403, 'outsider': 403, 'original_preserved': True})
proof['result'] = 'passed'
except Exception as error:
proof.update({'result': 'failed', 'failure': str(error)})
raise
finally:
proof['events'] = events
proof['receipts'] = receipts
proof['completed_at'] = datetime.now(timezone.utc).isoformat()
if path.exists():
path.rename(PRIVATE / ('independent-documents-http-qa-' + datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S') + '.json'))
path.write_text(json.dumps(proof, ensure_ascii=False, indent=2), encoding='utf-8')
print(json.dumps({'result': proof['result'], 'target': EXPECTED, 'land_types': len(proof['land']), 'room_types': len(proof['room']), 'events': len(events), 'active_writes': 0}))