<commit_before>#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Path hack
import os, sys
sys.path.insert(0, os.path.abspath('..'))

import unittest
import time
import json
from collections import defaultdict

import requests

import requests_cache

CACHE_BACKEND = 'sqlite'
CACHE_NAME = 'requests_cache_test'
FAST_SAVE = False



class CacheTestCase(unittest.TestCase):

    def setUp(self):
        requests_cache.configure(CACHE_NAME, backend=CACHE_BACKEND, fast_save=FAST_SAVE)
        requests_cache.clear()

    def test_speedup_and_undo_redo_patch(self):
        delay = 1
        def long_request():
            t = time.time()
            for i in range(5):
                r = requests.get('http://httpbin.org/delay/%s' % delay)
            delta = time.time() - t
            self.assertLess(delta, delay * 3)
        long_request()
        requests_cache.undo_patch()
        t = time.time()
        r = requests.get('http://httpbin.org/delay/%s' % delay)
        delta = time.time() - t
        self.assertGreaterEqual(delta, delay)
        requests_cache.redo_patch()
        long_request()

    def test_expire_cache(self):
        delay = 1
        url = 'http://httpbin.org/delay/%s' % delay
        requests_cache.configure(CACHE_NAME, backend=CACHE_BACKEND, expire_after=0.001)
        t = time.time()
        r = requests.get(url)
        delta = time.time() - t
        self.assertGreaterEqual(delta, delay)
        time.sleep(0.5)
        t = time.time()
        r = requests.get(url)
        delta = time.time() - t
        self.assertGreaterEqual(delta, delay)

    def test_delete_urls(self):
        url = 'http://httpbin.org/redirect/3'
        r = requests.get(url)
        for i in range(1, 4):
            self.assert_(requests_cache.has_url('http://httpbin.org/redirect/%s' % i))
        requests_cache.delete_url(url)
        self.assert_(not requests_cache.has_url(url))

    def test_unregistered_backend(self):
        with self.assertRaises(ValueError):
            requests_cache.configure(CACHE_NAME, backend='nonexistent')

    def test_async_compatibility(self):
        try:
            from requests import async
        except Exception:
            self.fail('gevent is not installed')
        n = 3
        def long_running():
            t = time.time()
            rs = [async.get('http://httpbin.org/delay/%s' % i) for i in range(n + 1)]
            async.map(rs)
            return time.time() - t
        # cache it
        delta = long_running()
        self.assertGreaterEqual(delta, n)
        # fast from cache
        delta = 0
        for i in range(n):<commit_msg>skip async test if gevent is not installed<commit_after>#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Path hack
import os, sys
sys.path.insert(0, os.path.abspath('..'))

import unittest
import time
import json
from collections import defaultdict

import requests

import requests_cache

CACHE_BACKEND = 'sqlite'
CACHE_NAME = 'requests_cache_test'
FAST_SAVE = False



class CacheTestCase(unittest.TestCase):

    def setUp(self):
        requests_cache.configure(CACHE_NAME, backend=CACHE_BACKEND, fast_save=FAST_SAVE)
        requests_cache.clear()

    def test_speedup_and_undo_redo_patch(self):
        delay = 1
        def long_request():
            t = time.time()
            for i in range(5):
                r = requests.get('http://httpbin.org/delay/%s' % delay)
            delta = time.time() - t
            self.assertLess(delta, delay * 3)
        long_request()
        requests_cache.undo_patch()
        t = time.time()
        r = requests.get('http://httpbin.org/delay/%s' % delay)
        delta = time.time() - t
        self.assertGreaterEqual(delta, delay)
        requests_cache.redo_patch()
        long_request()

    def test_expire_cache(self):
        delay = 1
        url = 'http://httpbin.org/delay/%s' % delay
        requests_cache.configure(CACHE_NAME, backend=CACHE_BACKEND, expire_after=0.001)
        t = time.time()
        r = requests.get(url)
        delta = time.time() - t
        self.assertGreaterEqual(delta, delay)
        time.sleep(0.5)
        t = time.time()
        r = requests.get(url)
        delta = time.time() - t
        self.assertGreaterEqual(delta, delay)

    def test_delete_urls(self):
        url = 'http://httpbin.org/redirect/3'
        r = requests.get(url)
        for i in range(1, 4):
            self.assert_(requests_cache.has_url('http://httpbin.org/redirect/%s' % i))
        requests_cache.delete_url(url)
        self.assert_(not requests_cache.has_url(url))

    def test_unregistered_backend(self):
        with self.assertRaises(ValueError):
            requests_cache.configure(CACHE_NAME, backend='nonexistent')

    def test_async_compatibility(self):
        try:
            from requests import async
        except Exception:
            self.skipTest('gevent is not installed')
        n = 3
        def long_running():
            t = time.time()
            rs = [async.get('http://httpbin.org/delay/%s' % i) for i in range(n + 1)]
            async.map(rs)
            return time.time() - t
        # cache it
        delta = long_running()
        self.assertGreaterEqual(delta, n)
        # fast from cache
        delta = 0
        for i in range(n):
{'fraction_non_alphanumeric': 0.11686586985391766, 'fraction_numerical': 0.006829823562891292, 'mean_word_length': 2.2068126520681264, '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': '310284', 'n_tokens_mistral': 1652, 'n_tokens_neox': 1613, 'n_words': 348}