# -*- coding: utf-8 -*-
from datetime import timedelta

from odoo import fields
from odoo.addons.base.models.ir_cron import CompletionStatus
from odoo.tests import TransactionCase, tagged


@tagged('post_install', '-at_install')
class TestCronHealth(TransactionCase):

    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        cls.Log = cls.env['do.cron.health.log']
        cls.IrCron = cls.env['ir.cron']
        # A dedicated test cron with no demo/other logs, so counts are
        # deterministic even when demo data is installed.
        cls.cron = cls.env['ir.cron'].create({
            'name': 'DCH Test Cron',
            'model_id': cls.env['ir.model']._get('res.partner').id,
            'state': 'code',
            'code': 'model.browse([])',
            'interval_number': 1,
            'interval_type': 'days',
            'active': False,
        })
        icp = cls.env['ir.config_parameter'].sudo()
        icp.set_param('do_ops_alert_base.enabled', 'True')
        icp.set_param('do_ops_alert_base.email_enabled', 'False')
        icp.set_param('do_ops_alert_base.webhook_enabled', 'False')
        icp.set_param('do_ops_alert_base.throttle_minutes', '60')
        icp.set_param('do_cron_health.enabled', 'True')
        icp.set_param('do_cron_health.stale_minutes', '1')
        icp.set_param('do_cron_health.critical_failures', '3')
        icp.set_param('do_cron_health.retention_days', '30')

    def _mklog(self, **vals):
        v = {
            'cron_id': self.cron.id,
            'name': self.cron.cron_name,
            'start_datetime': fields.Datetime.now(),
            'state': 'success',
        }
        v.update(vals)
        return self.Log.create(v)

    def test_state_for_status_mapping(self):
        m = self.IrCron._dch_state_for_status
        self.assertEqual(m(CompletionStatus.FULLY_DONE, None), ('success', False))
        self.assertEqual(m(CompletionStatus.PARTIALLY_DONE, 'tb'), ('partial', 'tb'))
        self.assertEqual(m(CompletionStatus.FAILED, 'boom'), ('failed', 'boom'))
        state, err = m(None, None)  # super() raised without a status
        self.assertEqual(state, 'failed')
        self.assertTrue(err)

    def test_computed_health(self):
        self._mklog(state='success', duration=1.0)
        self._mklog(state='failed', error_message='X',
                    start_datetime=fields.Datetime.now() + timedelta(seconds=1))
        self.cron.invalidate_recordset()
        self.assertEqual(self.cron.do_health_log_count, 2)
        self.assertEqual(self.cron.do_last_state, 'failed')
        self.assertEqual(self.cron.do_recent_failure_count, 1)
        self.assertEqual(self.cron.do_health_status, 'warning')

    def test_watchdog_marks_stale(self):
        log = self._mklog(state='running',
                          start_datetime=fields.Datetime.now() - timedelta(minutes=5))
        self.Log._cron_watchdog()
        self.assertEqual(log.state, 'stale')
        self.assertTrue(log.end_datetime)

    def test_watchdog_fires_alert(self):
        self._mklog(state='failed', error_message='kaboom')
        self.cron.invalidate_recordset()
        self.assertEqual(self.cron.do_health_status, 'warning')
        self.Log._dch_run_alerts()
        state = self.env['ops.alert.state'].sudo().search(
            [('key', '=', 'cron_health:%s' % self.cron.id)])
        self.assertTrue(state, "an alert (throttle state row) should have been recorded")
        self.assertEqual(state.last_state, 'warning')

    def test_retention_gc(self):
        old = self._mklog(state='success',
                          start_datetime=fields.Datetime.now() - timedelta(days=40))
        keep = self._mklog(state='success')
        self.Log._gc_cron_health_logs()
        self.assertFalse(old.exists())
        self.assertTrue(keep.exists())
