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

90 lines
5.7 KiB
Python

"""Real PDF lifecycle and scoped downloads on QA18097; sanitized receipts only."""
import base64
import hashlib
import json
import subprocess
import uuid
import io
import sys
from datetime import datetime, timezone
from portal_profiles_http_session import Session, PRIVATE, EXPECTED, REASON, events, PHP, ROOT
sys.path.insert(0, str(PRIVATE / 'pdf-proof-tooling'))
import pypdf
source = json.loads((PRIVATE / 'opportunity-semantic-http-qa.json').read_text())
assert source['result'] == 'passed' and source['target'] == EXPECTED
record_id = next(c['id'] for c in source['cases'] if c['category'] == 'concession')
path = PRIVATE / 'opportunity-semantic-documents-http-qa.json'
proof = {'target': EXPECTED, 'id': record_id, 'started_at': datetime.now(timezone.utc).isoformat(),
'active_writes': 0, 'interface_demonstrated': False}
def page(s):
return s.call('GET', f'/minhas-oportunidades/{record_id}/edit', inertia=True)['props']
def resources(s):
return s.call('GET', f'/minhas-oportunidades/{record_id}/resources', inertia=True)['props']['resources']
try:
promoter = Session('promoter').login()
label = 'Synthetic QA PDF - no official or legal value - ' + str(uuid.uuid4())
expression = "require 'tests/Support/SyntheticDocument.php';echo base64_encode(Tests\\Support\\SyntheticDocument::pdf('" + label + "'));"
data = base64.b64decode(subprocess.check_output([PHP, '-r', expression], cwd=ROOT))
reader = pypdf.PdfReader(io.BytesIO(data), strict=True)
assert len(reader.pages) == 1 and label in reader.pages[0].extract_text()
existing = next((d for d in resources(promoter)['documents'] if d['payload'].get('title') == 'PDF sintético QA válido — sem valor oficial'), None)
if existing and existing['deleted_at'] is not None:
promoter.call('POST', f"/minhas-oportunidades/{record_id}/resources/documents/{existing['id']}/lifecycle", {
'etag': page(promoter)['etag'], 'version': existing['version'], 'action': 'restore', 'reason': REASON, 'confirm': True})
uploaded = {'id': existing['id']} if existing else promoter.call('POST', f'/minhas-oportunidades/{record_id}/resources/documents', {
'etag': page(promoter)['etag'], 'creation_key': str(uuid.uuid4()), 'reason': REASON,
'payload': json.dumps({'title': 'PDF sintético QA válido — sem valor oficial', 'category': 'qa-synthetic',
'classification': 'restricted', 'visibility': 'internal'})
}, multipart=data, status=201)
document_id = uploaded['id']
proof['document_resource_id'] = document_id
if existing:
data = promoter.call('GET', f'/minhas-oportunidades/{record_id}/resources/documents/{document_id}/download', binary=True)
reader = pypdf.PdfReader(io.BytesIO(data), strict=True)
assert len(reader.pages) == 1 and 'no official or legal value' in reader.pages[0].extract_text()
row = next(d for d in resources(promoter)['documents'] if d['id'] == document_id)
digest = hashlib.sha256(data).hexdigest()
assert row['file']['media_type'] == 'application/pdf' and row['file']['malware_scan_status'] == 'clean'
assert row['file']['sha256'].strip() == digest
url = f'/minhas-oportunidades/{record_id}/resources/documents/{document_id}/download'
assert promoter.call('GET', url, binary=True) == data
for key in ['validator', 'publisher']:
assert Session(key).login().call('GET', url, binary=True) == data
outsider = Session('outsider').login()
outsider.call('GET', url, status=403, binary=True)
outsider.call('POST', f'/minhas-oportunidades/{record_id}/resources/documents/{document_id}/lifecycle', {
'etag': page(promoter)['etag'], 'version': row['version'], 'action': 'delete', 'reason': REASON, 'confirm': True}, status=403)
promoter.call('POST', f'/minhas-oportunidades/{record_id}/resources/documents/{document_id}/lifecycle', {
'etag': page(promoter)['etag'], 'version': row['version'], 'action': 'delete', 'reason': REASON, 'confirm': True})
deleted = next(d for d in resources(promoter)['documents'] if d['id'] == document_id)
assert deleted['deleted_at'] is not None
assert document_id not in [d['id'] for d in page(promoter)['canonicalOptions']['resources']['documents']]
promoter.call('GET', url, status=404, binary=True)
promoter.call('POST', f'/minhas-oportunidades/{record_id}/resources/documents/{document_id}/lifecycle', {
'etag': page(promoter)['etag'], 'version': deleted['version'], 'action': 'restore', 'reason': REASON, 'confirm': True})
new_session = Session('promoter').login()
assert new_session.call('GET', url, binary=True) == data
restored = next(d for d in resources(new_session)['documents'] if d['id'] == document_id)
assert restored['deleted_at'] is None and restored['file']['sha256'].strip() == digest
assert document_id in [d['id'] for d in page(new_session)['canonicalOptions']['resources']['documents']]
proof.update({'result': 'passed', 'document_resource_id': document_id, 'media_type': 'application/pdf',
'strict_parse': True, 'pages': 1, 'sha256': digest, 'clamd_real_clean': True,
'download_allowed_profiles': [21, 24, 25], 'foreign_organization_download': 403, 'foreign_organization_delete': 403,
'deleted_download': 404, 'inactive_not_offered_by_editor': True, 'restored_original_identical': True,
'recovered_in_new_mfa_session': True})
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', 'clamd_real_clean', 'restored_original_identical', 'active_writes']}))