"""Real HTTP/session evidence in guarded loopback QA; never visual acceptance.""" import hashlib import http.cookiejar import json import re import shutil import urllib.error import urllib.parse import urllib.request import uuid from datetime import datetime, timedelta, timezone from pathlib import Path ROOT = Path(__file__).resolve().parent.parent OUT = ROOT / 'storage/app/private/investmoz-v2/portal-audit' BASE = 'http://127.0.0.1:18097' backup = json.loads((OUT.parent / 'opportunity-resources-backup-proof.json').read_text(encoding='utf-8-sig')) target = backup['restored_target'] assert backup['restore_verified'] and re.fullmatch(r'desk_restore_\d{8}_\d{6}_test', target['database']) assert int(target['port']) == 55441 EXPECTED = target['database'] + ':55441' events = [] class Session: def __init__(self, label): self.label = label self.jar = http.cookiejar.CookieJar() self.opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(self.jar)) self.call('GET', '/login', status=200, html=True) def identity(self): values = [c.value for c in self.jar if c.name == 'opportunity_resources_qa'] assert len(values) == 1, 'QA session cookie missing' return hashlib.sha256(values[0].encode()).hexdigest() def call(self, method, path, data=None, status=200, html=False, extra=None): assert path.startswith('/') and not path.startswith('//') headers = {'Accept': 'text/html' if html else 'application/json', 'Content-Type': 'application/json'} for cookie in self.jar: if cookie.name == 'XSRF-TOKEN': headers['X-XSRF-TOKEN'] = urllib.parse.unquote(cookie.value) headers.update(extra or {}) request = urllib.request.Request(BASE + path, data=None if data is None else json.dumps(data).encode(), headers=headers, method=method) try: response = self.opener.open(request) except urllib.error.HTTPError as error: response = error assert response.headers.get('X-Investmoz-QA-Target') == EXPECTED, 'QA identity absent or changed' raw = response.read() assert response.status == status, f'{self.label} {method} {path}: expected {status}, observed {response.status}; response body withheld' events.append({'actor': self.label, 'method': method, 'url': path, 'status': response.status, 'qa_target_verified': True}) return None if html or not raw else json.loads(raw) def login(self, email, password): self.call('POST', '/login', {'email': email, 'password': password}) # Verify isolation before constructing any mutation. owner = Session('investor') owner.login('investor-path-qa@example.invalid', 'Investor-Path-QA-2026!') manager = Session('manager') manager.login('opportunity-manager@example.invalid', 'Opportunity-QA-2026!') installation = manager.call('GET', '/management/balcao-install') assert installation['pathInstalled'] and not installation['pathQaVerified'] scenarios = [] for index, name in enumerate(['A', 'B']): view_id = owner.call('POST', '/investimentos', { 'title': f'Cenário {name} HTTP QA — ' + ('terra e financiamento' if index else 'sem terra nem financiamento externo'), 'client_key': f'61000919-0000-4000-8000-00000000000{index + 3}', 'reuse_consent': True, }, status=201)['id'] url = '/investimentos/' + view_id view = owner.call('GET', url) assert len(view['steps']) == 19 and len(view['phases']) == 5 if view['facts']['land_needed'] == 'unknown': owner.call('POST', url + '/aplicabilidade', {'version': view['case']['version'], 'land_needed': 'yes' if index else 'no', 'external_funding': 'yes' if index else 'no', 'viability': 'proceed', 'reason': 'Declaração sintética QA; nenhuma decisão externa.'}) draft = owner.call('GET', '/api/v1/simulation-drafts/current?context=' + view_id)['data'] if not draft: snapshot = {'version': 1, 'data': {'budgetPeriod': 'Mensal', 'loan': {'principal': '100000', 'rate': '0', 'months': '24', 'rateType': 'nominal', 'periodMonths': '1', 'gracePeriods': '0', 'upfrontFee': '0', 'recurringFee': '0'}, 'breakEven': {'fixed': '1000', 'price': '100', 'variable': '50'}, 'funding': {'total': '100000', 'secured': '30000' if index else '100000'}, 'incomes': [{'label': 'Receita sintética QA', 'amount': '100'}], 'expenses': [{'label': 'Custo sintético QA', 'amount': '50'}]}} draft = owner.call('POST', '/api/v1/simulation-drafts', {'context': view_id, 'personal_snapshot': {'fiscal_residency_country': 'PT'}, 'economic_snapshot': {'purpose': 'simulate_only', 'project_title': name + ' HTTP QA', 'project_summary': 'Descrição ' + name + ' HTTP QA', 'currency': 'MZN', 'funds_origin_countries': ['PT', 'ZA'], 'financial_tools_snapshot': json.dumps(snapshot)}}, status=201, extra={'Idempotency-Key': str(uuid.uuid4())})['data'] stored_snapshot = json.loads(draft['economic_snapshot']['financial_tools_snapshot']) if stored_snapshot['data']['incomes'] == [] and stored_snapshot['data']['expenses'] == []: # Correct only this tool's synthetic incomplete fixture by a new draft/result version. # Old frozen results remain immutable and must report incompatibility, not invented zero. assert draft['economic_snapshot']['project_title'] == name + ' HTTP QA' stored_snapshot['data']['incomes'] = [{'label': 'Receita sintética QA', 'amount': '100'}] stored_snapshot['data']['expenses'] = [{'label': 'Custo sintético QA', 'amount': '50'}] draft = owner.call('PUT', '/api/v1/simulation-drafts/' + draft['id'], {'context': view_id, 'personal_snapshot': draft['personal_snapshot'], 'economic_snapshot': {**draft['economic_snapshot'], 'financial_tools_snapshot': json.dumps(stored_snapshot)}}, extra={'If-Match': '"sfd-' + draft['id'] + '-v' + str(draft['version_no']) + '"'})['data'] view = owner.call('GET', url) result = owner.call('POST', url + '/resultados', {'version': view['case']['version'], 'draft_id': draft['id'], 'draft_version': draft['version_no']})['result'] view = owner.call('GET', url) if not view['requests']: assert view['services'], 'Published QA support service required' owner.call('POST', '/balcao-virtual/dossiers/' + view_id + '/pedidos', {'service_version_id': view['services'][0]['id'], 'client_key': str(uuid.uuid4())}, status=201) view = owner.call('GET', url) if index and next(s for s in view['steps'] if s['code'] == 'S10')['saved_state'] != 'waiting_external': owner.call('POST', url + '/etapas/S10', {'version': view['case']['version'], 'state': 'waiting_external', 'action': 'prepare', 'reason': 'Aguarda informação do terreno; mercado pode avançar.'}) view = owner.call('GET', url) if not view['tasks']: owner.call('POST', url + '/tarefas', {'version': view['case']['version'], 'code': 'S02', 'title': 'Mercado independente ' + name + ' HTTP QA', 'due_at': (datetime.now(timezone.utc) + timedelta(days=7)).isoformat(), 'consent': True}, status=201) task = owner.call('GET', url)['tasks'][0] managed = manager.call('GET', url) assert managed['canReview'] and not managed['canWrite'] and managed['results'] == [] manager.call('GET', '/api/v1/simulation-drafts/' + draft['id'], status=404) for state in ['in_progress', 'done']: task = next(t for t in manager.call('GET', url)['tasks'] if t['id'] == task['id']) if task['status'] != 'done': manager.call('POST', '/balcao-virtual/dossiers/' + view_id + '/tarefas/' + task['id'], {'version': task['version_no'], 'status': state, 'reason': 'Acompanhamento interno sintético; tarefa independente.'}) scenarios.append({'scenario': name, 'id': view_id, 'url': url, 'draft_id': draft['id'], 'result_id': result['id'], 'ui_passed': False}) prior_session = owner.identity() owner.call('POST', '/logout', status=204) owner.call('GET', scenarios[0]['url'], status=401) owner = Session('investor-new-session') # fresh cookie jar; not merely a new request owner.login('investor-path-qa@example.invalid', 'Investor-Path-QA-2026!') assert owner.identity() != prior_session for entry in scenarios: view = owner.call('GET', entry['url']) recovered = next(r for r in view['results'] if r['id'] == entry['result_id']) assert recovered['draft_id'] == entry['draft_id'] assert view['tasks'][0]['status'] == 'done' and view['requests'][0]['status'] == 'draft' request = owner.call('GET', '/balcao-virtual/pedidos/' + view['requests'][0]['id'])['request'] assert request['payload']['description'] == 'Descrição ' + entry['scenario'] + ' HTTP QA' draft = owner.call('GET', '/api/v1/simulation-drafts/current?context=' + entry['id'])['data'] assert draft['id'] == entry['draft_id'] and draft['personal_snapshot']['fiscal_residency_country'] == 'PT' assert draft['economic_snapshot']['funds_origin_countries'] == ['PT', 'ZA'] steps = {s['code']: s for s in view['steps']} assert steps['S10']['state'] == ('dispensed' if entry['scenario'] == 'A' else 'waiting_external') assert steps['S13']['applicability'] == ('not_needed_for_preparation' if entry['scenario'] == 'A' else 'applicable') entry['new_authenticated_session_recovered'] = True entry['independent_task_done'] = True entry['expected_break_even_units'] = 20 entry['expected_funding_gap'] = 0 if entry['scenario'] == 'A' else 70000 entry['recovered_financial_result'] = {key: recovered[key] for key in ['id', 'engine', 'financial_snapshot', 'draft_id', 'draft_version']} entry['prior_result_ids_preserved'] = [r['id'] for r in view['results'] if r['id'] != entry['result_id']] owner.call('GET', '/api/v1/simulation-drafts/' + scenarios[0]['draft_id'] + '?context=' + scenarios[1]['id'], status=403) owner.call('POST', scenarios[1]['url'] + '/resultados', {'version': owner.call('GET', scenarios[1]['url'])['case']['version'], 'draft_id': scenarios[0]['draft_id'], 'draft_version': 1}, status=403) stranger = Session('outsider') stranger.login('investor-path-outsider-qa@example.invalid', 'Outsider-Path-QA-2026!') for entry in scenarios: stranger.call('GET', entry['url'], status=403) stranger.call('GET', '/api/v1/simulation-drafts/' + entry['draft_id'], status=404) assert stranger.call('GET', '/investimentos')['contexts'] == [] geometry_proof = json.loads((OUT / 'geometry-consumers-path-qa.json').read_text(encoding='utf-8-sig')) assert geometry_proof['target'] == target and geometry_proof['real_geometries_checked'] == 361 geometry_checks = [] for selected in geometry_proof['selector_examples']: query = urllib.parse.urlencode({'territory_id': selected['id']}) private = manager.call('GET', '/management/territory-geometries?' + query) public = owner.call('GET', '/api/v1/investment/territory-geometries?' + query) stranger.call('GET', '/management/territory-geometries?' + query, status=403) assert private['selected_geometry_available'] and private['investment_site'] is False assert public['territory_registered'] and public['context_geometry_count'] == 0 geometry_checks.append({'territory': selected, 'manager_selected_geometry': True, 'public_context_count': 0, 'ui_passed': False}) proof = {'captured_at': datetime.now(timezone.utc).isoformat(), 'target': target, 'base_url': BASE, 'real_http_sessions': True, 'session_cookie_changed': True, 'scenarios': scenarios, 'geometry_http_profiles': geometry_checks, 'requests': events, 'api_persistence_passed': True, 'integrated_ui_ab_passed': False, 'active_writes': 0, 'activation_ready': False, 'installation_preview_qa': {key: installation[key] for key in ['code_hash', 'baseDeskInstalled', 'pathInstalled', 'pathEnabled', 'pathQaVerified', 'pendingMigrations']}} proof_path = OUT / 'investor-path-http-qa.json' if proof_path.exists(): shutil.copy2(proof_path, OUT / ('investor-path-http-qa-' + datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S-%f') + '.json')) proof_path.write_text(json.dumps(proof, ensure_ascii=False, indent=2), encoding='utf-8') print(json.dumps({'target': target, 'scenarios': [{key: value for key, value in scenario.items() if key != 'recovered_financial_result'} for scenario in scenarios], 'requests': len(events), 'ui_passed': False, 'active_writes': 0}, ensure_ascii=False))