<commit_before>import unittest
import requests_mock
from dns.resolver import Resolver, NXDOMAIN, Timeout
from requests.exceptions import RequestException
from mock import patch, mock_open, MagicMock
from consul.base import ACLPermissionDenied
from e2j2.tags import base64_tag, consul_tag, file_tag, json_tag, jsonfile_tag, vault_tag, dns_tag, escape_tag
from e2j2.tags import list_tag as list_tag
from e2j2.exceptions import E2j2Exception
from json.decoder import JSONDecodeError

class TestParsers(unittest.TestCase):
    
    def setUp(self):
        pass

    def test_parse_json(self):

        # render json string
        self.assertEqual(json_tag.parse('{"foo": "bar"}'), {'foo': 'bar'})

        # render json string with hash rockets
        self.assertEqual(json_tag.parse('{"foo"=>"bar"}'), {'foo': 'bar'})

        # invalid json
        with self.assertRaisesRegex(E2j2Exception, 'invalid JSON'):
            json_tag.parse('<invalid>')

    def test_parse_json_file(self):
        with patch('builtins.open'):
            with patch('e2j2.tags.jsonfile_tag.json.load', return_value={'foo': 'bar'}):
                self.assertEqual(jsonfile_tag.parse('myfile'), {'foo': 'bar'})

        with patch('builtins.open') as open_mock:
            open_mock.side_effect = IOError()
            with self.assertRaisesRegex(E2j2Exception, 'IOError raised while reading file'):
                jsonfile_tag.parse('myfile')

        with patch('builtins.open'):
            with patch('e2j2.tags.jsonfile_tag.json.load') as jsonload_mock:
                jsonload_mock.side_effect = JSONDecodeError('', '', 0)
                with self.assertRaisesRegex(E2j2Exception, 'invalid JSON'):
                    jsonfile_tag.parse('myfile')

    def test_base64(self):
        self.assertEqual(base64_tag.parse('Zm9vYmFy'), 'foobar')

        # TypeError
        with self.assertRaisesRegex(E2j2Exception, 'invalid BASE64 string'):
            base64_tag.parse(123456)

        # binascii.Error
        with self.assertRaisesRegex(E2j2Exception, 'invalid BASE64 string'):
            base64_tag.parse('123456')

    def test_consul(self):
        config = {'url': 'https://foobar', 'token': '<KEY>'}

        # test setup with config dict
        consul_kv = consul_tag.ConsulKV(config=config)
        self.assertEqual(consul_kv.scheme, 'https')
        self.assertEqual(consul_kv.host, 'foobar')
        self.assertEqual(consul_kv.port, 443)
        self.assertEqual(consul_kv.token, '<KEY>')

        # test setup with default values
        consul_kv = consul_tag.ConsulKV(config={})
        self.assertEqual(consul_kv.scheme, 'http')
        self.assertEqual(consul_kv.host, '127.0.0.1')
        self.assertEqual(consul_kv.port, 8500)
        self.assertEqual(consul_kv.token, None)

        # test get kv
        with patch.object(consul_kv.client.kv, 'get', return_value=(None, 'value')) as get_mock:
            consul_kv.get('foo/bar')
            get_mock.assert_called_with(key='foo/bar', recurse=False)

        # permission denied
        consul_kv.get = MagicMock(side_effect = ACLPermissionDenied)
        with patch('e2j2.tags.consul_tag.ConsulKV', return_value=consul_kv):
            with self.assertRaisesRegex(E2j2Exception, '^access denied connecting to:'):
                consul_tag.parse(config, 'foo/bar')

        # key not found
        consul_kv.get = MagicMock(return_value=[])
        with patch('e2j2.tags.consul_tag.ConsulKV', return_value=consul_kv):
            with self.assertRaisesRegex(E2j2Exception, 'key not found'):
                consul_tag.parse(config, 'foo/bar')

        # get nested consul entry
        value = [{'Key': 'key/folder/subkey', 'Value': 'foobar'}]
        consul_kv.get = MagicMock(return_value=value)
        with patch('e2j2.tags.consul_tag.ConsulKV', return_value=consul_kv):
            self.assertEqual(consul_tag.parse(config, 'key/folder/subkey'), 'foobar')<commit_msg>add test for consul tag<commit_after>import unittest
import requests_mock
from dns.resolver import Resolver, NXDOMAIN, Timeout
from requests.exceptions import RequestException
from mock import patch, mock_open, MagicMock
from consul.base import ACLPermissionDenied
from e2j2.tags import base64_tag, consul_tag, file_tag, json_tag, jsonfile_tag, vault_tag, dns_tag, escape_tag
from e2j2.tags import list_tag as list_tag
from e2j2.exceptions import E2j2Exception
from json.decoder import JSONDecodeError

