"""Check aggregate counts, averages and share denominators against original rows.""" import hashlib import json import sqlite3 from collections import defaultdict from datetime import datetime from decimal import Decimal from pathlib import Path LAB = Path(__file__).resolve().parents[1] EVIDENCE = LAB / 'evidence/chapter-10' ZERO = Decimal('0') CENT = Decimal('0.01') TOLERANCE = Decimal('0.000000000001') HASH = 'bdf635be69850bd3be09c9a2dbeef7ddfb80036bd3ef3381383cd03b61e4a61a' def load(path): return json.loads(Path(path).read_text(encoding='utf-8-sig')) def near(actual, expected): assert abs(Decimal(str(actual)) - expected) < TOLERANCE, (actual, expected) def main(): manifest = load(EVIDENCE / 'query-manifest.json') cases = {item['name']: item for item in manifest['queries']} assert len(cases) == 11 def data(name): assert load(cases[name]['validation'])['success'] is True response = load(cases[name]['response']) assert response['success'] is True if cases[name].get('complete', True): assert not response['data'].get('hasNext') assert not response['data'].get('pagination', {}).get('hasMore') return response['data'] database = LAB / 'data/Chinook_Sqlite.sqlite' original = LAB.parent / '2026-10-13/open-data/chinook-v1.4.5/Chinook_Sqlite.sqlite' for path in [original, database]: assert hashlib.sha256(path.read_bytes()).hexdigest() == HASH evolution = load(EVIDENCE / 'model-evolution.json') assert evolution['before'] == load(LAB / 'evidence/chapter-08/acceptance.json')['modelSha256'] model_hashes = evolution['after'] for path, digest in model_hashes.items(): assert hashlib.sha256((LAB / path).read_bytes()).hexdigest() == digest if path != 'models\\query\\ChinookInvoiceQueryModel.qm': assert digest == evolution['before'][path] validation = load(evolution['validationResponse']) assert validation['success'] and validation['data']['validFiles'] == 4 field = load(evolution['describeResponse'])['data']['data']['fields']['averageAmount'] assert field['predefined'] is True and field['calculated'] is True assert field['usage'] == 'predefined_formula' with sqlite3.connect(database.as_uri() + '?mode=ro&immutable=1', uri=True) as db: db.row_factory = sqlite3.Row invoices = [dict(row) for row in db.execute('SELECT * FROM Invoice')] all_customers = {row[0] for row in db.execute('SELECT CustomerId FROM Customer')} year = [row for row in invoices if '2012-01-01' <= row['InvoiceDate'] < '2013-01-01'] buyers = {row['CustomerId'] for row in year} total = sum((Decimal(str(row['Total'])) for row in year), ZERO) average = total / len(year) totals = data('01-totals-and-counts')['items'][0] near(totals['totalAmount'], total) assert totals['billCount'] == len(year) == 83 assert totals['buyerCount'] == len(buyers) == 47 assert buyers <= all_customers and len(all_customers) == 59 actual_average = data('02-average')['items'][0] assert actual_average['billCount'] == len(year) near(actual_average['totalAmount'], total) near(actual_average['temporaryAverage'], average) native = data('08-native-aggregates')['items'][0] near(native['directAverage'], average) near(native['lowestAmount'], min(Decimal(str(row['Total'])) for row in year)) near(native['highestAmount'], max(Decimal(str(row['Total'])) for row in year)) country_amounts = defaultdict(lambda: ZERO) country_counts = defaultdict(int) for row in year: country_amounts[row['BillingCountry']] += Decimal(str(row['Total'])) country_counts[row['BillingCountry']] += 1 averages = data('04-country-average')['items'] assert [row['billingCountry'] for row in averages] == sorted(country_amounts) for row in averages: key = row['billingCountry'] near(row['totalAmount'], country_amounts[key]) assert row['billCount'] == country_counts[key] near(row['temporaryAverage'], country_amounts[key] / country_counts[key]) simple_country_average = sum((country_amounts[k] / country_counts[k] for k in country_amounts), ZERO) / len(country_amounts) assert abs(simple_country_average - average) > Decimal('0.01') def check_shares(name, amounts, selected=None): rows = data(name)['items'] denominator = sum(amounts.values(), ZERO) order = sorted(amounts, key=lambda key: (-amounts[key], key)) if selected: order = order[:selected] assert [row['billingCountry'] for row in rows] == order for row in rows: key = row['billingCountry'] near(row['totalAmount'], amounts[key]) near(row['amountShare'], amounts[key] / denominator) if selected is None: assert abs(sum((Decimal(str(row['amountShare'])) for row in rows), ZERO) - 1) < TOLERANCE return rows shares = check_shares('03-country-share', country_amounts) subset = {key: country_amounts[key] for key in ['USA', 'Canada']} restricted = check_shares('05-share-two-countries', subset) top_three = check_shares('06-share-top-three', country_amounts, selected=3) assert top_three == shares[:3] empty = data('07-empty-average')['items'] assert not any('1990-01-01' <= row['InvoiceDate'] < '1991-01-01' for row in invoices) assert len(empty) == 1 and empty[0]['billCount'] == 0 assert empty[0].get('totalAmount') is None and empty[0].get('temporaryAverage') is None predefined = data('09-predefined-average')['items'] assert len(predefined) == 1 near(predefined[0]['averageAmount'], average) near(predefined[0]['averageAmount'], Decimal(str(actual_average['temporaryAverage']))) country_predefined = data('10-predefined-country-average')['items'] assert [row['billingCountry'] for row in country_predefined] == sorted(country_amounts) for row, temporary in zip(country_predefined, averages): key = row['billingCountry'] near(row['averageAmount'], country_amounts[key] / country_counts[key]) near(row['averageAmount'], Decimal(str(temporary['temporaryAverage']))) fixed_empty = data('11-predefined-empty-average')['items'] assert len(fixed_empty) == 1 and fixed_empty[0]['billCount'] == 0 assert fixed_empty[0].get('averageAmount') is None baseline = dict(amount=str(total), billCount=len(year), purchasingCustomers=len(buyers), allCustomers=len(all_customers), averageAmount=str(average), simpleMeanOfCountryAverages=str(simple_country_average), twoCountryDenominator=str(sum(subset.values(), ZERO)), countries=[dict(country=key, amount=str(country_amounts[key]), bills=country_counts[key], share=str(country_amounts[key] / total)) for key in sorted(country_amounts, key=lambda key: (-country_amounts[key], key))], emptyResponse=empty, predefinedEmptyResponse=fixed_empty, predefinedAverageMatchesTemporary=True, topThreeKeepWholeYearDenominator=True, calculation='Read-only original Invoice rows, Python sets/dictionaries and Decimal.') (EVIDENCE / 'baseline.json').write_text(json.dumps(baseline, ensure_ascii=False, indent=2), encoding='utf-8') acceptance = dict(checkedAt=datetime.now().astimezone().isoformat(), success=True, runtimeUrl='http://127.0.0.1:18067', namespace='chinook', dataSource='chinook-sqlite', model='ChinookInvoiceQueryModel', securityMode='none-dev-test-only', passingQueryCases=11, modelSha256=model_hashes, changedModelFiles=['models/query/ChinookInvoiceQueryModel.qm'], checks=['SUM and COUNT', 'COUNT_DISTINCT purchasing customers', 'Calculated ratio-of-aggregates average', 'Native AVG/MIN/MAX', '20 grouped country averages', 'Whole-year and filtered share denominators', 'Top 3 keeps denominator', 'Zero-count empty average is absent, not zero', 'Predefined QM average: overall, 20 country groups and empty scope match temporary formula', 'Model validation and predefined describe', 'Database unchanged; only intended QM calculation added'], numericTolerance=str(TOLERANCE), executionMode='Current Codex session generated payloads and executed public CLI; no separate clean Codex prompt run.', cleanSessionPromptVerified=False, textSqlBenchmarkExecuted=False, productionPermissionsVerified=False, browserFapVerified=False) (EVIDENCE / 'acceptance.json').write_text(json.dumps(acceptance, ensure_ascii=False, indent=2), encoding='utf-8') print(json.dumps(acceptance, ensure_ascii=False, indent=2)) print('Baseline:', json.dumps({k:v for k,v in baseline.items() if k != 'countries'}, ensure_ascii=False)) if __name__ == '__main__': main()