"""Read adapter and §27 semantics on QA18097. No credentials, cookies or MFA output.""" import copy import json import uuid import sys from datetime import datetime, timezone from portal_profiles_http_session import Session, PRIVATE, EXPECTED, REASON, events receipt_path = PRIVATE / 'opportunity-response-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, 'responses': [], 'semantics': []} resume_semantics = '--semantics-only' in sys.argv if resume_semantics: previous = json.loads((PRIVATE / 'opportunity-response-http-qa.json').read_text(encoding='utf-8-sig')) assert len(previous['responses']) == 10 and previous['active_writes'] == 0 proof['responses'] = previous['responses'] proof['response_read_scope_from_attempt_at'] = previous['completed_at'] proof['response_scope_note'] = 'Ten read/recovery criteria passed in the preserved preceding attempt; its subsequent harness correction failed. New semantic receipts below are separate.' def save_receipts(): receipt_path.write_text(json.dumps(receipts, indent=2), encoding='utf-8') def page(session, ident): return session.call('GET', f'/minhas-oportunidades/{ident}/view', inertia=True)['props'] def persist(session, ident, payload): current = page(session, ident) return session.call('PUT', f'/minhas-oportunidades/{ident}/draft', {'etag': current['etag'], 'canonical': payload, 'reason': REASON}) def submit(session, ident, status=200): current = page(session, ident) return session.call('POST', f'/minhas-oportunidades/{ident}/workflow', {'etag': current['etag'], 'action': 'submit', 'reason': REASON}, status=status) def clone(session, source_id, key): source = page(session, source_id) payload = copy.deepcopy(source['values']['canonical']) if key not in receipts: row = session.call('POST', '/minhas-oportunidades/drafts', {'creation_key': str(uuid.uuid4()), 'organization_id': source['profileOrganization'], 'opportunity_category': payload['opportunity_category'], 'reason': REASON}, status=201) receipts[key] = row['id']; save_receipts() ident = receipts[key] existing = page(session, ident) mapping = {} for kind in ['locations', 'documents']: for source_row in source['canonicalOptions']['resources'][kind]: if source_row['id'] not in payload[kind]: continue row = next((r for r in existing['canonicalOptions']['resources'][kind] if r['payload'].get('title') == 'QA response copy ' + source_row['id']), None) if row is None: resource_payload = {**source_row['payload'], 'title': 'QA response copy ' + source_row['id']} body = {'etag': page(session, ident)['etag'], 'creation_key': str(uuid.uuid4()), 'reason': REASON, 'payload': resource_payload} url = f'/minhas-oportunidades/{ident}/resources/{kind}' if kind == 'documents': raw = session.call('GET', f'/minhas-oportunidades/{source_id}/resources/documents/{source_row["id"]}/download', binary=True) body['payload'] = json.dumps(resource_payload) session.call('POST', url, body, multipart=('qa-response-original.png', 'image/png', raw), status=201) else: session.call('POST', url, body, status=201) existing = page(session, ident) row = next(r for r in existing['canonicalOptions']['resources'][kind] if r['payload'].get('title') == resource_payload['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) payload['official_title'] = 'Synthetic QA response semantics ' + str(ident) if 'land_rights' in payload: payload['land_rights']['duat_number'] = 'QA-RESPONSE-' + str(ident) return ident, payload try: owner = Session('promoter').login() reviewer = Session('validator').login() outsider = Session('outsider').login() fresh = Session('promoter').login() for ident in ([] if resume_semantics else range(30, 40)): current = page(owner, ident) original = current['values']['canonical'] response = owner.call('GET', f'/minhas-oportunidades/{ident}/response') assert response == current['canonicalResponse'] assert response['data']['id'] == ident assert not response['compatibility']['normative_response_complete'] assert not response['validation_result']['revalidated_on_read'] assert 'completion_percentage' not in response['data'] assert response['resource_references'] == {k: original.get(k, []) for k in ['documents', 'images', 'locations']} assert response == fresh.call('GET', f'/minhas-oportunidades/{ident}/response') for kind in ['documents', 'images', 'locations']: assert [r['id'] for r in response['data'][kind]] == original.get(kind, []) proof['responses'].append({'id': ident, 'adapter_version': response['adapter_version'], 'ids_preserved': True, 'new_mfa_recovery': True, 'inertia_consumer_matches_endpoint': True, 'validation_not_invented': True, 'normative_response_complete': False}) reviewer.call('GET', '/minhas-oportunidades/33/response') outsider.call('GET', '/minhas-oportunidades/33/response', status=403) Session('outsider').call('GET', '/minhas-oportunidades/33/response', status=401) land_id, land = clone(owner, 32, 'land_duplicate') current = page(owner, land_id) if current['workflowState'] == 'draft': land['land_rights']['parcels'] = [copy.deepcopy(land['land_rights']['parcels'][0]) for _ in range(2)] for parcel in land['land_rights']['parcels']: parcel['cadastral_number'] = 'QA-DUPLICATE-' + str(land_id) persist(owner, land_id, land) failure = submit(owner, land_id, status=422) assert 'canonical.land_rights.parcels.1.cadastral_number' in failure['errors'] recovered = page(Session('promoter').login(), land_id) assert recovered['values']['canonical']['land_rights']['parcels'] == land['land_rights']['parcels'] land['land_rights']['parcels'] = copy.deepcopy(land['land_rights']['parcels']) land['land_rights']['parcels'][1]['cadastral_number'] += '-SECOND' persist(owner, land_id, land) submit(owner, land_id) proof['semantics'].append({'id': land_id, 'rule': '§27 cadastral number unique within opportunity', 'invalid_submission': 422, 'field': 'canonical.land_rights.parcels.1.cadastral_number', 'invalid_draft_recovered': True, 'corrected_submission': 'submitted'}) business_id, business = clone(owner, 33, 'succession') if page(owner, business_id)['workflowState'] == 'draft': business['visibility'] = 'internal' business['images'] = [] business['business_transaction']['transaction_type'] = 'succession' persist(owner, business_id, business) submit(owner, business_id) response = Session('promoter').login().call('GET', f'/minhas-oportunidades/{business_id}/response') assert response['data']['business_transaction']['transaction_type'] == 'succession' assert response['data']['status'] == 'submitted' proof['semantics'].append({'id': business_id, 'rule': 'original business_sale succession modality', 'saved': True, 'submitted': True, 'new_mfa_recovery': 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() dest = PRIVATE / 'opportunity-response-http-qa.json' if dest.exists(): dest.rename(PRIVATE / ('opportunity-response-http-qa-' + datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S') + '.json')) dest.write_text(json.dumps(proof, ensure_ascii=False, indent=2), encoding='utf-8') print(json.dumps({'result': proof['result'], 'responses': len(proof['responses']), 'semantics': len(proof['semantics']), 'events': len(events), 'active_writes': 0}))