# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.

import collections
import secrets
import textwrap
import threading
from concurrent.futures import ThreadPoolExecutor
from datetime import timedelta
from unittest.mock import call, patch
from freezegun import freeze_time

import odoo
from odoo import api, fields
from odoo.tests.common import BaseCase, TransactionCase, RecordCapturer, get_db_name, tagged
from odoo.tools import mute_logger


class CronMixinCase:
    def capture_triggers(self, cron_id=None):
        """
        Get a context manager to get all cron triggers created during
        the context lifetime. While in the context, it exposes the
        triggers created so far from the beginning of the context. When
        the context exits, it doesn't capture new triggers anymore.

        The triggers are accessible on the `records` attribute of the
        returned object.

        :param cron_id: An optional cron record id (int) or xmlid (str)
                        to only capture triggers for that cron.
        """
        if isinstance(cron_id, str):  # xmlid case
            cron_id = self.env.ref(cron_id).id

        return RecordCapturer(
            model=self.env['ir.cron.trigger'].sudo(),
            domain=[('cron_id', '=', cron_id)] if cron_id else []
        )

    @classmethod
    def _get_cron_data(cls, env, priority=5):
        unique = secrets.token_urlsafe(8)
        return {
            'name': f'Dummy cron for TestIrCron {unique}',
            'state': 'code',
            'code': '',
            'model_id': env.ref('base.model_res_partner').id,
            'model_name': 'res.partner',
            'user_id': env.uid,
            'active': True,
            'interval_number': 1,
            'interval_type': 'days',
            'numbercall': -1,
            'doall': False,
            'nextcall': fields.Datetime.now() + timedelta(hours=1),
            'lastcall': False,
            'priority': priority,
        }

    @classmethod
    def _get_partner_data(cls, env):
        unique = secrets.token_urlsafe(8)
        return {'name': f'Dummy partner for TestIrCron {unique}'}


