<commit_before>"""Unit Tests for the pydisque module."""

import unittest
import json
import time
import random
import six
from pydisque.client import Client


class TestDisque(unittest.TestCase):

    """TestCase class for pydisque."""

    def setUp(self):
        """Setup the tests."""
        self.client = Client(['localhost:7711'])
        self.client.connect()

    def test_publish_and_receive(self):
        """Test the most important functions of pydisque."""
        t1 = str(time.time())
        self.client.add_job("test_q", t1, timeout=100)
        jobs = self.client.get_job(['test_q'])
        assert len(jobs) == 1
        for queue_name, job_id, job in jobs:
            assert job == six.b(t1)
            self.client.ack_job(job_id)
        assert len(self.client.get_job(['test_q'], timeout=100)) == 0

    def test_nack(self):
        """Fetch the queue, return a job, check that it's back."""
        t1 = str(time.time())
        self.client.add_job("test_nack_q", str(t1), timeout=100)
        jobs = self.client.get_job(['test_nack_q'])
        # NACK the first read
        assert len(jobs) == 1
        for queue_name, job_id, job in jobs:
            assert len(jobs) == 1
            assert job == six.b(t1)
            self.client.nack_job(job_id)
        # this time ACK it
        jobs = self.client.get_job(['test_nack_q'])
        assert len(jobs) == 1
        for queue_name, job_id, job in jobs:
            assert job == six.b(t1)
            self.client.ack_job(job_id)
        assert len(self.client.get_job(['test_nack_q'], timeout=100)) == 0

    def test_qscan(self):
        """<commit_msg>Bug Hunt.<commit_after>"""
Unit Tests for the pydisque module.

Currently, most of these tests require a fresh instance of
Disque to be valid and pass.
"""

import unittest
import json
import time
import random
import six
from pydisque.client import Client


class TestDisque(unittest.TestCase):

    """TestCase class for pydisque."""

    def setUp(self):
        """Setup the tests."""
        self.client = Client(['localhost:7711'])
        self.client.connect()

    def test_publish_and_receive(self):
        """Test the most important functions of pydisque."""
        t1 = str(time.time())
        self.client.add_job("test_q", t1, timeout=100)
        jobs = self.client.get_job(['test_q'])
        assert len(jobs) == 1
        for queue_name, job_id, job in jobs:
            assert job == six.b(t1)
            self.client.ack_job(job_id)
        assert len(self.client.get_job(['test_q'], timeout=100)) == 0

    def test_nack(self):
        """Fetch the queue, return a job, check that it's back."""
        t1 = str(time.time())
        self.client.add_job("test_nack_q", str(t1), timeout=100)
        jobs = self.client.get_job(['test_nack_q'])
        # NACK the first read
        assert len(jobs) == 1
        for queue_name, job_id, job in jobs:
            assert len(jobs) == 1
            assert job == six.b(t1)
            self.client.nack_job(job_id)
        # this time ACK it
        jobs = self.client.get_job(['test_nack_q'])
        assert len(jobs) == 1
        for queue_name, job_id, job in jobs:
            assert job == six.b(t1)
            self.client.ack_job(job_id)
        assert len(self.client.get_job(['test_nack_q'], timeout=100)) == 0

    def test_qpeek(self):
        """
        Test qpeek
        
        Ran into some problems with an ENQUEUE/DEQUEUE test that
        was using qpeek, checking core functionality of qpeek().
        """
        queuename = "test_qpeek-%d" % random.randint(1000, 1000000)
        job_id = self.client.add_job(queuename, "Peek A Boo")

        peeked = self.client.qpeek(queuename, 1)
        assert peeked[0][1] == job_id

    def test_qscan(self):
        """
{'fraction_non_alphanumeric': 0.1260570824524313, 'fraction_numerical': 0.019027484143763214, 'mean_word_length': 2.38248436103664, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '770043', 'n_tokens_mistral': 1273, 'n_tokens_neox': 1201, 'n_words': 316}