wuyang
81 lines
5.1 KiB
Python
81 lines
5.1 KiB
Python
import datetime as dt
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
|
|
spec = importlib.util.spec_from_file_location('report', Path(__file__).with_name('report.py'))
|
|
report = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(report)
|
|
|
|
|
|
class AccountingTests(unittest.TestCase):
|
|
def test_monthly_allocation_across_month_boundary(self):
|
|
budget = {'subscription_usd': 200, 'billing_period': 'monthly', 'project_allocation_fraction': 0.5}
|
|
value = report.subscription_allocation(budget, dt.date(2026, 10, 29), dt.date(2026, 11, 5))
|
|
self.assertEqual(value, round((200 / 31 * 3 + 200 / 30 * 4) * 0.5, 2))
|
|
|
|
def test_unknown_period_or_share_is_not_zero(self):
|
|
self.assertIsNone(report.subscription_allocation({'subscription_usd': 200}, dt.date(2026, 11, 2), dt.date(2026, 11, 9)))
|
|
|
|
def test_friday_twenty_hour_cutoff(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path, _ = report.build_report(Path(directory), now=dt.datetime(2026, 11, 6, 20, tzinfo=report.TZ))
|
|
value = report.read_json(path)
|
|
self.assertEqual(value['period_start'], '2026-10-30T20:00:00+08:00')
|
|
self.assertEqual(value['period_end'], '2026-11-06T20:00:00+08:00')
|
|
|
|
def test_cross_month_proration_includes_partial_days(self):
|
|
budget = {'subscription_usd': 200, 'billing_period': 'monthly', 'project_allocation_fraction': 1}
|
|
value = report.subscription_allocation(budget, dt.datetime(2026, 10, 30, 20, tzinfo=report.TZ), dt.datetime(2026, 11, 6, 20, tzinfo=report.TZ))
|
|
self.assertEqual(value, round(200 / 31 * 28 / 24 + 200 / 30 * 140 / 24, 2))
|
|
|
|
def test_duplicate_usage_events_are_not_counted_twice(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / 'test.jsonl'
|
|
meta = {'type': 'session_meta', 'payload': {'id': 'pawplay'}}
|
|
event = {'timestamp': '2026-11-03T08:00:00Z', 'type': 'event_msg', 'payload': {'type': 'token_count', 'info': {'total_token_usage': {'input_tokens': 100, 'cached_input_tokens': 80, 'output_tokens': 10, 'reasoning_output_tokens': 4, 'total_tokens': 110}}}}
|
|
import json
|
|
path.write_text('\n'.join(json.dumps(e) for e in [meta, event, event]))
|
|
result = report.codex_usage(directory, 'pawplay', dt.datetime(2026, 11, 2, tzinfo=report.TZ), dt.datetime(2026, 11, 9, tzinfo=report.TZ))
|
|
self.assertEqual(result['total_tokens'], 110)
|
|
self.assertEqual(result['uncached_input_tokens'], 20)
|
|
self.assertEqual(result['output_tokens'], 10)
|
|
|
|
def test_final_reports_are_immutable(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
now = dt.datetime(2026, 11, 9, 9, tzinfo=report.TZ)
|
|
first, created = report.build_report(root, now=now)
|
|
self.assertTrue(created)
|
|
original = first.read_bytes()
|
|
second, created = report.build_report(root, now=now + dt.timedelta(hours=1))
|
|
self.assertFalse(created)
|
|
self.assertEqual(original, second.read_bytes())
|
|
|
|
def test_scheduled_usage_is_period_scoped_and_preserves_failed_runs(self):
|
|
import json
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
(root / 'one.summary.json').write_text(json.dumps({'id':'20261103T141341Z-a','status':'complete','usage':[{'input_tokens':100,'cached_input_tokens':80,'output_tokens':10}]}))
|
|
(root / 'two.summary.json').write_text(json.dumps({'id':'20261104T141341Z-b','status':'failed','usage':[]}))
|
|
(root / 'paid.summary.json').write_text(json.dumps({'id':'20261104T151341Z-d','status':'complete','auth':'api_key','api_fallback_used':True,'usage':[{'input_tokens':33,'output_tokens':5}]}))
|
|
(root / 'old.summary.json').write_text(json.dumps({'id':'20261003T141341Z-c','status':'complete','usage':[{'input_tokens':900}]}))
|
|
value=report.scheduled_codex_usage(root,dt.datetime(2026,11,3,tzinfo=report.TZ),dt.datetime(2026,11,5,tzinfo=report.TZ))
|
|
self.assertEqual((value['runs'],value['failed_runs'],value['input_tokens'],value['cached_input_tokens']), (3,1,100,80))
|
|
self.assertEqual(value['api_runs_separate'],1);self.assertEqual(value['api_observed_usage']['input_tokens'],33)
|
|
|
|
def test_harness_ledger_counts_only_selected_period(self):
|
|
import json
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / 'harness-runs.jsonl'
|
|
entries = [{'at_utc':'2026-11-03T14:00:00Z','usage':[{'inputTokens':10,'outputTokens':4}]}, {'at_utc':'2026-10-03T14:00:00Z','usage':[{'inputTokens':100}]}]
|
|
path.write_text('\n'.join(json.dumps(e) for e in entries))
|
|
value=report.harness_run_usage(path,dt.datetime(2026,11,3,tzinfo=report.TZ),dt.datetime(2026,11,5,tzinfo=report.TZ))
|
|
self.assertEqual((value['runs'],value['input_tokens'],value['output_tokens']),(1,10,4))
|
|
self.assertIsNone(value['cash_settlement_usd'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|