236 lines
15 KiB
Python
236 lines
15 KiB
Python
"""Follow dashboard destinations on the existing guarded QA server. No visual claim."""
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import http.cookiejar
|
|
import json
|
|
import re
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
root = Path(__file__).resolve().parent.parent
|
|
sys.stdout.reconfigure(encoding='utf-8')
|
|
private = root / 'storage/app/private/investmoz-v2/portal-audit'
|
|
php = Path.home() / '.config/investmoz/php-8.5.11/php.exe'
|
|
target = json.loads((private.parent / 'opportunity-resources-backup-proof.json').read_text(encoding='utf-8-sig'))['restored_target']
|
|
assert re.fullmatch(r'desk_restore_\d{8}_\d{6}_test', target['database']) and int(target['port']) == 55441
|
|
expected = target['database'] + ':55441'
|
|
manifest_path = root / 'public/build-qa-opportunities/manifest.json'
|
|
manifest = json.loads(manifest_path.read_text(encoding='utf-8'))
|
|
version = subprocess.check_output([str(php), '-r', "echo hash_file('xxh128','public/build-qa-opportunities/manifest.json');"], cwd=root, text=True).strip()
|
|
actors = json.loads((private / 'resource-http-actors-private.json').read_text(encoding='utf-8-sig'))
|
|
events = []
|
|
|
|
|
|
def snapshot():
|
|
return json.loads(subprocess.check_output([str(php), 'tools/portal-dashboard-qa-snapshot.php'], cwd=root, text=True))
|
|
|
|
|
|
def new_session():
|
|
global jar, opener
|
|
jar = http.cookiejar.CookieJar()
|
|
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
|
|
|
|
|
|
def call(method, path, data=None, status=200, inertia=False, raw=False):
|
|
assert path.startswith('/') and not path.startswith('//')
|
|
headers = {'Accept': 'application/json' if inertia or data is not None else 'text/html', 'Content-Type': 'application/json'}
|
|
for cookie in jar:
|
|
if cookie.name == 'XSRF-TOKEN':
|
|
headers['X-XSRF-TOKEN'] = urllib.parse.unquote(cookie.value)
|
|
if inertia:
|
|
headers.update({'X-Inertia': 'true', 'X-Inertia-Version': version})
|
|
request = urllib.request.Request('http://127.0.0.1:18097' + path, data=None if data is None else json.dumps(data).encode(), headers=headers, method=method)
|
|
try:
|
|
response = opener.open(request)
|
|
except urllib.error.HTTPError as error:
|
|
response = error
|
|
assert response.headers.get('X-Investmoz-QA-Target') == expected, 'QA header mismatch; refused'
|
|
body = response.read()
|
|
assert response.status == status, f'{method} {path.split("?")[0]}: expected {status}, observed {response.status}; body withheld'
|
|
events.append({'method': method, 'path': path.split('?')[0], 'status': response.status, 'qa_target_verified': True})
|
|
return body if raw else json.loads(body) if body else None
|
|
|
|
|
|
def login(role):
|
|
call('GET', '/login', raw=True)
|
|
account = actors[role]
|
|
result = call('POST', '/login', {k: account[k] for k in ['email', 'password']})
|
|
required = bool(result.get('two_factor'))
|
|
if required:
|
|
assert account['totp_secret']
|
|
digest = hmac.new(base64.b32decode(account['totp_secret']), struct.pack('>Q', int(time.time()) // 30), hashlib.sha1).digest()
|
|
offset = digest[-1] & 15
|
|
code = f'{(struct.unpack(">I", digest[offset:offset + 4])[0] & 0x7fffffff) % 1000000:06d}'
|
|
call('POST', '/two-factor-challenge', {'code': code}, status=204)
|
|
return required
|
|
|
|
|
|
if '--report-access-only' in sys.argv:
|
|
before = snapshot()
|
|
new_session()
|
|
login('manager')
|
|
report = call('GET', '/qa-auditoria/pedidos-abertos', raw=True)
|
|
assert hashlib.sha256(report).hexdigest() == hashlib.sha256((private / 'pedidos-abertos.html').read_bytes()).hexdigest()
|
|
assert report.count(b'<article ') == 3349
|
|
call('POST', '/logout', {}, status=204)
|
|
new_session()
|
|
login('outsider')
|
|
call('GET', '/qa-auditoria/pedidos-abertos', status=403, raw=True)
|
|
call('POST', '/logout', {}, status=204)
|
|
assert before == snapshot()
|
|
proof = {'captured_at': datetime.now(timezone.utc).isoformat(), 'target': target, 'authenticated_report_url': 'http://127.0.0.1:18097/qa-auditoria/pedidos-abertos',
|
|
'requirements_presented': 3349, 'manager_200': True, 'outsider_403': True, 'report_bytes_sha256': hashlib.sha256(report).hexdigest(),
|
|
'records_changed': 0, 'active_writes': 0, 'browser_visual_observed': False, 'requests': events}
|
|
(private / 'dashboard-open-requests-http-qa.json').write_text(json.dumps(proof, indent=2), encoding='utf-8')
|
|
print(json.dumps({k: v for k, v in proof.items() if k != 'requests'}))
|
|
sys.exit(0)
|
|
|
|
if '--real-estate-draft-only' in sys.argv:
|
|
before = snapshot()
|
|
new_session()
|
|
login('manager')
|
|
dashboard = call('GET', '/dashboard', inertia=True)['props']
|
|
assert call('GET', dashboard['opportunityManagement']['create']['href'], inertia=True)['props']['record'] is None
|
|
key = str(uuid.uuid4())
|
|
data = {'creation_key': key, 'opportunity_category': 'real_estate', 'title': 'QA contrato imobiliário incompleto',
|
|
'real_estate': {'transaction_type': 'lease', 'land_area': 250.5, 'area_unit': 'sqm'}, 'reason': 'Rascunho parcial imobiliário sem validação final nem decisão externa'}
|
|
draft = call('POST', '/management/opportunities/drafts', data, status=201)
|
|
assert call('POST', '/management/opportunities/drafts', data, status=201)['id'] == draft['id']
|
|
path = f'/management/opportunities/{draft["id"]}/edit'
|
|
assert call('GET', path, inertia=True)['props']['values']['real_estate'] == data['real_estate']
|
|
call('POST', '/logout', {}, status=204)
|
|
new_session()
|
|
login('manager')
|
|
assert call('GET', path, inertia=True)['props']['values']['real_estate'] == data['real_estate']
|
|
updated = call('PUT', f'/management/opportunities/{draft["id"]}/draft', {'etag': draft['etag'], 'real_estate': {'rent': 1000, 'rent_periodicity': 'monthly'}, 'reason': 'Complemento parcial de renda preservando os campos já guardados'})
|
|
values = call('GET', path, inertia=True)['props']['values']['real_estate']
|
|
assert values == {**data['real_estate'], 'rent': 1000, 'rent_periodicity': 'monthly'}
|
|
for invalid in [{'price': 0}, {'price': '100'}, {'id': str(uuid.uuid4())}, {'transaction_type': 'sell_duat'}, {'utilities': ['invalid scalar']}]:
|
|
call('PUT', f'/management/opportunities/{draft["id"]}/draft', {'etag': updated['etag'], 'real_estate': invalid, 'reason': 'Ensaio negativo sem alterar os dados válidos do rascunho'}, status=422)
|
|
assert call('GET', path, inertia=True)['props']['values']['real_estate'] == values
|
|
call('POST', '/logout', {}, status=204)
|
|
new_session()
|
|
login('outsider')
|
|
call('GET', path, status=403, raw=True)
|
|
call('POST', '/logout', {}, status=204)
|
|
after = snapshot()
|
|
assert before['tables'] == after['tables'] and before['published'] == after['published'] and before['account18_sha256'] == after['account18_sha256']
|
|
assert all(after['opportunities'][key] == sha for key, sha in before['opportunities'].items())
|
|
assert set(after['opportunities']) - set(before['opportunities']) == {str(draft['id'])}
|
|
proof = {'captured_at': datetime.now(timezone.utc).isoformat(), 'target': target, 'opportunity_id': draft['id'],
|
|
'source_schema_sha256': hashlib.sha256((root / 'resources/data/investmoz-opportunity-modular.schema.json').read_bytes()).hexdigest(),
|
|
'module': 'real_estate', 'source_patch_properties': 24, 'incomplete_draft_preserved': True, 'new_session_recovered': True,
|
|
'patch_preserves_previous_properties': True, 'five_invalid_inputs_422_without_mutation': True, 'replay_same_id': True, 'outsider_403': True,
|
|
'existing_records_originals_dossiers_preserved': True, 'canonical_bindings_complete': False, 'full_contract_conditions_validated': False,
|
|
'module_form_implemented': False, 'workflow_implemented': False, 'browser_visual_observed': False, 'active_writes': 0, 'requests': events}
|
|
(private / 'real-estate-draft-http-qa.json').write_text(json.dumps(proof, indent=2), encoding='utf-8')
|
|
print(json.dumps({k: v for k, v in proof.items() if k != 'requests'}))
|
|
sys.exit(0)
|
|
|
|
if '--mfa-manager-recovery-only' in sys.argv:
|
|
retained = json.loads((private / 'dashboard-opportunities-http-qa.json').read_text(encoding='utf-8'))
|
|
assert retained['target'] == target and retained['created_saved_edited_archived_restored']
|
|
actors['manager'] = json.loads((private / 'i0-qa-actor-private.json').read_text(encoding='utf-8-sig'))
|
|
before = snapshot()
|
|
recovered = []
|
|
for session in range(2):
|
|
new_session()
|
|
assert login('manager'), 'Expected the existing manager MFA configuration'
|
|
dashboard = call('GET', '/dashboard', inertia=True)['props']
|
|
current_html = call('GET', '/dashboard', raw=True).decode()
|
|
assert '/build-qa-opportunities/' in current_html and manifest['resources/js/pages/dashboard.tsx']['file'] in current_html
|
|
assert dashboard['auth']['canManagePortal'] and dashboard['opportunityManagement']['installed']
|
|
management = call('GET', dashboard['opportunityManagement']['manage']['href'] + '?q=' + urllib.parse.quote(retained['reference']), inertia=True)
|
|
assert any(r['id'] == retained['opportunity_id'] for r in management['props']['records']['data'])
|
|
path = f'/management/opportunities/{retained["opportunity_id"]}/edit'
|
|
first = call('GET', path, inertia=True)['props']
|
|
assert first['record']['status'] == 'draft'
|
|
assert call('GET', path, inertia=True)['props']['values'] == first['values']
|
|
recovered.append(first['values'])
|
|
call('POST', '/logout', {}, status=204)
|
|
assert recovered[0] == recovered[1] and before == snapshot()
|
|
proof = {'captured_at': datetime.now(timezone.utc).isoformat(), 'target': target, 'manager_id': dashboard['auth']['user']['id'],
|
|
'mfa_both_sessions': True, 'dashboard_destinations_followed': True, 'opportunity_id': retained['opportunity_id'],
|
|
'reload_logout_new_session_recovered': True, 'records_changed': 0, 'browser_visual_observed': False,
|
|
'current_qa_assets_in_html_verified': True, 'qa_manifest_sha256': hashlib.sha256(manifest_path.read_bytes()).hexdigest(),
|
|
'active_writes': 0, 'requests': events}
|
|
(private / 'dashboard-manager-mfa-http-qa.json').write_text(json.dumps(proof, indent=2), encoding='utf-8')
|
|
print(json.dumps({k: v for k, v in proof.items() if k != 'requests'}))
|
|
sys.exit(0)
|
|
|
|
before = snapshot()
|
|
assert before['target'] == target
|
|
new_session()
|
|
manager_mfa_required = login('manager')
|
|
html = call('GET', '/dashboard', raw=True).decode()
|
|
assert '/build-qa-opportunities/' in html and '/build/assets/' not in html
|
|
dashboard_file = manifest['resources/js/pages/dashboard.tsx']['file']
|
|
assert dashboard_file in html
|
|
page = call('GET', '/dashboard', inertia=True)
|
|
assert page['component'] == 'dashboard' and page['version'] == version
|
|
access = page['props']['opportunityManagement']
|
|
assert page['props']['auth']['canManagePortal'] and access['installed']
|
|
assert access['manage']['label'] == 'Gestão de Oportunidades' and access['create']['label'] == 'Nova oportunidade'
|
|
management = call('GET', access['manage']['href'], inertia=True)
|
|
assert management['component'] == 'management/opportunities'
|
|
create = call('GET', access['create']['href'], inertia=True)
|
|
assert create['component'] == 'management/opportunity-draft' and create['props']['record'] is None
|
|
title = 'QA dashboard opportunity ' + uuid.uuid4().hex[:8]
|
|
draft = call('POST', '/management/opportunities/drafts', {'creation_key': str(uuid.uuid4()), 'title': title, 'reason': 'Criação sintética através dos destinos do dashboard QA'}, status=201)
|
|
id = draft['id']
|
|
url = f'/management/opportunities/{id}/edit'
|
|
assert call('GET', url, inertia=True)['props']['values']['title'] == title
|
|
call('POST', '/logout', {}, status=204)
|
|
new_session()
|
|
second_mfa = login('manager')
|
|
assert call('GET', '/dashboard', inertia=True)['props']['opportunityManagement'] == access
|
|
assert call('GET', url, inertia=True)['props']['values']['title'] == title
|
|
edited_title = title + ' edited'
|
|
updated = call('PUT', f'/management/opportunities/{id}/draft', {'etag': draft['etag'], 'title': edited_title, 'reason': 'Edição após recuperar o rascunho noutra sessão QA'})
|
|
assert call('GET', url, inertia=True)['props']['values']['title'] == edited_title
|
|
action = {'etag': updated['etag'], 'action': 'archive', 'confirm': True, 'reason': 'Arquivo explícito com conservação do rascunho e histórico QA'}
|
|
call('POST', f'/management/opportunities/{id}/lifecycle', action)
|
|
archived = call('GET', url, inertia=True)['props']
|
|
assert archived['record']['status'] == 'archived' and archived['record']['deleted_at'] is None and archived['values']['title'] == edited_title
|
|
records = call('GET', access['manage']['href'] + '?status=archived&q=' + urllib.parse.quote(title), inertia=True)['props']['records']['data']
|
|
assert [r['id'] for r in records] == [id]
|
|
call('POST', f'/management/opportunities/{id}/lifecycle', {**action, 'action': 'restore', 'etag': archived['etag']})
|
|
restored = call('GET', url, inertia=True)['props']
|
|
assert restored['record']['status'] == 'draft' and restored['record']['deleted_at'] is None and restored['values']['title'] == edited_title
|
|
assert len(restored['history']) == 4
|
|
call('POST', '/logout', {}, status=204)
|
|
new_session()
|
|
login('outsider')
|
|
outside = call('GET', '/dashboard', inertia=True)['props']
|
|
assert outside['opportunityManagement'] is None and not outside['auth']['canManagePortal']
|
|
call('GET', access['manage']['href'], status=403, raw=True)
|
|
call('GET', url, status=403, raw=True)
|
|
call('POST', f'/management/opportunities/{id}/lifecycle', {**action, 'etag': restored['etag']}, status=403, raw=True)
|
|
call('POST', '/logout', {}, status=204)
|
|
after = snapshot()
|
|
assert before['tables'] == after['tables'] and before['published'] == after['published']
|
|
assert before['account18_sha256'] == after['account18_sha256']
|
|
assert all(after['opportunities'][key] == sha for key, sha in before['opportunities'].items())
|
|
assert set(after['opportunities']) - set(before['opportunities']) == {str(id)}
|
|
proof = {'captured_at': datetime.now(timezone.utc).isoformat(), 'target': target, 'dashboard_url': 'http://127.0.0.1:18097/dashboard', 'dashboard_links_followed': access,
|
|
'opportunity_id': id, 'reference': restored['record']['reference'], 'new_session_recovered': True, 'created_saved_edited_archived_restored': True,
|
|
'history_events': len(restored['history']), 'manager_mfa_required': manager_mfa_required, 'second_session_mfa_required': second_mfa,
|
|
'outsider_denied': True, 'existing_opportunities_preserved': len(before['opportunities']), 'related_tables_originals_dossiers_preserved': after['tables'],
|
|
'account18_preserved': True, 'published_before': before['published'], 'published_after': after['published'],
|
|
'qa_assets_verified_in_html': True, 'qa_manifest_sha256': hashlib.sha256(manifest_path.read_bytes()).hexdigest(), 'asset_version': version,
|
|
'browser_visual_observed': False, 'active_writes': 0, 'requests': events, 'qa_accounts_access': after['accounts']}
|
|
path = private / 'dashboard-opportunities-http-qa.json'
|
|
if path.exists():
|
|
path.rename(private / ('dashboard-opportunities-http-qa-' + datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S') + '.json'))
|
|
path.write_text(json.dumps(proof, indent=2, ensure_ascii=False), encoding='utf-8')
|
|
print(json.dumps({k: v for k, v in proof.items() if k not in ['requests', 'related_tables_originals_dossiers_preserved', 'asset_version', 'qa_accounts_access']}, ensure_ascii=False))
|