<commit_before>'''
Created on Feb 18, 2015

@author: SK
'''
import unittest
import urllib
from gglsbl3 import protocol
import time
import httpretty
from nose.tools import *
from tests import logger
log = logger.Logger("protocoltest").get()


class BaseProtocolTest(unittest.TestCase):

    def setUp(self):
        self.api_key = "abcdefg"
        self.client = protocol.BaseProtocolClient(api_key=self.api_key, discard_fair_use_policy=False)

    def tearDown(self):
        pass

    def testBaseProtocolClientInit(self):
        assert_equal(self.client.config["url_args"]["key"], "abcdefg")
        assert_false(self.client.discard_fair_use_policy)

    def testSetNextCallTimeOut(self, delay=30):
        self.client.set_next_call_timeout(delay)
        # log.debug("next call timestamp is: "+ str(self.client._next_call_timestamp))
        assert_equal(self.client._next_call_timestamp, int(time.time()) + delay)

    def testGetFairUseDelay(self):
        self.testSetNextCallTimeOut(10)
        delay = self.client.get_fair_use_delay()
        assert_in(delay, [9, 10])
        self.client._error_count = 1
        assert_equal(self.client.get_fair_use_delay(), 60)
        self.client._error_count = 2
        log.debug("delay is: " + str(self.client.get_fair_use_delay()))
        assert_less_equal(self.client.get_fair_use_delay(), 28800)
        self.client._error_count = 20
        assert_less_equal(self.client.get_fair_use_delay(), 28800)

    def testFairUseDelay(self):
        self.client.set_next_call_timeout(2)
        time1 = time.time()
        self.client.fair_use_delay()
        time2 = time.time()
        delta = time2 - time1
        assert_equal(int(round(delta)), 2)

    def testDiscardFairUse(self):
        client = protocol.BaseProtocolClient(api_key=self.api_key, discard_fair_use_policy=True)
        client.set_next_call_timeout(2)
        time1 = time.time()
        client.fair_use_delay()
        time2 = time.time()
        delta = time2 - time1
        assert_equal(int(round(delta)), 0)

    def test_make_url(self):
        url = self.client.make_url("downloads")
        assert_true(url.startswith("https://"))

    def test_api_call(self):
        httpretty.enable()  # enable HTTPretty so that it will monkey patch the socket module
        try:
            url = "https://test.com/download"
            body = b"This is the response"
            httpretty.register_uri(httpretty.POST, url,
                                   body=body)
            payload = "some unicode string"
            result = self.client.api_call(url, payload)
            assert_equal(result, body)
            assert_not_equal(result, body.decode("ascii"))
        except:
            raise
        finally:
            httpretty.disable()
            httpretty.reset()

    def test_api_call_rror(self):
        httpretty.enable()  # enable HTTPretty so that it will monkey patch the socket module
        try:
            url = "https://test.com/download"
            body = b"This is the response"
            httpretty.register_uri(httpretty.POST, url, body=body, status=401)
            result = self.client.api_call(url)
            assert_equal(result, body)
            assert_not_equal(result, body.decode("ascii"))
        except urllib.error.HTTPError:
            pass  # intended
        except:
            raise
        else:
            raise Exception("a exception should have been raised during the test")
        finally:
            httpretty.disable()
            httpretty.reset()


