Add private operations portal and weekly accounting reports

2026-11-03 16:45:50 +08:00
parent 10c4487463
commit 20aea13054
18 changed files with 958 additions and 0 deletions
@@ -0,0 +1,58 @@
+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())
+
+
+if __name__ == '__main__':
+    unittest.main()