<commit_before># Copyright (C) 2014 Red Hat, Inc.
#
# Author: <NAME> <<EMAIL>>
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

import fixtures
from mock import MagicMock
from requests.auth import AuthBase
from designate import tests
from designate import utils
from designate.tests.test_backend import BackendTestMixin
from designate.openstack.common import jsonutils as json
from designate.backend import impl_ipa

ipamethods = {"dnszone_add": {}, "dnszone_mod": {},
              "dnszone_del": {}, "dnsrecord_add": {},
              "dnsrecord_mod": {}, "dnsrecord_del": {},
              }


class MockIPAAuth(AuthBase):
    def __init__(self, hostname, keytab):
        self.count = 0

    def refresh_auth(self):
        self.count += 1


class MockResponse(object):
    def __init__(self, status_code, jsontext):
        self.status_code = status_code
        self.text = jsontext


class MockRequest(object):
    def __init__(self, testcase):
        self.headers = {}
        self.myauth = MockIPAAuth("ignore", "ignore")
        self.testcase = testcase
        self.error = None
        self.needauth = False

    @property
    def auth(self):
        # always return the mock object
        return self.myauth

    @auth.setter
    def auth(self, val):
        # disallow setting
        pass

    def post(self, jsonurl, data):
        # convert json data string to dict
        ddict = json.loads(data)
        # check basic parameters
        self.testcase.assertIn('method', ddict)
        meth = ddict['method']
        self.testcase.assertIn(meth, ipamethods)
        self.testcase.assertIn('params', ddict)
        self.testcase.assertIsInstance(ddict['params'], list)
        self.testcase.assertEqual(len(ddict['params']), 2)
        self.testcase.assertIsInstance(ddict['params'][0], list)
        self.testcase.assertIsInstance(ddict['params'][1], dict)
        self.testcase.assertIn('version', ddict['params'][1])
        # check method specific parameters
        if meth.startswith('dnsrecord_'):
            self.testcase.assertEqual(len(ddict['params'][0]), 2)
            # domain params end with a .
            param1 = ddict['params'][0][0]
            self.testcase.assertEqual(param1[-1], ".")
        elif meth.startswith('dnszone_'):
            self.testcase.assertEqual(len(ddict['params'][0]), 1)
            param1 = ddict['params'][0][0]
            self.testcase.assertEqual(param1[-1], ".")

        rc = {}
        if self.needauth:
            self.needauth = False  # reset
            return MockResponse(401, json.dumps(rc))
        if self.error:
            rc['error'] = {'code': self.error}
            self.error = None  # reset
        else:
            rc['error'] = None
        return MockResponse(200, json.dumps(rc))


class IPABackendTestCase(tests.TestCase, BackendTestMixin):

    def get_record_fixture(self, recordset_type, fixture=0, values={}):
        """override to ensure all records have a recordset_id"""
        return super(IPABackendTestCase, self).get_record_fixture(
            recordset_type, fixture,
            values={
                'recordset_id': utils.generate_uuid()
            }
        )

    def setUp(self):
        super(IPABackendTestCase, self).setUp()
        self.request = MockRequest(self)
        # make requests return our mock object

        def getSession():
            return self.request

        # replace requests.Session() with our mock version
        self.useFixture(fixtures.MonkeyPatch('requests.Session', getSession))

        self.config(backend_driver='ipa', group='service:agent')
        self.backend = self.get_backend_driver()
        self.CONF['backend:ipa'].ipa_auth_driver_class = \
            "designate.tests.test_backend.test_ipa.MockIPAAuth"
        self.backend.start()
        self.central_service = self.start_service('central')
        # Since some CRUD methods in impl_ipa call central's find_servers
        # and find_records method, mock it up to return our fixture.
        self.backend.central_service.find_servers = MagicMock(
            return_value=[self.get_server_fixture()])
        self.backend.central_service.find_records = MagicMock(
            return_value=[self.get_record_fixture('A')])

    def test_create_server(self):
        context = self.get_context()
        server = self.get_server_fixture()
        self.backend.create_server(context, server)

    def test_update_server(self):
        context = self.get_context()
        server = self.get_server_fixture()
        self.backend.create_server(context, server)
        self.backend.update_server(context, server)

    def test_delete_server(self):
        context = self.get_context()
        server = self.get_server_fixture()
        self.backend.create_server(context, server)
        self.backend.delete_server(context, server)

    def test_create_domain(self):
        context = self.get_context()
        server = self.get_server_fixture()
        domain = self.get_domain_fixture()
        self.backend.create_server(context, server)
        self.backend.create_domain(context, domain)

    def test_update_domain(self):
        context = self.get_context()<commit_msg>Fix FreeIPA tests mock's<commit_after># Copyright (C) 2014 Red Hat, Inc.
