135 lines
9.8 KiB
Python
135 lines
9.8 KiB
Python
"""Separate authenticated QA disclosure, real originals, and sector dossier continuity."""
|
|
import hashlib
|
|
import json
|
|
import sys
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from portal_profiles_http_session import BASE, EXPECTED, PRIVATE, ROOT, Session, actors, events
|
|
|
|
proof = {'started_at': datetime.now(timezone.utc).isoformat(), 'target': EXPECTED, 'base': BASE,
|
|
'interface_observed': False, 'active_writes': 0, 'phase': 'initialization'}
|
|
proof_path = PRIVATE / 'legacy-sectors-http-qa.json'
|
|
previous = json.loads(proof_path.read_text(encoding='utf-8')) if proof_path.exists() else {}
|
|
if previous:
|
|
archive = PRIVATE / ('legacy-sectors-http-attempt-' + previous['started_at'].replace(':', '-') + '.json')
|
|
if not archive.exists():
|
|
archive.write_text(json.dumps(previous, ensure_ascii=False, indent=2), encoding='utf-8')
|
|
|
|
def save():
|
|
proof_path.write_text(json.dumps(proof, ensure_ascii=False, indent=2), encoding='utf-8')
|
|
|
|
def run():
|
|
receipt = json.loads((PRIVATE / 'legacy-sectors-verify-qa.json').read_text(encoding='utf-8'))
|
|
manifest = json.loads((ROOT / 'storage/app/private/document-import-staging/legacy-setores-20261010/manifest.json').read_text(encoding='utf-8'))
|
|
assert receipt['target'] == {'database': 'desk_restore_20261009_174726_test', 'port': 55441}
|
|
assert len(receipt['resources']) == 22 and all(c['clean_scan_for_hash'] and c['stored_original_equal'] for c in receipt['checks'])
|
|
actors['library_manager'] = json.loads((PRIVATE / 'resource-http-actors-private.json').read_text(encoding='utf-8'))['manager']
|
|
anonymous = Session('outsider')
|
|
manager = Session('library_manager').login()
|
|
rows = receipt['resources']
|
|
proof['phase'] = 'separate_authenticated_disclosure'
|
|
proof['publication_actions'] = []
|
|
for row in rows:
|
|
record = next(r for r in manifest['documents'] if r['external_id'] == row['external_id'])
|
|
assert hashlib.sha256(manager.call('GET', f"/management/documents/{row['document_id']}/download", binary=True)).hexdigest() == row['sha256'].strip()
|
|
if record['external_id'].startswith('legacy-tempo-post-171994-'):
|
|
assert not row['published']
|
|
anonymous.call('GET', f"/documentos/{row['id']}/view", status=404, binary=True)
|
|
proof['empty_finance_private'] = {'resource_id': row['id'], 'external_id': row['external_id'], 'published': False}
|
|
continue
|
|
if not row['published']:
|
|
anonymous.call('GET', f"/documentos/{row['id']}/view", status=404, binary=True)
|
|
data = {k: record[k] for k in ['title', 'issuer', 'category', 'topic', 'document_type', 'source_reference', 'source_url', 'language', 'version_label', 'description']}
|
|
data.update({'document_id': row['document_id'], 'document_date': record.get('document_date'), 'metadata_version': row['metadata_version'],
|
|
'territorial_scope': 'unknown', 'source_confirmation': 'identified', 'legal_status': 'not_applicable',
|
|
'rights_status': 'authorized', 'data_classification': 'public', 'independent_review': False,
|
|
'legislation': False, 'essential': False,
|
|
'source_reference': record['source_reference'] + '; utilização e divulgação do legado exclusivamente em QA autorizadas pelo responsável em 10/10/2026; sem habilitação ativa.'})
|
|
if record['external_id'].startswith('legacy-tempo-post-'):
|
|
source = json.loads((ROOT / 'storage/app/private/document-import-staging' / record['relative_path']).read_text(encoding='utf-8'))
|
|
image = next((r for r in rows if r['sha256'].strip() == source.get('image_sha256')), None)
|
|
if image:
|
|
data['associations'] = [{'target_type': 'document', 'target_key': str(image['id']), 'relationship': 'related_to',
|
|
'applicability': 'support', 'basis_reference': 'Thumbnail identificada no exemplar original do post: ' + image['source_url'], 'review_status': 'confirmed'}]
|
|
manager.call('PUT', f"/management/portal-services/resources/{row['id']}", data, status=302, binary=True)
|
|
manager.call('POST', f"/management/portal-services/resources/{row['id']}", {'published': True}, binary=True)
|
|
proof['publication_actions'].append({'resource_id': row['id'], 'actor_id': actors['library_manager']['id'], 'authenticated': True,
|
|
'mfa_challenge_observed': any(e['actor_id'] == actors['library_manager']['id'] and e['path'] == '/two-factor-challenge' and e['status'] == 204 for e in events)})
|
|
save()
|
|
assert hashlib.sha256(anonymous.call('GET', f"/documentos/{row['id']}/download", binary=True)).hexdigest() == row['sha256'].strip()
|
|
|
|
proof['phase'] = 'public_consumers'
|
|
articles = {}
|
|
for group, expected in [('energia-mineracao', 3), ('turismo', 12), ('infraestruturas', 1)]:
|
|
landing = anonymous.call('GET', '/explorar/' + group, inertia=True)['props']
|
|
assert len(landing['articles']) == expected
|
|
for brief in landing['articles']:
|
|
article = anonymous.call('GET', brief['href'], inertia=True)['props']['article']
|
|
assert article['html'].strip() and article['image'] and not article['calculation_use']
|
|
assert article['territorial_references'] == []
|
|
assert hashlib.sha256(anonymous.call('GET', article['image']['href'], binary=True)).hexdigest() == article['image']['sha256']
|
|
anonymous.call('GET', article['library_href'], inertia=True)
|
|
anonymous.call('GET', article['start_href'], status=401, binary=True)
|
|
articles[brief['href']] = article
|
|
missing = anonymous.call('GET', '/explorar/' + group + '?q=zzzz-no-content-qa', inertia=True)['props']
|
|
assert not missing['match'] and missing['articles'] == []
|
|
hit = anonymous.call('GET', '/explorar/energia-mineracao?q=Energia', inertia=True)['props']
|
|
assert hit['match'] and any(a['href'] == '/energia' for a in hit['articles'])
|
|
assert articles['/energia']['id'] != articles['/mineracao']['id']
|
|
assert 'href="/energia"' in articles['/energia-mineracao']['html'] and 'href="/mineracao"' in articles['/energia-mineracao']['html']
|
|
assert '<h2' in articles['/infraestruturas']['html']
|
|
anonymous.call('GET', '/mercado-financeiro', status=404, binary=True)
|
|
anonymous.call('GET', '/explorar/mercado-financeiro', status=404, binary=True)
|
|
proof['public'] = {'new_nonempty_articles': len(articles), 'distinct_energy_mining': True,
|
|
'tourism_editorial_only': True, 'finance_empty_page_unavailable': True,
|
|
'image_and_original_hashes': True, 'search_hit_and_miss': True, 'infra_original_heading_preserved': True,
|
|
'urls': list(articles)}
|
|
save()
|
|
|
|
proof['phase'] = 'dossier_continuity'
|
|
owner = Session('promoter').login()
|
|
other = Session('outsider').login()
|
|
proof['dossiers'] = previous.get('dossiers', {})
|
|
for href in ['/energia', '/mineracao', '/turismo', '/infraestruturas']:
|
|
article = articles[href]
|
|
choice = owner.call('GET', article['start_href'], inertia=True)['props']
|
|
assert choice['contentSource']['id'] == article['id']
|
|
entry = proof['dossiers'].setdefault(href, {})
|
|
if not entry.get('id'):
|
|
entry['id'] = owner.call('POST', '/investimentos', {'title': 'Synthetic QA — ideia ' + article['title'],
|
|
'client_key': str(uuid.uuid4()), 'reuse_consent': True, 'content_resource_id': article['id']}, status=201)['id']
|
|
save()
|
|
case = entry['id']
|
|
first = owner.call('GET', '/investimentos/' + case)
|
|
assert len(first['phases']) == 5 and len(first['steps']) == 19
|
|
assert len(first['content_sources']) == 1 and first['content_sources'][0]['resource_id'] == article['id']
|
|
owner.call('POST', f'/investimentos/{case}/conteudos', {'version': first['case']['version'], 'resource_id': article['id']})
|
|
repeated = owner.call('GET', '/investimentos/' + case)
|
|
assert repeated['case']['version'] == first['case']['version'] and repeated['content_sources'] == first['content_sources']
|
|
other.call('GET', '/investimentos/' + case, status=403, binary=True)
|
|
other.call('POST', f'/investimentos/{case}/conteudos', {'version': first['case']['version'], 'resource_id': article['id']}, status=403, binary=True)
|
|
assert other.call('GET', href + '?context=' + case, inertia=True)['props']['contentContext'] is None
|
|
assert anonymous.call('GET', href + '?context=' + case, inertia=True)['props']['contentContext'] is None
|
|
entry.update(resource_id=article['id'], return_href=href, other_account_denied=True, reference_replay_idempotent=True)
|
|
save()
|
|
owner.call('POST', '/logout', {}, status=204)
|
|
recovered = Session('promoter').login()
|
|
for href, entry in proof['dossiers'].items():
|
|
view = recovered.call('GET', '/investimentos/' + entry['id'])
|
|
assert len(view['content_sources']) == 1 and view['content_sources'][0]['return_href'] == href
|
|
assert recovered.call('GET', href + '?context=' + entry['id'], inertia=True)['props']['contentContext']['id'] == entry['id']
|
|
entry['new_mfa_session_recovered'] = True
|
|
proof.update(result='passed', completed_at=datetime.now(timezone.utc).isoformat(), events=events, actor_id=21)
|
|
save()
|
|
print(json.dumps({'result': 'passed', 'new_articles': len(articles), 'publication_actions': len(proof['publication_actions']),
|
|
'dossiers': len(proof['dossiers']), 'events': len(events), 'finance_private': True, 'active_writes': 0}))
|
|
|
|
try:
|
|
run()
|
|
except Exception as error:
|
|
proof.update(result='failed', error_type=type(error).__name__, events=events)
|
|
save()
|
|
print(json.dumps({'result': 'failed', 'phase': proof['phase'], 'error_type': type(error).__name__, 'details': 'private sanitized receipt'}))
|
|
sys.exit(1)
|