class TestParsers(unittest.TestCase):
    
    def setUp(self):
        pass

    def test_parse_json(self):

        # render json string
        self.assertEqual(json_tag.parse('{"foo": "bar"}'), {'foo': 'bar'})

        # render json string with hash rockets
        self.assertEqual(json_tag.parse('{"foo"=>"bar"}'), {'foo': 'bar'})

        # invalid json
        with self.assertRaisesRegex(E2j2Exception, 'invalid JSON'):
            json_tag.parse('<invalid>')

    def test_parse_json_file(self):
        with patch('builtins.open'):
            with patch('e2j2.tags.jsonfile_tag.json.load', return_value={'foo': 'bar'}):
                self.assertEqual(jsonfile_tag.parse('myfile'), {'foo': 'bar'})

        with patch('builtins.open') as open_mock:
            open_mock.side_effect = IOError()
            with self.assertRaisesRegex(E2j2Exception, 'IOError raised while reading file'):
                jsonfile_tag.parse('myfile')

        with patch('builtins.open'):
            with patch('e2j2.tags.jsonfile_tag.json.load') as jsonload_mock:
                jsonload_mock.side_effect = JSONDecodeError('', '', 0)
                with self.assertRaisesRegex(E2j2Exception, 'invalid JSON'):
                    jsonfile_tag.parse('myfile')

    def test_base64(self):
        self.assertEqual(base64_tag.parse('Zm9vYmFy'), 'foobar')

        # TypeError
        with self.assertRaisesRegex(E2j2Exception, 'invalid BASE64 string'):
            base64_tag.parse(123456)

        # binascii.Error
        with self.assertRaisesRegex(E2j2Exception, 'invalid BASE64 string'):
            base64_tag.parse('123456')

    def test_consul(self):
        config = {'url': 'https://foobar', 'token': '<KEY>'}

        # test setup with config dict
        consul_kv = consul_tag.ConsulKV(config=config)
        self.assertEqual(consul_kv.scheme, 'https')
        self.assertEqual(consul_kv.host, 'foobar')
        self.assertEqual(consul_kv.port, 443)
        self.assertEqual(consul_kv.token, 'aabbccd<PASSWORD>')

        # test setup with default values
        consul_kv = consul_tag.ConsulKV(config={})
        self.assertEqual(consul_kv.scheme, 'http')
        self.assertEqual(consul_kv.host, '127.0.0.1')
        self.assertEqual(consul_kv.port, 8500)
        self.assertEqual(consul_kv.token, None)

        # test get kv
        with patch.object(consul_kv.client.kv, 'get', return_value=(None, 'value')) as get_mock:
            consul_kv.get('foo/bar')
            get_mock.assert_called_with(key='foo/bar', recurse=False)

        # permission denied
        consul_kv.get = MagicMock(side_effect = ACLPermissionDenied)
        with patch('e2j2.tags.consul_tag.ConsulKV', return_value=consul_kv):
            with self.assertRaisesRegex(E2j2Exception, '^access denied connecting to:'):
                consul_tag.parse(config, 'foo/bar')

        # permission denied
        consul_kv.get = MagicMock(side_effect = AssertionError('FOOBAR error'))
        with patch('e2j2.tags.consul_tag.ConsulKV', return_value=consul_kv):
            with self.assertRaisesRegex(E2j2Exception, 'FOOBAR error'):
                consul_tag.parse(config, 'foo/bar')

        # key not found
        consul_kv.get = MagicMock(return_value=[])
        with patch('e2j2.tags.consul_tag.ConsulKV', return_value=consul_kv):
            with self.assertRaisesRegex(E2j2Exception, 'key not found'):
                consul_tag.parse(config, 'foo/bar')

        # get nested consul entry
        value = [{'Key': 'key/folder/subkey', 'Value': 'foobar'}]
        consul_kv.get = MagicMock(return_value=value)
        with patch('e2j2.tags.consul_tag.ConsulKV', return_value=consul_kv):
            self.assertEqual(consul_tag.parse(config, 'key/folder/subkey'), 'foobar')
{'fraction_non_alphanumeric': 0.11902439024390243, 'fraction_numerical': 0.018292682926829267, 'mean_word_length': 3.04588061174149, 'pattern_counts': {'":': 2, '<': 9, '<?xml version=': 0, '>': 11, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, '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': '1413073', 'n_tokens_mistral': 2687, 'n_tokens_neox': 2568, 'n_words': 486}