#
# Author: <NAME> <<EMAIL>>
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import fixtures
from requests.auth import AuthBase
from designate import tests
from designate import utils
from designate.tests.test_backend import BackendTestMixin
from designate.openstack.common import jsonutils as json
from designate.openstack.common.fixture import mockpatch
from designate.backend import impl_ipa

ipamethods = {"dnszone_add": {}, "dnszone_mod": {},
              "dnszone_del": {}, "dnsrecord_add": {},
              "dnsrecord_mod": {}, "dnsrecord_del": {},
              }


class MockIPAAuth(AuthBase):
    def __init__(self, hostname, keytab):
        self.count = 0

    def refresh_auth(self):
        self.count += 1


class MockResponse(object):
    def __init__(self, status_code, jsontext):
        self.status_code = status_code
        self.text = jsontext


class MockRequest(object):
    def __init__(self, testcase):
        self.headers = {}
        self.myauth = MockIPAAuth("ignore", "ignore")
        self.testcase = testcase
        self.error = None
        self.needauth = False

    @property
    def auth(self):
        # always return the mock object
        return self.myauth

    @auth.setter
    def auth(self, val):
        # disallow setting
        pass

    def post(self, jsonurl, data):
        # convert json data string to dict
        ddict = json.loads(data)
        # check basic parameters
        self.testcase.assertIn('method', ddict)
        meth = ddict['method']
        self.testcase.assertIn(meth, ipamethods)
        self.testcase.assertIn('params', ddict)
        self.testcase.assertIsInstance(ddict['params'], list)
        self.testcase.assertEqual(len(ddict['params']), 2)
        self.testcase.assertIsInstance(ddict['params'][0], list)
        self.testcase.assertIsInstance(ddict['params'][1], dict)
        self.testcase.assertIn('version', ddict['params'][1])
        # check method specific parameters
        if meth.startswith('dnsrecord_'):
            self.testcase.assertEqual(len(ddict['params'][0]), 2)
            # domain params end with a .
            param1 = ddict['params'][0][0]
            self.testcase.assertEqual(param1[-1], ".")
        elif meth.startswith('dnszone_'):
            self.testcase.assertEqual(len(ddict['params'][0]), 1)
            param1 = ddict['params'][0][0]
            self.testcase.assertEqual(param1[-1], ".")

        rc = {}
        if self.needauth:
            self.needauth = False  # reset
            return MockResponse(401, json.dumps(rc))
        if self.error:
            rc['error'] = {'code': self.error}
            self.error = None  # reset
        else:
            rc['error'] = None
        return MockResponse(200, json.dumps(rc))


class IPABackendTestCase(tests.TestCase, BackendTestMixin):

    def get_record_fixture(self, recordset_type, fixture=0, values={}):
        """override to ensure all records have a recordset_id"""
        return super(IPABackendTestCase, self).get_record_fixture(
            recordset_type, fixture,
            values={
                'recordset_id': utils.generate_uuid()
            }
        )

    def setUp(self):
        super(IPABackendTestCase, self).setUp()
        self.request = MockRequest(self)
        # make requests return our mock object

        def getSession():
            return self.request

        # replace requests.Session() with our mock version
        self.useFixture(fixtures.MonkeyPatch('requests.Session', getSession))

        self.config(backend_driver='ipa', group='service:agent')
        self.backend = self.get_backend_driver()
        self.CONF['backend:ipa'].ipa_auth_driver_class = \
            "designate.tests.test_backend.test_ipa.MockIPAAuth"
        self.backend.start()
        self.central_service = self.start_service('central')

        # Since some CRUD methods in impl_ipa call central's find_servers
        # and find_records method, mock it up to return our fixture.
        self.useFixture(mockpatch.PatchObject(
            self.backend.central_service,
            'find_servers',
            return_value=[self.get_server_fixture()]
        ))

        self.useFixture(mockpatch.PatchObject(
            self.backend.central_service,
            'find_records',
            return_value=[self.get_record_fixture('A')]
        ))

    def test_create_server(self):
        context = self.get_context()
        server = self.get_server_fixture()
        self.backend.create_server(context, server)

    def test_update_server(self):
        context = self.get_context()
        server = self.get_server_fixture()
        self.backend.create_server(context, server)
        self.backend.update_server(context, server)

    def test_delete_server(self):
        context = self.get_context()
        server = self.get_server_fixture()
        self.backend.create_server(context, server)
        self.backend.delete_server(context, server)

    def test_create_domain(self):
        context = self.get_context()
        server = self.get_server_fixture()
        domain = self.get_domain_fixture()
        self.backend.create_server(context, server)
        self.backend.create_domain(context, domain)

    def test_update_domain(self):
        context = self.get_context()
{'fraction_non_alphanumeric': 0.10713049498096228, 'fraction_numerical': 0.006057459328487366, 'mean_word_length': 2.725660863958736, 'pattern_counts': {'":': 12, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '3003362', 'n_tokens_mistral': 3337, 'n_tokens_neox': 3279, 'n_words': 776}