175 lines
12 KiB
Python
175 lines
12 KiB
Python
"""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))
|