# Copyright (C) The Arvados Authors. All rights reserved.
#
# SPDX-License-Identifier: Apache-2.0

from __future__ import print_function
from __future__ import absolute_import

from builtins import str
from builtins import range
import hashlib
import mock
import os
import random
import shutil
import sys
import tempfile
import threading
import unittest

import arvados
import arvados.cache
from . import run_test_server


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


class CacheTestThread(threading.Thread):
    def __init__(self, dir):
        super(CacheTestThread, self).__init__()
        self._dir = dir

    def run(self):
        c = arvados.cache.SafeHTTPCache(self._dir)
        url = 'http://example.com/foo'
        self.ok = True
        for x in range(16):
            try:
                data_in = _random(128)
                data_in = hashlib.md5(data_in).hexdigest().encode() + b"\n" + data_in
                c.set(url, data_in)
                data_out = c.get(url)
                digest, _, content = data_out.partition(b"\n")
                if digest != hashlib.md5(content).hexdigest().encode():
                    self.ok = False
            except Exception as err:
                self.ok = False
                print("cache failed: {}: {}".format(type(err), err), file=sys.stderr)
                raise


class CacheTest(unittest.TestCase):
    def setUp(self):
        self._dir = tempfile.mkdtemp()

    def tearDown(self):
        shutil.rmtree(self._dir)

    def test_cache_create_error(self):
        _, filename = tempfile.mkstemp()
        home_was = os.environ['HOME']
        os.environ['HOME'] = filename
        try:
            c = arvados.http_cache('test')
            self.assertEqual(None, c)
        finally:
            os.environ['HOME'] = home_was
            os.unlink(filename)

    def test_cache_crud(self):
        c = arvados.cache.SafeHTTPCache(self._dir, max_age=0)
        url = 'https://example.com/foo?bar=baz'
        data1 = _random(256)
        data2 = _random(128)
        self.assertEqual(None, c.get(url))
        c.delete(url)
        c.set(url, data1)
        self.assertEqual(data1, c.get(url))
        c.delete(url)
        self.assertEqual(None, c.get(url))
        c.set(url, data1)
        c.set(url, data2)
        self.assertEqual(data2, c.get(url))

    def test_cache_threads(self):
        threads = []
        for _ in range(64):
            t = CacheTestThread(dir=self._dir)
            t.start()
            threads.append(t)
        for t in threads:
            t.join()
            self.assertTrue(t.ok)


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

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