sashatrask
1087 lines
43 KiB
Python
1087 lines
43 KiB
Python
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
APP_DIR = ROOT / 'app'
|
|
sys.path.insert(0, str(APP_DIR))
|
|
|
|
import docker_depth_report
|
|
from docker_depth_report import build_docker_depth_report
|
|
from scanner_db import ScannerDB, utc_now_iso
|
|
|
|
|
|
SCHEMA = '''
|
|
CREATE TABLE docker_depth_experiments (
|
|
id INTEGER PRIMARY KEY,
|
|
experiment_key TEXT NOT NULL,
|
|
source TEXT NOT NULL,
|
|
state TEXT NOT NULL,
|
|
query_count INTEGER NOT NULL,
|
|
target_count INTEGER NOT NULL,
|
|
selection_count INTEGER NOT NULL
|
|
);
|
|
CREATE TABLE docker_depth_experiment_queries (
|
|
id INTEGER PRIMARY KEY,
|
|
experiment_id INTEGER NOT NULL,
|
|
source TEXT NOT NULL,
|
|
query_ordinal INTEGER NOT NULL,
|
|
query TEXT NOT NULL,
|
|
query_sha256 TEXT NOT NULL,
|
|
required_repository_count INTEGER NOT NULL,
|
|
selected_repository_count INTEGER NOT NULL
|
|
);
|
|
CREATE TABLE docker_depth_experiment_repositories (
|
|
id INTEGER PRIMARY KEY,
|
|
experiment_id INTEGER NOT NULL,
|
|
query_ordinal INTEGER NOT NULL,
|
|
source TEXT NOT NULL,
|
|
query TEXT NOT NULL,
|
|
repository_queue_id INTEGER NOT NULL,
|
|
replacement_repository_queue_id INTEGER,
|
|
repository_rank INTEGER NOT NULL,
|
|
work_state TEXT NOT NULL,
|
|
resolver_attempts INTEGER NOT NULL
|
|
);
|
|
CREATE TABLE docker_depth_experiment_targets (
|
|
id INTEGER PRIMARY KEY,
|
|
experiment_id INTEGER NOT NULL,
|
|
target_queue_id INTEGER NOT NULL,
|
|
manifest_id INTEGER NOT NULL,
|
|
state TEXT NOT NULL
|
|
);
|
|
CREATE TABLE docker_depth_experiment_selections (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
experiment_id INTEGER NOT NULL,
|
|
query_ordinal INTEGER NOT NULL,
|
|
experiment_repository_id INTEGER NOT NULL,
|
|
experiment_target_id INTEGER NOT NULL,
|
|
image_rank INTEGER NOT NULL
|
|
);
|
|
CREATE TABLE docker_depth_experiment_scan_bindings (
|
|
id INTEGER PRIMARY KEY,
|
|
experiment_target_id INTEGER NOT NULL,
|
|
reservation_id INTEGER NOT NULL,
|
|
target_scan_id INTEGER,
|
|
attempt INTEGER NOT NULL,
|
|
state TEXT NOT NULL
|
|
);
|
|
CREATE TABLE target_queue (
|
|
id INTEGER PRIMARY KEY,
|
|
source TEXT NOT NULL,
|
|
platform TEXT NOT NULL,
|
|
query TEXT,
|
|
target TEXT NOT NULL,
|
|
normalized_target TEXT NOT NULL
|
|
);
|
|
CREATE TABLE result_reservations (
|
|
id INTEGER PRIMARY KEY,
|
|
queue_id INTEGER NOT NULL,
|
|
source TEXT NOT NULL,
|
|
platform TEXT NOT NULL,
|
|
query TEXT,
|
|
target TEXT NOT NULL,
|
|
normalized_target TEXT NOT NULL,
|
|
claim_lease_token TEXT NOT NULL,
|
|
scan_event_id TEXT NOT NULL,
|
|
state TEXT NOT NULL
|
|
);
|
|
CREATE TABLE target_scans (
|
|
id INTEGER PRIMARY KEY,
|
|
scan_event_id TEXT,
|
|
queue_id INTEGER,
|
|
result_reservation_id INTEGER,
|
|
claim_lease_token TEXT,
|
|
source TEXT,
|
|
normalized_target TEXT,
|
|
scan_type TEXT,
|
|
status TEXT,
|
|
duration_sec REAL,
|
|
error_count INTEGER,
|
|
query TEXT,
|
|
target TEXT,
|
|
raw_result_json TEXT
|
|
);
|
|
CREATE TABLE docker_image_manifests (
|
|
id INTEGER PRIMARY KEY,
|
|
target_queue_id INTEGER NOT NULL,
|
|
repository TEXT,
|
|
layer_count INTEGER NOT NULL
|
|
);
|
|
CREATE TABLE docker_manifest_layers (
|
|
id INTEGER PRIMARY KEY,
|
|
manifest_id INTEGER NOT NULL,
|
|
position_from_base INTEGER NOT NULL,
|
|
position_from_top INTEGER NOT NULL,
|
|
layer_digest TEXT NOT NULL
|
|
);
|
|
CREATE TABLE findings (
|
|
id INTEGER PRIMARY KEY,
|
|
target_scan_id INTEGER NOT NULL,
|
|
finding_fingerprint TEXT,
|
|
detector_secret_hash TEXT,
|
|
raw_secret TEXT,
|
|
redacted_secret TEXT,
|
|
raw_finding_json TEXT
|
|
);
|
|
CREATE TABLE docker_finding_layer_attributions (
|
|
id INTEGER PRIMARY KEY,
|
|
scan_binding_id INTEGER NOT NULL,
|
|
finding_id INTEGER NOT NULL,
|
|
manifest_layer_id INTEGER,
|
|
attribution_state TEXT NOT NULL,
|
|
reported_layer_digest TEXT,
|
|
position_from_base INTEGER,
|
|
position_from_top INTEGER
|
|
);
|
|
CREATE TABLE keycheck_credentials (
|
|
id INTEGER PRIMARY KEY,
|
|
secret_text TEXT,
|
|
secret_json TEXT,
|
|
key_masked TEXT
|
|
);
|
|
CREATE TABLE keycheck_candidates (
|
|
id INTEGER PRIMARY KEY,
|
|
credential_id INTEGER NOT NULL,
|
|
finding_id INTEGER NOT NULL,
|
|
target_scan_id INTEGER NOT NULL,
|
|
state TEXT NOT NULL,
|
|
attempts INTEGER NOT NULL,
|
|
keycheck_result_id INTEGER,
|
|
metadata_json TEXT
|
|
);
|
|
CREATE TABLE keycheck_results (
|
|
id INTEGER PRIMARY KEY,
|
|
candidate_id INTEGER,
|
|
credential_id INTEGER,
|
|
status TEXT NOT NULL,
|
|
status_group TEXT NOT NULL,
|
|
message TEXT,
|
|
source_line TEXT,
|
|
metadata_json TEXT
|
|
);
|
|
CREATE TABLE keycheck_current_state (
|
|
credential_id INTEGER PRIMARY KEY,
|
|
status TEXT NOT NULL,
|
|
status_group TEXT NOT NULL,
|
|
last_result_id INTEGER NOT NULL,
|
|
metadata_json TEXT
|
|
);
|
|
CREATE TABLE errors (
|
|
id INTEGER PRIMARY KEY,
|
|
target_scan_id INTEGER NOT NULL,
|
|
category TEXT,
|
|
summary TEXT,
|
|
raw_error TEXT
|
|
);
|
|
'''
|
|
|
|
|
|
class DockerDepthReportTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.conn = sqlite3.connect(':memory:')
|
|
self.conn.row_factory = sqlite3.Row
|
|
self.conn.executescript(SCHEMA)
|
|
self.conn.execute(
|
|
'''INSERT INTO docker_depth_experiments(
|
|
id, experiment_key, source, state, query_count,
|
|
target_count, selection_count
|
|
) VALUES (1, 'depth-fixture', 'dockerhub', 'completed', 0, 0, 0)'''
|
|
)
|
|
|
|
def tearDown(self):
|
|
self.conn.close()
|
|
|
|
def test_rows_fail_closed_before_materialization_bound_is_exceeded(self):
|
|
with self.assertRaisesRegex(RuntimeError, 'row bound is exceeded'):
|
|
docker_depth_report._rows(
|
|
self.conn,
|
|
'SELECT 1 AS value UNION ALL SELECT 2 UNION ALL SELECT 3',
|
|
(),
|
|
max_rows=2,
|
|
)
|
|
|
|
def add_query(
|
|
self, query_id, ordinal, *, query=None, attempts=1, required=1,
|
|
selected=None, repository_queue_id=None, add_repository=True,
|
|
repository_state='resolved'):
|
|
query = query or f'query-{query_id}'
|
|
selected = required if selected is None else selected
|
|
repository_queue_id = repository_queue_id or 2000 + query_id
|
|
self.conn.execute(
|
|
'''INSERT INTO docker_depth_experiment_queries(
|
|
id, experiment_id, source, query_ordinal, query,
|
|
query_sha256, required_repository_count,
|
|
selected_repository_count
|
|
) VALUES (?, 1, 'dockerhub', ?, ?, ?, ?, ?)''',
|
|
(query_id, ordinal, query, f'{query_id:064x}', required, selected),
|
|
)
|
|
if add_repository:
|
|
self.conn.execute(
|
|
'''INSERT INTO docker_depth_experiment_repositories(
|
|
id, experiment_id, query_ordinal, source, query,
|
|
repository_queue_id, repository_rank, work_state,
|
|
resolver_attempts
|
|
) VALUES (?, 1, ?, 'dockerhub', ?, ?, 1, ?, ?)''',
|
|
(
|
|
query_id, ordinal, query, repository_queue_id,
|
|
repository_state, attempts,
|
|
),
|
|
)
|
|
self.conn.execute(
|
|
'UPDATE docker_depth_experiments SET query_count = query_count + 1 WHERE id = 1'
|
|
)
|
|
|
|
def add_target(self, target_id, query_ranks, *, layer_count=1, repository='repo'):
|
|
target_queue_id = 1000 + target_id
|
|
target = f'{repository}@sha256:{target_id:064x}'
|
|
self.conn.execute(
|
|
'''INSERT INTO target_queue(
|
|
id, source, platform, query, target, normalized_target
|
|
) VALUES (?, 'dockerhub', 'docker', 'fixture', ?, ?)''',
|
|
(target_queue_id, target, target),
|
|
)
|
|
self.conn.execute(
|
|
'''INSERT INTO docker_image_manifests(
|
|
id, target_queue_id, repository, layer_count
|
|
) VALUES (?, ?, ?, ?)''',
|
|
(target_id, target_queue_id, repository, layer_count),
|
|
)
|
|
self.conn.execute(
|
|
'''INSERT INTO docker_depth_experiment_targets(
|
|
id, experiment_id, target_queue_id, manifest_id, state
|
|
) VALUES (?, 1, ?, ?, 'done')''',
|
|
(target_id, target_queue_id, target_id),
|
|
)
|
|
for query_id, rank in query_ranks.items():
|
|
query = self.conn.execute(
|
|
'''SELECT query_ordinal FROM docker_depth_experiment_queries
|
|
WHERE id = ?''',
|
|
(query_id,),
|
|
).fetchone()
|
|
self.conn.execute(
|
|
'''INSERT INTO docker_depth_experiment_selections(
|
|
experiment_id, query_ordinal, experiment_repository_id,
|
|
experiment_target_id, image_rank
|
|
) VALUES (1, ?, ?, ?, ?)''',
|
|
(query['query_ordinal'], query_id, target_id, rank),
|
|
)
|
|
self.conn.execute(
|
|
'''UPDATE docker_depth_experiments
|
|
SET target_count = target_count + 1,
|
|
selection_count = selection_count + ?
|
|
WHERE id = 1''',
|
|
(len(query_ranks),),
|
|
)
|
|
|
|
def add_scan(
|
|
self, target_id, scan_id, *, binding_id=None, attempt=1,
|
|
duration=1.0, error_count=0, status='clean', target=None,
|
|
binding_state='completed', reservation_state='acknowledged'):
|
|
binding_id = scan_id if binding_id is None else binding_id
|
|
queue_id = 1000 + target_id
|
|
reservation_id = 5000 + binding_id
|
|
queue = self.conn.execute(
|
|
'SELECT * FROM target_queue WHERE id = ?', (queue_id,),
|
|
).fetchone()
|
|
target = str(target or queue['target'])
|
|
normalized_target = target.lower()
|
|
self.conn.execute(
|
|
'''UPDATE target_queue SET target = ?, normalized_target = ?
|
|
WHERE id = ?''',
|
|
(target, normalized_target, queue_id),
|
|
)
|
|
self.conn.execute(
|
|
'''INSERT INTO result_reservations(
|
|
id, queue_id, source, platform, query, target,
|
|
normalized_target, claim_lease_token, scan_event_id, state
|
|
) VALUES (?, ?, 'dockerhub', 'docker', 'fixture', ?, ?, ?, ?, ?)''',
|
|
(
|
|
reservation_id, queue_id, target, normalized_target,
|
|
f'lease-{binding_id}', f'event-{binding_id}', reservation_state,
|
|
),
|
|
)
|
|
self.conn.execute(
|
|
'''INSERT INTO target_scans(
|
|
id, scan_event_id, queue_id, result_reservation_id,
|
|
claim_lease_token, source, query, target, normalized_target,
|
|
scan_type, status, duration_sec, error_count, raw_result_json
|
|
) VALUES (?, ?, ?, ?, ?, 'dockerhub', 'fixture', ?, ?,
|
|
'docker', ?, ?, ?, 'private result')''',
|
|
(
|
|
scan_id, f'event-{binding_id}', queue_id, reservation_id,
|
|
f'lease-{binding_id}', target, normalized_target, status,
|
|
duration, error_count,
|
|
),
|
|
)
|
|
self.conn.execute(
|
|
'''INSERT INTO docker_depth_experiment_scan_bindings(
|
|
id, experiment_target_id, reservation_id, target_scan_id,
|
|
attempt, state
|
|
) VALUES (?, ?, ?, ?, ?, ?)''',
|
|
(
|
|
binding_id, target_id, reservation_id, scan_id, attempt,
|
|
binding_state,
|
|
),
|
|
)
|
|
return binding_id
|
|
|
|
def add_scanless_binding(
|
|
self, target_id, binding_id, *, attempt, binding_state,
|
|
reservation_state):
|
|
queue_id = 1000 + target_id
|
|
reservation_id = 5000 + binding_id
|
|
queue = self.conn.execute(
|
|
'SELECT * FROM target_queue WHERE id = ?', (queue_id,),
|
|
).fetchone()
|
|
self.conn.execute(
|
|
'''INSERT INTO result_reservations(
|
|
id, queue_id, source, platform, query, target,
|
|
normalized_target, claim_lease_token, scan_event_id, state
|
|
) VALUES (?, ?, 'dockerhub', 'docker', 'fixture', ?, ?, ?, ?, ?)''',
|
|
(
|
|
reservation_id, queue_id, queue['target'],
|
|
queue['normalized_target'], f'lease-{binding_id}',
|
|
f'event-{binding_id}', reservation_state,
|
|
),
|
|
)
|
|
self.conn.execute(
|
|
'''INSERT INTO docker_depth_experiment_scan_bindings(
|
|
id, experiment_target_id, reservation_id, target_scan_id,
|
|
attempt, state
|
|
) VALUES (?, ?, ?, NULL, ?, ?)''',
|
|
(binding_id, target_id, reservation_id, attempt, binding_state),
|
|
)
|
|
|
|
def add_finding(
|
|
self, finding_id, scan_id, fingerprint, detector_hash,
|
|
*, secret='private-secret'):
|
|
self.conn.execute(
|
|
'''INSERT INTO findings(
|
|
id, target_scan_id, finding_fingerprint,
|
|
detector_secret_hash, raw_secret, redacted_secret,
|
|
raw_finding_json
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?)''',
|
|
(
|
|
finding_id, scan_id, fingerprint, detector_hash, secret,
|
|
f'redacted-{secret}', json.dumps({'excerpt': secret}),
|
|
),
|
|
)
|
|
|
|
def add_layer(self, layer_id, target_id, base, top, digest='sha256:layer'):
|
|
self.conn.execute(
|
|
'''INSERT INTO docker_manifest_layers(
|
|
id, manifest_id, position_from_base, position_from_top,
|
|
layer_digest
|
|
) VALUES (?, ?, ?, ?, ?)''',
|
|
(layer_id, target_id, base, top, digest),
|
|
)
|
|
|
|
def add_attribution(
|
|
self, attribution_id, binding_id, finding_id, *, state,
|
|
layer_id=None, base=None, top=None, digest=None):
|
|
self.conn.execute(
|
|
'''INSERT INTO docker_finding_layer_attributions(
|
|
id, scan_binding_id, finding_id, manifest_layer_id,
|
|
attribution_state, reported_layer_digest,
|
|
position_from_base, position_from_top
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)''',
|
|
(
|
|
attribution_id, binding_id, finding_id, layer_id, state,
|
|
digest, base, top,
|
|
),
|
|
)
|
|
|
|
def add_credential(self, credential_id, secret='private-credential'):
|
|
self.conn.execute(
|
|
'''INSERT INTO keycheck_credentials(
|
|
id, secret_text, secret_json, key_masked
|
|
) VALUES (?, ?, ?, ?)''',
|
|
(credential_id, secret, json.dumps({'secret': secret}), f'***{secret}'),
|
|
)
|
|
|
|
def add_result(
|
|
self, result_id, credential_id, status, status_group,
|
|
*, candidate_id=None, sensitive='private-result-evidence'):
|
|
self.conn.execute(
|
|
'''INSERT INTO keycheck_results(
|
|
id, candidate_id, credential_id, status, status_group,
|
|
message, source_line, metadata_json
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)''',
|
|
(
|
|
result_id, candidate_id, credential_id, status, status_group,
|
|
sensitive, sensitive, json.dumps({'evidence': sensitive}),
|
|
),
|
|
)
|
|
|
|
def add_candidate(
|
|
self, candidate_id, credential_id, finding_id, scan_id, *,
|
|
state='completed', attempts=1, result_id=None):
|
|
self.conn.execute(
|
|
'''INSERT INTO keycheck_candidates(
|
|
id, credential_id, finding_id, target_scan_id, state,
|
|
attempts, keycheck_result_id, metadata_json
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, 'private candidate evidence')''',
|
|
(
|
|
candidate_id, credential_id, finding_id, scan_id,
|
|
state, attempts, result_id,
|
|
),
|
|
)
|
|
|
|
def report(self, **selector):
|
|
self.conn.commit()
|
|
if not selector:
|
|
selector = {'experiment_id': 1}
|
|
return build_docker_depth_report(self.conn, **selector)
|
|
|
|
def test_shared_attribution_is_separate_from_physical_totals(self):
|
|
self.add_query(1, 0)
|
|
self.add_query(2, 1)
|
|
self.add_target(10, {1: 3, 2: 4})
|
|
self.add_scan(10, 100)
|
|
|
|
report = self.report(experiment_key='depth-fixture')
|
|
|
|
self.assertEqual(report['physical']['totals']['target_count'], 1)
|
|
self.assertEqual(
|
|
report['physical']['rank_buckets']['ranks_1_3']['target_count'], 1,
|
|
)
|
|
self.assertEqual(
|
|
report['physical']['rank_buckets']['ranks_4_10']['target_count'], 0,
|
|
)
|
|
self.assertEqual(report['per_query']['1']['totals']['target_count'], 1)
|
|
self.assertEqual(report['per_query']['2']['totals']['target_count'], 1)
|
|
self.assertEqual(
|
|
report['per_query']['1']['rank_buckets']['ranks_1_3']['target_count'], 1,
|
|
)
|
|
self.assertEqual(
|
|
report['per_query']['2']['rank_buckets']['ranks_4_10']['target_count'], 1,
|
|
)
|
|
self.assertEqual(report['overlap']['targets']['shared_physical_count'], 1)
|
|
self.assertEqual(report['overlap']['targets']['attribution_credit_count'], 2)
|
|
self.assertEqual(report['overlap']['scans']['overlap_credit_count'], 1)
|
|
|
|
def test_zero_member_query_reports_scarcity_without_missing_planned_work(self):
|
|
self.add_query(
|
|
1, 0, required=10, selected=0, add_repository=False,
|
|
)
|
|
|
|
coverage = self.report()['per_query']['1']['coverage']
|
|
|
|
self.assertEqual(coverage['required_repository_count'], 10)
|
|
self.assertEqual(coverage['cohort_repository_count'], 0)
|
|
self.assertEqual(coverage['unavailable_repository_count'], 10)
|
|
self.assertEqual(coverage['repository_count'], 0)
|
|
self.assertEqual(coverage['missing_cohort_repository_count'], 0)
|
|
|
|
def test_image_unavailable_repository_is_reported_without_a_target(self):
|
|
self.add_query(
|
|
1, 0, required=10, selected=1, repository_state='skipped',
|
|
)
|
|
|
|
report = self.report()
|
|
physical = report['physical']['coverage']
|
|
query = report['per_query']['1']['coverage']
|
|
|
|
self.assertEqual(physical['image_unavailable_repository_count'], 1)
|
|
self.assertEqual(
|
|
physical['image_unavailable_repository_membership_count'], 1,
|
|
)
|
|
self.assertEqual(
|
|
physical['queries_with_image_unavailable_repositories'], 1,
|
|
)
|
|
self.assertEqual(query['image_unavailable_repository_count'], 1)
|
|
self.assertEqual(query['target_count'], 0)
|
|
|
|
def test_shared_repository_uses_physical_queue_identity_and_membership_credits(self):
|
|
self.add_query(1, 0, repository_queue_id=9000)
|
|
self.add_query(2, 1, repository_queue_id=9000)
|
|
self.add_target(10, {1: 2, 2: 3})
|
|
|
|
report = self.report()
|
|
coverage = report['physical']['coverage']
|
|
|
|
self.assertEqual(coverage['repository_count'], 1)
|
|
self.assertEqual(coverage['repository_query_membership_credit_count'], 2)
|
|
self.assertEqual(coverage['repository_overlap_credit_count'], 1)
|
|
self.assertEqual(coverage['shared_repository_count'], 1)
|
|
self.assertEqual(coverage['selected_repository_count'], 1)
|
|
self.assertEqual(
|
|
coverage['selected_repository_query_membership_credit_count'], 2,
|
|
)
|
|
self.assertEqual(coverage['selected_repository_overlap_credit_count'], 1)
|
|
self.assertEqual(report['per_query']['1']['coverage']['repository_count'], 1)
|
|
self.assertEqual(report['per_query']['2']['coverage']['repository_count'], 1)
|
|
self.assertEqual(
|
|
report['overlap']['repositories'],
|
|
{
|
|
'physical_count': 1,
|
|
'query_membership_credit_count': 2,
|
|
'overlap_credit_count': 1,
|
|
'shared_physical_count': 1,
|
|
},
|
|
)
|
|
|
|
def test_replacement_repository_is_used_for_physical_overlap(self):
|
|
self.add_query(1, 0, repository_queue_id=9001)
|
|
self.add_query(2, 1, repository_queue_id=9002)
|
|
self.conn.execute(
|
|
'''UPDATE docker_depth_experiment_repositories
|
|
SET replacement_repository_queue_id = 9999 WHERE id IN (1, 2)'''
|
|
)
|
|
self.add_target(10, {1: 1, 2: 1})
|
|
|
|
coverage = self.report()['physical']['coverage']
|
|
|
|
self.assertEqual(coverage['repository_count'], 1)
|
|
self.assertEqual(coverage['shared_repository_count'], 1)
|
|
self.assertEqual(coverage['selected_repository_count'], 1)
|
|
|
|
def test_queries_match_the_current_scanner_schema(self):
|
|
with tempfile.TemporaryDirectory() as directory, mock.patch.dict(
|
|
os.environ, {'SCANNER_DB_URL': '', 'DATABASE_URL': ''}):
|
|
db = ScannerDB(db_path=os.path.join(directory, 'scanner.db'), db_url='')
|
|
try:
|
|
now = utc_now_iso()
|
|
digest = 'a' * 64
|
|
experiment_id = db.conn.execute(
|
|
'''INSERT INTO docker_depth_experiments(
|
|
experiment_key, source, state, config_sha256,
|
|
ordered_queries_sha256, selector_version,
|
|
selector_sha256, provenance_policy_sha256,
|
|
query_count, repositories_per_query,
|
|
images_per_repository, target_limit, created_at,
|
|
updated_at
|
|
) VALUES (
|
|
'schema-smoke', 'dockerhub', 'held', ?, ?,
|
|
'selector-v1', ?, ?, 1, 1, 1, 1, ?, ?
|
|
)''',
|
|
(digest, digest, digest, digest, now, now),
|
|
).lastrowid
|
|
db.conn.commit()
|
|
|
|
report = build_docker_depth_report(
|
|
db.conn, experiment_id=experiment_id,
|
|
)
|
|
|
|
self.assertEqual(report['experiment']['id'], experiment_id)
|
|
self.assertEqual(report['physical']['totals']['target_count'], 0)
|
|
self.assertEqual(report['per_query'], {})
|
|
finally:
|
|
db.close()
|
|
|
|
def test_rank_three_and_four_are_in_different_buckets(self):
|
|
self.add_query(1, 0)
|
|
self.add_target(10, {1: 1})
|
|
self.add_target(20, {1: 3})
|
|
self.add_target(30, {1: 4})
|
|
self.add_target(40, {1: 10})
|
|
|
|
report = self.report()
|
|
|
|
buckets = report['physical']['rank_buckets']
|
|
self.assertEqual(buckets['ranks_1_3']['target_count'], 2)
|
|
self.assertEqual(buckets['ranks_4_10']['target_count'], 2)
|
|
|
|
def test_minimum_rank_yield_uses_rank_three_at_the_bucket_boundary(self):
|
|
self.add_query(1, 0)
|
|
self.add_target(10, {1: 4})
|
|
self.add_target(20, {1: 3})
|
|
self.add_scan(10, 100)
|
|
self.add_scan(20, 200)
|
|
fingerprint = 'f' * 64
|
|
detector_hash = 'd' * 64
|
|
self.add_finding(1000, 100, fingerprint, detector_hash)
|
|
self.add_finding(2000, 200, fingerprint, detector_hash)
|
|
self.add_credential(50)
|
|
self.add_candidate(1, 50, 1000, 100)
|
|
self.add_candidate(2, 50, 2000, 200)
|
|
|
|
marginal = self.report()['physical']['marginal_minimum_rank_yield']
|
|
|
|
for metric in ('findings', 'detector_secret_identities', 'credentials'):
|
|
self.assertEqual(marginal[metric]['deduplicated_total'], 1)
|
|
self.assertEqual(marginal[metric]['ranks_1_3'], 1)
|
|
self.assertEqual(marginal[metric]['ranks_4_10'], 0)
|
|
|
|
def test_duplicate_identities_have_one_minimum_rank_yield(self):
|
|
self.add_query(1, 0)
|
|
self.add_target(10, {1: 2})
|
|
self.add_target(20, {1: 7})
|
|
self.add_scan(10, 100)
|
|
self.add_scan(20, 200)
|
|
fingerprint = 'f' * 64
|
|
detector_hash = 'd' * 64
|
|
self.add_finding(1000, 100, fingerprint, detector_hash)
|
|
self.add_finding(2000, 200, fingerprint, detector_hash)
|
|
self.add_credential(50)
|
|
self.add_candidate(1, 50, 1000, 100, state='pending')
|
|
self.add_candidate(2, 50, 2000, 200, state='pending')
|
|
|
|
report = self.report()
|
|
totals = report['physical']['totals']
|
|
marginal = report['physical']['marginal_minimum_rank_yield']
|
|
|
|
self.assertEqual(totals['findings']['occurrence_count'], 2)
|
|
self.assertEqual(totals['findings']['deduplicated_count'], 1)
|
|
self.assertEqual(totals['credentials']['deduplicated_count'], 1)
|
|
self.assertEqual(marginal['findings']['ranks_1_3'], 1)
|
|
self.assertEqual(marginal['findings']['ranks_4_10'], 0)
|
|
self.assertEqual(marginal['detector_secret_identities']['ranks_1_3'], 1)
|
|
self.assertEqual(marginal['credentials']['ranks_1_3'], 1)
|
|
self.assertEqual(marginal['credentials']['ranks_4_10'], 0)
|
|
|
|
def test_pending_frozen_and_current_verification_are_separate(self):
|
|
self.add_query(1, 0)
|
|
self.add_target(10, {1: 1})
|
|
self.add_scan(10, 100)
|
|
self.add_finding(1000, 100, 'a' * 64, 'b' * 64)
|
|
self.add_finding(1001, 100, 'c' * 64, 'd' * 64)
|
|
self.add_credential(50)
|
|
self.add_credential(60)
|
|
self.add_result(500, 50, 'VALID', 'alive', candidate_id=5)
|
|
self.add_result(501, 50, 'DEAD', 'dead')
|
|
self.add_candidate(5, 50, 1000, 100, result_id=500)
|
|
self.add_candidate(6, 60, 1001, 100, state='pending', attempts=0)
|
|
self.conn.execute(
|
|
'''INSERT INTO keycheck_current_state(
|
|
credential_id, status, status_group, last_result_id,
|
|
metadata_json
|
|
) VALUES (50, 'DEAD', 'dead', 501, 'private current evidence')'''
|
|
)
|
|
|
|
keychecks = self.report()['physical']['totals']['keychecks']
|
|
|
|
self.assertEqual(keychecks['pending_candidate_count'], 1)
|
|
self.assertEqual(keychecks['missing_frozen_result_candidate_count'], 1)
|
|
self.assertEqual(keychecks['frozen']['credential_count'], 1)
|
|
self.assertEqual(keychecks['frozen']['missing_credential_count'], 1)
|
|
self.assertEqual(keychecks['frozen']['status_counts'], {'VALID': 1})
|
|
self.assertEqual(keychecks['current']['credential_count'], 1)
|
|
self.assertEqual(keychecks['current']['missing_credential_count'], 1)
|
|
self.assertEqual(keychecks['current']['missing_backing_result_count'], 0)
|
|
self.assertEqual(keychecks['current']['status_counts'], {'DEAD': 1})
|
|
|
|
def test_keycheck_outcomes_require_candidate_and_credential_identity(self):
|
|
self.add_query(1, 0)
|
|
self.add_target(10, {1: 1})
|
|
self.add_scan(10, 100)
|
|
self.add_finding(1000, 100, 'a' * 64, 'b' * 64)
|
|
self.add_credential(50)
|
|
self.add_credential(60)
|
|
self.add_result(500, 60, 'VALID', 'alive', candidate_id=5)
|
|
self.add_result(501, 60, 'DEAD', 'dead', candidate_id=6)
|
|
self.add_candidate(5, 50, 1000, 100, result_id=500)
|
|
self.conn.execute(
|
|
'''INSERT INTO keycheck_current_state(
|
|
credential_id, status, status_group, last_result_id,
|
|
metadata_json
|
|
) VALUES (50, 'DEAD', 'dead', 501, 'private current evidence')'''
|
|
)
|
|
|
|
keychecks = self.report()['physical']['totals']['keychecks']
|
|
|
|
self.assertEqual(keychecks['missing_frozen_result_candidate_count'], 1)
|
|
self.assertEqual(keychecks['frozen']['result_count'], 0)
|
|
self.assertEqual(keychecks['frozen']['credential_count'], 0)
|
|
self.assertEqual(keychecks['current']['credential_count'], 1)
|
|
self.assertEqual(keychecks['current']['missing_backing_result_count'], 1)
|
|
self.assertEqual(
|
|
keychecks['current']['status_counts'], {'invalid_or_unknown': 1},
|
|
)
|
|
|
|
def test_duplicate_layer_digest_positions_remain_exact_rows(self):
|
|
self.add_query(1, 0)
|
|
self.add_target(10, {1: 1}, layer_count=3)
|
|
binding_id = self.add_scan(10, 100)
|
|
self.add_finding(1000, 100, 'a' * 64, 'b' * 64)
|
|
digest = 'sha256:duplicate'
|
|
self.add_layer(1, 10, 1, 3, digest)
|
|
self.add_layer(2, 10, 3, 1, digest)
|
|
self.add_attribution(
|
|
1, binding_id, 1000, state='exact', layer_id=1,
|
|
base=1, top=3, digest=digest,
|
|
)
|
|
self.add_attribution(
|
|
2, binding_id, 1000, state='exact', layer_id=2,
|
|
base=3, top=1, digest=digest,
|
|
)
|
|
|
|
totals = self.report()['physical']['totals']
|
|
layers = totals['layers']
|
|
|
|
self.assertEqual(totals['declared_manifest_layer_count'], 3)
|
|
self.assertEqual(layers['exact_attribution_count'], 2)
|
|
self.assertEqual(layers['exact_manifest_layer_count'], 2)
|
|
self.assertEqual(layers['positions_from_base'], {'1': 1, '3': 1})
|
|
self.assertEqual(layers['positions_from_top'], {'1': 1, '3': 1})
|
|
self.assertEqual(layers['unattributed_finding_occurrence_count'], 0)
|
|
|
|
def test_stale_or_cross_queue_bindings_cannot_admit_evidence(self):
|
|
self.add_query(1, 0)
|
|
for target_id, rank in ((10, 1), (20, 2), (30, 3), (40, 4)):
|
|
self.add_target(target_id, {1: rank})
|
|
self.add_scan(target_id, target_id * 10)
|
|
self.add_finding(
|
|
target_id * 100, target_id * 10,
|
|
f'{target_id:064x}', f'{target_id + 1:064x}',
|
|
)
|
|
self.conn.execute(
|
|
'UPDATE result_reservations SET queue_id = 999999 WHERE id = 5200'
|
|
)
|
|
self.conn.execute('UPDATE target_scans SET queue_id = 999998 WHERE id = 300')
|
|
self.conn.execute(
|
|
'UPDATE target_scans SET result_reservation_id = 5100 WHERE id = 400'
|
|
)
|
|
self.add_credential(50)
|
|
self.add_candidate(1, 50, 2000, 200)
|
|
self.conn.execute(
|
|
'''INSERT INTO errors(id, target_scan_id, category, summary, raw_error)
|
|
VALUES (1, 100, 'timeout', 'private', 'private'),
|
|
(2, 200, 'network', 'private', 'private')'''
|
|
)
|
|
|
|
totals = self.report()['physical']['totals']
|
|
|
|
self.assertEqual(totals['target_count'], 4)
|
|
self.assertEqual(totals['targets_with_bindings'], 3)
|
|
self.assertEqual(totals['targets_with_scans'], 1)
|
|
self.assertEqual(totals['scan_bindings']['attempt_count'], 3)
|
|
self.assertEqual(totals['scans']['count'], 1)
|
|
self.assertEqual(totals['findings']['occurrence_count'], 1)
|
|
self.assertEqual(totals['keychecks']['candidate_count'], 0)
|
|
self.assertEqual(totals['scans']['errors']['row_count'], 1)
|
|
self.assertEqual(
|
|
totals['scans']['errors']['category_counts'], {'timeout': 1},
|
|
)
|
|
|
|
def test_retries_duration_errors_and_coverage_are_aggregated_once(self):
|
|
self.add_query(1, 0, attempts=3)
|
|
self.add_target(10, {1: 1})
|
|
self.add_scan(10, 100, attempt=1, duration=2.0, error_count=1)
|
|
self.add_scan(10, 200, attempt=2, duration=3.0)
|
|
self.conn.execute(
|
|
'''INSERT INTO errors(id, target_scan_id, summary, raw_error)
|
|
VALUES (1, 100, 'private summary', 'private raw error')'''
|
|
)
|
|
|
|
report = self.report()
|
|
totals = report['physical']['totals']
|
|
coverage = report['per_query']['1']['coverage']
|
|
|
|
self.assertEqual(totals['scan_bindings']['attempt_count'], 2)
|
|
self.assertEqual(totals['scan_bindings']['retry_count'], 1)
|
|
self.assertEqual(totals['scan_bindings']['maximum_attempt'], 2)
|
|
self.assertEqual(totals['scans']['count'], 2)
|
|
self.assertEqual(totals['scans']['duration']['total_seconds'], 5.0)
|
|
self.assertEqual(totals['scans']['errors']['row_count'], 1)
|
|
self.assertEqual(totals['scans']['errors']['reported_count'], 1)
|
|
self.assertEqual(totals['scans']['errors']['scan_count'], 1)
|
|
self.assertEqual(coverage['repository_count'], 1)
|
|
self.assertEqual(coverage['selected_repository_count'], 1)
|
|
self.assertEqual(coverage['resolver_attempt_count'], 3)
|
|
self.assertEqual(coverage['resolver_retry_count'], 2)
|
|
|
|
def test_scanless_refund_and_retry_attempts_remain_visible(self):
|
|
self.add_query(1, 0)
|
|
self.add_target(10, {1: 1})
|
|
self.add_scanless_binding(
|
|
10, 100, attempt=1, binding_state='released',
|
|
reservation_state='refunded',
|
|
)
|
|
self.add_scanless_binding(
|
|
10, 200, attempt=2, binding_state='scanning',
|
|
reservation_state='ready',
|
|
)
|
|
self.conn.execute(
|
|
"UPDATE docker_depth_experiment_targets SET state = 'scanning' WHERE id = 10"
|
|
)
|
|
|
|
totals = self.report()['physical']['totals']
|
|
|
|
self.assertEqual(totals['targets_with_bindings'], 1)
|
|
self.assertEqual(totals['targets_with_scans'], 0)
|
|
self.assertEqual(totals['scan_bindings']['attempt_count'], 2)
|
|
self.assertEqual(totals['scan_bindings']['retry_count'], 1)
|
|
self.assertEqual(totals['scan_bindings']['refunded_attempt_count'], 1)
|
|
self.assertEqual(totals['scan_bindings']['maximum_attempt'], 2)
|
|
self.assertEqual(
|
|
totals['scan_bindings']['state_counts'],
|
|
{'released': 1, 'scanning': 1},
|
|
)
|
|
self.assertEqual(
|
|
totals['scan_bindings']['reservation_state_counts'],
|
|
{'ready': 1, 'refunded': 1},
|
|
)
|
|
|
|
def test_canonical_unknown_and_foundry_statuses_are_not_collapsed(self):
|
|
self.add_query(1, 0)
|
|
self.add_target(10, {1: 1})
|
|
self.add_scan(10, 100)
|
|
self.add_finding(1000, 100, 'a' * 64, 'b' * 64)
|
|
self.add_finding(1001, 100, 'c' * 64, 'd' * 64)
|
|
self.add_credential(50)
|
|
self.add_credential(60)
|
|
self.add_result(500, 50, 'UNKNOWN', 'unknown', candidate_id=5)
|
|
self.add_result(
|
|
600, 60, 'FOUNDRY_BAD_ENDPOINT', 'unknown', candidate_id=6,
|
|
)
|
|
self.add_candidate(5, 50, 1000, 100, result_id=500)
|
|
self.add_candidate(6, 60, 1001, 100, result_id=600)
|
|
self.conn.execute(
|
|
'''INSERT INTO keycheck_current_state(
|
|
credential_id, status, status_group, last_result_id,
|
|
metadata_json
|
|
) VALUES (50, 'UNKNOWN', 'unknown', 500, '{}'),
|
|
(60, 'FOUNDRY_BAD_ENDPOINT', 'unknown', 600, '{}')'''
|
|
)
|
|
self.conn.execute(
|
|
"INSERT INTO errors(id, target_scan_id, category) VALUES (1, 100, 'unknown')"
|
|
)
|
|
|
|
totals = self.report()['physical']['totals']
|
|
|
|
self.assertEqual(
|
|
totals['keychecks']['frozen']['status_counts'],
|
|
{'FOUNDRY_BAD_ENDPOINT': 1, 'UNKNOWN': 1},
|
|
)
|
|
self.assertEqual(
|
|
totals['keychecks']['current']['status_group_counts'], {'unknown': 2},
|
|
)
|
|
self.assertEqual(
|
|
totals['scans']['errors']['category_counts'], {'unknown': 1},
|
|
)
|
|
|
|
def test_unattributed_findings_and_secret_sentinels_never_leak(self):
|
|
sentinel = 'SECRET_SENTINEL_DO_NOT_LEAK'
|
|
self.add_query(1, 0, query=f'query-{sentinel}')
|
|
self.add_target(10, {1: 1}, repository=f'repository-{sentinel}')
|
|
binding_id = self.add_scan(10, 100, target=f'target-{sentinel}')
|
|
self.add_finding(1000, 100, 'a' * 64, 'b' * 64, secret=sentinel)
|
|
self.add_attribution(
|
|
1, binding_id, 1000, state='unattributed',
|
|
)
|
|
self.add_credential(50, secret=sentinel)
|
|
self.add_result(
|
|
500, 50, 'VALID', 'alive', candidate_id=5, sensitive=sentinel,
|
|
)
|
|
self.add_candidate(5, 50, 1000, 100, result_id=500)
|
|
self.conn.execute(
|
|
'INSERT INTO errors(id, target_scan_id, summary, raw_error) VALUES (1, 100, ?, ?)',
|
|
(sentinel, sentinel),
|
|
)
|
|
|
|
report = self.report()
|
|
serialized = json.dumps(report, sort_keys=True)
|
|
|
|
self.assertNotIn(sentinel, serialized)
|
|
layers = report['physical']['totals']['layers']
|
|
self.assertEqual(layers['unattributed_attribution_count'], 1)
|
|
self.assertEqual(layers['unattributed_finding_occurrence_count'], 1)
|
|
self.assertEqual(layers['unattributed_deduplicated_finding_count'], 1)
|
|
|
|
def test_untrusted_enum_and_error_sentinel_is_collapsed(self):
|
|
sentinel = 'SECRET_SENTINEL_DO_NOT_LEAK'
|
|
self.add_query(1, 0)
|
|
self.add_target(10, {1: 1})
|
|
binding_id = self.add_scan(10, 100)
|
|
self.add_finding(1000, 100, 'a' * 64, 'b' * 64)
|
|
self.add_attribution(1, binding_id, 1000, state=sentinel)
|
|
self.add_credential(50)
|
|
self.add_result(500, 50, sentinel, sentinel, candidate_id=5)
|
|
self.add_candidate(5, 50, 1000, 100, state=sentinel, result_id=500)
|
|
self.conn.execute(
|
|
'''INSERT INTO keycheck_current_state(
|
|
credential_id, status, status_group, last_result_id,
|
|
metadata_json
|
|
) VALUES (50, ?, ?, 500, 'private current evidence')''',
|
|
(sentinel, sentinel),
|
|
)
|
|
self.conn.execute(
|
|
'''INSERT INTO errors(id, target_scan_id, category, summary, raw_error)
|
|
VALUES (1, 100, ?, 'private', 'private')''',
|
|
(sentinel,),
|
|
)
|
|
self.conn.execute(
|
|
'''UPDATE docker_depth_experiments
|
|
SET experiment_key = ?, source = ?, state = ? WHERE id = 1''',
|
|
(sentinel, sentinel, sentinel),
|
|
)
|
|
self.conn.execute(
|
|
'''UPDATE docker_depth_experiment_queries
|
|
SET source = ?, query_sha256 = ? WHERE id = 1''',
|
|
(sentinel, sentinel),
|
|
)
|
|
self.conn.execute(
|
|
'''UPDATE docker_depth_experiment_repositories
|
|
SET source = ?, work_state = ? WHERE id = 1''',
|
|
(sentinel, sentinel),
|
|
)
|
|
self.conn.execute(
|
|
'UPDATE docker_depth_experiment_targets SET state = ? WHERE id = 10',
|
|
(sentinel,),
|
|
)
|
|
self.conn.execute(
|
|
'''UPDATE docker_depth_experiment_scan_bindings
|
|
SET state = ? WHERE id = ?''',
|
|
(sentinel, binding_id),
|
|
)
|
|
self.conn.execute(
|
|
'UPDATE target_scans SET status = ? WHERE id = 100',
|
|
(sentinel,),
|
|
)
|
|
|
|
report = self.report()
|
|
totals = report['physical']['totals']
|
|
|
|
self.assertNotIn(sentinel, json.dumps(report, sort_keys=True))
|
|
self.assertEqual(report['experiment']['key'], 'invalid_or_unknown')
|
|
self.assertEqual(report['experiment']['source'], 'invalid_or_unknown')
|
|
self.assertEqual(report['experiment']['state'], 'invalid_or_unknown')
|
|
self.assertEqual(report['per_query']['1']['query']['source'], 'invalid_or_unknown')
|
|
self.assertEqual(report['per_query']['1']['query']['sha256'], 'invalid_or_unknown')
|
|
self.assertEqual(totals['target_state_counts'], {'invalid_or_unknown': 1})
|
|
self.assertEqual(
|
|
totals['scan_bindings']['state_counts'], {'invalid_or_unknown': 1},
|
|
)
|
|
self.assertEqual(totals['scans']['status_counts'], {'invalid_or_unknown': 1})
|
|
self.assertEqual(
|
|
totals['scans']['errors']['category_counts'],
|
|
{'invalid_or_unknown': 1},
|
|
)
|
|
self.assertEqual(
|
|
totals['keychecks']['candidate_state_counts'],
|
|
{'invalid_or_unknown': 1},
|
|
)
|
|
self.assertEqual(
|
|
totals['keychecks']['frozen']['status_counts'],
|
|
{'invalid_or_unknown': 1},
|
|
)
|
|
self.assertEqual(
|
|
totals['keychecks']['current']['status_group_counts'],
|
|
{'invalid_or_unknown': 1},
|
|
)
|
|
self.assertEqual(totals['layers']['invalid_or_unknown_attribution_count'], 1)
|
|
|
|
def test_sqlite_report_uses_one_snapshot_during_concurrent_commit(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = os.path.join(directory, 'report.db')
|
|
reader = sqlite3.connect(path)
|
|
reader.row_factory = sqlite3.Row
|
|
reader.execute('PRAGMA journal_mode=WAL')
|
|
reader.executescript(SCHEMA)
|
|
reader.execute(
|
|
'''INSERT INTO docker_depth_experiments(
|
|
id, experiment_key, source, state, query_count,
|
|
target_count, selection_count
|
|
) VALUES (1, 'snapshot', 'dockerhub', 'completed', 0, 0, 0)'''
|
|
)
|
|
reader.commit()
|
|
writer = sqlite3.connect(path)
|
|
writer.row_factory = sqlite3.Row
|
|
|
|
class ConcurrentConnection:
|
|
def __init__(self):
|
|
self.changed = False
|
|
|
|
@property
|
|
def in_transaction(self):
|
|
return reader.in_transaction
|
|
|
|
def execute(self, sql, params=None):
|
|
cursor = reader.execute(sql, tuple(params or ()))
|
|
if (
|
|
not self.changed
|
|
and 'FROM docker_depth_experiment_queries' in sql
|
|
):
|
|
writer.execute(
|
|
'UPDATE docker_depth_experiments SET target_count = 99 WHERE id = 1'
|
|
)
|
|
writer.commit()
|
|
self.changed = True
|
|
return cursor
|
|
|
|
def rollback(self):
|
|
return reader.rollback()
|
|
|
|
try:
|
|
report = build_docker_depth_report(
|
|
ConcurrentConnection(), experiment_id=1,
|
|
)
|
|
self.assertEqual(
|
|
report['experiment']['persisted_counts']['targets'], 0,
|
|
)
|
|
self.assertEqual(
|
|
writer.execute(
|
|
'SELECT target_count FROM docker_depth_experiments WHERE id = 1'
|
|
).fetchone()['target_count'],
|
|
99,
|
|
)
|
|
self.assertFalse(reader.in_transaction)
|
|
self.assertEqual(
|
|
report['snapshot']['consistency'], 'sqlite_transaction',
|
|
)
|
|
finally:
|
|
writer.close()
|
|
reader.close()
|
|
|
|
def test_postgres_report_transaction_is_read_only_and_rolled_back(self):
|
|
class Connection:
|
|
is_postgres = True
|
|
|
|
def __init__(self):
|
|
self.statements = []
|
|
self.rollbacks = 0
|
|
|
|
def execute(self, sql, params=None):
|
|
self.statements.append(str(sql))
|
|
return object()
|
|
|
|
def rollback(self):
|
|
self.rollbacks += 1
|
|
|
|
connection = Connection()
|
|
fixture = {
|
|
'experiment': {'state': 'completed'},
|
|
'physical': {}, 'per_query': {}, 'overlap': {},
|
|
}
|
|
with mock.patch.object(
|
|
docker_depth_report, '_build_docker_depth_report_snapshot',
|
|
return_value=fixture,
|
|
):
|
|
report = build_docker_depth_report(connection, experiment_id=1)
|
|
self.assertEqual(
|
|
connection.statements,
|
|
['BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY'],
|
|
)
|
|
self.assertEqual(connection.rollbacks, 1)
|
|
self.assertEqual(report['snapshot']['consistency'], 'repeatable_read')
|
|
|
|
def test_raw_postgres_connection_is_rejected(self):
|
|
class Connection:
|
|
def execute(self, _sql, _params=None):
|
|
raise AssertionError('raw connection must be rejected before SQL')
|
|
|
|
Connection.__module__ = 'psycopg'
|
|
with self.assertRaisesRegex(ValueError, 'DatabaseConnection'):
|
|
build_docker_depth_report(Connection(), experiment_id=1)
|
|
|
|
def test_finding_position_fanout_is_aggregated_before_materialization(self):
|
|
source = (APP_DIR / 'docker_depth_report.py').read_text(encoding='utf-8')
|
|
self.assertIn('cursor.fetchmany(512)', source)
|
|
self.assertIn('COUNT(a.id) AS attribution_count', source)
|
|
self.assertIn(
|
|
'GROUP BY rb.target_id, ml.id, ml.position_from_base,', source,
|
|
)
|
|
self.assertNotIn('ORDER BY rb.target_id, f.id, a.id', source)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|