class TestIrCron(TransactionCase, CronMixinCase):

    @classmethod
    def setUpClass(cls):
        super().setUpClass()

        freezer = freeze_time(cls.cr.now())
        cls.frozen_datetime = freezer.start()
        cls.addClassCleanup(freezer.stop)

        cls.cron = cls.env['ir.cron'].create(cls._get_cron_data(cls.env))
        cls.partner = cls.env['res.partner'].create(cls._get_partner_data(cls.env))

    def setUp(self):
        self.partner.write(self._get_partner_data(self.env))
        self.cron.write(self._get_cron_data(self.env))
        self.env['ir.cron.trigger'].search(
            [('cron_id', '=', self.cron.id)]
        ).unlink()

    def test_cron_direct_trigger(self):
        self.cron.code = textwrap.dedent(f"""\
            model.search(
                [("id", "=", {self.partner.id})]
            ).write(
                {{"name": "You have been CRONWNED"}}
            )
        """)

        self.cron.method_direct_trigger()

        self.assertEqual(self.cron.lastcall, fields.Datetime.now())
        self.assertEqual(self.partner.name, 'You have been CRONWNED')

    def test_cron_no_job_ready(self):
        self.cron.nextcall = fields.Datetime.now() + timedelta(days=1)
        self.cron.flush_recordset()

        ready_jobs = self.registry['ir.cron']._get_all_ready_jobs(self.cr)
        self.assertNotIn(self.cron.id, [job['id'] for job in ready_jobs])

    def test_cron_ready_by_nextcall(self):
        self.cron.nextcall = fields.Datetime.now()
        self.cron.flush_recordset()

        ready_jobs = self.registry['ir.cron']._get_all_ready_jobs(self.cr)
        self.assertIn(self.cron.id, [job['id'] for job in ready_jobs])

    def test_cron_ready_by_trigger(self):
        self.cron._trigger()
        self.env['ir.cron.trigger'].flush_model()

        ready_jobs = self.registry['ir.cron']._get_all_ready_jobs(self.cr)
        self.assertIn(self.cron.id, [job['id'] for job in ready_jobs])

    def test_cron_unactive_never_ready(self):
        self.cron.active = False
        self.cron.nextcall = fields.Datetime.now()
        self.cron._trigger()
        self.cron.flush_recordset()
        self.env['ir.cron.trigger'].flush_model()

        ready_jobs = self.registry['ir.cron']._get_all_ready_jobs(self.cr)
        self.assertNotIn(self.cron.id, [job['id'] for job in ready_jobs])

    def test_cron_numbercall0_never_ready(self):
        self.cron.numbercall = 0
        self.cron.nextcall = fields.Datetime.now()
        self.cron._trigger()
        self.cron.flush_recordset()
        self.env['ir.cron.trigger'].flush_model()

        ready_jobs = self.registry['ir.cron']._get_all_ready_jobs(self.cr)
        self.assertNotIn(self.cron.id, [job['id'] for job in ready_jobs])

    def test_cron_ready_jobs_order(self):
        cron_avg = self.cron.copy()
        cron_avg.priority = 5  # average priority

        cron_high = self.cron.copy()
        cron_high.priority = 0  # highest priority

        cron_low = self.cron.copy()
        cron_low.priority = 10  # lowest priority

        crons = cron_high | cron_avg | cron_low  # order is important
        crons.write({'nextcall': fields.Datetime.now()})
        crons.flush_recordset()
        ready_jobs = self.registry['ir.cron']._get_all_ready_jobs(self.cr)

        self.assertEqual(
            [job['id'] for job in ready_jobs if job['id'] in crons._ids],
            list(crons._ids),
        )

    def test_cron_skip_unactive_triggers(self):
        # Situation: an admin disable the cron and another user triggers
        # the cron to be executed *now*, the cron shouldn't be ready and
        # the trigger should not be stored.

        self.cron.active = False
        self.cron.nextcall = fields.Datetime.now() + timedelta(days=2)
        self.cron.flush_recordset()
        with self.capture_triggers() as capture:
            self.cron._trigger()

        ready_jobs = self.registry['ir.cron']._get_all_ready_jobs(self.cr)
        self.assertNotIn(self.cron.id, [job['id'] for job in ready_jobs],
            "the cron shouldn't be ready")
        self.assertFalse(capture.records, "trigger should has been skipped")

    def test_cron_keep_future_triggers(self):
        # Situation: yesterday an admin disabled the cron, while the
        # cron was disabled, another user triggered it to run today.
        # In case the cron as been re-enabled before "today", it should
        # run.

        # go yesterday
        self.frozen_datetime.tick(delta=timedelta(days=-1))

        # admin disable the cron
        self.cron.active = False
        self.cron.nextcall = fields.Datetime.now() + timedelta(days=10)
        self.cron.flush_recordset()

        # user triggers the cron to run *tomorrow of yesterday (=today)
        with self.capture_triggers() as capture:
            self.cron._trigger(at=fields.Datetime.now() + timedelta(days=1))

        # admin re-enable the cron
        self.cron.active = True
        self.cron.flush_recordset()

        # go today, check the cron should run
        self.frozen_datetime.tick(delta=timedelta(days=1))
        ready_jobs = self.registry['ir.cron']._get_all_ready_jobs(self.cr)
        self.assertIn(self.cron.id, [job['id'] for job in ready_jobs],
            "cron should be ready")
        self.assertTrue(capture.records, "trigger should has been kept")

    def test_cron_process_job(self):

        Setup = collections.namedtuple('Setup', ['doall', 'numbercall', 'missedcall', 'trigger'])
        Expect = collections.namedtuple('Expect', ['call_count', 'call_left', 'active'])

        matrix = [
            (Setup(doall=False, numbercall=-1, missedcall=2, trigger=False),
             Expect(call_count=1, call_left=-1, active=True)),
            (Setup(doall=True, numbercall=-1, missedcall=2, trigger=False),
             Expect(call_count=2, call_left=-1, active=True)),
            (Setup(doall=False, numbercall=3, missedcall=2, trigger=False),
             Expect(call_count=1, call_left=2, active=True)),
            (Setup(doall=True, numbercall=3, missedcall=2, trigger=False),
             Expect(call_count=2, call_left=1, active=True)),
            (Setup(doall=True, numbercall=3, missedcall=4, trigger=False),
             Expect(call_count=3, call_left=0, active=False)),
            (Setup(doall=True, numbercall=3, missedcall=0, trigger=True),
             Expect(call_count=1, call_left=2, active=True)),
        ]

        for setup, expect in matrix:
            with self.subTest(setup=setup, expect=expect):
                self.cron.write({
                    'active': True,
                    'doall': setup.doall,
                    'numbercall': setup.numbercall,
                    'nextcall': fields.Datetime.now() - timedelta(days=setup.missedcall - 1),
                })
                with self.capture_triggers(self.cron.id) as capture:
                    if setup.trigger:
                        self.cron._trigger()

                self.cron.flush_recordset()
                capture.records.flush_recordset()
                self.registry.enter_test_mode(self.cr)
                try:
                    with patch.object(self.registry['ir.cron'], '_callback') as callback:
                        self.registry['ir.cron']._process_job(
                            self.registry.db_name,
                            self.registry.cursor(),
                            self.cron.read(load=None)[0]
                        )
                finally:
                    self.registry.leave_test_mode()
                self.cron.invalidate_recordset()
                capture.records.invalidate_recordset()

                self.assertEqual(callback.call_count, expect.call_count)
                self.assertEqual(self.cron.numbercall, expect.call_left)
                self.assertEqual(self.cron.active, expect.active)
                self.assertEqual(self.cron.lastcall, fields.Datetime.now())
                self.assertEqual(self.cron.nextcall, fields.Datetime.now() + timedelta(days=1))
                self.assertEqual(self.env['ir.cron.trigger'].search_count([
                    ('cron_id', '=', self.cron.id),
                    ('call_at', '<=', fields.Datetime.now())]
                ), 0)


