|
5 | 5 | import unittest.mock as mock |
6 | 6 | import io |
7 | 7 | import json |
| 8 | +import importlib |
8 | 9 | import requests |
9 | 10 | import redis |
10 | 11 | from minio.error import S3Error, InvalidResponseError |
@@ -530,5 +531,45 @@ def test_readiness_check_partial_failure(self): |
530 | 531 | self.assertEqual(result, 200) |
531 | 532 |
|
532 | 533 |
|
| 534 | +class TestOpenSenseExceptionBranches(unittest.TestCase): |
| 535 | + """Covers exception branches in opensense module.""" |
| 536 | + |
| 537 | + def test_set_cached_temperature_redis_error_is_swallowed(self): |
| 538 | + """_set_cached_temperature should swallow RedisError (pass).""" |
| 539 | + with mock.patch('app.opensense.REDIS_AVAILABLE', True), \ |
| 540 | + mock.patch('app.opensense.REDIS_CLIENT') as mock_client: |
| 541 | + mock_client.setex.side_effect = redis.RedisError("boom") |
| 542 | + # Should not raise |
| 543 | + opensense._set_cached_temperature("value") |
| 544 | + mock_client.setex.assert_called_once() |
| 545 | + |
| 546 | + def test_get_cached_temperature_redis_error_returns_none(self): |
| 547 | + """_get_cached_temperature should return None on RedisError.""" |
| 548 | + with mock.patch('app.opensense.REDIS_AVAILABLE', True), \ |
| 549 | + mock.patch('app.opensense.REDIS_CLIENT') as mock_client: |
| 550 | + mock_client.get.side_effect = redis.RedisError("boom") |
| 551 | + result = opensense._get_cached_temperature() |
| 552 | + self.assertIsNone(result) |
| 553 | + mock_client.get.assert_called_once_with("temperature_data") |
| 554 | + |
| 555 | + def test_module_level_redis_init_errors_are_caught_all_types(self): |
| 556 | + """create_redis_client exceptions at import-time are handled (all types).""" |
| 557 | + for exc in ( |
| 558 | + redis.ConnectionError("c"), |
| 559 | + redis.TimeoutError("t"), |
| 560 | + OSError("o"), |
| 561 | + ImportError("i")): |
| 562 | + with self.subTest(exc=type(exc).__name__), \ |
| 563 | + mock.patch('app.config.create_redis_client', side_effect=exc), \ |
| 564 | + mock.patch('builtins.print'): |
| 565 | + reloaded = importlib.reload(opensense) |
| 566 | + self.assertIsNone(reloaded.REDIS_CLIENT) |
| 567 | + self.assertFalse(reloaded.REDIS_AVAILABLE) |
| 568 | + # Restore module for other tests |
| 569 | + with mock.patch('app.config.create_redis_client', return_value=(None, False)), \ |
| 570 | + mock.patch('builtins.print'): |
| 571 | + importlib.reload(opensense) |
| 572 | + |
| 573 | + |
533 | 574 | if __name__ == '__main__': |
534 | 575 | unittest.main() |
0 commit comments