<commit_before>package test.jobs

import kotlinx.coroutines.runBlocking
import org.junit.Assert
import org.junit.Test
import slatekit.common.Status
import slatekit.core.queues.InMemoryQueue
import slatekit.core.queues.QueueStringConverter
import slatekit.core.queues.WrappedAsyncQueue
import slatekit.integration.jobs.JobQueue
import slatekit.jobs.*


class Job_Manage_Queue_Tests : JobTestSupport {

    fun sampleQueue():Queue{
        val source = InMemoryQueue<String>("q1", QueueStringConverter())
        val queue = JobQueue(source.name, Priority.Mid, WrappedAsyncQueue(source))
        return queue
    }


    @Test
    fun can_start_all_workers() {
        runBlocking {
            val queue = sampleQueue()
            (1..10).forEach { queue.send(it.toString()) }
            val manager = run(2, queue, Action.Start)
            runBlocking {
                val worker1 = manager.workers.all.first()
                manager.respond() // Start worker1
                ensure(manager.workers, true, 1, 1, 0, worker1.id, Status.Running, 4, Action.Process, 0)

                val worker2 = manager.workers.all.last()
                manager.respond() // Start worker2val worker = manager.workers.all.first()
                ensure(manager.workers, true, 1, 1, 0, worker2.id, Status.Running, 5, Action.Process, 0)
            }
        }
    }


    @Test
    fun can_pause_worker() {
        runBlocking {
            val queue = sampleQueue()
            (1..10).forEach { queue.send(it.toString()) }
            val manager = run(2, queue, Action.Start)
            runBlocking {
                val worker1 = manager.workers.all.first()
                val worker2 = manager.workers.all.last()
                manager.request(Action.Pause, worker2.id, "test")
                manager.respond() // Start worker1
                manager.respond() // Start worker2
                manager.respond() // Pause worker1
                Assert.assertEquals(Status.Running, worker1.status())
                Assert.assertEquals(Status.Paused, worker2.status())
            }
        }
    }


    @Test
    fun can_pause_worker_due_to_empty_queue() {
        runBlocking {
            val queue = sampleQueue()
            (1..2).forEach { queue.send(it.toString()) }
            val manager = run(2, queue, Action.Start)
            runBlocking {

                manager.respond() // Start worker1
                manager.respond() // Start worker2

                val worker1 = manager.workers.all.first()
                val worker2 = manager.workers.all.last()
                println("worker 1 id: " + worker1.id.id)
                println("worker 2 id: " + worker2.id.id)

                // Backoff at 2, status is
                val worker1BackoffBefore = manager.workers.get(worker1.id)!!.backoffs.curr()
                val worker2BackoffBefore = manager.workers.get(worker2.id)!!.backoffs.curr()
                Assert.assertEquals(2, worker1BackoffBefore)
                Assert.assertEquals(2, worker2BackoffBefore)
                Assert.assertEquals(Status.Running, worker1.status())
                Assert.assertEquals(Status.Running, worker2.status())

                // Backed off at, moved to 4
                manager.respond() // Backoff worker1
                manager.respond() // Backoff worker1
                val worker1Backoff2 = manager.workers.get(worker1.id)!!.backoffs.curr()
                val worker2Backoff2 = manager.workers.get(worker2.id)!!.backoffs.curr()
                Assert.assertEquals(4, worker1Backoff2)
                Assert.assertEquals(4, worker2Backoff2)
                Assert.assertEquals(Status.Paused, worker1.status())
                Assert.assertEquals(Status.Paused, worker2.status())

                // Back off at 4 moved to 8
                (manager.coordinator as MockCoordinatorWithChannel).resume()
                (manager.coordinator as MockCoordinatorWithChannel).resume()
                manager.respond() // Resume worker 1
                manager.respond() // Resume worker 2
                val worker1Backoff3 = manager.workers.get(worker1.id)!!.backoffs.curr()
                val worker2Backoff3 = manager.workers.get(worker2.id)!!.backoffs.curr()
                Assert.assertEquals(8, worker1Backoff3)
                Assert.assertEquals(8, worker2Backoff3)
                Assert.assertEquals(Status.Paused, worker1.status())
                Assert.assertEquals(Status.Paused, worker2.status())
                Assert.assertEquals("Backoff", worker1.note())
                Assert.assertEquals("Backoff", worker2.note())

                // Ensure its back to running
                (3..4).forEach { queue.send(it.toString()) }
                (manager.coordinator as MockCoordinatorWithChannel).resume()
                (manager.coordinator as MockCoordinatorWithChannel).resume()
                manager.respond() // Resume worker 1
                manager.respond() // Resume worker 2
                val worker1Backoff4 = manager.workers.get(worker1.id)!!.backoffs.curr()
                val worker2Backoff4 = manager.workers.get(worker2.id)!!.backoffs.curr()
                Assert.assertEquals(2, worker1Backoff4)
                Assert.assertEquals(2, worker2Backoff4)
                Assert.assertEquals(Status.Running, worker1.status())
                Assert.assertEquals(Status.Running, worker2.status())
            }
        }
    }