class FullHashProtocolTest(unittest.TestCase):

    def setUp(self):
        self.api_key = "abcdef"
        self.client = protocol.FullHashProtocolClient(self.api_key)

    def tearDown(self):
        pass

    def testSetNextCallTimeOut(self, delay=30):
        self.client.set_next_call_timeout(delay)
        # log.debug("next call timestamp is: "+ str(self.client._next_call_timestamp))
        assert_equal(self.client._next_call_timestamp, int(time.time()) + delay)

    def testGetHashes(self):
        httpretty.enable()  # enable HTTPretty so that it will monkey patch the socket module
        try:
            url = "https://safebrowsing.google.com/safebrowsing/gethash"
            hash_prefix = bytes(bytearray.fromhex('24b24191'))
            body = b'600\ngoog-malware-shavar:32:1:m\n$\xb2A\x91\xaf\xc2\xd5\x8b\xdfh\xc8R\x82Y\x9do\xbb\x84\x92\xf9\xa2h,\x02\xf4j\x8dQy\x1e\r\xff2\n\x08\x02'
            httpretty.register_uri(httpretty.POST, url, body=body)
            result = self.client.getHashes([hash_prefix])
            log.info("got result: {res}".format(res=result))
            # log.debug(httpretty.last_request())
            # log.debug(dir(httpretty.last_request()))
            expected = {'hashes': {b'goog-malware-shavar': [b'$\xb2A\x91\xaf\xc2\xd5\x8b\xdfh\xc8R\x82Y\x9do\xbb\x84\x92\xf9\xa2h,\x02\xf4j\x8dQy\x1e\r\xff']}, 'cache_lifetime': 600, 'metadata': {b'goog-malware-shavar': [2]}} # integer is returned now as metadata instead of unparsed protobuf bytes
            assert_equal(result, expected)
        except:
            raise
        finally:
            httpretty.disable()
            httpretty.reset()

    def testFullHashClient(self):
        res = self.client._parseHashEntry(b'goog-malware-shavar:32:1:m\n$\xb2A\x91\xaf\xc2\xd5\x8b\xdfh\xc8R\x82Y\x9do\xbb\x84\x92\xf9\xa2h,\x02\xf4j\x8dQy\x1e\r\xff2\n\x08\x02')
        assert_equal(len(res), 2)

    def testGetFairUseDelay(self):
        self.testSetNextCallTimeOut(10)
        delay = self.client.get_fair_use_delay()
        assert_in(delay, [9, 10])

    def testDiscardFairUse(self):<commit_msg>added refactored method name in test<commit_after>'''
Created on Feb 18, 2015

@author: SK
'''
import unittest
import urllib
from gglsbl3 import protocol
import time
import httpretty
from nose.tools import *
from tests import logger
log = logger.Logger("protocoltest").get()


class BaseProtocolTest(unittest.TestCase):

    def setUp(self):
        self.api_key = "abcdefg"
        self.client = protocol.BaseProtocolClient(api_key=self.api_key, discard_fair_use_policy=False)

    def tearDown(self):
        pass

    def testBaseProtocolClientInit(self):
        assert_equal(self.client.config["url_args"]["key"], "abcdefg")
        assert_false(self.client.discard_fair_use_policy)

    def testSetNextCallTimeOut(self, delay=30):
        self.client.set_next_call_timeout(delay)
        # log.debug("next call timestamp is: "+ str(self.client._next_call_timestamp))
        assert_equal(self.client._next_call_timestamp, int(time.time()) + delay)

    def testGetFairUseDelay(self):
        self.testSetNextCallTimeOut(10)
        delay = self.client.get_fair_use_delay()
        assert_in(delay, [9, 10])
        self.client._error_count = 1
        assert_equal(self.client.get_fair_use_delay(), 60)
        self.client._error_count = 2
        log.debug("delay is: " + str(self.client.get_fair_use_delay()))
        assert_less_equal(self.client.get_fair_use_delay(), 28800)
        self.client._error_count = 20
        assert_less_equal(self.client.get_fair_use_delay(), 28800)

    def testFairUseDelay(self):
        self.client.set_next_call_timeout(2)
        time1 = time.time()
        self.client.fair_use_delay()
        time2 = time.time()
        delta = time2 - time1
        assert_equal(int(round(delta)), 2)

    def testDiscardFairUse(self):
        client = protocol.BaseProtocolClient(api_key=self.api_key, discard_fair_use_policy=True)
        client.set_next_call_timeout(2)
        time1 = time.time()
        client.fair_use_delay()
        time2 = time.time()
        delta = time2 - time1
        assert_equal(int(round(delta)), 0)

    def test_make_url(self):
        url = self.client.make_url("downloads")
        assert_true(url.startswith("https://"))

    def test_api_call(self):
        httpretty.enable()  # enable HTTPretty so that it will monkey patch the socket module
        try:
            url = "https://test.com/download"
            body = b"This is the response"
            httpretty.register_uri(httpretty.POST, url,
                                   body=body)
            payload = "some unicode string"
            result = self.client.api_call(url, payload)
            assert_equal(result, body)
            assert_not_equal(result, body.decode("ascii"))
        except:
            raise
        finally:
            httpretty.disable()
            httpretty.reset()

    def test_api_call_rror(self):
        httpretty.enable()  # enable HTTPretty so that it will monkey patch the socket module
        try:
            url = "https://test.com/download"
            body = b"This is the response"
            httpretty.register_uri(httpretty.POST, url, body=body, status=401)
            result = self.client.api_call(url)
            assert_equal(result, body)
            assert_not_equal(result, body.decode("ascii"))
        except urllib.error.HTTPError:
            pass  # intended
        except:
            raise
        else:
            raise Exception("a exception should have been raised during the test")
        finally:
            httpretty.disable()
            httpretty.reset()


