"""Verify semantic filters against raw read-only Invoice/Customer 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-09' HASH = 'bdf635be69850bd3be09c9a2dbeef7ddfb80036bd3ef3381383cd03b61e4a61a' def load(path): return json.loads(Path(path).read_text(encoding='utf-8-sig')) def money(value): return Decimal(str(value)).quantize(Decimal('0.01')) def main(): manifest = load(EVIDENCE / 'query-manifest.json') cases = {case['name']: case for case in manifest['queries']} 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 model_hashes = load(LAB / 'evidence/chapter-08/acceptance.json')['modelSha256'] # Chapter 10 intentionally extends this QM; use its explicit before/after # contract when replaying earlier queries on the later accepted model. evolution_path = LAB / 'evidence/chapter-10/model-evolution.json' if evolution_path.exists(): evolution = load(evolution_path) assert evolution['before'] == model_hashes model_hashes = evolution['after'] for path, digest in model_hashes.items(): assert hashlib.sha256((LAB / path).read_bytes()).hexdigest() == digest, path 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')] customers = {row['CustomerId']: dict(row) for row in db.execute('SELECT * FROM Customer')} def rows(name): assert load(cases[name]['validation'])['success'] is True, name result = load(cases[name]['response']) assert result['success'] is True assert not result['data'].get('hasNext') assert not result['data'].get('pagination', {}).get('hasMore') return result['data']['items'] def check_details(name, expected): actual = rows(name) expected = sorted(expected, key=lambda item: item['InvoiceId']) assert [row['invoiceId'] for row in actual] == [row['InvoiceId'] for row in expected], name for output, original in zip(actual, expected): assert output['invoiceDate'] == original['InvoiceDate'], name assert output['billingCountry'] == original['BillingCountry'], name assert money(output['invoiceAmount']) == Decimal(str(original['Total'])), name return dict(count=len(actual), invoiceIds=[row['invoiceId'] for row in actual], amount=str(sum((Decimal(str(row['Total'])) for row in expected), Decimal('0')))) values = [row['billingCountry'] for row in rows('01-country-values')] assert values == sorted({row['BillingCountry'] for row in invoices}) year = [row for row in invoices if '2012-01-01' <= row['InvoiceDate'] < '2013-01-01'] usa = [row for row in year if row['BillingCountry'] == 'USA'] selected = [row for row in year if row['BillingCountry'] in {'USA', 'Canada'} and Decimal(str(row['Total'])) > 10] single = [row for row in year if Decimal(str(row['Total'])) > 20] baseline = dict(countryValues=values, usa2012=check_details('02-usa-2012', usa), twoCountriesOver10=check_details('03-two-countries-in', selected), singleInvoiceOver20=check_details('06-single-invoice-over-20', single)) check_details('04-two-countries-or', selected) assert rows('03-two-countries-in') == rows('04-two-countries-or') # The fixtures include matching countries/amounts outside 2012; equality # against original rows therefore also checks AND/OR scope boundaries. assert any(row['BillingCountry'] in {'USA', 'Canada'} and Decimal(str(row['Total'])) > 10 and row not in year for row in invoices) def sums(source): result = defaultdict(lambda: Decimal('0')) for row in source: result[row['CustomerId']] += Decimal(str(row['Total'])) return result def check_customers(name, expected): actual = rows(name) order = sorted(expected, key=lambda key: (-expected[key], key)) assert [row['customerId'] for row in actual] == order, name for row in actual: key = row['customerId'] assert money(row['totalAmount']) == expected[key], name assert row['customer$fullName'] == customers[key]['FirstName'] + ' ' + customers[key]['LastName'] return [dict(customerId=key, name=customers[key]['FirstName'] + ' ' + customers[key]['LastName'], amount=str(expected[key])) for key in order] yearly = sums(year) qualifying = {key: value for key, value in yearly.items() if value > 20} baseline['customerHavingOver20'] = check_customers('05-customer-having', qualifying) baseline['filterBeforeSum'] = check_customers('07-wrong-filter-before-sum', sums(single)) assert not any('1990-01-01' <= row['InvoiceDate'] < '1991-01-01' for row in invoices) assert rows('08-empty-range') == [] baseline['emptyRange'] = dict(success=True, rows=0) baseline['calculation'] = 'Read-only raw SQLite rows, Python predicates and Decimal; expected values do not come from semantic output.' (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=len(cases), countryValues=len(values), checks=['Country literal values', 'Every matching invoice identity/date/country/amount', 'IN equals explicit OR with date and amount ANDed', 'All customer sums, threshold membership and stable order', 'Source-row filtering differs from HAVING', 'Empty result succeeds', 'Database and accepted model-version hashes match'], executionMode='Current Codex session generated payloads and executed public CLI; no separate clean Codex prompt run.', cleanSessionPromptVerified=False, productionPermissionsVerified=False, browserFapVerified=False) (EVIDENCE / 'acceptance.json').write_text(json.dumps(acceptance, ensure_ascii=False, indent=2), encoding='utf-8') print(json.dumps(baseline, ensure_ascii=False, indent=2)) print(json.dumps(acceptance, ensure_ascii=False, indent=2)) if __name__ == '__main__': main()