Codex
70 lines
2.6 KiB
Python
70 lines
2.6 KiB
Python
import contextlib
|
|
import io
|
|
import json
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from scripts import run_operator_digest
|
|
|
|
|
|
class RunOperatorDigestTests(unittest.TestCase):
|
|
def test_main_outputs_json_digest(self) -> None:
|
|
payload = {
|
|
"status": "completed",
|
|
"summary_counts": {"recent_alert_count": 2},
|
|
"review_queue": [{"queue_type": "failed_retry"}],
|
|
"operator_summary": "Digest alerts=2.",
|
|
}
|
|
output = io.StringIO()
|
|
with patch("scripts.run_operator_digest.OperatorDigestService") as service_cls:
|
|
service_cls.return_value.build_digest.return_value = payload
|
|
with contextlib.redirect_stdout(output):
|
|
exit_code = run_operator_digest.main(
|
|
env={
|
|
"OPERATOR_DIGEST_OUTPUT_JSON": "1",
|
|
"OPERATOR_DIGEST_INSTRUMENT_ID": "2",
|
|
"OPERATOR_DIGEST_LIMIT": "5",
|
|
"OPERATOR_DIGEST_VISUAL_LIMIT": "7",
|
|
"OPERATOR_DIGEST_TOP_LIMIT": "3",
|
|
}
|
|
)
|
|
|
|
self.assertEqual(exit_code, 0)
|
|
service_cls.return_value.build_digest.assert_called_once_with(
|
|
instrument_id=2,
|
|
alert_limit=5,
|
|
visual_review_limit=7,
|
|
top_limit=3,
|
|
)
|
|
self.assertEqual(json.loads(output.getvalue())["review_queue"][0]["queue_type"], "failed_retry")
|
|
|
|
def test_main_outputs_text_digest(self) -> None:
|
|
payload = {
|
|
"status": "completed",
|
|
"summary_counts": {
|
|
"recent_alert_count": 2,
|
|
"suppressed_delivery_count": 1,
|
|
"failed_or_retry_delivery_count": 1,
|
|
"visual_mismatch_count": 1,
|
|
"pending_feedback_count": 3,
|
|
},
|
|
"review_queue": [{"priority": "high", "queue_type": "visual_mismatch", "visual_review_id": 31}],
|
|
"operator_summary": "Digest alerts=2.",
|
|
"warnings": ["visual_feedback_unavailable"],
|
|
}
|
|
output = io.StringIO()
|
|
with patch("scripts.run_operator_digest.OperatorDigestService") as service_cls:
|
|
service_cls.return_value.build_digest.return_value = payload
|
|
with contextlib.redirect_stdout(output):
|
|
exit_code = run_operator_digest.main(env={})
|
|
|
|
self.assertEqual(exit_code, 0)
|
|
text = output.getvalue()
|
|
self.assertIn("Operator digest status: completed", text)
|
|
self.assertIn("Review queue items: 1", text)
|
|
self.assertIn("visual_feedback_unavailable", text)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|