    @Test
    fun can_resume_worker() {
        runBlocking {
            val queue = sampleQueue()
            (1..10).forEach { queue.send(it.toString()) }
            val manager = run(2, queue, Action.Start)
            runBlocking {
                val worker1 = manager.workers.all.first()
                val worker2 = manager.workers.all.last()
                manager.request(Action.Pause, worker2.id, "test")
                manager.respond() // Start worker1
                manager.respond() // Start worker2
                manager.respond() // Pause worker1
                manager.respond() // Process worker1
                manager.respond() // Process worker2
                (manager.coordinator as MockCoordinatorWithChannel).resume()
                manager.respond() // Process worker1
                manager.respond() // Resume worker1
                Assert.assertEquals(Status.Running, worker1.status())
                Assert.assertEquals(Status.Running, worker2.status())
            }
        }
    }


    @Test
    fun can_stop_worker() {
        runBlocking {
            val queue = sampleQueue()
            (1..10).forEach { queue.send(it.toString()) }
            val manager = run(2, queue, Action.Start)
            runBlocking {
                val worker1 = manager.workers.all.first()
                val worker2 = manager.workers.all.last()
                manager.request(Action.Stop, worker2.id, "test")
                manager.respond() // Start worker1
                manager.respond() // Start worker2
                manager.respond() // Pause worker1
                Assert.assertEquals(Status.Running, worker1.status())
                Assert.assertEquals(Status.Stopped, worker2.status())
            }
        }
    }
}<commit_msg>FIX: Unit-tests compilation issues<commit_after>package test.jobs

import kotlinx.coroutines.runBlocking
import org.junit.Assert
import org.junit.Test
import slatekit.common.Status
import slatekit.core.queues.InMemoryQueue
import slatekit.core.queues.QueueStringConverter
import slatekit.core.queues.WrappedAsyncQueue
import slatekit.integration.jobs.JobQueue
import slatekit.jobs.*


class Job_Manage_Queue_Tests : JobTestSupport {

    fun sampleQueue():Queue{
        val source = InMemoryQueue<String>("q1", QueueStringConverter())
        val queue = JobQueue(source.name, Priority.Mid, WrappedAsyncQueue(source))
        return queue
    }


    @Test
    fun can_start_all_workers() {
        runBlocking {
            val queue = sampleQueue()
            (1..10).forEach { queue.send(it.toString()) }
            val manager = run(2, queue, Action.Start)
            runBlocking {
                val worker1 = manager.ctx.workers.first()
                manager.respond() // Start worker1
                ensure(manager.workers, true, 1, 1, 0, worker1.id, Status.Running, 4, Action.Process, 0)

                val worker2 = manager.ctx.workers.last()
                manager.respond() // Start worker2val worker = manager.workers.all.first()
                ensure(manager.workers, true, 1, 1, 0, worker2.id, Status.Running, 5, Action.Process, 0)
            }
        }
    }


    @Test
    fun can_pause_worker() {
        runBlocking {
            val queue = sampleQueue()
            (1..10).forEach { queue.send(it.toString()) }
            val manager = run(2, queue, Action.Start)
            runBlocking {
                val worker1 = manager.ctx.workers.first()
                val worker2 = manager.ctx.workers.last()
                manager.request(Action.Pause, worker2.id, "test")
                manager.respond() // Start worker1
                manager.respond() // Start worker2
                manager.respond() // Pause worker1
                Assert.assertEquals(Status.Running, worker1.status())
                Assert.assertEquals(Status.Paused, worker2.status())
            }
        }
    }


