sashatrask
1490 lines
61 KiB
Python
1490 lines
61 KiB
Python
import json
|
|
import os
|
|
import sys
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
from unittest import mock
|
|
|
|
|
|
APP_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'app'))
|
|
if APP_DIR not in sys.path:
|
|
sys.path.insert(0, APP_DIR)
|
|
|
|
from result_bundle import FORMAT_VERSION
|
|
from jsonl_projector import JsonlProjector, SerializedStream
|
|
from scan_execution import (
|
|
PACKAGE_DETECTOR_POLICY, PROTOCOL_VERSION, QueueDispositionPolicy,
|
|
remote_execution_identity,
|
|
)
|
|
from scanner_db import (
|
|
ScanEventConflictError,
|
|
ScannerDB,
|
|
fixed_lease_window,
|
|
remote_assignment_execution_plan,
|
|
validate_remote_result_execution_plan,
|
|
)
|
|
|
|
|
|
TOKEN_SHA256 = 'a' * 64
|
|
|
|
|
|
def _scan_policy():
|
|
return {
|
|
'drop_detectors': [],
|
|
'strict_git_provider_token_filter': True,
|
|
'trufflehog_stdout_max_mb': 32,
|
|
'trufflehog_stderr_max_mb': 8,
|
|
'result_bundle_max_event_bytes': 1024 * 1024,
|
|
'trufflehog_max_findings_per_target': 20000,
|
|
'trufflehog_job_memory_limit_bytes': 0,
|
|
'trufflehog_windows_job_cpu_weight': 0,
|
|
'trufflehog_windows_memory_priority': 0,
|
|
'trufflehog_diagnostic_max_lines': 2000,
|
|
'trufflehog_diagnostic_max_line_chars': 8192,
|
|
'trufflehog_diagnostic_max_line_bytes': 8192,
|
|
'trufflehog_diagnostic_max_errors': 200,
|
|
'trufflehog_diagnostic_max_warnings': 200,
|
|
'trufflehog_diagnostic_max_unclassified': 20,
|
|
}
|
|
|
|
|
|
def _execution_snapshot():
|
|
scan_kwargs = {
|
|
'timeout_sec': 30.0,
|
|
'trufflehog_config': PACKAGE_DETECTOR_POLICY,
|
|
}
|
|
policy = QueueDispositionPolicy()
|
|
limits = {'candidate_max_items': 10, 'candidate_max_bytes': 4096}
|
|
effective, execution = remote_execution_identity(
|
|
'github', scan_kwargs, scan_kwargs, policy, limits, _scan_policy(),
|
|
)
|
|
return {
|
|
'schema': 1,
|
|
'compatibility': {
|
|
'protocol_version': PROTOCOL_VERSION,
|
|
'bundle_format_version': FORMAT_VERSION,
|
|
'platform_tag': 'windows-x86_64',
|
|
'code_manifest_sha256': 'd' * 64,
|
|
'effective_config_sha256': effective,
|
|
'detector_policy_sha256': 'e' * 64,
|
|
},
|
|
'execution': execution,
|
|
'planning': {
|
|
'kind': 'exact_git_v1',
|
|
'git_baseline_depth': 100,
|
|
'git_ref_resolution_attempts': 2,
|
|
'git_ref_resolution_timeout_sec': 10.0,
|
|
'git_ref_resolution_max_bytes': 1 << 20,
|
|
},
|
|
'credential_ref': {'source': 'github', 'auth_entry': 'primary'},
|
|
}
|
|
|
|
|
|
def _effective_config_sha256():
|
|
return _execution_snapshot()['compatibility']['effective_config_sha256']
|
|
|
|
|
|
def _direct_execution_snapshot(queue_source, platform, planning_kind):
|
|
scan_kwargs = {
|
|
'timeout_sec': 30.0,
|
|
'trufflehog_config': PACKAGE_DETECTOR_POLICY,
|
|
}
|
|
policy = QueueDispositionPolicy()
|
|
limits = {'candidate_max_items': 10, 'candidate_max_bytes': 4096}
|
|
effective, execution = remote_execution_identity(
|
|
platform, scan_kwargs, scan_kwargs, policy, limits, _scan_policy(),
|
|
)
|
|
return {
|
|
'schema': 1,
|
|
'compatibility': {
|
|
'protocol_version': PROTOCOL_VERSION,
|
|
'bundle_format_version': FORMAT_VERSION,
|
|
'platform_tag': 'windows-x86_64',
|
|
'code_manifest_sha256': 'd' * 64,
|
|
'effective_config_sha256': effective,
|
|
'detector_policy_sha256': 'e' * 64,
|
|
},
|
|
'execution': execution,
|
|
'planning': {'kind': planning_kind},
|
|
'credential_ref': {'source': queue_source, 'auth_entry': ''},
|
|
}
|
|
|
|
|
|
class _Cursor:
|
|
def __init__(self, row=None, rowcount=1):
|
|
self.row = row
|
|
self.rowcount = rowcount
|
|
|
|
def fetchone(self):
|
|
return self.row
|
|
|
|
|
|
class _RowsCursor(_Cursor):
|
|
def __init__(self, rows=None, rowcount=1):
|
|
super().__init__(rowcount=rowcount)
|
|
self.rows = list(rows or [])
|
|
|
|
def fetchall(self):
|
|
return self.rows
|
|
|
|
|
|
class _ProvisionConnection:
|
|
is_postgres = True
|
|
|
|
def __init__(self):
|
|
self.statements = []
|
|
self.commits = 0
|
|
|
|
def execute(self, statement, params=()):
|
|
self.statements.append((statement, params))
|
|
if 'FROM remote_worker_users WHERE user_key' in statement:
|
|
return _Cursor({'id': 7})
|
|
if 'FROM remote_worker_devices WHERE device_key' in statement:
|
|
return _Cursor({
|
|
'id': 11, 'user_id': 7, 'device_key': 'device',
|
|
'revoked_at': '2026-09-17T00:00:00+00:00',
|
|
'created_at': 'created', 'updated_at': 'updated',
|
|
})
|
|
return _Cursor()
|
|
|
|
def commit(self):
|
|
self.commits += 1
|
|
|
|
def rollback(self):
|
|
raise AssertionError('provisioning unexpectedly rolled back')
|
|
|
|
|
|
class _ResolutionConnection:
|
|
is_postgres = True
|
|
|
|
def __init__(self):
|
|
self.statements = []
|
|
|
|
def execute(self, statement, params=()):
|
|
self.statements.append((statement, params))
|
|
normalized = ' '.join(statement.split())
|
|
if normalized.startswith('SELECT * FROM target_queue'):
|
|
return _Cursor({
|
|
'status': 'in_progress', 'lease_token': 'lease',
|
|
'current_result_reservation_id': 41, 'claim_event_id': 'event',
|
|
})
|
|
if normalized.startswith('SELECT * FROM pipeline_capacity'):
|
|
return _Cursor({
|
|
'bundle_items': 1, 'bundle_bytes': 100,
|
|
'projection_items': 1, 'projection_bytes': 100,
|
|
'keycheck_items': 1, 'keycheck_bytes': 100,
|
|
})
|
|
return _Cursor()
|
|
|
|
|
|
class _ProjectionCapacityConnection:
|
|
is_postgres = True
|
|
|
|
def __init__(self):
|
|
self.job = {
|
|
'id': 71, 'status': 'leased', 'lease_token': 'projection-lease',
|
|
'capacity_bytes': 2, 'updated_at': 'before',
|
|
}
|
|
self.capacity = {'projection_bytes': 2}
|
|
self.commits = 0
|
|
self.rollbacks = 0
|
|
|
|
def execute(self, statement, params=()):
|
|
normalized = ' '.join(statement.split())
|
|
if normalized.startswith('SELECT * FROM projection_jobs'):
|
|
if (
|
|
self.job['status'] == 'leased'
|
|
and self.job['lease_token'] == params[1]
|
|
):
|
|
return _Cursor(dict(self.job))
|
|
return _Cursor()
|
|
if normalized.startswith('SELECT 1 AS present FROM projection_appends'):
|
|
return _Cursor()
|
|
if normalized.startswith('SELECT * FROM pipeline_capacity'):
|
|
return _Cursor(dict(self.capacity))
|
|
if normalized.startswith("UPDATE projection_jobs SET status = 'pending'"):
|
|
self.job.update({
|
|
'status': 'pending', 'lease_token': None,
|
|
'capacity_bytes': self.job['capacity_bytes'],
|
|
})
|
|
return _Cursor(rowcount=1)
|
|
if normalized.startswith('UPDATE projection_jobs SET capacity_bytes'):
|
|
self.job['capacity_bytes'] = params[0]
|
|
self.job['updated_at'] = params[1]
|
|
return _Cursor(rowcount=1)
|
|
if normalized.startswith('UPDATE pipeline_capacity'):
|
|
self.capacity['projection_bytes'] += params[0]
|
|
return _Cursor(rowcount=1)
|
|
raise AssertionError(normalized)
|
|
|
|
def commit(self):
|
|
self.commits += 1
|
|
|
|
def rollback(self):
|
|
self.rollbacks += 1
|
|
|
|
|
|
class _ReconcileConnection:
|
|
is_postgres = True
|
|
|
|
def __init__(self, intent, reservation=None):
|
|
self.intent = intent
|
|
self.reservation = reservation
|
|
self.statements = []
|
|
self.commits = 0
|
|
self.rollbacks = 0
|
|
|
|
def execute(self, statement, params=()):
|
|
self.statements.append((statement, params))
|
|
normalized = ' '.join(statement.split())
|
|
if normalized.startswith('SELECT * FROM admission_intents'):
|
|
return _Cursor(self.intent)
|
|
if normalized.startswith('SELECT d.id AS device_id'):
|
|
return _Cursor({'device_id': 11, 'user_id': 7})
|
|
if normalized.startswith('SELECT r.*, q.attempts'):
|
|
if self.reservation is None:
|
|
return _Cursor()
|
|
row = dict(self.reservation)
|
|
row.update({
|
|
'bound_queue_id': row['queue_id'],
|
|
'bound_queue_source': row['source'],
|
|
'bound_queue_platform': row['platform'],
|
|
'bound_queue_query': row['query'],
|
|
'bound_queue_target': row['target'],
|
|
'bound_queue_normalized_target': row['normalized_target'],
|
|
})
|
|
return _Cursor(row)
|
|
return _Cursor()
|
|
|
|
def commit(self):
|
|
self.commits += 1
|
|
|
|
def rollback(self):
|
|
self.rollbacks += 1
|
|
|
|
|
|
class _DispatchGateConnection:
|
|
is_postgres = True
|
|
is_sqlite = False
|
|
|
|
def __init__(self, *, discovery_paused=0, dispatch_paused=0, drain_state='normal'):
|
|
self.discovery_paused = discovery_paused
|
|
self.dispatch_paused = dispatch_paused
|
|
self.drain_state = drain_state
|
|
self.intent_sha256 = None
|
|
self.remote_user_id = None
|
|
self.remote_device_id = None
|
|
self.intent_state = 'pending'
|
|
self.resolution_detail = None
|
|
self.statements = []
|
|
self.commits = 0
|
|
self.rollbacks = 0
|
|
|
|
def execute(self, statement, params=()):
|
|
params = tuple(params or ())
|
|
self.statements.append((statement, params))
|
|
normalized = ' '.join(statement.split())
|
|
if normalized.startswith('INSERT INTO admission_intents'):
|
|
self.intent_sha256 = params[1]
|
|
self.remote_user_id = params[2]
|
|
self.remote_device_id = params[3]
|
|
return _Cursor(rowcount=1)
|
|
if normalized.startswith('SELECT intent_sha256, state, remote_user_id'):
|
|
return _Cursor({
|
|
'intent_sha256': self.intent_sha256,
|
|
'state': self.intent_state,
|
|
'remote_user_id': self.remote_user_id,
|
|
'remote_device_id': self.remote_device_id,
|
|
})
|
|
if normalized.startswith('SELECT * FROM admission_intents'):
|
|
return _Cursor({
|
|
'intent_sha256': self.intent_sha256,
|
|
'state': self.intent_state,
|
|
'remote_user_id': self.remote_user_id,
|
|
'remote_device_id': self.remote_device_id,
|
|
})
|
|
if normalized == 'SELECT * FROM runtime_operations_control WHERE id = 1 FOR SHARE':
|
|
return _Cursor({
|
|
'revision': 0,
|
|
'discovery_paused': self.discovery_paused,
|
|
'dispatch_paused': self.dispatch_paused,
|
|
'drain_state': self.drain_state,
|
|
'actor': 'system:migration',
|
|
'operation_id': None,
|
|
'created_at': '2026-09-19T00:00:00+00:00',
|
|
'updated_at': '2026-09-19T00:00:00+00:00',
|
|
})
|
|
if normalized.startswith('SELECT * FROM remote_worker_users'):
|
|
return _Cursor({'id': 7, 'active_assignment_cap': 1})
|
|
if normalized.startswith('SELECT d.id AS device_id'):
|
|
return _Cursor({'device_id': 11, 'user_id': 7})
|
|
if normalized.startswith('SELECT r.*, q.attempts'):
|
|
return _Cursor()
|
|
if normalized.startswith('SELECT COUNT(*) AS value FROM result_reservations'):
|
|
return _Cursor({'value': 0})
|
|
if normalized.startswith('SELECT state, lease_expires_at, supervisor_instance_id'):
|
|
return _Cursor({
|
|
'state': 'ready',
|
|
'lease_expires_at': '2099-01-01T00:00:00+00:00',
|
|
'supervisor_instance_id': 'supervisor',
|
|
})
|
|
if normalized.startswith('SELECT id, query, target, normalized_target, attempts'):
|
|
return _Cursor()
|
|
if normalized.startswith("UPDATE admission_intents SET state = 'aborted'"):
|
|
self.intent_state = 'aborted'
|
|
if 'dispatch_gate_closed' in normalized:
|
|
self.resolution_detail = 'dispatch_gate_closed'
|
|
elif 'no_claimable_target' in normalized:
|
|
self.resolution_detail = 'no_claimable_target'
|
|
return _Cursor(rowcount=1)
|
|
raise AssertionError(normalized)
|
|
|
|
def commit(self):
|
|
self.commits += 1
|
|
|
|
def rollback(self):
|
|
self.rollbacks += 1
|
|
|
|
|
|
def _remote_reservation(snapshot_fields):
|
|
return {
|
|
'id': 41, 'reservation_token': 'f' * 32,
|
|
'bundle_id': '1' * 32, 'scan_event_id': '2' * 32,
|
|
'queue_id': 9, 'claim_lease_owner': 'remote:11',
|
|
'claim_lease_token': 'lease', 'claim_batch': 'f' * 32,
|
|
'attempts': 1, 'declared_bundle_bytes': 1 << 20,
|
|
'ready_relative_path': 'ready/11/bundle.trb',
|
|
'source': 'github', 'platform': 'github', 'query': 'q',
|
|
'target': 'https://github.com/example/project.git',
|
|
'normalized_target': 'https://github.com/example/project',
|
|
'run_id': None, 'cycle_id': None,
|
|
'producer_instance_id': 'supervisor', 'producer_pid': 123,
|
|
'producer_creation_time': 'created', 'producer_executable': 'python',
|
|
'assignment_kind': 'remote', 'remote_user_id': 7,
|
|
'remote_device_id': 11, 'remote_issued_at': 'issued',
|
|
'remote_expires_at': '2099-01-01T00:00:00+00:00',
|
|
'remote_result_upload_body_timeout_seconds': 1800,
|
|
'remote_effective_config_sha256': _effective_config_sha256(),
|
|
'remote_client_compat_sha256': 'c' * 64,
|
|
'remote_resolution_json': None,
|
|
'git_scan_plan_json': None, 'git_scan_plan_sha256': None,
|
|
'docker_layer_plan_json': None, 'docker_layer_plan_sha256': None,
|
|
**snapshot_fields,
|
|
}
|
|
|
|
|
|
def _reservation_for_snapshot(snapshot, source, platform, target, normalized_target):
|
|
effective_sha256 = snapshot['compatibility']['effective_config_sha256']
|
|
mapped = ScannerDB._remote_assignment_mapping({
|
|
'user_id': 7,
|
|
'device_id': 11,
|
|
'effective_config_sha256': effective_sha256,
|
|
'client_compat_sha256': 'c' * 64,
|
|
'token_sha256': TOKEN_SHA256,
|
|
'result_upload_body_timeout_seconds': 1800,
|
|
'execution_snapshot': snapshot,
|
|
})
|
|
return _remote_reservation({
|
|
'source': source,
|
|
'platform': platform,
|
|
'target': target,
|
|
'normalized_target': normalized_target,
|
|
'remote_effective_config_sha256': effective_sha256,
|
|
'remote_execution_snapshot_json': mapped['execution_snapshot_json'],
|
|
'remote_execution_snapshot_sha256': mapped['execution_snapshot_sha256'],
|
|
})
|
|
|
|
|
|
class RemoteWorkerDBTests(unittest.TestCase):
|
|
@staticmethod
|
|
def _reserve_with_gate(connection, *, remote=True):
|
|
db = object.__new__(ScannerDB)
|
|
db.conn = connection
|
|
return db.reserve_and_claim_target(
|
|
'github', 'github',
|
|
{'pid': 123, 'creation_time': 'created', 'executable': sys.executable},
|
|
'supervisor', 1 << 20, 4096, 10, 4096,
|
|
reservation_token='f' * 32,
|
|
bundle_id='1' * 32,
|
|
scan_event_id='2' * 32,
|
|
remote_assignment={
|
|
'user_id': 7,
|
|
'device_id': 11,
|
|
'effective_config_sha256': _effective_config_sha256(),
|
|
'client_compat_sha256': 'c' * 64,
|
|
'token_sha256': TOKEN_SHA256,
|
|
'result_upload_body_timeout_seconds': 1800,
|
|
'execution_snapshot': _execution_snapshot(),
|
|
} if remote else None,
|
|
)
|
|
|
|
def test_effective_dispatch_gate_aborts_pending_remote_intent_before_admission(self):
|
|
for dispatch_paused, drain_state in (
|
|
(1, 'normal'), (0, 'draining'), (0, 'drained'),
|
|
):
|
|
with self.subTest(dispatch_paused=dispatch_paused, drain_state=drain_state):
|
|
connection = _DispatchGateConnection(
|
|
dispatch_paused=dispatch_paused,
|
|
drain_state=drain_state,
|
|
)
|
|
self.assertIsNone(self._reserve_with_gate(connection))
|
|
self.assertEqual(connection.resolution_detail, 'dispatch_gate_closed')
|
|
sql = '\n'.join(statement for statement, _params in connection.statements)
|
|
self.assertIn('FOR SHARE', sql)
|
|
self.assertNotIn('FROM target_queue', sql)
|
|
self.assertNotIn('INSERT INTO result_reservations', sql)
|
|
self.assertNotIn('UPDATE pipeline_capacity', sql)
|
|
self.assertEqual(connection.rollbacks, 0)
|
|
|
|
def test_projection_capacity_expands_once_or_defers_without_quarantine(self):
|
|
full = _ProjectionCapacityConnection()
|
|
db = object.__new__(ScannerDB)
|
|
db.conn = full
|
|
self.assertIs(
|
|
db.expand_projection_job_capacity(
|
|
71, 'projection-lease', 5, 4,
|
|
),
|
|
False,
|
|
)
|
|
self.assertEqual(full.job['status'], 'pending')
|
|
self.assertEqual(full.job['capacity_bytes'], 2)
|
|
self.assertEqual(full.capacity['projection_bytes'], 2)
|
|
|
|
available = _ProjectionCapacityConnection()
|
|
db.conn = available
|
|
expanded = db.expand_projection_job_capacity(
|
|
71, 'projection-lease', 5, 10,
|
|
)
|
|
self.assertEqual(expanded['capacity_bytes'], 5)
|
|
self.assertEqual(available.capacity['projection_bytes'], 5)
|
|
replayed = db.expand_projection_job_capacity(
|
|
71, 'projection-lease', 5, 10,
|
|
)
|
|
self.assertEqual(replayed['capacity_bytes'], 5)
|
|
self.assertEqual(available.capacity['projection_bytes'], 5)
|
|
|
|
def test_projector_defers_capacity_without_append_completion_or_quarantine(self):
|
|
calls = []
|
|
|
|
class DB:
|
|
def claim_projection_job(self, generation, lease_token, lease_seconds):
|
|
return {
|
|
'id': 71, 'lease_token': 'job-lease',
|
|
'capacity_bytes': 2,
|
|
}
|
|
|
|
def expand_projection_job_capacity(
|
|
self, job_id, lease_token, actual_bytes, projection_max_bytes,
|
|
):
|
|
calls.append(
|
|
('expand', job_id, lease_token, actual_bytes,
|
|
projection_max_bytes)
|
|
)
|
|
return False
|
|
|
|
def complete_projection_job(self, *_args):
|
|
raise AssertionError('deferred projection completed')
|
|
|
|
def quarantine_projection_job(self, *_args, **_kwargs):
|
|
raise AssertionError('capacity backpressure was quarantined')
|
|
|
|
projector = object.__new__(JsonlProjector)
|
|
projector.db = DB()
|
|
projector.lease = {'generation': 1, 'lease_token': 'projector-lease'}
|
|
projector.lease_seconds = 30
|
|
projector.projection_max_bytes = 4
|
|
projector.reconcile_terminal_temps = lambda max_pages=1: None
|
|
projector._serialize = lambda _job: [
|
|
SerializedStream('scan_results', 'fixture.tmp', 5, 'a' * 64, 1),
|
|
]
|
|
projector._delete_registered_artifact = lambda path, artifact_id: calls.append(
|
|
('delete', path, artifact_id)
|
|
)
|
|
projector._rollback_database = lambda: calls.append(('rollback',))
|
|
|
|
self.assertTrue(projector.process_one())
|
|
self.assertEqual(calls, [
|
|
('expand', 71, 'job-lease', 5, 4),
|
|
('delete', 'fixture.tmp', 0),
|
|
])
|
|
|
|
def test_discovery_pause_does_not_close_dispatch(self):
|
|
connection = _DispatchGateConnection(discovery_paused=1)
|
|
self.assertIsNone(self._reserve_with_gate(connection))
|
|
self.assertEqual(connection.resolution_detail, 'no_claimable_target')
|
|
sql = '\n'.join(statement for statement, _params in connection.statements)
|
|
self.assertIn('FROM remote_worker_users', sql)
|
|
self.assertIn('FROM target_queue', sql)
|
|
|
|
def test_docker_direct_claim_query_does_not_require_access_evidence(self):
|
|
connection = _DispatchGateConnection()
|
|
db = object.__new__(ScannerDB)
|
|
db.conn = connection
|
|
snapshot = _direct_execution_snapshot(
|
|
'dockerhub', 'docker', 'docker_direct_v1',
|
|
)
|
|
result = db.reserve_and_claim_target(
|
|
'dockerhub', 'docker',
|
|
{'pid': 123, 'creation_time': 'created', 'executable': sys.executable},
|
|
'supervisor', 1 << 20, 4096, 10, 4096,
|
|
reservation_token='f' * 32,
|
|
bundle_id='1' * 32,
|
|
scan_event_id='2' * 32,
|
|
remote_assignment={
|
|
'user_id': 7,
|
|
'device_id': 11,
|
|
'effective_config_sha256': (
|
|
snapshot['compatibility']['effective_config_sha256']
|
|
),
|
|
'client_compat_sha256': 'c' * 64,
|
|
'token_sha256': TOKEN_SHA256,
|
|
'result_upload_body_timeout_seconds': 1800,
|
|
'execution_snapshot': snapshot,
|
|
},
|
|
)
|
|
self.assertIsNone(result)
|
|
claim_sql = next(
|
|
statement for statement, _params in connection.statements
|
|
if 'FROM target_queue' in statement
|
|
)
|
|
self.assertNotIn('docker_anonymous_', claim_sql)
|
|
|
|
def test_huggingface_direct_claim_query_does_not_require_public_evidence(self):
|
|
connection = _DispatchGateConnection()
|
|
db = object.__new__(ScannerDB)
|
|
db.conn = connection
|
|
snapshot = _direct_execution_snapshot(
|
|
'huggingface', 'huggingface', 'huggingface_space_v1',
|
|
)
|
|
result = db.reserve_and_claim_target(
|
|
'huggingface', 'huggingface',
|
|
{'pid': 123, 'creation_time': 'created', 'executable': sys.executable},
|
|
'supervisor', 1 << 20, 4096, 10, 4096,
|
|
reservation_token='f' * 32,
|
|
bundle_id='1' * 32,
|
|
scan_event_id='2' * 32,
|
|
remote_assignment={
|
|
'user_id': 7,
|
|
'device_id': 11,
|
|
'effective_config_sha256': (
|
|
snapshot['compatibility']['effective_config_sha256']
|
|
),
|
|
'client_compat_sha256': 'c' * 64,
|
|
'token_sha256': TOKEN_SHA256,
|
|
'result_upload_body_timeout_seconds': 1800,
|
|
'execution_snapshot': snapshot,
|
|
},
|
|
)
|
|
self.assertIsNone(result)
|
|
claim_sql = next(
|
|
statement for statement, _params in connection.statements
|
|
if 'FROM target_queue' in statement
|
|
)
|
|
self.assertNotIn('huggingface_public_', claim_sql)
|
|
|
|
def test_dispatch_pause_does_not_gate_local_reservation(self):
|
|
connection = _DispatchGateConnection(dispatch_paused=1)
|
|
self.assertIsNone(self._reserve_with_gate(connection, remote=False))
|
|
self.assertEqual(connection.resolution_detail, 'no_claimable_target')
|
|
sql = '\n'.join(statement for statement, _params in connection.statements)
|
|
self.assertNotIn('runtime_operations_control', sql)
|
|
|
|
def test_fixed_lease_window_uses_one_issuance_clock(self):
|
|
issued = datetime(2026, 9, 18, 12, 30, 0, tzinfo=timezone.utc)
|
|
issued_at, expires_at = fixed_lease_window(24 * 60 * 60, now=issued)
|
|
self.assertEqual(issued_at, '2026-09-18T12:30:00+00:00')
|
|
self.assertEqual(expires_at, '2026-09-19T12:30:00+00:00')
|
|
|
|
def test_remote_assignment_requires_presented_token_digest(self):
|
|
with self.assertRaisesRegex(ValueError, 'token_sha256'):
|
|
ScannerDB._remote_assignment_mapping({
|
|
'user_id': 7, 'device_id': 11,
|
|
'effective_config_sha256': 'b' * 64,
|
|
'client_compat_sha256': 'c' * 64,
|
|
})
|
|
mapped = ScannerDB._remote_assignment_mapping({
|
|
'user_id': 7, 'device_id': 11,
|
|
'effective_config_sha256': _effective_config_sha256(),
|
|
'client_compat_sha256': 'c' * 64,
|
|
'token_sha256': TOKEN_SHA256,
|
|
'result_upload_body_timeout_seconds': 1800,
|
|
'execution_snapshot': _execution_snapshot(),
|
|
})
|
|
self.assertEqual(mapped['token_sha256'], TOKEN_SHA256)
|
|
self.assertEqual(mapped['result_upload_body_timeout_seconds'], 1800)
|
|
snapshot = json.loads(mapped['execution_snapshot_json'])
|
|
self.assertNotIn('token', snapshot['execution']['scan_kwargs'])
|
|
self.assertIn('"auth_entry":"primary"', mapped['execution_snapshot_json'])
|
|
self.assertRegex(mapped['execution_snapshot_sha256'], r'^[0-9a-f]{64}$')
|
|
|
|
def test_remote_assignment_requires_bounded_persisted_upload_timeout(self):
|
|
for timeout in (None, True, 29, 86401, '1800'):
|
|
with self.subTest(timeout=timeout), self.assertRaisesRegex(
|
|
ValueError, 'upload body timeout',
|
|
):
|
|
ScannerDB._remote_assignment_mapping({
|
|
'user_id': 7, 'device_id': 11,
|
|
'effective_config_sha256': _effective_config_sha256(),
|
|
'client_compat_sha256': 'c' * 64,
|
|
'token_sha256': TOKEN_SHA256,
|
|
'result_upload_body_timeout_seconds': timeout,
|
|
'execution_snapshot': _execution_snapshot(),
|
|
})
|
|
|
|
def test_remote_assignment_rejects_noncanonical_or_secret_snapshot(self):
|
|
snapshot = _execution_snapshot()
|
|
snapshot['execution']['scan_kwargs']['token'] = 'must-not-persist'
|
|
with self.assertRaisesRegex(Exception, 'private|match'):
|
|
ScannerDB._remote_assignment_mapping({
|
|
'user_id': 7, 'device_id': 11,
|
|
'effective_config_sha256': _effective_config_sha256(),
|
|
'client_compat_sha256': 'c' * 64,
|
|
'token_sha256': TOKEN_SHA256,
|
|
'result_upload_body_timeout_seconds': 1800,
|
|
'execution_snapshot': snapshot,
|
|
})
|
|
|
|
def test_remote_assignment_rejects_effective_config_identity_drift(self):
|
|
with self.assertRaisesRegex(ValueError, 'effective configuration'):
|
|
ScannerDB._remote_assignment_mapping({
|
|
'user_id': 7,
|
|
'device_id': 11,
|
|
'effective_config_sha256': 'b' * 64,
|
|
'client_compat_sha256': 'c' * 64,
|
|
'token_sha256': TOKEN_SHA256,
|
|
'result_upload_body_timeout_seconds': 1800,
|
|
'execution_snapshot': _execution_snapshot(),
|
|
})
|
|
|
|
def test_remote_execution_plans_validate_all_three_contexts(self):
|
|
git_reservation = _reservation_for_snapshot(
|
|
_execution_snapshot(), 'github', 'github',
|
|
'https://github.com/example/project.git',
|
|
'https://github.com/example/project',
|
|
)
|
|
self.assertEqual(remote_assignment_execution_plan(git_reservation), {
|
|
'kind': 'exact_git_v1',
|
|
'execution_target': 'https://github.com/example/project.git',
|
|
'bound_plan': None,
|
|
})
|
|
self.assertEqual(
|
|
validate_remote_result_execution_plan(git_reservation, {}),
|
|
remote_assignment_execution_plan(git_reservation),
|
|
)
|
|
|
|
docker_target = 'library/alpine@sha256:' + ('a' * 64)
|
|
docker_reservation = _reservation_for_snapshot(
|
|
_direct_execution_snapshot(
|
|
'dockerhub', 'docker', 'docker_direct_v1',
|
|
),
|
|
'dockerhub', 'docker', docker_target, docker_target,
|
|
)
|
|
self.assertEqual(remote_assignment_execution_plan(docker_reservation), {
|
|
'kind': 'docker_direct_v1',
|
|
'execution_target': docker_target,
|
|
'bound_plan': None,
|
|
})
|
|
self.assertEqual(
|
|
validate_remote_result_execution_plan(docker_reservation, {}),
|
|
remote_assignment_execution_plan(docker_reservation),
|
|
)
|
|
|
|
hf_reservation = _reservation_for_snapshot(
|
|
_direct_execution_snapshot(
|
|
'huggingface', 'huggingface', 'huggingface_space_v1',
|
|
),
|
|
'huggingface', 'huggingface', 'ExampleOrg/Space-One',
|
|
'exampleorg/space-one',
|
|
)
|
|
self.assertEqual(remote_assignment_execution_plan(hf_reservation), {
|
|
'kind': 'huggingface_space_v1',
|
|
'execution_target': 'ExampleOrg/Space-One',
|
|
'bound_plan': None,
|
|
})
|
|
self.assertEqual(
|
|
validate_remote_result_execution_plan(hf_reservation, {}),
|
|
remote_assignment_execution_plan(hf_reservation),
|
|
)
|
|
|
|
def test_direct_execution_plans_fail_closed_on_context_or_metadata_drift(self):
|
|
docker_target = 'library/alpine@sha256:' + ('a' * 64)
|
|
reservation = _reservation_for_snapshot(
|
|
_direct_execution_snapshot(
|
|
'dockerhub', 'docker', 'docker_direct_v1',
|
|
),
|
|
'dockerhub', 'docker', docker_target, docker_target,
|
|
)
|
|
for updates in (
|
|
{'source': 'github'},
|
|
{'platform': 'gitlab'},
|
|
{'target': 'library/alpine:latest', 'normalized_target': 'library/alpine:latest'},
|
|
{'remote_effective_config_sha256': 'f' * 64},
|
|
{'git_scan_plan_json': '{}'},
|
|
{'docker_layer_plan_json': '{}'},
|
|
):
|
|
with self.subTest(updates=updates):
|
|
changed = dict(reservation)
|
|
changed.update(updates)
|
|
with self.assertRaises(ScanEventConflictError):
|
|
remote_assignment_execution_plan(changed)
|
|
for metadata in (
|
|
{'git_scan_plan': {}},
|
|
{'git_scan_execution': {}},
|
|
{'docker_layer_plan': {}},
|
|
{'docker_layer_execution': {}},
|
|
):
|
|
with self.subTest(metadata=metadata):
|
|
with self.assertRaises(ScanEventConflictError):
|
|
validate_remote_result_execution_plan(reservation, metadata)
|
|
|
|
def test_provisioning_does_not_implicitly_reactivate_device(self):
|
|
db = object.__new__(ScannerDB)
|
|
db.conn = _ProvisionConnection()
|
|
|
|
result = db.provision_remote_worker_device(
|
|
'user', 'device', TOKEN_SHA256, 2,
|
|
)
|
|
|
|
device_upsert = next(
|
|
statement for statement, _params in db.conn.statements
|
|
if 'INSERT INTO remote_worker_devices' in statement
|
|
)
|
|
self.assertNotIn('revoked_at = NULL', device_upsert)
|
|
self.assertTrue(result['revoked'])
|
|
self.assertEqual(db.conn.commits, 1)
|
|
|
|
def test_device_token_issue_is_dedicated_and_never_updates_user_quota(self):
|
|
class IssueConnection:
|
|
is_postgres = True
|
|
|
|
def __init__(self):
|
|
self.statements = []
|
|
self.commits = 0
|
|
self.rollbacks = 0
|
|
|
|
def execute(self, statement, params=()):
|
|
self.statements.append((statement, params))
|
|
normalized = ' '.join(statement.split())
|
|
if normalized.startswith('SELECT id, user_key FROM remote_worker_users'):
|
|
return _Cursor({'id': 7, 'user_key': 'alice'})
|
|
if normalized.startswith('SELECT id, user_id, device_key, revoked_at'):
|
|
return _Cursor({
|
|
'id': 11, 'user_id': 7, 'device_key': 'laptop',
|
|
'revoked_at': None,
|
|
})
|
|
return _Cursor(rowcount=1)
|
|
|
|
def commit(self):
|
|
self.commits += 1
|
|
|
|
def rollback(self):
|
|
self.rollbacks += 1
|
|
|
|
db = object.__new__(ScannerDB)
|
|
db.conn = IssueConnection()
|
|
|
|
result = db.issue_remote_worker_device(
|
|
'alice', 'laptop', TOKEN_SHA256, rotate=True,
|
|
)
|
|
|
|
sql = '\n'.join(statement for statement, _params in db.conn.statements)
|
|
update = next(
|
|
(statement, params) for statement, params in db.conn.statements
|
|
if 'UPDATE remote_worker_devices SET token_sha256' in statement
|
|
)
|
|
self.assertNotIn('active_assignment_cap', sql)
|
|
self.assertNotIn('UPDATE remote_worker_users', sql)
|
|
self.assertEqual(update[1][0], TOKEN_SHA256)
|
|
self.assertNotIn('token_sha256', result)
|
|
self.assertEqual(result['device_key'], 'laptop')
|
|
self.assertEqual(db.conn.commits, 1)
|
|
self.assertEqual(db.conn.rollbacks, 0)
|
|
|
|
def test_admin_snapshot_is_bounded_and_uses_only_authoritative_worker_records(self):
|
|
class SnapshotConnection:
|
|
is_postgres = True
|
|
|
|
def __init__(self):
|
|
self.statements = []
|
|
self.commits = 0
|
|
|
|
def execute(self, statement, params=()):
|
|
self.statements.append((statement, params))
|
|
normalized = ' '.join(statement.split())
|
|
if normalized.startswith('SELECT user_key, active_assignment_cap'):
|
|
return _RowsCursor([{
|
|
'user_key': 'alice', 'active_assignment_cap': 2,
|
|
'disabled': False,
|
|
}])
|
|
if normalized.startswith('SELECT d.device_key'):
|
|
return _RowsCursor([{
|
|
'device_key': 'laptop', 'user_key': 'alice',
|
|
'last_contact_at': 'contact', 'revoked': False,
|
|
'unfinished_count': 1, 'completed_count': 2,
|
|
'failed_count': 3, 'expired_count': 4,
|
|
}])
|
|
if normalized.startswith('SELECT r.id AS reservation_id'):
|
|
return _RowsCursor([{
|
|
'reservation_id': 41, 'queue_id': 9, 'user_key': 'alice',
|
|
'active_assignment_cap': 2,
|
|
'device_key': 'laptop', 'source': 'github',
|
|
'target': 'https://secret@github.com/example/project?token=hidden',
|
|
'issued_at': 'issued', 'finished_at': 'finished',
|
|
'duration_seconds': 17, 'assignment_outcome': 'accepted',
|
|
'scan_outcome': 'error',
|
|
'accepted': True, 'ingested': False,
|
|
'assignment_code': None,
|
|
}])
|
|
if normalized.startswith('SELECT id AS queue_id'):
|
|
return _RowsCursor([{
|
|
'queue_id': 12, 'source': 'github',
|
|
'target': 'https://secret@github.com/example/deferred?token=hidden',
|
|
'available_after': 'later',
|
|
}])
|
|
raise AssertionError(normalized)
|
|
|
|
def commit(self):
|
|
self.commits += 1
|
|
|
|
def rollback(self):
|
|
raise AssertionError('snapshot unexpectedly rolled back')
|
|
|
|
db = object.__new__(ScannerDB)
|
|
db.conn = SnapshotConnection()
|
|
|
|
snapshot = db.admin_remote_worker_snapshot(limit=7)
|
|
|
|
sql = '\n'.join(statement for statement, _params in db.conn.statements)
|
|
normalized_lower = ' '.join(sql.lower().split())
|
|
for table in (
|
|
'remote_worker_users', 'remote_worker_devices',
|
|
'result_reservations', 'result_bundles', 'target_queue',
|
|
):
|
|
self.assertIn(table, normalized_lower)
|
|
for resolution in ('bundle_accepted', 'prebundle_report', 'expired'):
|
|
self.assertIn(resolution, normalized_lower)
|
|
self.assertIn("else 'unfinished'", normalized_lower)
|
|
self.assertIn('extract(epoch', normalized_lower)
|
|
self.assertIn('remote_issued_at', normalized_lower)
|
|
self.assertIn('remote_resolved_at', normalized_lower)
|
|
self.assertIn('normalized_target as target', normalized_lower)
|
|
self.assertIn('committed_at', normalized_lower)
|
|
self.assertNotIn('token_sha256', normalized_lower)
|
|
self.assertNotIn('last_error_detail', normalized_lower)
|
|
self.assertNotIn('finding', normalized_lower)
|
|
self.assertNotIn('credential', normalized_lower)
|
|
self.assertNotIn('online', normalized_lower)
|
|
self.assertTrue(all(params == (7,) for _statement, params in db.conn.statements))
|
|
self.assertEqual(snapshot['workers'][0]['last_contact_at'], 'contact')
|
|
self.assertEqual(snapshot['assignments'][0]['assignment_outcome'], 'accepted')
|
|
self.assertEqual(snapshot['assignments'][0]['scan_outcome'], 'error')
|
|
self.assertTrue(snapshot['assignments'][0]['accepted'])
|
|
self.assertFalse(snapshot['assignments'][0]['ingested'])
|
|
self.assertEqual(
|
|
snapshot['assignments'][0]['target'],
|
|
'https://github.com/example/project',
|
|
)
|
|
self.assertEqual(
|
|
snapshot['deferred_queue'][0]['target'],
|
|
'https://github.com/example/deferred',
|
|
)
|
|
self.assertNotIn('secret', repr(snapshot))
|
|
self.assertNotIn('hidden', repr(snapshot))
|
|
self.assertEqual(db.conn.commits, 1)
|
|
|
|
def test_admin_diagnostic_grouping_is_deterministic_and_retains_occurrences(self):
|
|
class DiagnosticConnection:
|
|
is_postgres = True
|
|
|
|
def __init__(self):
|
|
self.statement = None
|
|
self.params = None
|
|
|
|
def execute(self, statement, params=()):
|
|
self.statement = ' '.join(statement.split())
|
|
self.params = params
|
|
rows = []
|
|
for diagnostic_id, reservation_id, fingerprint in (
|
|
(3, 43, 'stable-fingerprint'),
|
|
(2, 42, 'stable-fingerprint'),
|
|
(1, 41, 'other-fingerprint'),
|
|
):
|
|
rows.append({
|
|
'id': diagnostic_id,
|
|
'diagnostic_uid': str(diagnostic_id) * 64,
|
|
'reservation_id': reservation_id,
|
|
'target_scan_id': None, 'source': 'gitlab',
|
|
'phase': 'scanning', 'kind': 'exception',
|
|
'category': 'rate_limit', 'code': 'provider.rate_limit',
|
|
'summary': 'rate limited', 'retryable': 1,
|
|
'occurred_at': f'2026-09-20T00:0{diagnostic_id}:00Z',
|
|
'received_at': f'2026-09-20T00:0{diagnostic_id}:01Z',
|
|
'envelope_json': json.dumps({
|
|
'exception': {'fingerprint': fingerprint},
|
|
}),
|
|
'fingerprint': fingerprint,
|
|
'fingerprint_count': (
|
|
2 if fingerprint == 'stable-fingerprint' else 1
|
|
),
|
|
'affected_assignment_count': (
|
|
2 if fingerprint == 'stable-fingerprint' else 1
|
|
),
|
|
'matched_occurrence_count': 4,
|
|
'user_key': 'alice', 'device_key': 'laptop',
|
|
'assignment_outcome': (
|
|
'accepted' if reservation_id == 41 else 'prebundle_failed'
|
|
),
|
|
'scan_outcome': (
|
|
'error' if reservation_id == 41 else 'unavailable'
|
|
),
|
|
})
|
|
page_limit, offset = params[-2:]
|
|
return _RowsCursor(rows[offset:offset + page_limit])
|
|
|
|
def commit(self):
|
|
pass
|
|
|
|
def rollback(self):
|
|
raise AssertionError('group query unexpectedly rolled back')
|
|
|
|
db = object.__new__(ScannerDB)
|
|
db.conn = DiagnosticConnection()
|
|
result = db.admin_worker_diagnostic_groups(
|
|
limit=1, occurrence_offset=0,
|
|
filters={'category': 'rate_limit', 'retryable': True},
|
|
)
|
|
|
|
self.assertEqual(len(result['groups']), 1)
|
|
group = result['groups'][0]
|
|
self.assertEqual(group['fingerprint'], 'stable-fingerprint')
|
|
self.assertEqual(group['count'], 2)
|
|
self.assertEqual(group['page_occurrence_count'], 1)
|
|
self.assertEqual(group['affected_assignment_count'], 2)
|
|
self.assertEqual(group['affected_assignments'], [43])
|
|
self.assertEqual(
|
|
[item['reservation_id'] for item in group['occurrences']], [43],
|
|
)
|
|
self.assertEqual(result['matched_occurrence_count'], 4)
|
|
self.assertEqual(result['page_occurrence_count'], 1)
|
|
self.assertTrue(result['has_next'])
|
|
self.assertEqual(result['next_occurrence_offset'], 1)
|
|
first_statement = db.conn.statement
|
|
first_params = db.conn.params
|
|
second = db.admin_worker_diagnostic_groups(
|
|
limit=1, occurrence_offset=1,
|
|
filters={'category': 'rate_limit', 'retryable': True},
|
|
)
|
|
self.assertEqual(
|
|
second['groups'][0]['occurrences'][0]['reservation_id'], 42,
|
|
)
|
|
self.assertTrue(second['has_previous'])
|
|
self.assertEqual(second['previous_occurrence_offset'], 0)
|
|
self.assertEqual(second['next_occurrence_offset'], 2)
|
|
self.assertIn('wd.category = ?', first_statement)
|
|
self.assertIn('wd.retryable = ?', first_statement)
|
|
self.assertIn('COUNT(DISTINCT reservation_id)', first_statement)
|
|
self.assertIn('LIMIT ? OFFSET ?', first_statement)
|
|
self.assertEqual(first_params, ('rate_limit', 1, 2, 0))
|
|
|
|
def test_admin_duration_metrics_use_completed_edges_percentiles_and_sample_labels(self):
|
|
class MetricConnection:
|
|
is_postgres = True
|
|
|
|
def __init__(self):
|
|
self.statement = None
|
|
self.params = None
|
|
|
|
def execute(self, statement, params=()):
|
|
self.statement = ' '.join(statement.split())
|
|
self.params = params
|
|
row = {
|
|
'source': 'dockerhub', 'phase': 'cleaning',
|
|
'outcome': 'accepted', 'sample_count': 4,
|
|
'p50_seconds': 10, 'p95_seconds': 20, 'p99_seconds': 21,
|
|
'total_group_count': 21,
|
|
}
|
|
page_limit, offset = params[-2:]
|
|
count = max(0, min(page_limit, 21 - offset))
|
|
return _RowsCursor([dict(row) for _ in range(count)])
|
|
|
|
def commit(self):
|
|
pass
|
|
|
|
def rollback(self):
|
|
raise AssertionError('metric query unexpectedly rolled back')
|
|
|
|
db = object.__new__(ScannerDB)
|
|
db.conn = MetricConnection()
|
|
result = db.admin_worker_duration_metrics(
|
|
limit=20,
|
|
filters={
|
|
'source': 'dockerhub', 'phase': 'cleaning',
|
|
'since': '2026-09-20T00:00:00+00:00',
|
|
},
|
|
)
|
|
|
|
metric = result['metrics'][0]
|
|
self.assertEqual(metric['sample_count'], 4)
|
|
self.assertFalse(metric['sufficient'])
|
|
self.assertEqual(metric['minimum_sample_count'], 5)
|
|
self.assertEqual(result['total_group_count'], 21)
|
|
self.assertTrue(result['has_next'])
|
|
self.assertEqual(result['next_metric_offset'], 20)
|
|
first_statement = db.conn.statement
|
|
first_params = db.conn.params
|
|
continuation = db.admin_worker_duration_metrics(
|
|
limit=20, offset=20,
|
|
filters={
|
|
'source': 'dockerhub', 'phase': 'cleaning',
|
|
'since': '2026-09-20T00:00:00+00:00',
|
|
},
|
|
)
|
|
self.assertEqual(continuation['page_group_count'], 1)
|
|
self.assertTrue(continuation['has_previous'])
|
|
self.assertFalse(continuation['has_next'])
|
|
self.assertEqual(continuation['previous_metric_offset'], 0)
|
|
self.assertIn('PERCENTILE_CONT(0.50)', first_statement)
|
|
self.assertIn('PERCENTILE_CONT(0.95)', first_statement)
|
|
self.assertIn('PERCENTILE_CONT(0.99)', first_statement)
|
|
self.assertIn('NULLIF(phase_started_at, \'\')', first_statement)
|
|
self.assertIn('pe.next_phase_started_at IS NOT NULL', first_statement)
|
|
self.assertIn(
|
|
'COALESCE( pe.next_phase_started_at, rs.remote_resolved_at )',
|
|
first_statement,
|
|
)
|
|
self.assertIn('- pe.phase_started_at::timestamptz', first_statement)
|
|
self.assertNotIn('- pe.event_timestamp::timestamptz', first_statement)
|
|
self.assertIn('completed_at >= ?', first_statement)
|
|
self.assertNotIn('r.remote_issued_at >= ?', first_statement)
|
|
self.assertIn(
|
|
'(r.remote_resolved_at IS NULL OR r.remote_resolved_at >= ?)',
|
|
first_statement,
|
|
)
|
|
self.assertEqual(first_params, (
|
|
'dockerhub', '2026-09-20T00:00:00+00:00', 'cleaning',
|
|
'2026-09-20T00:00:00+00:00', 21, 0,
|
|
))
|
|
|
|
def test_admin_worker_filter_accepts_trimmed_128_character_device_keys(self):
|
|
key = 'Ω' + (' /[]+#:' * 18) + 'Z'
|
|
self.assertEqual(len(key), 128)
|
|
class FilterConnection:
|
|
is_postgres = True
|
|
|
|
def execute(self, statement, params=()):
|
|
normalized = ' '.join(statement.split())
|
|
if normalized.startswith('SELECT user_key, active_assignment_cap'):
|
|
return _RowsCursor([])
|
|
if normalized.startswith('SELECT d.device_key'):
|
|
return _RowsCursor([])
|
|
if normalized.startswith('SELECT r.id AS reservation_id'):
|
|
self.assignment_params = params
|
|
return _RowsCursor([])
|
|
if normalized.startswith('SELECT id AS queue_id'):
|
|
return _RowsCursor([])
|
|
raise AssertionError(normalized)
|
|
|
|
def commit(self):
|
|
pass
|
|
|
|
def rollback(self):
|
|
raise AssertionError('filter query unexpectedly rolled back')
|
|
|
|
db = object.__new__(ScannerDB)
|
|
db.conn = FilterConnection()
|
|
snapshot = db.admin_remote_worker_snapshot(
|
|
limit=2, filters={'worker': ' ' + key + ' '},
|
|
)
|
|
self.assertEqual(snapshot['filters']['worker'], key)
|
|
self.assertEqual(db.conn.assignment_params, (key, 2))
|
|
|
|
def test_assignment_detail_bound_keeps_latest_progress_chronological_and_times_final_phase(self):
|
|
assignment = {
|
|
'id': 41, 'queue_id': 9, 'source': 'gitlab',
|
|
'normalized_target': 'group/project', 'user_key': 'alice',
|
|
'device_key': 'rack / east', 'active_assignment_cap': 2,
|
|
'assignment_outcome': 'unfinished', 'scan_outcome': 'unavailable',
|
|
'remote_issued_at': '2026-09-20T00:00:00Z',
|
|
'remote_resolved_at': None,
|
|
'remote_expires_at': '2026-09-21T00:00:00Z',
|
|
'authority_at': '2026-09-20T00:05:00Z',
|
|
'last_error_code': None, 'last_error_detail': None,
|
|
'target_scan_id': None, 'remote_execution_snapshot_json': json.dumps({
|
|
'compatibility': {'protocol_version': 2},
|
|
}),
|
|
'remote_resolution_json': None,
|
|
'remote_diagnostic_projection_version': None,
|
|
'remote_diagnostic_count': None,
|
|
}
|
|
events = []
|
|
for sequence, minute, phase in (
|
|
(2, 1, 'preparing'), (3, 2, 'scanning'), (4, 4, 'cleaning'),
|
|
):
|
|
timestamp = f'2026-09-20T00:0{minute}:00Z'
|
|
events.append({
|
|
'id': sequence, 'reservation_id': 41, 'sequence': sequence,
|
|
'phase': phase, 'event_timestamp': timestamp,
|
|
'phase_started_at': timestamp, 'received_at': timestamp,
|
|
'event_json': json.dumps({
|
|
'phase': phase, 'timestamp': timestamp,
|
|
'phase_started_at': timestamp, 'scan_deadline_at': None,
|
|
}),
|
|
'total_event_count': 4,
|
|
})
|
|
|
|
class DetailConnection:
|
|
is_postgres = True
|
|
|
|
def execute(self, statement, params=()):
|
|
normalized = ' '.join(statement.split())
|
|
if normalized.startswith('SELECT r.*'):
|
|
return _Cursor(assignment)
|
|
if normalized.startswith('SELECT * FROM ( SELECT e.*'):
|
|
self.progress_sql = normalized
|
|
self.progress_params = params
|
|
return _RowsCursor(events)
|
|
if normalized.startswith('SELECT * FROM worker_diagnostics'):
|
|
return _RowsCursor([])
|
|
raise AssertionError(normalized)
|
|
|
|
def commit(self):
|
|
pass
|
|
|
|
def rollback(self):
|
|
raise AssertionError('detail query unexpectedly rolled back')
|
|
|
|
db = object.__new__(ScannerDB)
|
|
db.conn = DetailConnection()
|
|
detail = db.admin_worker_assignment_detail(
|
|
41, event_limit=2, diagnostic_limit=2,
|
|
)
|
|
|
|
self.assertIn('ORDER BY e.sequence DESC', db.conn.progress_sql)
|
|
self.assertEqual(db.conn.progress_params, (41, 3))
|
|
self.assertTrue(detail['progress']['truncated'])
|
|
self.assertEqual(detail['progress']['total_event_count'], 4)
|
|
self.assertEqual(detail['progress']['omitted_older_event_count'], 2)
|
|
self.assertEqual(
|
|
[event['sequence'] for event in detail['progress']['events']], [3, 4],
|
|
)
|
|
self.assertEqual(detail['progress']['current_phase'], 'cleaning')
|
|
self.assertEqual(detail['progress']['phase_age_seconds'], 60.0)
|
|
self.assertEqual(detail['progress']['last_progress_age_seconds'], 60.0)
|
|
self.assertEqual(
|
|
[(item['phase'], item['duration_seconds'], item['complete'])
|
|
for item in detail['durations'] if item['phase'] != 'end_to_end'],
|
|
[('scanning', 120.0, True), ('cleaning', 60.0, False)],
|
|
)
|
|
|
|
def test_assignment_detail_prefers_durable_receipt_scan_deadline(self):
|
|
receipt_deadline = '2026-09-20T00:30:00Z'
|
|
assignment = {
|
|
'id': 41, 'queue_id': 9, 'source': 'gitlab',
|
|
'normalized_target': 'group/project', 'user_key': 'alice',
|
|
'device_key': 'rack / east', 'active_assignment_cap': 0,
|
|
'assignment_outcome': 'bundle_accepted', 'scan_outcome': 'succeeded',
|
|
'remote_issued_at': '2026-09-20T00:00:00Z',
|
|
'remote_resolved_at': '2026-09-20T00:05:00Z',
|
|
'remote_expires_at': '2026-09-21T00:00:00Z',
|
|
'authority_at': '2026-09-20T00:05:00Z',
|
|
'last_error_code': None, 'last_error_detail': None,
|
|
'target_scan_id': None,
|
|
'remote_execution_snapshot_json': json.dumps({
|
|
'compatibility': {'protocol_version': 2},
|
|
}),
|
|
'remote_resolution_json': json.dumps({
|
|
'resolution': 'bundle_accepted',
|
|
'deadlines': {'scan_deadline_at': receipt_deadline},
|
|
}),
|
|
'remote_diagnostic_projection_version': 1,
|
|
'remote_diagnostic_count': 0,
|
|
}
|
|
latest_timestamp = '2026-09-20T00:04:00Z'
|
|
events = [{
|
|
'id': 1, 'reservation_id': 41, 'sequence': 1,
|
|
'phase': 'awaiting_receipt', 'event_timestamp': latest_timestamp,
|
|
'phase_started_at': latest_timestamp, 'received_at': latest_timestamp,
|
|
'event_json': json.dumps({
|
|
'phase': 'awaiting_receipt', 'timestamp': latest_timestamp,
|
|
'phase_started_at': latest_timestamp, 'scan_deadline_at': None,
|
|
}),
|
|
'total_event_count': 1,
|
|
}]
|
|
|
|
class DetailConnection:
|
|
is_postgres = True
|
|
|
|
def execute(self, statement, params=()):
|
|
normalized = ' '.join(statement.split())
|
|
if normalized.startswith('SELECT r.*'):
|
|
return _Cursor(assignment)
|
|
if normalized.startswith('SELECT * FROM ( SELECT e.*'):
|
|
return _RowsCursor(events)
|
|
if normalized.startswith('SELECT * FROM worker_diagnostics'):
|
|
return _RowsCursor([])
|
|
raise AssertionError(normalized)
|
|
|
|
def commit(self):
|
|
pass
|
|
|
|
def rollback(self):
|
|
raise AssertionError('detail query unexpectedly rolled back')
|
|
|
|
db = object.__new__(ScannerDB)
|
|
db.conn = DetailConnection()
|
|
detail = db.admin_worker_assignment_detail(41)
|
|
|
|
self.assertEqual(detail['deadlines']['scan_deadline_at'], receipt_deadline)
|
|
|
|
def test_admin_worker_filters_reject_unknown_values_before_sql(self):
|
|
class NoSQLConnection:
|
|
is_postgres = True
|
|
|
|
def execute(self, _statement, _params=()):
|
|
raise AssertionError('invalid filters must not execute SQL')
|
|
|
|
db = object.__new__(ScannerDB)
|
|
db.conn = NoSQLConnection()
|
|
with self.assertRaisesRegex(ValueError, 'filters'):
|
|
db.admin_remote_worker_snapshot(filters={'assignment_outcome': 'completed'})
|
|
|
|
def test_admin_requeue_rejects_unbounded_or_duplicate_ids_before_sql(self):
|
|
class RequeueConnection:
|
|
is_postgres = True
|
|
|
|
def execute(self, statement, params=()):
|
|
raise AssertionError('invalid selection must not execute SQL')
|
|
|
|
db = object.__new__(ScannerDB)
|
|
db.conn = RequeueConnection()
|
|
with self.assertRaisesRegex(ValueError, 'exceeds'):
|
|
db.admin_requeue_deferred_targets([1, 1], max_items=2)
|
|
with self.assertRaisesRegex(ValueError, 'exceeds'):
|
|
db.admin_requeue_deferred_targets([1, 2, 3], max_items=2)
|
|
|
|
def test_admin_source_queue_discard_preserves_issued_and_scanned_rows(self):
|
|
class Cursor:
|
|
rowcount = 7
|
|
|
|
class QueueConnection:
|
|
is_postgres = True
|
|
|
|
def __init__(self):
|
|
self.statement = None
|
|
self.params = None
|
|
self.committed = False
|
|
|
|
def execute(self, statement, params=()):
|
|
self.statement = statement
|
|
self.params = params
|
|
return Cursor()
|
|
|
|
def commit(self):
|
|
self.committed = True
|
|
|
|
def rollback(self):
|
|
raise AssertionError('queue discard unexpectedly rolled back')
|
|
|
|
db = object.__new__(ScannerDB)
|
|
db.conn = QueueConnection()
|
|
affected = db.admin_discard_queued_source('gitlab')
|
|
|
|
self.assertEqual(affected, 7)
|
|
self.assertTrue(db.conn.committed)
|
|
self.assertIn("status IN ('pending', 'deferred', 'cold')", db.conn.statement)
|
|
self.assertIn('current_result_reservation_id IS NULL', db.conn.statement)
|
|
self.assertIn('target_scan_id IS NULL', db.conn.statement)
|
|
self.assertIn('NOT EXISTS (', db.conn.statement)
|
|
self.assertIn('FROM result_reservations', db.conn.statement)
|
|
self.assertIn('FROM target_scans', db.conn.statement)
|
|
self.assertNotIn("status = 'in_progress'", db.conn.statement)
|
|
self.assertEqual(db.conn.params[-1], 'gitlab')
|
|
|
|
def test_terminal_resolution_leaves_artifacts_cleanup_eligible(self):
|
|
db = object.__new__(ScannerDB)
|
|
db.conn = _ResolutionConnection()
|
|
db._release_docker_blob_leases_locked = lambda *_args, **_kwargs: None
|
|
db._transition_docker_depth_binding_locked = lambda *_args, **_kwargs: None
|
|
row = {
|
|
'id': 41, 'queue_id': 9, 'claim_lease_token': 'lease',
|
|
'scan_event_id': 'event', 'bundle_id': 'bundle',
|
|
'declared_bundle_bytes': 100, 'reserved_bundle_bytes': 100,
|
|
'reserved_projection_items': 1,
|
|
'reserved_projection_bytes': 100, 'reserved_candidate_items': 1,
|
|
'reserved_candidate_bytes': 100,
|
|
}
|
|
|
|
receipt = db._resolve_remote_scanning_locked(
|
|
row, 'expired', None, 'deadline expired',
|
|
'2026-09-17T00:00:00+00:00',
|
|
)
|
|
|
|
artifact_update = next(
|
|
statement for statement, _params in db.conn.statements
|
|
if 'UPDATE pipeline_artifacts' in statement
|
|
)
|
|
self.assertIn("state IN ('expected','present')", artifact_update)
|
|
self.assertNotIn("SET state = 'deleted'", artifact_update)
|
|
self.assertEqual(receipt['resolution'], 'expired')
|
|
|
|
def test_old_persisted_receipt_remains_exact_without_invented_observability(self):
|
|
old_receipt = {
|
|
'receipt_id': 'a' * 64,
|
|
'resolution': 'prebundle_report',
|
|
'reservation_id': 41,
|
|
'bundle_id': 'b' * 32,
|
|
'scan_event_id': 'c' * 32,
|
|
'resolved_at': '2026-09-17T00:00:00+00:00',
|
|
'failure_code': 'client_process_failed',
|
|
'payload_sha256': 'd' * 64,
|
|
}
|
|
row = {
|
|
'remote_resolution_json': json.dumps(
|
|
old_receipt, ensure_ascii=True, sort_keys=True, separators=(',', ':'),
|
|
),
|
|
'remote_receipt_id': old_receipt['receipt_id'],
|
|
}
|
|
|
|
restored = ScannerDB._remote_resolution_from_row(row)
|
|
|
|
self.assertEqual(restored, old_receipt)
|
|
self.assertNotIn('deadlines', restored)
|
|
self.assertNotIn('latest_progress', restored)
|
|
self.assertNotIn('diagnostics', restored)
|
|
|
|
def test_pending_request_reconciliation_aborts_without_claim(self):
|
|
intent = {
|
|
'state': 'pending', 'remote_device_id': 11,
|
|
'remote_user_id': 7, 'reservation_id': None,
|
|
}
|
|
db = object.__new__(ScannerDB)
|
|
db.conn = _ReconcileConnection(intent)
|
|
|
|
result = db.reconcile_remote_assignment_request(
|
|
'f' * 32, 11, TOKEN_SHA256,
|
|
)
|
|
|
|
self.assertEqual(result, {'state': 'aborted'})
|
|
update = next(
|
|
statement for statement, _params in db.conn.statements
|
|
if 'UPDATE admission_intents' in statement
|
|
)
|
|
self.assertIn('serialized_remote_recovery_observed_no_commit', update)
|
|
self.assertEqual(db.conn.commits, 1)
|
|
self.assertEqual(db.conn.rollbacks, 0)
|
|
|
|
def test_committed_request_reconciliation_returns_exact_snapshot(self):
|
|
snapshot = _execution_snapshot()
|
|
mapped = ScannerDB._remote_assignment_mapping({
|
|
'user_id': 7, 'device_id': 11,
|
|
'effective_config_sha256': _effective_config_sha256(),
|
|
'client_compat_sha256': 'c' * 64,
|
|
'token_sha256': TOKEN_SHA256,
|
|
'result_upload_body_timeout_seconds': 1800,
|
|
'execution_snapshot': snapshot,
|
|
})
|
|
reservation = _remote_reservation({
|
|
'remote_execution_snapshot_json': mapped['execution_snapshot_json'],
|
|
'remote_execution_snapshot_sha256': mapped['execution_snapshot_sha256'],
|
|
})
|
|
intent = {
|
|
'state': 'committed', 'remote_device_id': 11,
|
|
'remote_user_id': 7, 'reservation_id': 41,
|
|
}
|
|
db = object.__new__(ScannerDB)
|
|
db.conn = _ReconcileConnection(intent, reservation)
|
|
|
|
result = db.reconcile_remote_assignment_request(
|
|
'f' * 32, 11, TOKEN_SHA256,
|
|
)
|
|
|
|
self.assertEqual(result['state'], 'committed')
|
|
self.assertEqual(result['claim']['reservation_id'], 41)
|
|
self.assertEqual(
|
|
result['claim']['remote_result_upload_body_timeout_seconds'], 1800,
|
|
)
|
|
self.assertEqual(result['execution_snapshot'], snapshot)
|
|
self.assertEqual(
|
|
result['execution_snapshot_sha256'],
|
|
mapped['execution_snapshot_sha256'],
|
|
)
|
|
self.assertEqual(result['execution_plan'], {
|
|
'kind': 'exact_git_v1',
|
|
'execution_target': reservation['target'],
|
|
'bound_plan': None,
|
|
})
|
|
self.assertIsNone(result['git_plan'])
|
|
self.assertIsNone(result['receipt'])
|
|
reconciliation_sql = next(
|
|
statement for statement, _params in db.conn.statements
|
|
if 'FROM result_reservations r' in statement
|
|
)
|
|
self.assertIn('FOR UPDATE OF r, q', reconciliation_sql)
|
|
self.assertEqual(db.conn.commits, 1)
|
|
self.assertEqual(db.conn.rollbacks, 0)
|
|
|
|
def test_generic_claim_renewal_refuses_remote_reservations(self):
|
|
class RenewalConnection:
|
|
is_postgres = True
|
|
|
|
def __init__(self):
|
|
self.rollbacks = 0
|
|
|
|
def execute(self, statement, params=()):
|
|
if 'FROM result_reservations r' in statement:
|
|
return _Cursor({
|
|
'id': 41, 'state': 'scanning',
|
|
'assignment_kind': 'remote',
|
|
})
|
|
return _Cursor()
|
|
|
|
def rollback(self):
|
|
self.rollbacks += 1
|
|
|
|
db = object.__new__(ScannerDB)
|
|
db.conn = RenewalConnection()
|
|
db._lock_docker_depth_experiment_for_reservation = lambda *_args: None
|
|
|
|
self.assertFalse(db.renew_result_claim(41, 'lease', lease_seconds=86400))
|
|
self.assertEqual(db.conn.rollbacks, 1)
|
|
|
|
def test_reaper_continues_after_one_expiry_failure(self):
|
|
class RowsCursor:
|
|
def fetchall(self):
|
|
return [{'id': 41}, {'id': 42}]
|
|
|
|
class ReaperConnection:
|
|
is_postgres = True
|
|
|
|
def __init__(self):
|
|
self.commits = 0
|
|
|
|
def execute(self, statement, params=()):
|
|
return RowsCursor()
|
|
|
|
def commit(self):
|
|
self.commits += 1
|
|
|
|
calls = []
|
|
db = object.__new__(ScannerDB)
|
|
db.conn = ReaperConnection()
|
|
|
|
def expire(reservation_id, now=None):
|
|
calls.append((reservation_id, now))
|
|
if reservation_id == 41:
|
|
raise RuntimeError('corrupt fixture')
|
|
return {'reservation_id': reservation_id, 'resolution': 'expired'}
|
|
|
|
db.expire_remote_assignment = expire
|
|
with mock.patch('scanner_db.logger.error') as logged:
|
|
receipts = db.reap_expired_remote_assignments(limit=2)
|
|
|
|
self.assertEqual([item[0] for item in calls], [41, 42])
|
|
self.assertEqual(receipts, [{'reservation_id': 42, 'resolution': 'expired'}])
|
|
self.assertEqual(db.conn.commits, 1)
|
|
logged.assert_called_once()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|