Skip to content

fix: preventing task_rety() from overwriting completed Tests #414

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lrodden merged 1 commit into from
Apr 12, 2026
Merged

fix: preventing task_rety() from overwriting completed Tests #414

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions app/api/tasks.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1585,6 +1585,11 @@ def run_test(
# get test
test = Test.objects.get(id=test_id)

# idempotency guard: once a Test is complete, do not execute it again.
if test.time_completed is not None:
logger.info(f'skipping run_test for completed test_id: {str(test_id)}')
return None

# define objects for flowrun
objects = [_flow_obj(
parent=str(test.page.id),
Expand Down
126 changes: 78 additions & 48 deletions app/api/utils/scanner.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
from .updater import update_flowrun
from .manager import record_task
from .tester import Tester
from django.core.cache import cache
from datetime import datetime
from cursion import settings
import os, asyncio, uuid, boto3, random, time
Expand Down Expand Up @@ -377,53 +378,83 @@ def check_scan_completion(
# start Test if test_id present
if test_id is not None:

# update flowrun
if flowrun_id and flowrun_id != 'None':
time.sleep(random.uniform(0.1, 5))
update_flowrun(**{
'flowrun_id': str(flowrun_id),
'node_index': node_index,
'message': f'starting test comparison algorithm for {scan.page.page_url} | test_id: {str(test_id)}',
'objects': objects
})

# get task_id from scan.system
task_id = None
for task in scan.system['tasks']:
if task.get('component') == sender:
task_id = task.get('task_id')

# record task data in test
record_task(
resource_type='test',
resource_id=str(test_id),
task_id=str(task_id),
task_method='run_test',
kwargs={
'test_id': str(test_id),
'alert_id': str(alert_id) if alert_id is not None else None,
'flowrun_id': str(flowrun_id) if flowrun_id is not None else None,
'node_index': str(node_index) if node_index is not None else None,
'track_id': track_id
}
)

print('\n---------------\nScan Complete\nStarting Test...\n---------------\n')
test = Test.objects.get(id=test_id)
updated_test = Tester(test=test).run_test()

# update flowrun
if flowrun_id and flowrun_id != 'None':
objects[-1]['status'] = updated_test.status
update_flowrun(**{
'flowrun_id': str(flowrun_id),
'node_index': node_index,
'message': (
f'test for {scan.page.page_url} completed with status: '+
f'{"❌ FAILED" if updated_test.status == 'failed' else "✅ PASSED"} | test_id: {str(test_id)}'
),
'objects': objects
})
# derive tracking identity from Test system data
run_track_id = str(test_id)
try:
test = Test.objects.get(id=test_id)
first_task = ((test.system or {}).get('tasks') or [{}])[0]
task_kwargs = first_task.get('kwargs') or {}
run_track_id = str(task_kwargs.get('track_id') or test_id)
except Exception:
test = None

# if Test already completed, do not re-run it.
if test is not None and test.time_completed is not None:
if flowrun_id and flowrun_id != 'None':
objects[-1]['status'] = test.status
update_flowrun(**{
'flowrun_id': str(flowrun_id),
'node_index': node_index,
'message': f'skipping run_test for completed test_id: {str(test_id)}',
'objects': objects
})
return scan

# avoid duplicate test launches from concurrent component completions
run_test_lock_key = f'scanner:run_test:{str(test_id)}'
if not cache.add(run_test_lock_key, '1', timeout=600):
return scan

try:
# update flowrun
if flowrun_id and flowrun_id != 'None':
time.sleep(random.uniform(0.1, 5))
update_flowrun(**{
'flowrun_id': str(flowrun_id),
'node_index': node_index,
'message': f'starting test comparison algorithm for {scan.page.page_url} | test_id: {str(test_id)}',
'objects': objects
})

# get task_id from scan.system
task_id = None
for task in scan.system['tasks']:
if task.get('component') == sender:
task_id = task.get('task_id')

# record task data in test
record_task(
resource_type='test',
resource_id=str(test_id),
task_id=str(task_id),
task_method='run_test',
kwargs={
'test_id': str(test_id),
'alert_id': str(alert_id) if alert_id is not None else None,
'flowrun_id': str(flowrun_id) if flowrun_id is not None else None,
'node_index': str(node_index) if node_index is not None else None,
'track_id': run_track_id
}
)

print('\n---------------\nScan Complete\nStarting Test...\n---------------\n')
test = Test.objects.get(id=test_id)
updated_test = Tester(test=test).run_test()

# update flowrun
if flowrun_id and flowrun_id != 'None':
objects[-1]['status'] = updated_test.status
update_flowrun(**{
'flowrun_id': str(flowrun_id),
'node_index': node_index,
'message': (
f'test for {scan.page.page_url} completed with status: '+
f'{"❌ FAILED" if updated_test.status == 'failed' else "✅ PASSED"} | test_id: {str(test_id)}'
),
'objects': objects
})
finally:
cache.delete(run_test_lock_key)

if alert_id is not None and alert_id != 'None':
print('running alert from `cursion.check_scan_completion`')
Expand Down Expand Up @@ -756,4 +787,3 @@ def _yellowlab(
return scan



5 changes: 5 additions & 0 deletions app/api/utils/updater.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,11 @@ def update_flowrun(**kwargs) -> object:
# stale copies of nodes/edges/logs.
flowrun = FlowRun.objects.select_for_update().get(id=flowrun_id)

# Ignore stale worker updates after completion; late async tasks should
# not mutate finished runs.
if flowrun.time_completed is not None:
return flowrun

# set timestamp
timestamp = timezone.now().strftime('%Y-%m-%d %H:%M:%S.%f')

Expand Down