krkn/tests/test_cerberus_setup.py

"""
Test suite for krkn/cerberus/setup.py

Run this test file individually with:
  python -m unittest tests/test_cerberus_setup.py -v

Or with coverage:
  python3 -m coverage run -a -m unittest tests/test_cerberus_setup.py -v

Generated with help from Claude Code
"""
import unittest
from unittest.mock import patch, MagicMock, Mock
import sys
import json
import requests
from krkn.cerberus import setup as cerberus_setup


class TestCerberusSetup(unittest.TestCase):
    """Test suite for cerberus/setup.py module"""

    def setUp(self):
        """Reset global variables before each test"""
        cerberus_setup.cerberus_url = None
        cerberus_setup.exit_on_failure = False
        cerberus_setup.cerberus_enabled = False
        cerberus_setup.check_application_routes = ""

    def test_set_url_with_cerberus_enabled(self):
        """Test set_url when cerberus is enabled"""
        config = {
            "kraken": {"exit_on_failure": True},
            "cerberus": {
                "cerberus_enabled": True,
                "cerberus_url": "http://cerberus.example.com",
                "check_application_routes": "route1,route2"
            }
        }

        cerberus_setup.set_url(config)

        self.assertEqual(cerberus_setup.cerberus_url, "http://cerberus.example.com")
        self.assertTrue(cerberus_setup.exit_on_failure)
        self.assertTrue(cerberus_setup.cerberus_enabled)
        self.assertEqual(cerberus_setup.check_application_routes, "route1,route2")

    def test_set_url_with_cerberus_disabled(self):
        """Test set_url when cerberus is disabled"""
        config = {
            "kraken": {"exit_on_failure": False},
            "cerberus": {"cerberus_enabled": False}
        }

        cerberus_setup.set_url(config)

        self.assertFalse(cerberus_setup.cerberus_enabled)
        self.assertFalse(cerberus_setup.exit_on_failure)
        self.assertIsNone(cerberus_setup.cerberus_url)

    def test_set_url_with_defaults(self):
        """Test set_url with missing optional fields (should use defaults)"""
        config = {
            "kraken": {},
            "cerberus": {}
        }

        cerberus_setup.set_url(config)

        self.assertFalse(cerberus_setup.exit_on_failure)
        self.assertFalse(cerberus_setup.cerberus_enabled)

    @patch.object(cerberus_setup, 'http_session')
    def test_get_status_cerberus_disabled(self, mock_session):
        """Test get_status when cerberus is disabled makes no HTTP calls"""
        cerberus_setup.cerberus_enabled = False

        result = cerberus_setup.get_status(0, 100)

        self.assertTrue(result)
        mock_session.get.assert_not_called()

    @patch.object(cerberus_setup, 'http_session')
    def test_get_status_cerberus_enabled_healthy(self, mock_session):
        """Test get_status when cerberus is enabled and cluster is healthy"""
        cerberus_setup.cerberus_enabled = True
        cerberus_setup.cerberus_url = "http://cerberus.example.com"

        mock_response = MagicMock()
        mock_response.content = b"True"
        mock_session.get.return_value = mock_response

        result = cerberus_setup.get_status(0, 100)

        self.assertTrue(result)
        mock_session.get.assert_called_once_with("http://cerberus.example.com", timeout=60)

    @patch.object(cerberus_setup, 'http_session')
    def test_get_status_cerberus_enabled_unhealthy(self, mock_session):
        """Test get_status when cerberus is enabled and cluster is unhealthy"""
        cerberus_setup.cerberus_enabled = True
        cerberus_setup.cerberus_url = "http://cerberus.example.com"

        mock_response = MagicMock()
        mock_response.content = b"False"
        mock_session.get.return_value = mock_response

        with self.assertRaises(SystemExit) as cm:
            cerberus_setup.get_status(0, 100)

        self.assertEqual(cm.exception.code, 1)
        mock_session.get.assert_called_once_with("http://cerberus.example.com", timeout=60)

    def test_get_status_no_url_provided(self):
        """Test get_status when cerberus is enabled but URL is not provided"""
        cerberus_setup.cerberus_enabled = True
        cerberus_setup.cerberus_url = None

        with self.assertRaises(SystemExit) as cm:
            cerberus_setup.get_status(0, 100)

        self.assertEqual(cm.exception.code, 1)

    @patch.object(cerberus_setup, 'http_session')
    def test_get_status_cerberus_healthy_returns_true(self, mock_session):
        """Test get_status returns True when cerberus reports healthy.
        Note: check_application_routes is shadowed locally in get_status()
        (pre-existing issue), so route-check branch is not exercised here."""
        cerberus_setup.cerberus_enabled = True
        cerberus_setup.cerberus_url = "http://cerberus.example.com"
        cerberus_setup.check_application_routes = "route1,route2"

        def mock_get_side_effect(url, timeout):
            mock_response = MagicMock()
            if "/history?" in url:
                mock_response.content = json.dumps({"history": {"failures": []}}).encode()
            else:
                mock_response.content = b"True"
            return mock_response

        mock_session.get.side_effect = mock_get_side_effect

        result = cerberus_setup.get_status(0, 100)

        self.assertTrue(result)
        self.assertEqual(mock_session.get.call_count, 2)

    @patch.object(cerberus_setup, 'http_session')
    def test_get_status_with_application_routes_check_failure(self, mock_session):
        """Test get_status when cerberus returns False (unhealthy)"""
        cerberus_setup.cerberus_enabled = True
        cerberus_setup.cerberus_url = "http://cerberus.example.com"

        mock_response = MagicMock()
        mock_response.content = b"False"
        mock_session.get.return_value = mock_response

        with self.assertRaises(SystemExit) as cm:
            cerberus_setup.get_status(0, 100)

        self.assertEqual(cm.exception.code, 1)

    @patch('krkn.cerberus.setup.get_status')
    def test_publish_kraken_status_healthy_exit_on_failure_false(self, mock_get_status):
        """Test publish_kraken_status when cluster is healthy and exit_on_failure is False"""
        cerberus_setup.exit_on_failure = False
        mock_get_status.return_value = True

        cerberus_setup.publish_kraken_status(0, 100)

        mock_get_status.assert_called_once_with(0, 100)

    @patch('krkn.cerberus.setup.get_status')
    def test_publish_kraken_status_healthy_exit_on_failure_true(self, mock_get_status):
        """Test publish_kraken_status when cluster is healthy and exit_on_failure is True"""
        cerberus_setup.exit_on_failure = True
        mock_get_status.return_value = True

        with self.assertRaises(SystemExit) as cm:
            cerberus_setup.publish_kraken_status(0, 100)

        self.assertEqual(cm.exception.code, 1)
        mock_get_status.assert_called_once_with(0, 100)

    @patch('krkn.cerberus.setup.get_status')
    def test_publish_kraken_status_unhealthy_exit_on_failure_false(self, mock_get_status):
        """Test publish_kraken_status when cluster is unhealthy and exit_on_failure is False"""
        cerberus_setup.exit_on_failure = False
        mock_get_status.return_value = False

        cerberus_setup.publish_kraken_status(0, 100)

        mock_get_status.assert_called_once_with(0, 100)

    @patch('krkn.cerberus.setup.get_status')
    def test_publish_kraken_status_unhealthy_exit_on_failure_true(self, mock_get_status):
        """Test publish_kraken_status when cluster is unhealthy and exit_on_failure is True"""
        cerberus_setup.exit_on_failure = True
        mock_get_status.return_value = False

        with self.assertRaises(SystemExit) as cm:
            cerberus_setup.publish_kraken_status(0, 100)

        self.assertEqual(cm.exception.code, 1)
        mock_get_status.assert_called_once_with(0, 100)

    @patch.object(cerberus_setup, 'http_session')
    def test_application_status_no_failures(self, mock_session):
        """Test application_status when there are no route failures"""
        cerberus_setup.cerberus_url = "http://cerberus.example.com"

        mock_response = MagicMock()
        mock_response.content = json.dumps({
            "history": {
                "failures": []
            }
        }).encode()
        mock_session.get.return_value = mock_response

        status, failed_routes = cerberus_setup.application_status(0, 6000)

        self.assertTrue(status)
        self.assertEqual(failed_routes, set())
        expected_url = "http://cerberus.example.com/history?loopback=100.0"
        mock_session.get.assert_called_once_with(expected_url, timeout=60)

    @patch.object(cerberus_setup, 'http_session')
    def test_application_status_with_route_failures(self, mock_session):
        """Test application_status when there are route failures"""
        cerberus_setup.cerberus_url = "http://cerberus.example.com"

        mock_response = MagicMock()
        mock_response.content = json.dumps({
            "history": {
                "failures": [
                    {"component": "route", "name": "route1"},
                    {"component": "route", "name": "route2"},
                    {"component": "pod", "name": "pod1"},  # Non-route: should be ignored
                    {"component": "route", "name": "route1"},  # Duplicate: deduped by set()
                ]
            }
        }).encode()
        mock_session.get.return_value = mock_response

        status, failed_routes = cerberus_setup.application_status(0, 6000)

        self.assertFalse(status)
        self.assertEqual(failed_routes, {"route1", "route2"})

    @patch.object(cerberus_setup, 'http_session')
    def test_application_status_with_non_route_failures(self, mock_session):
        """Test application_status when there are non-route failures only"""
        cerberus_setup.cerberus_url = "http://cerberus.example.com"

        mock_response = MagicMock()
        mock_response.content = json.dumps({
            "history": {
                "failures": [
                    {"component": "pod", "name": "pod1"},
                    {"component": "node", "name": "node1"},
                ]
            }
        }).encode()
        mock_session.get.return_value = mock_response

        status, failed_routes = cerberus_setup.application_status(0, 6000)

        self.assertTrue(status)
        self.assertEqual(failed_routes, set())

    def test_application_status_no_url_provided(self):
        """Test application_status when cerberus URL is not provided"""
        cerberus_setup.cerberus_url = None

        with self.assertRaises(SystemExit) as cm:
            cerberus_setup.application_status(0, 100)

        self.assertEqual(cm.exception.code, 1)

    @patch.object(cerberus_setup, 'http_session')
    def test_application_status_request_exception(self, mock_session):
        """Test application_status when request raises an exception"""
        cerberus_setup.cerberus_url = "http://cerberus.example.com"

        mock_session.get.side_effect = Exception("Connection error")

        with self.assertRaises(SystemExit) as cm:
            cerberus_setup.application_status(0, 6000)

        self.assertEqual(cm.exception.code, 1)

    @patch.object(cerberus_setup, 'http_session')
    def test_application_status_duration_calculation(self, mock_session):
        """Test application_status correctly calculates duration in minutes"""
        cerberus_setup.cerberus_url = "http://cerberus.example.com"

        mock_response = MagicMock()
        mock_response.content = json.dumps({"history": {"failures": []}}).encode()
        mock_session.get.return_value = mock_response

        cerberus_setup.application_status(0, 300)

        expected_url = "http://cerberus.example.com/history?loopback=5.0"
        mock_session.get.assert_called_once_with(expected_url, timeout=60)

    def test_http_session_is_singleton(self):
        """Test that http_session is a requests.Session and the same object across accesses"""
        session1 = cerberus_setup.http_session
        session2 = cerberus_setup.http_session
        self.assertIsInstance(session1, requests.Session)
        self.assertIs(session1, session2)

    def test_http_session_reused_across_calls(self):
        """Test that application_status reuses the module-level http_session"""
        cerberus_setup.cerberus_url = "http://cerberus.example.com"
        mock_response = MagicMock()
        mock_response.content = json.dumps({"history": {"failures": []}}).encode()
        original_session = cerberus_setup.http_session

        with patch.object(cerberus_setup.http_session, 'get', return_value=mock_response):
            cerberus_setup.application_status(0, 300)
            self.assertIs(cerberus_setup.http_session, original_session)

            cerberus_setup.application_status(0, 600)
            self.assertIs(cerberus_setup.http_session, original_session)

    def test_http_session_atexit_registered(self):
        """Test that http_session.close is registered via atexit for cleanup"""
        import atexit
        # atexit._run_exitfuncs is internal, so verify registration via the module code
        # The atexit handler should have been registered at module import time
        # We verify by checking the atexit registry contains our session's close
        registered = False
        # atexit callbacks are stored internally; verify by re-registering and checking no error
        # Best we can do without poking internals: verify the session is closeable
        session = cerberus_setup.http_session
        self.assertTrue(callable(getattr(session, 'close', None)))
        # Verify atexit module was imported and used in setup.py
        import inspect
        source = inspect.getsource(cerberus_setup)
        self.assertIn('atexit.register(http_session.close)', source)


if __name__ == '__main__':
    unittest.main()