package com.atlassian.bitbucket.internal.mirroring.mirror;

import com.atlassian.bitbucket.mirroring.mirror.IntegrationState;
import com.atlassian.bitbucket.mockito.MockitoUtils;
import com.atlassian.bitbucket.server.ApplicationPropertiesService;
import com.atlassian.scheduler.JobRunner;
import com.atlassian.scheduler.JobRunnerRequest;
import com.atlassian.scheduler.SchedulerService;
import com.atlassian.scheduler.SchedulerServiceException;
import com.atlassian.scheduler.config.JobConfig;
import com.atlassian.scheduler.config.RunMode;
import com.atlassian.scheduler.config.Schedule;
import com.atlassian.util.concurrent.Promise;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;

import java.util.List;

import static com.atlassian.bitbucket.internal.mirroring.mirror.MirrorConstants.*;
import static com.atlassian.bitbucket.internal.mirroring.mirror.UpstreamRegistrationJob.UPSTREAM_REGISTRATION_JOB_ID;
import static com.atlassian.bitbucket.internal.mirroring.mirror.UpstreamRegistrationJob.UPSTREAM_REGISTRATION_JOB_RUNNER_KEY;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.*;

@RunWith(MockitoJUnitRunner.class)
public class UpstreamRegistrationJobTest {
    @Mock
    private JobRunnerRequest request;
    @Mock
    private Promise<Void> promise;
    private ApplicationPropertiesService propertiesService;
    @Mock
    private SchedulerService schedulerService;
    @Mock
    private InternalUpstreamServer upstream;
    @Mock
    private InternalUpstreamService upstreamService;
    private UpstreamRegistrationJob job;

    @Before
    public void setUp() {
        propertiesService = mock(ApplicationPropertiesService.class, MockitoUtils.returnArg(1));
        job = new UpstreamRegistrationJob(upstreamService, propertiesService, schedulerService);

        when(upstream.getState()).thenReturn(IntegrationState.INITIALIZING);
        when(upstreamService.get()).thenReturn(upstream);
        when(upstreamService.register()).thenReturn(promise);
    }
    
    @Test
    public void testDestroyUnregistersJobRunnerButNotJob() throws SchedulerServiceException {
        when(upstreamService.isMirror()).thenReturn(true);

        job.onStart();
        job.onStop();

        verify(schedulerService).registerJobRunner(eq(UPSTREAM_REGISTRATION_JOB_RUNNER_KEY), any(JobRunner.class));
        verify(schedulerService).scheduleJob(eq(UPSTREAM_REGISTRATION_JOB_ID), any(JobConfig.class));
        verify(schedulerService).unregisterJobRunner(eq(UPSTREAM_REGISTRATION_JOB_RUNNER_KEY));
        verifyNoMoreInteractions(schedulerService);
    }

    @Test
    public void testInitDoesNotScheduleJobInDefaultMode() {
        job.onStart();

        verify(upstreamService).isMirror();
        verifyZeroInteractions(schedulerService);
    }

    @Test
    public void testInitSchedulesJobInMirrorMode() throws SchedulerServiceException {
        when(upstreamService.isMirror()).thenReturn(true);

        // initialize the job - this should register the job runner and schedule the job
        job.onStart();

        verify(schedulerService).registerJobRunner(eq(UPSTREAM_REGISTRATION_JOB_RUNNER_KEY),
                any(JobRunner.class));
        ArgumentCaptor<JobConfig> configCaptor = ArgumentCaptor.forClass(JobConfig.class);
        verify(schedulerService).scheduleJob(eq(UPSTREAM_REGISTRATION_JOB_ID), configCaptor.capture());

        JobConfig config = configCaptor.getValue();
        assertEquals(UPSTREAM_REGISTRATION_JOB_RUNNER_KEY, config.getJobRunnerKey());
        assertEquals(RunMode.RUN_ONCE_PER_CLUSTER, config.getRunMode());

        Schedule schedule = config.getSchedule();
        assertNotNull(schedule);
        assertEquals(Schedule.Type.INTERVAL, schedule.getType());
        assertNotNull(schedule.getIntervalScheduleInfo().getFirstRunTime());
        assertEquals(DEFAULT_UPSTREAM_REGO_RETRY_INTERVAL_MS, schedule.getIntervalScheduleInfo().getIntervalInMillis());

        verifyNoMoreInteractions(schedulerService);
    }

    @Test
    public void testRunJobFailureTriggerReschedulesJob() throws Exception {
        when(propertiesService.getPluginProperty(PROPERTY_UPSTREAM_REGO_RETRY_INTERVAL, DEFAULT_UPSTREAM_REGO_RETRY_INTERVAL_MS))
                .thenReturn(1000L);
        when(propertiesService.getPluginProperty(PROPERTY_UPSTREAM_REGO_RETRY_BACKOFF_FACTOR, DEFAULT_UPSTREAM_REGO_RETRY_EXP_FACTOR))
                .thenReturn(10d);
        when(propertiesService.getPluginProperty(PROPERTY_UPSTREAM_REGO_RETRY_MAX, DEFAULT_UPSTREAM_REGO_RETRY_MAX_INTERVAL_MS))
                .thenReturn(30000L);

        doThrow(new IllegalStateException("to trigger rescheduling")).when(promise).claim();

        // run the job twice
        JobRunner runner = initAndGetJobRunner();
        runner.runJob(request);
        runner.runJob(request);

        ArgumentCaptor<JobConfig> configCaptor = ArgumentCaptor.forClass(JobConfig.class);
        verify(schedulerService, times(3)).scheduleJob(eq(UPSTREAM_REGISTRATION_JOB_ID), configCaptor.capture());

        // job should have been scheduled three times, each time with a factor 10 increase in interval, but
        // maxing out at 30s
        List<JobConfig> configs = configCaptor.getAllValues();
        assertEquals(1000L,
                configs.get(0).getSchedule().getIntervalScheduleInfo().getIntervalInMillis());
        assertEquals(10000L,
                configs.get(1).getSchedule().getIntervalScheduleInfo().getIntervalInMillis());
        assertEquals(30000L,
                configs.get(2).getSchedule().getIntervalScheduleInfo().getIntervalInMillis());
    }

    @Test
    public void testRunJobSuccessUnschedulesJob() throws SchedulerServiceException {
        JobRunner runner = initAndGetJobRunner();

        runner.runJob(request);

        verify(upstreamService).register();
        verify(schedulerService).unscheduleJob(UPSTREAM_REGISTRATION_JOB_ID);
    }

    @Test
    public void testRunJobUnschedulesJobWhenAlreadyRegistered() throws SchedulerServiceException {
        JobRunner runner = initAndGetJobRunner();

        when(upstream.getState()).thenReturn(IntegrationState.INSTALLED);
        runner.runJob(request);

        verify(upstreamService, never()).register();
        verify(schedulerService).unscheduleJob(UPSTREAM_REGISTRATION_JOB_ID);
    }

    private JobRunner initAndGetJobRunner() throws SchedulerServiceException {
        when(upstreamService.isMirror()).thenReturn(true);

        job.onStart();
        ArgumentCaptor<JobRunner> jobRunnerCaptor = ArgumentCaptor.forClass(JobRunner.class);
        verify(schedulerService).registerJobRunner(eq(UPSTREAM_REGISTRATION_JOB_RUNNER_KEY),
                jobRunnerCaptor.capture());

        return jobRunnerCaptor.getValue();
    }
}