88 lines
8.4 KiB
Python
88 lines
8.4 KiB
Python
"""Public reading and authenticated library release/dossier continuity on the guarded QA server."""
|
|
import hashlib
|
|
import json
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from portal_profiles_http_session import BASE, EXPECTED, PRIVATE, Session, actors, events
|
|
|
|
receipt = json.loads((PRIVATE / 'legacy-import-qa.json').read_text(encoding='utf-8'))
|
|
manifest = json.loads(Path('storage/app/private/document-import-staging/legacy-agronegocio/manifest.json').read_text(encoding='utf-8'))
|
|
rows = receipt['resources']
|
|
article = next(r for r in rows if r['file_name'] == 'agronegocio-source.txt')
|
|
image = next(r for r in rows if r['file_name'] == 'Agronegocio.png')
|
|
proof = {'url': BASE, 'target': EXPECTED, 'started_at': datetime.now(timezone.utc).isoformat(), 'interface_observed': False, 'active_writes': 0}
|
|
anonymous = Session('outsider')
|
|
actors['library_manager'] = json.loads((PRIVATE / 'resource-http-actors-private.json').read_text(encoding='utf-8'))['manager']
|
|
manager = Session('library_manager').login()
|
|
for row in rows:
|
|
record = next(r for r in manifest['documents'] if r['source_url'] == row['source_url'])
|
|
if not row['published']:
|
|
anonymous.call('GET', f"/documentos/{row['id']}/view", status=404, binary=True)
|
|
# Separate authenticated metadata/disclosure and publication actions; importer never publishes.
|
|
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 exclusivamente no piloto QA autorizadas pelo responsável em 10/10/2026; sem habilitação ativa.'})
|
|
if row['id'] == article['id']:
|
|
data['associations'] = [{'target_type': 'theme', 'target_key': 'agronegocio', 'relationship': 'supports', 'applicability': 'support', 'basis_reference': row['source_url'] + '; título e conteúdo do artigo', 'review_status': 'confirmed'},
|
|
{'target_type': 'document', 'target_key': str(image['id']), 'relationship': 'related_to', 'applicability': 'support', 'basis_reference': 'thumbnail do post 171985: ' + image['source_url'], 'review_status': 'confirmed'},
|
|
{'target_type': 'stage', 'target_key': 'S01', 'relationship': 'supports', 'applicability': 'support', 'basis_reference': 'Associação editorial do piloto QA: compreender o setor para sistematizar a ideia; não é requisito legal.', '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)
|
|
|
|
landing = anonymous.call('GET', '/explorar/agronegocio', inertia=True)['props']
|
|
content = anonymous.call('GET', '/agronegocio', inertia=True)['props']['article']
|
|
assert landing['legacyContentEnabled'] and content['id'] == article['id'] and content['image']
|
|
assert len(content['territorial_references']) == 6 and content['calculation_use'] is False
|
|
search_hit = anonymous.call('GET', '/explorar/agronegocio?q=corredores', inertia=True)['props']
|
|
search_miss = anonymous.call('GET', '/explorar/agronegocio?q=zzzz-no-content-qa', inertia=True)['props']
|
|
assert search_hit['match'] and not search_miss['match']
|
|
assert hashlib.sha256(anonymous.call('GET', content['image']['href'], binary=True)).hexdigest() == image['sha256'].strip()
|
|
assert hashlib.sha256(anonymous.call('GET', f"/documentos/{article['id']}/download", binary=True)).hexdigest() == article['sha256'].strip()
|
|
anonymous.call('GET', '/investimentos/a-partir-de/agronegocio', status=401, binary=True)
|
|
proof['public'] = {'resource_id': content['id'], 'image_resource_id': image['id'], 'anonymous_read': True, 'image_hash_verified': True,
|
|
'original_hash_verified': True, 'search_hit_and_miss': True, 'territorial_references': content['territorial_references'],
|
|
'published_sector_opportunities': len(content['opportunities']), 'no_opportunity_created_from_article': True}
|
|
for link in ['/carteira?kind=corridor', '/zonas', content['library_href']] + ['/oportunidades/' + o['slug'] for o in content['opportunities']]:
|
|
anonymous.call('GET', link, inertia=True)
|
|
proof['public']['consumer_links_checked'] = ['/carteira?kind=corridor', '/zonas', content['library_href']]
|
|
|
|
owner = Session('promoter').login()
|
|
choice = owner.call('GET', '/investimentos/a-partir-de/agronegocio', inertia=True)['props']
|
|
assert choice['contentSource']['id'] == article['id']
|
|
previous = PRIVATE / 'legacy-content-http-qa.json'
|
|
old = json.loads(previous.read_text(encoding='utf-8')) if previous.exists() else {}
|
|
case = old.get('dossier', {}).get('id') or owner.call('POST', '/investimentos', {'title': 'Synthetic QA — ideia agronegócio', 'client_key': str(uuid.uuid4()), 'reuse_consent': True, 'content_resource_id': article['id']}, status=201)['id']
|
|
first = owner.call('GET', '/investimentos/' + case)
|
|
assert first['content_sources'][0]['resource_id'] == article['id'] and len(first['steps']) == 19 and len(first['phases']) == 5
|
|
body = {'version': first['case']['version'], 'resource_id': article['id']}
|
|
owner.call('POST', f'/investimentos/{case}/conteudos', body)
|
|
replay = owner.call('GET', '/investimentos/' + case)
|
|
assert replay['case']['version'] == first['case']['version'] and len(replay['content_sources']) == 1
|
|
owner.call('POST', '/logout', {}, status=204)
|
|
new_session = Session('promoter').login()
|
|
recovered = new_session.call('GET', '/investimentos/' + case)
|
|
assert recovered['content_sources'] == first['content_sources']
|
|
empty = old.get('dossier', {}).get('independent_id') or new_session.call('POST', '/investimentos', {'title': 'Synthetic QA — outra ideia sem artigo', 'client_key': str(uuid.uuid4()), 'reuse_consent': True}, status=201)['id']
|
|
assert new_session.call('GET', '/investimentos/' + empty)['content_sources'] == []
|
|
continuing = old.get('dossier', {}).get('continue_id') or new_session.call('POST', '/investimentos', {'title': 'Synthetic QA — ideia existente para continuar', 'client_key': str(uuid.uuid4()), 'reuse_consent': True}, status=201)['id']
|
|
before_continue = new_session.call('GET', '/investimentos/' + continuing)
|
|
new_session.call('POST', f'/investimentos/{continuing}/conteudos', {'version': before_continue['case']['version'], 'resource_id': article['id']})
|
|
continued = Session('promoter').login().call('GET', '/investimentos/' + continuing)
|
|
assert len(continued['content_sources']) == 1 and continued['content_sources'][0]['resource_id'] == article['id']
|
|
other = Session('outsider').login()
|
|
other.call('GET', '/investimentos/' + case, status=403, binary=True)
|
|
other.call('POST', f'/investimentos/{case}/conteudos', body, status=403, binary=True)
|
|
other.call('PUT', f"/management/portal-services/resources/{article['id']}", {}, status=403, binary=True)
|
|
assert new_session.call('GET', '/agronegocio?context=' + case, inertia=True)['props']['contentContext']['id'] == case
|
|
assert other.call('GET', '/agronegocio?context=' + case, inertia=True)['props']['contentContext'] is None
|
|
assert anonymous.call('GET', '/agronegocio?context=' + case, inertia=True)['props']['contentContext'] is None
|
|
proof['public']['private_return_context_authorized_and_hidden'] = True
|
|
proof['dossier'] = {'id': case, 'independent_id': empty, 'continue_id': continuing, 'existing_context_attached_and_recovered': True, 'new_mfa_session_recovered': True, 'replay_single_reference_no_version_increment': True,
|
|
'outside_account_read_write_denied': True, 'public_source_return_href': recovered['content_sources'][0]['return_href'], 'phases': 5, 'steps': 19}
|
|
proof.update(result='passed', events=events, completed_at=datetime.now(timezone.utc).isoformat())
|
|
(PRIVATE / 'legacy-content-http-qa.json').write_text(json.dumps(proof, ensure_ascii=False, indent=2), encoding='utf-8')
|
|
print(json.dumps({'result': proof['result'], 'public_resources': [article['id'], image['id']], 'events': len(events), 'new_session_recovery': True, 'interface_observed': False, 'active_writes': 0}))
|