"""Tests for fetch_users() in app.cm_transfer_credit. fetch_users() is the bot's API-fetch boundary. It must: - Return the list on a clean 200 with a JSON list body. - Return [] on a 200 with empty list (caller distinguishes "no work" from "fetch failed" — empty list is NOT a failure). - Return None and send a Telegram alert on every error mode: HTTP error, network error, malformed JSON, contract violation. - Hit the dedicated /user/batch endpoint with a finite timeout. The previous implementation collapsed every error mode into "0 items" silently — these tests pin the new loud-skip behavior so the silent failure can't regress. """ import logging import unittest from unittest import mock import requests import app.cm_transfer_credit as transfer def _make_response(status_code=200, json_data=None, json_error=None, text=""): """Build a Mock that quacks like requests.Response for the paths fetch_users actually calls: status_code, text, raise_for_status(), json(). json_error: if set, response.json() raises this instead of returning json_data. Use to simulate malformed JSON bodies. """ response = mock.Mock(spec=requests.Response) response.status_code = status_code response.text = text if json_error is not None: response.json.side_effect = json_error else: response.json.return_value = json_data if 400 <= status_code < 600: response.raise_for_status.side_effect = requests.HTTPError( f"{status_code} Server Error", response=response ) else: response.raise_for_status.return_value = None return response class FetchUsersHappyPathTests(unittest.TestCase): def setUp(self): self.logger = mock.Mock(spec=logging.Logger) @mock.patch("app.cm_transfer_credit.notifier") @mock.patch("app.cm_transfer_credit.requests.get") def test_returns_list_on_200_with_list_body(self, mock_get, mock_notifier): rows = [ {"f_username": "13c1", "f_password": "p1", "t_username": "tA", "t_password": "pA"}, {"f_username": "13c2", "f_password": "p2", "t_username": "tB", "t_password": "pB"}, ] mock_get.return_value = _make_response(200, json_data=rows) result = transfer.fetch_users(self.logger) self.assertEqual(result, rows) mock_notifier.notify_generic_error.assert_not_called() self.logger.error.assert_not_called() @mock.patch("app.cm_transfer_credit.notifier") @mock.patch("app.cm_transfer_credit.requests.get") def test_empty_list_is_not_treated_as_failure(self, mock_get, mock_notifier): # An empty user table is a legitimate "no work" signal, not an # error. Caller must see [] (truthy len()==0) not None. mock_get.return_value = _make_response(200, json_data=[]) result = transfer.fetch_users(self.logger) self.assertEqual(result, []) self.assertIsNotNone(result) mock_notifier.notify_generic_error.assert_not_called() @mock.patch("app.cm_transfer_credit.notifier") @mock.patch("app.cm_transfer_credit.requests.get") def test_hits_batch_endpoint_with_30s_timeout(self, mock_get, mock_notifier): # Pins two contracts the bot relies on: # 1. URL ends in /user/batch (NOT /user/ — that returns # {"rows": [...]} and would re-trigger the original bug). # 2. A finite timeout exists, so a hung DB can't wedge the # worker for the full sleep window. mock_get.return_value = _make_response(200, json_data=[]) transfer.fetch_users(self.logger) mock_get.assert_called_once() url = mock_get.call_args.args[0] self.assertTrue(url.endswith("/user/batch"), f"expected /user/batch, got {url!r}") self.assertEqual(mock_get.call_args.kwargs.get("timeout"), 30) class FetchUsersErrorPathTests(unittest.TestCase): """Every error mode must: return None, log via local_logger.error, and call notifier.notify_generic_error exactly once.""" def setUp(self): self.logger = mock.Mock(spec=logging.Logger) def _assert_alerted_and_skipped(self, result, mock_notifier): self.assertIsNone(result, "fetch_users must return None on failure") self.assertEqual(self.logger.error.call_count, 1) mock_notifier.notify_generic_error.assert_called_once() @mock.patch("app.cm_transfer_credit.notifier") @mock.patch("app.cm_transfer_credit.requests.get") def test_http_500_alerts_and_returns_none(self, mock_get, mock_notifier): mock_get.return_value = _make_response(500, text="Internal Server Error") result = transfer.fetch_users(self.logger) self._assert_alerted_and_skipped(result, mock_notifier) @mock.patch("app.cm_transfer_credit.notifier") @mock.patch("app.cm_transfer_credit.requests.get") def test_connection_error_alerts_and_returns_none(self, mock_get, mock_notifier): mock_get.side_effect = requests.ConnectionError("api-server unreachable") result = transfer.fetch_users(self.logger) self._assert_alerted_and_skipped(result, mock_notifier) @mock.patch("app.cm_transfer_credit.notifier") @mock.patch("app.cm_transfer_credit.requests.get") def test_timeout_alerts_and_returns_none(self, mock_get, mock_notifier): mock_get.side_effect = requests.Timeout("read timeout") result = transfer.fetch_users(self.logger) self._assert_alerted_and_skipped(result, mock_notifier) @mock.patch("app.cm_transfer_credit.notifier") @mock.patch("app.cm_transfer_credit.requests.get") def test_malformed_json_alerts_and_returns_none(self, mock_get, mock_notifier): mock_get.return_value = _make_response( 200, json_error=ValueError("Expecting value: line 1 column 1 (char 0)"), text="500 Bad Gateway", ) result = transfer.fetch_users(self.logger) self._assert_alerted_and_skipped(result, mock_notifier) @mock.patch("app.cm_transfer_credit.notifier") @mock.patch("app.cm_transfer_credit.requests.get") def test_envelope_dict_alerts_and_returns_none(self, mock_get, mock_notifier): # Regression test for the actual bug: /user/ returns # {"rows": [...], "total": N} which the old silent guard # collapsed to "0 items". The new code must surface this loudly. envelope = {"rows": [{"f_username": "13c1"}], "total": 1} mock_get.return_value = _make_response(200, json_data=envelope) result = transfer.fetch_users(self.logger) self._assert_alerted_and_skipped(result, mock_notifier) # The alert message should mention what was actually returned so # an operator reading the Telegram alert can diagnose without # opening logs. alert_msg = mock_notifier.notify_generic_error.call_args.args[0] self.assertIn("dict", alert_msg, f"alert should name the bad type; got: {alert_msg!r}") if __name__ == "__main__": unittest.main()