"""Verify saved public CLI probes against read-only SQLite rows and Decimal totals.""" import hashlib import json import sqlite3 from collections import defaultdict from datetime import datetime, timedelta, timezone from decimal import Decimal from pathlib import Path ROOT = Path(__file__).resolve().parents[1] EVIDENCE = ROOT / 'evidence/chapter-07' DATABASE = ROOT / 'data/Chinook_Sqlite.sqlite' EXPECTED_HASH = 'bdf635be69850bd3be09c9a2dbeef7ddfb80036bd3ef3381383cd03b61e4a61a' ZERO = Decimal('0') def payload(name): value = json.loads((EVIDENCE / (name + '.json')).read_text(encoding='utf-8-sig')) assert value['success'], name return value def rows(name): value = payload(name)['data'] assert not value['truncated'], name assert value['dataSource'] == 'chinook-sqlite', name return value['rows'] def money(value): return Decimal(str(value)).quantize(Decimal('0.01')) def main(): source = ROOT.parent / '2026-10-13/open-data/chinook-v1.4.5/Chinook_Sqlite.sqlite' for path in [source, DATABASE]: assert hashlib.sha256(path.read_bytes()).hexdigest() == EXPECTED_HASH, path with sqlite3.connect(DATABASE.as_uri() + '?mode=ro&immutable=1', uri=True) as db: db.row_factory = sqlite3.Row data = {table: [dict(row) for row in db.execute('SELECT * FROM ' + table)] for table in ['Customer', 'Invoice', 'InvoiceLine', 'Track', 'Genre']} for table in data: metadata = payload('inspect-' + table)['data'] actual_columns = [row['name'] for row in db.execute('PRAGMA table_info(' + table + ')')] assert [column['name'] for column in metadata['columns']] == actual_columns, table actual_keys = {(row['from'], row['table'], row['to']) for row in db.execute('PRAGMA foreign_key_list(' + table + ')')} described_keys = {(item['column'], item['referencedTable'], item['referencedColumn']) for item in metadata['foreignKeys']} assert described_keys == actual_keys, table actual_pk = [row['name'] for row in sorted(db.execute('PRAGMA table_info(' + table + ')'), key=lambda row: row['pk']) if row['pk']] assert metadata['primaryKey']['columns'] == actual_pk, table invoice_by_id = {row['InvoiceId']: row for row in data['Invoice']} customer_by_id = {row['CustomerId']: row for row in data['Customer']} track_by_id = {row['TrackId']: row for row in data['Track']} genre_ids = {row['GenreId'] for row in data['Genre']} assert all(row['CustomerId'] in customer_by_id for row in data['Invoice']) assert all(row['InvoiceId'] in invoice_by_id and row['TrackId'] in track_by_id for row in data['InvoiceLine']) assert all(row['GenreId'] in genre_ids for row in data['Track']) for result in rows('04-core-relations'): assert result['null_keys'] == result['missing_targets'] == 0 lines_by_invoice = defaultdict(list) amounts_by_invoice = defaultdict(lambda: ZERO) for line in data['InvoiceLine']: lines_by_invoice[line['InvoiceId']].append(line) amounts_by_invoice[line['InvoiceId']] += Decimal(str(line['UnitPrice'])) * line['Quantity'] for invoice in data['Invoice']: assert lines_by_invoice[invoice['InvoiceId']], invoice['InvoiceId'] assert Decimal(str(invoice['Total'])) == amounts_by_invoice[invoice['InvoiceId']], invoice['InvoiceId'] total = sum((Decimal(str(row['Total'])) for row in data['Invoice']), ZERO) overview = rows('01-overview')[0] for table, field in [('Customer', 'customer_count'), ('Invoice', 'invoice_count'), ('InvoiceLine', 'invoice_line_count'), ('Track', 'track_count'), ('Genre', 'genre_count')]: assert overview[field] == len(data[table]), field dates = [row['InvoiceDate'] for row in data['Invoice']] assert overview['first_invoice_date'] == min(dates) and overview['last_invoice_date'] == max(dates) assert money(overview['invoice_amount']) == total reconciliation = rows('03-invoice-reconciliation')[0] assert reconciliation['checked_invoices'] == len(data['Invoice']) assert reconciliation['invoices_without_lines'] == reconciliation['mismatched_invoices'] == 0 assert money(reconciliation['invoice_amount']) == money(reconciliation['line_amount']) == total monthly = {} for invoice in data['Invoice']: datetime.fromisoformat(invoice['InvoiceDate']) if not ('2012-01-01' <= invoice['InvoiceDate'] < '2013-01-01'): continue month = invoice['InvoiceDate'][:7] item = monthly.setdefault(month, {'month': month, 'invoice_count': 0, 'line_count': 0, 'quantity': 0, 'customers': set(), 'amount': ZERO}) item['invoice_count'] += 1 item['customers'].add(invoice['CustomerId']) item['amount'] += Decimal(str(invoice['Total'])) item['line_count'] += len(lines_by_invoice[invoice['InvoiceId']]) item['quantity'] += sum(line['Quantity'] for line in lines_by_invoice[invoice['InvoiceId']]) heads = rows('06-monthly-invoices-2012') details = rows('07-monthly-lines-2012') assert [row['month'] for row in heads] == [row['month'] for row in details] == sorted(monthly) for head, detail in zip(heads, details): expected = monthly[head['month']] assert head['invoice_count'] == expected['invoice_count'] assert head['billed_customer_count'] == len(expected['customers']) assert detail['line_count'] == expected['line_count'] and detail['quantity'] == expected['quantity'] assert money(head['invoice_amount']) == money(detail['line_amount']) == expected['amount'] annual = sum((item['amount'] for item in monthly.values()), ZERO) annual_invoices = sum(item['invoice_count'] for item in monthly.values()) annual_lines = sum(item['line_count'] for item in monthly.values()) repeated = sum((Decimal(str(row['Total'])) * len(lines_by_invoice[row['InvoiceId']]) for row in data['Invoice'] if '2012-01-01' <= row['InvoiceDate'] < '2013-01-01'), ZERO) grain = rows('08-join-grain-2012')[0] assert grain['invoice_count'] == annual_invoices and grain['joined_rows'] == annual_lines assert money(grain['correct_invoice_amount']) == money(grain['correct_line_amount']) == annual assert money(grain['repeated_invoice_amount']) == repeated for row in rows('02-invoice-sample'): invoice = invoice_by_id[row['invoice_id']] assert row['invoice_date'] == invoice['InvoiceDate'] and row['customer_id'] == invoice['CustomerId'] assert row['line_count'] == len(lines_by_invoice[row['invoice_id']]) assert money(row['invoice_amount']) == money(row['line_amount']) == Decimal(str(invoice['Total'])) for row, raw in zip(rows('09-line-sample'), sorted(lines_by_invoice[252], key=lambda line: line['InvoiceLineId'])): assert row['line_id'] == raw['InvoiceLineId'] and row['quantity'] == raw['Quantity'] assert money(row['unit_price']) == Decimal(str(raw['UnitPrice'])) assert rows('10-genre-values') == [{'genre_id': row['GenreId'], 'genre_name': row['Name']} for row in sorted(data['Genre'], key=lambda row: row['GenreId'])] shapes = rows('05-data-shapes')[0] assert shapes['stored_date_types'] == 'text' and all(isinstance(value, str) for value in dates) assert shapes['non_midnight_invoice_dates'] == sum(datetime.fromisoformat(value).time().isoformat() != '00:00:00' for value in dates) == 0 assert shapes['minimum_quantity'] == min(row['Quantity'] for row in data['InvoiceLine']) == 1 assert shapes['maximum_quantity'] == max(row['Quantity'] for row in data['InvoiceLine']) == 1 assert shapes['country_differences'] == sum(row['BillingCountry'] != customer_by_id[row['CustomerId']]['Country'] for row in data['Invoice']) == 0 assert shapes['line_and_catalog_price_differences'] == sum(line['UnitPrice'] != track_by_id[line['TrackId']]['UnitPrice'] for line in data['InvoiceLine']) == 0 for field in ['unparseable_invoice_dates', 'missing_billing_countries', 'null_or_negative_invoice_amounts', 'invalid_line_values', 'invoice_amounts_beyond_two_decimals', 'line_prices_beyond_two_decimals']: assert shapes[field] == 0, field baseline = { 'dataSha256': EXPECTED_HASH, 'tableCounts': {table: len(value) for table, value in data.items()}, 'dateRange': [min(dates), max(dates)], 'allDataAmount': format(total, '.2f'), 'year2012': {'invoiceCount': annual_invoices, 'lineCount': annual_lines, 'amount': format(annual, '.2f'), 'incorrectJoinedInvoiceAmount': format(repeated, '.2f')}, 'monthly2012': [{'month': value['month'], 'invoiceCount': value['invoice_count'], 'lineCount': value['line_count'], 'quantity': value['quantity'], 'billedCustomerCount': len(value['customers']), 'amount': format(value['amount'], '.2f')} for value in sorted(monthly.values(), key=lambda item: item['month'])], 'amountUnit': 'Original sample unit; no conversion or currency inference', } report = { 'status': 'passed', 'checkedAt': datetime.now(timezone(timedelta(hours=8))).isoformat(), 'runtimeUrl': 'http://127.0.0.1:18067', 'namespace': 'chinook', 'datasource': 'chinook-sqlite', 'engine': payload('capabilities')['engine'], 'runtimeApiVersion': payload('capabilities')['runtimeApiVersion'], 'schemaVersion': payload('capabilities')['data']['schemaVersion'], 'securityMode': payload('capabilities')['data']['securityMode'], 'runtimeWarnings': payload('capabilities')['data'].get('warnings', []), 'coreTablesInspected': 5, 'readOnlySqlProbes': 10, 'perInvoiceAmountCheck': {'invoiceCount': len(data['Invoice']), 'mismatches': 0}, 'monthlyChecks2012': len(monthly), 'sourceAndCopyMatchPinnedFixture': True, 'independentCheck': 'Read-only sqlite3 raw rows, Python mappings and Decimal totals; no SQL aggregation copied into the independent calculation', 'scope': 'Physical data exploration and baseline only; no TM/QM or semantic query validation', } for filename, value in [('baseline.json', baseline), ('acceptance.json', report)]: (EVIDENCE / filename).write_text(json.dumps(value, ensure_ascii=False, indent=2) + '\n', encoding='utf-8') print(json.dumps({'acceptance': report, 'baseline': baseline}, ensure_ascii=False, indent=2)) if __name__ == '__main__': main()