140 lines
8.4 KiB
Python
140 lines
8.4 KiB
Python
"""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}))
|