fix: integrate token tracker auto-logging with orchestration (#634)
Some checks failed
Smoke Test / smoke (pull_request) Failing after 21s
Architecture Lint / Linter Tests (pull_request) Successful in 25s
Validate Config / YAML Lint (pull_request) Failing after 16s
Validate Config / JSON Validate (pull_request) Successful in 18s
Validate Config / Python Syntax & Import Check (pull_request) Failing after 59s
Validate Config / Python Test Suite (pull_request) Has been skipped
Validate Config / Shell Script Lint (pull_request) Failing after 1m2s
Validate Config / Cron Syntax Check (pull_request) Successful in 11s
Validate Config / Deploy Script Dry Run (pull_request) Successful in 10s
Validate Config / Playbook Schema Validation (pull_request) Successful in 23s
Architecture Lint / Lint Repository (pull_request) Failing after 22s
PR Checklist / pr-checklist (pull_request) Successful in 5m28s

This commit is contained in:
Alexander Whitestone
2026-04-22 20:38:06 -04:00
parent ae8c1d46ae
commit 213fe0ca91
2 changed files with 56 additions and 9 deletions

View File

@@ -80,6 +80,20 @@ class TestLogTokenUsage:
line = json.loads(log_file.read_text().strip())
assert line["pipeline"] == "knowledge-mine"
def test_records_to_token_tracker(self, tmp_path):
"""Should record total tokens to token_tracker for automatic pipeline logging."""
log_file = tmp_path / "token_usage.jsonl"
mock_conn = MagicMock()
mock_tracker = MagicMock()
with patch("orchestration.TOKEN_LOG", log_file), patch("orchestration.record_usage"), patch("orchestration.get_token_tracker_db", return_value=mock_conn), patch("orchestration.token_tracker_record_usage", mock_tracker):
from orchestration import log_token_usage
log_token_usage("knowledge_mine_task", {
"input_tokens": 10,
"output_tokens": 20,
})
mock_tracker.assert_called_once_with(mock_conn, "knowledge-mine", "knowledge_mine_task", 30)
class TestCheckBudget:
"""Test check_budget function."""