"""Accepted modular QA contracts on the actual isolated server; no auth material in receipts.""" import copy import hashlib import json import uuid import sys from datetime import datetime, timezone from portal_profiles_http_session import Session, PRIVATE, BASE, EXPECTED, REASON, events receipt_path = PRIVATE / 'opportunity-modular-receipts-qa.json' receipts = json.loads(receipt_path.read_text(encoding='utf-8-sig')) if receipt_path.exists() else {} proof = {'url': BASE, 'target': EXPECTED, 'started_at': datetime.now(timezone.utc).isoformat(), 'interface_observed': False, 'active_writes': 0, 'cases': []} def page(session, ident): return session.call('GET', f'/minhas-oportunidades/{ident}/view', inertia=True)['props'] def save(session, ident, payload): return session.call('PUT', f'/minhas-oportunidades/{ident}/draft', {'etag': page(session, ident)['etag'], 'canonical': payload, 'reason': REASON}) def action(session, ident, name, status=200, **extra): current = page(session, ident) if name == 'publish': extra.update({'confirm_publication': True, 'preview_hash': current['profileContext']['approval']['hash']}) return session.call('POST', f'/minhas-oportunidades/{ident}/workflow', {'etag': current['etag'], 'action': name, 'reason': REASON, **extra}, status=status) def prepare(session, category, mode): source = page(session, 33) original = source['values']['canonical'] exclude = ['images', 'review_due_date', 'module_profile_version'] + ([] if category == 'business_sale' else ['business_transaction']) payload = {k: copy.deepcopy(v) for k, v in original.items() if k not in exclude} payload.update({'opportunity_category': category, 'visibility': 'internal'}) key = category + ':' + mode if key not in receipts: created = session.call('POST', '/minhas-oportunidades/drafts', {'creation_key': str(uuid.uuid4()), 'organization_id': source['profileOrganization'], 'opportunity_category': category, 'canonical': {'opportunity_category': category}, 'reason': REASON}, status=201) receipts[key] = created['id'] receipt_path.write_text(json.dumps(receipts, indent=2), encoding='utf-8') ident = receipts[key] current = page(session, ident) mapping = {} for kind in ['locations', 'documents']: for source_row in source['canonicalOptions']['resources'][kind]: if source_row['id'] not in original[kind]: continue title = 'Synthetic QA modular copy ' + source_row['id'] row = next((r for r in current['canonicalOptions']['resources'][kind] if r['payload'].get('title') == title), None) if row is None: meta = {**source_row['payload'], 'title': title} body = {'etag': page(session, ident)['etag'], 'creation_key': str(uuid.uuid4()), 'reason': REASON, 'payload': meta} url = f'/minhas-oportunidades/{ident}/resources/{kind}' if kind == 'documents': raw = session.call('GET', f'/minhas-oportunidades/33/resources/documents/{source_row["id"]}/download', binary=True) body['payload'] = json.dumps(meta) session.call('POST', url, body, multipart=('qa-modular-original.png', 'image/png', raw), status=201) else: session.call('POST', url, body, status=201) current = page(session, ident) row = next(r for r in current['canonicalOptions']['resources'][kind] if r['payload'].get('title') == title) mapping[source_row['id']] = row['id'] def replace(value): if isinstance(value, dict): return {k: replace(v) for k, v in value.items()} if isinstance(value, list): return [replace(v) for v in value] return mapping.get(value, value) if isinstance(value, str) else value payload = replace(payload) document = payload['documents'][0] payload['official_title'] = 'Synthetic QA modular ' + str(ident) if category == 'business_sale': payload['module_profile_version'] = current['canonicalOptions']['historyVersion'] payload['business_transaction'].update({'history_status': 'no_completed_years', 'available_financial_years': 0, 'financial_periods': [], 'history_declaration': 'Entidade sintética QA recém-criada sem exercícios completos.', 'history_declaration_document_id': document, 'projections': {'scenario': 'Projeção sintética QA', 'source': 'Plano de teste QA', 'date': '2026-10-10', 'document_id': document}}) return ident, payload, document payload['module_profile_version'] = current['canonicalOptions']['modularProfiles'][category] m = {'mode': mode, 'object': 'Objeto sintético QA, sem compromisso real', 'object_document_id': document, 'representation_document_id': document} if category == 'concession': m.update({'granting_entity': 'Entidade sintética QA, não autoridade real', 'procedure_regime': 'Procedimento de teste QA', 'scope': 'Distrito de localização QA', 'term_value': 12, 'term_unit': 'months', 'remuneration_obligations': 'Obrigações sintéticas QA', 'participation_conditions': 'Condições sintéticas QA', 'land_required': False, 'construction_required': False, 'disclosure_authorization_document_id': document}) if mode == 'awarded_right': m['external_decision_document_id'] = document if category in ['financing', 'supply']: m.update({'conditions_status': 'proposed', 'terms': 'Condições indicativas QA, sem compromisso', 'scenario': 'Cenário QA', 'source': 'Fonte sintética QA', 'terms_date': '2026-10-10'}) if category == 'financing': m.update({'instrument': 'debt', 'term_value': 12, 'term_unit': 'months', 'repayment_periodicity': 'mensal', 'disbursement_calendar': 'Calendário indicativo QA'}) if mode == 'seeking_funding': payload.update({'total_investment': 1000, 'investment_breakdown': [{'label': 'QA', 'amount': 1000, 'currency': 'MZN'}], 'secured_financing': [], 'funding_sought': 500}) m.update({'use_of_funds': 'Preparação sintética QA', 'financial_plan_document_id': document}) else: m.update({'offered_capacity': 100000, 'eligibility_criteria': 'Critérios indicativos QA', 'capacity_document_id': document}) if category == 'supply': m.update({'quantity': 10, 'unit': 'unidade', 'destination': 'Distrito QA', 'start_date': '2026-10-11', 'end_date': '2026-11-11', 'acceptance_criteria': 'Critérios sintéticos QA', 'unit_price': 10, 'total_price': 100, 'advance_or_guarantee': False}) if mode == 'supply_offer': m['capacity_document_id'] = document if category == 'other': m.update({'subtype': 'qa-test-subtype', 'purpose': 'Teste QA', 'recipients': 'Destinatários sintéticos QA', 'expected_output': 'Resultado QA', 'risks': 'Riscos declarados QA'}) payload['module_details'] = m return ident, payload, document try: owner = Session('promoter').login() reviewer = Session('validator').login() publisher = Session('publisher').login() outsider = Session('outsider').login() cases = [('concession', 'intention_call'), ('concession', 'awarded_right'), ('financing', 'seeking_funding'), ('financing', 'funding_offer'), ('supply', 'purchase_request'), ('supply', 'supply_offer'), ('other', 'classified_other')] if '--history-only' in sys.argv: cases = [('business_sale', 'no_completed_years')] for category, mode in cases: # Every case uses normal fresh MFA; long suites must not outlive the 15-minute security window. owner = Session('promoter').login() reviewer = Session('validator').login() publisher = Session('publisher').login() ident, payload, document = prepare(owner, category, mode) current = page(owner, ident) initial_state = current['workflowState'] if current['workflowState'] == 'draft': invalid = copy.deepcopy(payload) field = None if mode == 'seeking_funding': invalid['funding_sought'] = 2000; field = 'canonical.funding_sought' if mode == 'funding_offer': invalid['total_investment'] = 100; field = 'canonical.total_investment' if mode == 'awarded_right': invalid['module_details']['external_decision_document_id'] = None; field = 'canonical.module_details.external_decision_document_id' if mode == 'supply_offer': invalid['module_details']['total_price'] = 99; field = 'canonical.module_details.total_price' if category == 'other': invalid['module_details']['subtype'] = 'unclassified-idea'; field = 'canonical.module_details.subtype' if field: save(owner, ident, invalid) assert page(Session('promoter').login(), ident)['values']['canonical']['module_details'] == invalid['module_details'] failure = action(owner, ident, 'submit', status=422) assert field in failure['errors'] # Clear fields left by a failed offer test explicitly; the merge never silently erases values. if mode == 'funding_offer': payload['total_investment'] = None save(owner, ident, payload) action(owner, ident, 'submit') current = page(owner, ident) if current['workflowState'] == 'submitted' and not current['profileContext'].get('correction_cycles'): action(reviewer, ident, 'request_correction', correction_items=['canonical.executive_summary']) current = page(owner, ident) if current['workflowState'] == 'needs_information': payload = current['values']['canonical'] payload['executive_summary'] += ' Correção QA preservada.' save(owner, ident, payload) action(owner, ident, 'submit') current = page(owner, ident) if current['workflowState'] in ['submitted', 'in_review']: for gate in current['canonicalOptions']['gates']: action(reviewer, ident, 'review_gate', gate=gate, gate_decision='approved', evidence_document_ids=[document]) action(owner, ident, 'approve', status=403) action(reviewer, ident, 'approve') if page(owner, ident)['workflowState'] == 'approved': action(publisher, ident, 'publish') recovered = Session('promoter').login().call('GET', f'/minhas-oportunidades/{ident}/response') if category == 'business_sale': assert recovered['data']['business_transaction']['history_status'] == mode assert recovered['data']['business_transaction']['financial_periods'] == [] else: assert recovered['data']['module_details']['mode'] == mode assert recovered['data']['status'] == 'published' and recovered['publication']['scope'] == 'restricted' outsider.call('GET', f'/minhas-oportunidades/{ident}/response', status=403) Session('outsider').call('GET', f'/minhas-oportunidades/{ident}/response', status=401) raw = owner.call('GET', f'/minhas-oportunidades/{ident}/resources/documents/{document}/download', binary=True) resource = next(d for d in recovered['data']['documents'] if d['id'] == document) assert resource['scan_recorded_status'] == 'clean' and resource['sha256'].strip() == hashlib.sha256(raw).hexdigest() proof['cases'].append({'id': ident, 'category': category, 'mode': mode, 'profile_version': recovered['data']['module_profile_version'], 'workflow': 'draft → submit → correction → resubmit → independent gates → approval → restricted release', 'new_mfa_recovery': True, 'same_opportunity_verified_original': True, 'other_organization': 403, 'anonymous': 401, 'public_catalog': False, 'initial_state_in_this_attempt': initial_state, 'subtype_scope': 'synthetic QA acceptance only' if category == 'other' else None}) print(json.dumps({'case': ident, 'category': category, 'mode': mode, 'result': 'passed', 'interface_observed': False}), flush=True) proof['result'] = 'passed' except Exception as error: proof.update({'result': 'failed', 'failure': str(error)}) raise finally: proof['events'] = events proof['completed_at'] = datetime.now(timezone.utc).isoformat() dest = PRIVATE / ('opportunity-history-http-qa.json' if '--history-only' in sys.argv else 'opportunity-modular-http-qa.json') if dest.exists(): previous = json.loads(dest.read_text(encoding='utf-8-sig')) archive = PRIVATE / ('opportunity-modular-http-qa-' + previous['completed_at'].replace(':', '').replace('+', '_') + '.json') if not archive.exists(): archive.write_bytes(dest.read_bytes()) dest.write_text(json.dumps(proof, ensure_ascii=False, indent=2), encoding='utf-8') print(json.dumps({'result': proof['result'], 'cases': len(proof['cases']), 'events': len(events), 'interface_observed': False, 'active_writes': 0}))