@tagged('-standard', '-at_install', 'post_install', 'database_breaking')
class TestIrCronConcurrent(BaseCase, CronMixinCase):

    @classmethod
    def setUpClass(cls):
        super().setUpClass()

        # Keep a reference on the real cron methods, those without patch
        cls.registry = odoo.registry(get_db_name())
        cls.cron_process_job = cls.registry['ir.cron']._process_job
        cls.cron_process_jobs = cls.registry['ir.cron']._process_jobs
        cls.cron_get_all_ready_jobs = cls.registry['ir.cron']._get_all_ready_jobs
        cls.cron_acquire_one_job = cls.registry['ir.cron']._acquire_one_job
        cls.cron_callback = cls.registry['ir.cron']._callback

    def setUp(self):
        super().setUp()

        with self.registry.cursor() as cr:
            env = api.Environment(cr, odoo.SUPERUSER_ID, {})
            env['ir.cron'].search([]).unlink()
            env['ir.cron.trigger'].search([]).unlink()

            self.cron1_data = env['ir.cron'].create(self._get_cron_data(env, priority=1)).read(load=None)[0]
            self.cron2_data = env['ir.cron'].create(self._get_cron_data(env, priority=2)).read(load=None)[0]
            self.partner_data = env['res.partner'].create(self._get_partner_data(env)).read(load=None)[0]
            self.cron_ids = [self.cron1_data['id'], self.cron2_data['id']]

    def test_cron_concurrency_1(self):
        """
        Two cron threads "th1" and "th2" wake up at the same time and
        see two jobs "job1" and "job2" that are ready (setup).

        Th1 acquire job1, before it can process and release its job, th2
        acquire a job too (setup). Th2 shouldn't be able to acquire job1
        as another thread is processing it, it should skips job1 and
        should acquire job2 instead (test). Both thread then process
        their job, update its `nextcall` and release it (setup).

        All the threads update and release their job before any thread
        attempt to acquire another job. (setup)

        The two thread each attempt to acquire a new job (setup), they
        should both fail to acquire any as each job's nextcall is in the
        future* (test).

        *actually, in their own transaction, the other job's nextcall is
        still "in the past" but any attempt to use that information
        would result in a serialization error. This tests ensure that
        that serialization error is correctly handled and ignored.
        """
        lock = threading.Lock()
        barrier = threading.Barrier(2)

        ###
        # Setup
        ###

        # Watchdog, if a thread was waiting at the barrier when the
        # other exited, it receives a BrokenBarrierError and exits too.
        def process_jobs(*args, **kwargs):
            try:
                self.cron_process_jobs(*args, **kwargs)
            finally:
                barrier.reset()

        # The two threads get the same list of jobs
        def get_all_ready_jobs(*args, **kwargs):
            jobs = self.cron_get_all_ready_jobs(*args, **kwargs)
            barrier.wait()
            return jobs

        # When a thread acquire a job, it processes it till the end
        # before another thread can acquire one.
        def acquire_one_job(*args, **kwargs):
            lock.acquire(timeout=1)
            try:
                with mute_logger('odoo.sql_db'):
                    job = self.cron_acquire_one_job(*args, **kwargs)
            except Exception:
                lock.release()
                raise
            if not job:
                lock.release()
            return job

        # When a thread is done processing its job, it waits for the
        # other thread to catch up.
        def process_job(*args, **kwargs):
            try:
                return_value = self.cron_process_job(*args, **kwargs)
            finally:
                lock.release()
            barrier.wait(timeout=1)
            return return_value

        # Set 2 jobs ready, process them in 2 different threads.
        with self.registry.cursor() as cr:
            env = api.Environment(cr, odoo.SUPERUSER_ID, {})
            env['ir.cron'].browse(self.cron_ids).write({
                'nextcall': fields.Datetime.now() - timedelta(hours=1)
            })

        ###
        # Run
        ###
        with patch.object(self.registry['ir.cron'], '_process_jobs', process_jobs), \
             patch.object(self.registry['ir.cron'], '_get_all_ready_jobs', get_all_ready_jobs), \
             patch.object(self.registry['ir.cron'], '_acquire_one_job', acquire_one_job), \
             patch.object(self.registry['ir.cron'], '_process_job', process_job), \
             patch.object(self.registry['ir.cron'], '_callback') as callback, \
             ThreadPoolExecutor(max_workers=2) as executor:
            fut1 = executor.submit(self.registry['ir.cron']._process_jobs, self.registry.db_name)
            fut2 = executor.submit(self.registry['ir.cron']._process_jobs, self.registry.db_name)
            fut1.result(timeout=2)
            fut2.result(timeout=2)

        ###
        # Validation
        ###

        self.assertEqual(len(callback.call_args_list), 2, 'Two jobs must have been processed.')
        self.assertEqual(callback.call_args_list, [
            call(
                self.cron1_data['name'],
                self.cron1_data['ir_actions_server_id'],
                self.cron1_data['id'],
            ),
            call(
                self.cron2_data['name'],
                self.cron2_data['ir_actions_server_id'],
                self.cron2_data['id'],
            ),
        ])
