64 lines
3.6 KiB
Python
64 lines
3.6 KiB
Python
"""Focused live QA checks for primary location and task lifecycle; no active target."""
|
|
import importlib.util
|
|
import json
|
|
import uuid
|
|
from pathlib import Path
|
|
from datetime import datetime, timezone
|
|
|
|
spec = importlib.util.spec_from_file_location('opportunity_workflow_http_qa', Path(__file__).with_name('opportunity-workflow-http-qa.py'))
|
|
m = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(m)
|
|
m.path = m.PRIVATE / 'opportunity-workflow-territory-tasks-http-qa.json'
|
|
proof = m.proof
|
|
|
|
def task(props, id):
|
|
return next(t for t in props['workflowTasks'] if t['id'] == id)
|
|
|
|
try:
|
|
id, payload, document = m.create('investment_project')
|
|
original = m.page(m.promoter, id)['canonicalOptions']['resources']['locations'][0]['payload']
|
|
districts = m.promoter.call('GET', '/minhas-oportunidades/territory-options?type=district&parent_id=' + original['province_id'])['data']
|
|
secondary = next(d for d in districts if d['id'] != original['district_id'])
|
|
extra = m.promoter.call('POST', f'/minhas-oportunidades/{id}/resources/locations', {'etag': m.page(m.promoter, id)['etag'],
|
|
'creation_key': str(uuid.uuid4()), 'reason': m.REASON, 'payload': {**original, 'district_id': secondary['id'], 'is_primary': False}}, status=201)
|
|
payload.update({'location_mode': 'multiple', 'locations': [extra['id'], payload['locations'][0]]})
|
|
m.save(m.promoter, id, payload)
|
|
m.transition(m.promoter, id, 'submit')
|
|
m.transition(m.validator, id, 'start_review')
|
|
initial_task = m.page(m.validator, id)['profileContext']['review_task_id']
|
|
m.transition(m.validator, id, 'request_correction', correction_items=['canonical.demand_statement'])
|
|
assert task(m.page(m.promoter, id), initial_task)['status'] == 'done'
|
|
payload['demand_statement'] += ' Correção sintética da revisão de tarefa e localização principal.'
|
|
m.save(m.promoter, id, payload)
|
|
m.transition(m.promoter, id, 'submit')
|
|
submitted = m.page(m.promoter, id)
|
|
assert 'assignment' not in submitted['profileContext']
|
|
current_task = submitted['profileContext']['review_task_id']
|
|
assert current_task != initial_task and task(submitted, current_task)['assignee_user_id'] is None
|
|
m.transition(m.validator, id, 'start_review')
|
|
m.review(id, document)
|
|
m.transition(m.publisher, id, 'publish')
|
|
published = m.page(m.promoter, id)
|
|
assert published['record']['primary_territory_id'] == original['district_id']
|
|
assert published['record']['district'] != secondary['name']
|
|
m.transition(m.publisher, id, 'suspend')
|
|
assert all(t['status'] == 'done' for t in m.page(m.promoter, id)['workflowTasks'])
|
|
proof['primary_and_tasks'] = {'id': id, 'primary_reference_second_in_array': True, 'published_primary_id': original['district_id'],
|
|
'correction_closed_previous_review': True, 'resubmission_unassigned_new_task': True, 'suspension_no_open_tasks': True}
|
|
|
|
rejected, rp, rd = m.create('real_estate')
|
|
m.transition(m.promoter, rejected, 'submit')
|
|
rejected_task = m.page(m.promoter, rejected)['profileContext']['review_task_id']
|
|
m.transition(m.validator, rejected, 'reject')
|
|
assert task(m.page(m.promoter, rejected), rejected_task)['status'] == 'done'
|
|
proof['rejection_task'] = {'id': rejected, 'review_closed_with_reason': True}
|
|
proof['result'] = 'passed'
|
|
proof['completed_at'] = datetime.now(timezone.utc).isoformat()
|
|
print(json.dumps({'result': 'passed', 'target': m.EXPECTED, 'primary_case_id': id, 'rejected_case_id': rejected,
|
|
'visual_pending': True, 'active_writes': 0}))
|
|
except Exception:
|
|
proof['result'] = 'failed'
|
|
raise
|
|
finally:
|
|
m.persist()
|