import copy import json from pathlib import Path import subprocess import sys import tempfile import unittest from odds_acquisition import evaluate ROOT = Path(__file__).resolve().parent class WorksheetTests(unittest.TestCase): def setUp(self): self.data = json.loads((ROOT / 'example.json').read_text()) def rows(self): return evaluate(self.data)['results'] def test_worked_baseline_and_sensitivity(self): a, b, fast_a, fast_b, large_a, large_b = self.rows() self.assertEqual((a['polls'], a['total_request_attempts'], a['api_credits'], a['modeled_total']), (14400, 30240, 90720, 200)) self.assertEqual((b['total_request_attempts'], b['billed_decimal_GB'], b['modeled_total']), (302400, 60.48, 560.96)) self.assertAlmostEqual(b['billed_GiB'], 56.32638931274414) self.assertEqual((fast_a['api_credits'], fast_a['quota_status'], fast_a['modeled_total']), (544320, 'exceeded', None)) self.assertEqual((fast_b['billed_decimal_GB'], fast_b['modeled_total']), (362.88, 1165.76)) self.assertEqual((large_b['billed_decimal_GB'], large_b['modeled_total']), (302.4, 1044.8)) self.assertEqual(large_a, {**a, 'scenario': large_a['scenario']}) def test_blank_preserves_unknowns(self): rows = evaluate(json.loads((ROOT / 'blank.json').read_text()))['results'] for r in rows: self.assertEqual(r['decision'], 'hold') for key in ('polls', 'scheduled_requests', 'total_request_attempts', 'modeled_total'): self.assertIsNone(r[key]) self.assertEqual(r['missing_cost_components'], ['fixed', 'usage', 'maintenance']) def test_each_no_or_unknown_gate_holds_even_with_known_costs(self): for gate in ('coverage', 'allowed_use', 'freshness', 'history'): for status in ('no', 'unknown'): with self.subTest(gate=gate, status=status): d = copy.deepcopy(self.data) d['candidates'][0]['gates'][gate]['status'] = status r = evaluate(d)['results'][0] self.assertEqual(r['decision'], 'hold') self.assertIn(f'{gate}:{status}', r['blockers']) self.assertEqual(r['modeled_total'], 200) def test_missing_cost_is_not_free_even_for_zero_rate(self): self.data['candidates'][0]['credits_per_attempt'] = None r = self.rows()[0] self.assertIsNone(r['cost_components']['usage']) self.assertIsNone(r['modeled_total']) self.assertEqual(r['quota_status'], 'unknown') def test_unknown_maintenance_or_fixed_cost_holds(self): for key in ('fixed_cost', 'maintenance_hours', 'hourly_rate', 'usage_price_per_unit'): with self.subTest(key=key): d = copy.deepcopy(self.data) d['candidates'][0]['costs'][key] = None self.assertIsNone(evaluate(d)['results'][0]['modeled_total']) self.assertEqual(evaluate(d)['results'][0]['decision'], 'hold') def test_quota_exact_limit_passes_one_less_fails_unknown_holds(self): for limit, expected in ((90720, 'pass'), (90719, 'exceeded'), (None, 'unknown')): self.data['candidates'][0]['quota']['limit'] = limit self.assertEqual(self.rows()[0]['quota_status'], expected) def test_daily_window_and_retry_round_up(self): self.data['workload'].update(active_days=2, hours_per_day=1) self.data['scenarios'][0]['poll_interval_seconds'] = 1000 self.data['candidates'][0]['units_per_poll'] = 1 r = self.rows()[0] self.assertEqual((r['polls_per_active_day'], r['scheduled_requests'], r['planned_retry_requests']), (4, 8, 1)) def test_zero_window_has_zero_usage_but_fixed_and_labor_cost(self): self.data['workload']['hours_per_day'] = 0 r = self.rows()[1] self.assertEqual((r['polls'], r['total_request_attempts'], r['modeled_total']), (0, 0, 440)) def test_invalid_numbers_fail(self): for value in (0, -1, True, '60', float('nan'), float('inf')): with self.subTest(value=value): self.data['scenarios'][0]['poll_interval_seconds'] = value with self.assertRaises(ValueError): self.rows() def test_fractional_units_and_impossible_daily_hours_fail(self): self.data['candidates'][0]['units_per_poll'] = 1.5 with self.assertRaises(ValueError): self.rows() self.data['candidates'][0]['units_per_poll'] = 2 self.data['workload']['hours_per_day'] = 25 with self.assertRaises(ValueError): self.rows() def test_no_evidence_or_invalid_status_fails(self): self.data['candidates'][0]['gates']['coverage']['note'] = '' with self.assertRaises(ValueError): self.rows() self.data['candidates'][0]['gates']['coverage'] = {'note': 'a note', 'status': 'maybe'} with self.assertRaises(ValueError): self.rows() def test_unknown_bytes_not_zero(self): self.data['scenarios'][0]['collector_billed_bytes_per_attempt'] = None r = self.rows()[1] self.assertIsNone(r['billed_decimal_GB']) self.assertIsNone(r['modeled_total']) self.assertEqual(r['decision'], 'hold') def test_nonobjects_and_boolean_schema_fail(self): for value in (None, [], 1, 'text', {**self.data, 'schema_version': True}): with self.subTest(value=type(value).__name__): with self.assertRaises(ValueError): evaluate(value) def test_cli_invalid_input_emits_no_result(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / 'input.json' invalid = [None, [], {**self.data, 'schema_version': True}] enormous = copy.deepcopy(self.data) enormous['candidates'][0]['planned_retry_fraction'] = 1e308 invalid.append(enormous) for value in invalid: path.write_text(json.dumps(value)) r = subprocess.run([sys.executable, str(ROOT / 'odds_acquisition.py'), str(path)], capture_output=True, text=True) self.assertEqual(r.returncode, 2) self.assertEqual(r.stdout, '') self.assertNotIn('Traceback', r.stderr) path.unlink() r = subprocess.run([sys.executable, str(ROOT / 'odds_acquisition.py'), str(path)], capture_output=True, text=True) self.assertEqual(r.returncode, 2) self.assertEqual(r.stdout, '') self.assertNotIn('Traceback', r.stderr) def test_cli_deterministic_and_matches_function(self): cmd = [sys.executable, str(ROOT / 'odds_acquisition.py'), str(ROOT / 'example.json')] first = subprocess.check_output(cmd) self.assertEqual(first, subprocess.check_output(cmd)) self.assertEqual(json.loads(first), evaluate(self.data)) if __name__ == '__main__': unittest.main()