class FullHashProtocolTest(unittest.TestCase):

    def setUp(self):
        self.api_key = "abcdef"
        self.client = protocol.FullHashProtocolClient(self.api_key)

    def tearDown(self):
        pass

    def testSetNextCallTimeOut(self, delay=30):
        self.client.set_next_call_timeout(delay)
        # log.debug("next call timestamp is: "+ str(self.client._next_call_timestamp))
        assert_equal(self.client._next_call_timestamp, int(time.time()) + delay)

    def testGetHashes(self):
        httpretty.enable()  # enable HTTPretty so that it will monkey patch the socket module
        try:
            url = "https://safebrowsing.google.com/safebrowsing/gethash"
            hash_prefix = bytes(bytearray.fromhex('24b24191'))
            body = b'600\ngoog-malware-shavar:32:1:m\n$\xb2A\x91\xaf\xc2\xd5\x8b\xdfh\xc8R\x82Y\x9do\xbb\x84\x92\xf9\xa2h,\x02\xf4j\x8dQy\x1e\r\xff2\n\x08\x02'
            httpretty.register_uri(httpretty.POST, url, body=body)
            result = self.client.getHashes([hash_prefix])
            log.info("got result: {res}".format(res=result))
            # log.debug(httpretty.last_request())
            # log.debug(dir(httpretty.last_request()))
            expected = {'hashes': {b'goog-malware-shavar': [b'$\xb2A\x91\xaf\xc2\xd5\x8b\xdfh\xc8R\x82Y\x9do\xbb\x84\x92\xf9\xa2h,\x02\xf4j\x8dQy\x1e\r\xff']}, 'cache_lifetime': 600, 'metadata': {b'goog-malware-shavar': [2]}} # integer is returned now as metadata instead of unparsed protobuf bytes
            assert_equal(result, expected)
        except:
            raise
        finally:
            httpretty.disable()
            httpretty.reset()

    def testFullHashClient(self):
        res = self.client._parse_hash_entry(b'goog-malware-shavar:32:1:m\n$\xb2A\x91\xaf\xc2\xd5\x8b\xdfh\xc8R\x82Y\x9do\xbb\x84\x92\xf9\xa2h,\x02\xf4j\x8dQy\x1e\r\xff2\n\x08\x02')
        assert_equal(len(res), 2)

    def testGetFairUseDelay(self):
        self.testSetNextCallTimeOut(10)
        delay = self.client.get_fair_use_delay()
        assert_in(delay, [9, 10])

    def testDiscardFairUse(self):
{'fraction_non_alphanumeric': 0.11487742278077706, 'fraction_numerical': 0.02584299495530578, 'mean_word_length': 2.624118024374599, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 8, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4296464', 'n_tokens_mistral': 3932, 'n_tokens_neox': 3713, 'n_words': 614}