<commit_before># -*- coding: utf-8 -*-
#
# Copyright (C) 2009 <NAME>
# All rights reserved.
#
# This software is licensed as described in the file COPYING, which
# you should have received as part of this distribution.

import doctest
import socket
import time
import unittest

from couchdb import http, util
from couchdb.tests import testutil


class SessionTestCase(testutil.TempDatabaseMixin, unittest.TestCase):

    def test_timeout(self):
        dbname, db = self.temp_db()
        timeout = 1
        session = http.Session(timeout=timeout)
        start = time.time()
        status, headers, body = session.request('GET', db.resource.url + '/_changes?feed=longpoll&since=1000&timeout=%s' % (timeout*2*1000,))
        self.assertRaises(socket.timeout, body.read)
        self.assertTrue(time.time() - start < timeout * 1.3)


class ResponseBodyTestCase(unittest.TestCase):
    def test_close(self):
        class TestStream(util.StringIO):
            def isclosed(self):
                return len(self.getvalue()) == self.tell()

        class Counter(object):
            def __init__(self):
                self.value = 0

            def __call__(self):
                self.value += 1

        counter = Counter()

        response = http.ResponseBody(TestStream('foobar'), counter)

        response.read(10) # read more than stream has. close() is called
        response.read() # steam ended. another close() call

        self.assertEqual(counter.value, 1)

    def test_double_iteration_over_same_response_body(self):
        class TestHttpResp(object):
            msg = {'transfer-encoding': 'chunked'}
            def __init__(self, fp):
                self.fp = fp

            def isclosed(self):
                return len(self.fp.getvalue()) == self.fp.tell()

        data = 'foobarbaz'
        data = '\n'.join([hex(len(data))[2:], data])
        response = http.ResponseBody(TestHttpResp(util.StringIO(data)),
                                     lambda *a, **k: None)
        self.assertEqual(list(response.iterchunks()), ['foobarbaz'])
        self.assertEqual(list(response.iterchunks()), [])


class CacheTestCase(testutil.TempDatabaseMixin, unittest.TestCase):

    def test_remove_miss(self):
        """Check that a cache remove miss is handled gracefully."""
        url = 'http://localhost:5984/foo'
        cache = http.Cache()
        cache.put(url, (None, None, None))
        cache.remove(url)
        cache.remove(url)


def suite():
    suite = unittest.TestSuite()
    suite.addTest(doctest.DocTestSuite(http))
    suite.addTest(unittest.makeSuite(SessionTestCase, 'test'))
    suite.addTest(unittest.makeSuite(ResponseBodyTestCase, 'test'))
    suite.addTest(unittest.makeSuite(CacheTestCase, 'test'))
    return suite


if __name__ == '__main__':
    unittest.main(defaultTest='suite')
<commit_msg>Python 3: HTTP protocol bits are represented as bytes.<commit_after># -*- coding: utf-8 -*-
#
# Copyright (C) 2009 <NAME>
# All rights reserved.
#
# This software is licensed as described in the file COPYING, which
# you should have received as part of this distribution.

import doctest
import socket
import time
import unittest

from couchdb import http, util
from couchdb.tests import testutil


class SessionTestCase(testutil.TempDatabaseMixin, unittest.TestCase):

    def test_timeout(self):
        dbname, db = self.temp_db()
        timeout = 1
        session = http.Session(timeout=timeout)
        start = time.time()
        status, headers, body = session.request('GET', db.resource.url + '/_changes?feed=longpoll&since=1000&timeout=%s' % (timeout*2*1000,))
        self.assertRaises(socket.timeout, body.read)
        self.assertTrue(time.time() - start < timeout * 1.3)


class ResponseBodyTestCase(unittest.TestCase):
    def test_close(self):
        class TestStream(util.StringIO):
            def isclosed(self):
                return len(self.getvalue()) == self.tell()

        class Counter(object):
            def __init__(self):
                self.value = 0

            def __call__(self):
                self.value += 1

        counter = Counter()

        response = http.ResponseBody(TestStream(b'foobar'), counter)

        response.read(10) # read more than stream has. close() is called
        response.read() # steam ended. another close() call

        self.assertEqual(counter.value, 1)

    def test_double_iteration_over_same_response_body(self):
        class TestHttpResp(object):
            msg = {'transfer-encoding': 'chunked'}
            def __init__(self, fp):
                self.fp = fp

            def isclosed(self):
                return len(self.fp.getvalue()) == self.fp.tell()

        data = b'foobarbaz'
        data = b'\n'.join([hex(len(data))[2:].encode('utf-8'), data])
        response = http.ResponseBody(TestHttpResp(util.StringIO(data)),
                                     lambda *a, **k: None)
        self.assertEqual(list(response.iterchunks()), [b'foobarbaz'])
        self.assertEqual(list(response.iterchunks()), [])


class CacheTestCase(testutil.TempDatabaseMixin, unittest.TestCase):

    def test_remove_miss(self):
        """Check that a cache remove miss is handled gracefully."""
        url = 'http://localhost:5984/foo'
        cache = http.Cache()
        cache.put(url, (None, None, None))
        cache.remove(url)
        cache.remove(url)


def suite():
    suite = unittest.TestSuite()
    suite.addTest(doctest.DocTestSuite(http))
    suite.addTest(unittest.makeSuite(SessionTestCase, 'test'))
    suite.addTest(unittest.makeSuite(ResponseBodyTestCase, 'test'))
    suite.addTest(unittest.makeSuite(CacheTestCase, 'test'))
    return suite


if __name__ == '__main__':
    unittest.main(defaultTest='suite')

{'fraction_non_alphanumeric': 0.12371850564726325, 'fraction_numerical': 0.009730668983492615, 'mean_word_length': 2.7571801566579635, 'pattern_counts': {'":': 0, '<': 7, '<?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': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5185910', 'n_tokens_mistral': 1728, 'n_tokens_neox': 1627, 'n_words': 386}