Statistics
| Branch: | Tag: | Revision:

arvados / sdk / python / tests / test_cache.py @ 76d9365a

History | View | Annotate | Download (2.79 KB)

1
from __future__ import print_function
2
from __future__ import absolute_import
3

    
4
from builtins import str
5
from builtins import range
6
import hashlib
7
import mock
8
import os
9
import random
10
import shutil
11
import sys
12
import tempfile
13
import threading
14
import unittest
15

    
16
import arvados
17
import arvados.cache
18
from . import run_test_server
19

    
20

    
21
def _random(n):
22
    return bytearray(random.getrandbits(8) for _ in range(n))
23

    
24

    
25
class CacheTestThread(threading.Thread):
26
    def __init__(self, dir):
27
        super(CacheTestThread, self).__init__()
28
        self._dir = dir
29

    
30
    def run(self):
31
        c = arvados.cache.SafeHTTPCache(self._dir)
32
        url = 'http://example.com/foo'
33
        self.ok = True
34
        for x in range(16):
35
            try:
36
                data_in = _random(128)
37
                data_in = bytes(hashlib.md5(data_in).hexdigest()) + bytes("\n") + bytes(data_in)
38
                c.set(url, data_in)
39
                data_out = c.get(url)
40
                digest, _, content = data_out.partition("\n")
41
                if digest != bytes(hashlib.md5(content).hexdigest()):
42
                    self.ok = False
43
            finally:
44
                pass
45
            #except Exception as err:
46
            #    self.ok = False
47
            #    print("cache failed: {}: {}".format(type(err), err), file=sys.stderr)
48

    
49

    
50
class CacheTest(unittest.TestCase):
51
    def setUp(self):
52
        self._dir = tempfile.mkdtemp()
53

    
54
    def tearDown(self):
55
        shutil.rmtree(self._dir)
56

    
57
    def test_cache_create_error(self):
58
        _, filename = tempfile.mkstemp()
59
        home_was = os.environ['HOME']
60
        os.environ['HOME'] = filename
61
        try:
62
            c = arvados.http_cache('test')
63
            self.assertEqual(None, c)
64
        finally:
65
            os.environ['HOME'] = home_was
66
            os.unlink(filename)
67

    
68
    def test_cache_crud(self):
69
        c = arvados.cache.SafeHTTPCache(self._dir, max_age=0)
70
        url = 'https://example.com/foo?bar=baz'
71
        data1 = _random(256)
72
        data2 = _random(128)
73
        self.assertEqual(None, c.get(url))
74
        c.delete(url)
75
        c.set(url, data1)
76
        self.assertEqual(data1, c.get(url))
77
        c.delete(url)
78
        self.assertEqual(None, c.get(url))
79
        c.set(url, data1)
80
        c.set(url, data2)
81
        self.assertEqual(data2, c.get(url))
82

    
83
    def test_cache_threads(self):
84
        threads = []
85
        for _ in range(64):
86
            t = CacheTestThread(dir=self._dir)
87
            t.start()
88
            threads.append(t)
89
        for t in threads:
90
            t.join()
91
            self.assertTrue(t.ok)
92

    
93

    
94
class CacheIntegrationTest(run_test_server.TestCaseWithServers):
95
    MAIN_SERVER = {}
96

    
97
    def test_cache_used_by_default_client(self):
98
        with mock.patch('arvados.cache.SafeHTTPCache.get') as getter:
99
            arvados.api('v1')._rootDesc.get('foobar')
100
            getter.assert_called()