    @Test
    fun can_pause_worker_due_to_empty_queue() {
        runBlocking {
            val queue = sampleQueue()
            (1..2).forEach { queue.send(it.toString()) }
            val manager = run(2, queue, Action.Start)
            runBlocking {

                manager.respond() // Start worker1
                manager.respond() // Start worker2

                val worker1 = manager.ctx.workers.first()
                val worker2 = manager.ctx.workers.last()
                println("worker 1 id: " + worker1.id.id)
                println("worker 2 id: " + worker2.id.id)

                // Backoff at 2, status is
                val worker1BackoffBefore = manager.workers.get(worker1.id)!!.backoffs.curr()
                val worker2BackoffBefore = manager.workers.get(worker2.id)!!.backoffs.curr()
                Assert.assertEquals(2, worker1BackoffBefore)
                Assert.assertEquals(2, worker2BackoffBefore)
                Assert.assertEquals(Status.Running, worker1.status())
                Assert.assertEquals(Status.Running, worker2.status())

                // Backed off at, moved to 4
                manager.respond() // Backoff worker1
                manager.respond() // Backoff worker1
                val worker1Backoff2 = manager.workers.get(worker1.id)!!.backoffs.curr()
                val worker2Backoff2 = manager.workers.get(worker2.id)!!.backoffs.curr()
                Assert.assertEquals(4, worker1Backoff2)
                Assert.assertEquals(4, worker2Backoff2)
                Assert.assertEquals(Status.Paused, worker1.status())
                Assert.assertEquals(Status.Paused, worker2.status())

                // Back off at 4 moved to 8
                (manager.coordinator as MockCoordinatorWithChannel).resume()
                (manager.coordinator as MockCoordinatorWithChannel).resume()
                manager.respond() // Resume worker 1
                manager.respond() // Resume worker 2
                val worker1Backoff3 = manager.workers.get(worker1.id)!!.backoffs.curr()
                val worker2Backoff3 = manager.workers.get(worker2.id)!!.backoffs.curr()
                Assert.assertEquals(8, worker1Backoff3)
                Assert.assertEquals(8, worker2Backoff3)
                Assert.assertEquals(Status.Paused, worker1.status())
                Assert.assertEquals(Status.Paused, worker2.status())
                Assert.assertEquals("Backoff", worker1.note())
                Assert.assertEquals("Backoff", worker2.note())

                // Ensure its back to running
                (3..4).forEach { queue.send(it.toString()) }
                (manager.coordinator as MockCoordinatorWithChannel).resume()
                (manager.coordinator as MockCoordinatorWithChannel).resume()
                manager.respond() // Resume worker 1
                manager.respond() // Resume worker 2
                val worker1Backoff4 = manager.workers.get(worker1.id)!!.backoffs.curr()
                val worker2Backoff4 = manager.workers.get(worker2.id)!!.backoffs.curr()
                Assert.assertEquals(2, worker1Backoff4)
                Assert.assertEquals(2, worker2Backoff4)
                Assert.assertEquals(Status.Running, worker1.status())
                Assert.assertEquals(Status.Running, worker2.status())
            }
        }
    }


    @Test
    fun can_resume_worker() {
        runBlocking {
            val queue = sampleQueue()
            (1..10).forEach { queue.send(it.toString()) }
            val manager = run(2, queue, Action.Start)
            runBlocking {
                val worker1 = manager.ctx.workers.first()
                val worker2 = manager.ctx.workers.last()
                manager.request(Action.Pause, worker2.id, "test")
                manager.respond() // Start worker1
                manager.respond() // Start worker2
                manager.respond() // Pause worker1
                manager.respond() // Process worker1
                manager.respond() // Process worker2
                (manager.coordinator as MockCoordinatorWithChannel).resume()
                manager.respond() // Process worker1
                manager.respond() // Resume worker1
                Assert.assertEquals(Status.Running, worker1.status())
                Assert.assertEquals(Status.Running, worker2.status())
            }
        }
    }


    @Test
    fun can_stop_worker() {
        runBlocking {
            val queue = sampleQueue()
            (1..10).forEach { queue.send(it.toString()) }
            val manager = run(2, queue, Action.Start)
            runBlocking {
                val worker1 = manager.ctx.workers.first()
                val worker2 = manager.ctx.workers.last()
                manager.request(Action.Stop, worker2.id, "test")
                manager.respond() // Start worker1
                manager.respond() // Start worker2
                manager.respond() // Pause worker1
                Assert.assertEquals(Status.Running, worker1.status())
                Assert.assertEquals(Status.Stopped, worker2.status())
            }
        }
    }
}
{'fraction_non_alphanumeric': 0.10709597180957646, 'fraction_numerical': 0.01906999239964071, 'mean_word_length': 1.811031268207419, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '20677', 'n_tokens_mistral': 3951, 'n_tokens_neox': 3677, 'n_words': 733}