import os.path
import shutil
import tempfile
import unittest
import mock

import koji
import kojihub

QP = kojihub.QueryProcessor


class TestGetRPM(unittest.TestCase):

    def setUp(self):
        self.exports = kojihub.RootExports()
        self.context = mock.patch('kojihub.context').start()
        self.get_external_repo_id = mock.patch('kojihub.get_external_repo_id').start()
        self.QueryProcessor = mock.patch('kojihub.QueryProcessor',
                                         side_effect=self.getQuery).start()
        self.queries = []

    def getQuery(self, *args, **kwargs):
        query = QP(*args, **kwargs)
        query.execute = mock.MagicMock()
        self.queries.append(query)
        return query

    def tearDown(self):
        mock.patch.stopall()

    def test_wrong_type_rpminfo(self):
        rpminfo = ['test-user']
        with self.assertRaises(koji.GenericError) as cm:
            kojihub.get_rpm(rpminfo)
        self.assertEqual(f"Invalid type for rpminfo: {type(rpminfo)}", str(cm.exception))

    def test_rpm_info_int(self):
        rpminfo = 123
        kojihub.get_rpm(rpminfo)

        self.assertEqual(len(self.queries), 1)
        query = self.queries[0]
        str(query)
        self.assertEqual(query.tables, ['rpminfo'])
        columns = ['rpminfo.id', 'build_id', 'buildroot_id', 'rpminfo.name', 'version', 'release',
                   'epoch', 'arch', 'external_repo_id', 'external_repo.name', 'payloadhash',
                   'size', 'buildtime', 'metadata_only', 'extra']
        self.assertEqual(set(query.columns), set(columns))
        self.assertEqual(query.clauses, ['external_repo_id = 0', "rpminfo.id=%(id)s"])
        self.assertEqual(query.joins,
                         ['external_repo ON rpminfo.external_repo_id = external_repo.id'])
        self.assertEqual(query.values, {'id': rpminfo})

    def test_rpm_info_str(self):
        rpminfo = 'testrpm-1.23-4.x86_64.rpm'
        kojihub.get_rpm(rpminfo, multi=True)

        self.assertEqual(len(self.queries), 1)
        query = self.queries[0]
        str(query)
        self.assertEqual(query.tables, ['rpminfo'])
        columns = ['rpminfo.id', 'build_id', 'buildroot_id', 'rpminfo.name', 'version', 'release',
                   'epoch', 'arch', 'external_repo_id', 'external_repo.name', 'payloadhash',
                   'size', 'buildtime', 'metadata_only', 'extra']
        self.assertEqual(set(query.columns), set(columns))
        self.assertEqual(query.clauses, ["rpminfo.name=%(name)s AND version=%(version)s "
                                         "AND release=%(release)s AND arch=%(arch)s"])
        self.assertEqual(query.joins,
                         ['external_repo ON rpminfo.external_repo_id = external_repo.id'])
        self.assertEqual(query.values, {'arch': 'x86_64', 'epoch': '', 'name': 'testrpm',
                                        'release': '4', 'src': False, 'version': '1.23'})

    def test_rpm_info_dict_location(self):
        rpminfo = {'id': 123, 'name': 'testrpm-1.23-4.x86_64.rpm', 'location': 'test-location'}
        self.get_external_repo_id.return_value = 125
        rpminfo_data = rpminfo.copy()
        rpminfo_data['external_repo_id'] = 125

        kojihub.get_rpm(rpminfo, multi=True)

        self.assertEqual(len(self.queries), 1)
        query = self.queries[0]
        str(query)
        self.assertEqual(query.tables, ['rpminfo'])
        columns = ['rpminfo.id', 'build_id', 'buildroot_id', 'rpminfo.name', 'version', 'release',
                   'epoch', 'arch', 'external_repo_id', 'external_repo.name', 'payloadhash',
                   'size', 'buildtime', 'metadata_only', 'extra']
        self.assertEqual(set(query.columns), set(columns))
        self.assertEqual(query.clauses,
                         ["external_repo_id = %(external_repo_id)i", "rpminfo.id=%(id)s"])
        self.assertEqual(query.joins,
                         ['external_repo ON rpminfo.external_repo_id = external_repo.id'])
        self.assertEqual(query.values, rpminfo_data)


class TestGetRPMHeaders(unittest.TestCase):

    def setUp(self):
        self.exports = kojihub.RootExports()
        self.exports.getLoggedInUser = mock.MagicMock()
        self.context = mock.patch('kojihub.context').start()
        self.cursor = mock.MagicMock()
        self.get_rpm = mock.patch('kojihub.get_rpm').start()
        self.get_build = mock.patch('kojihub.get_build').start()
        self.get_header_fields = mock.patch('koji.get_header_fields').start()
        self.tempdir = tempfile.mkdtemp()
        self.pathinfo = koji.PathInfo(self.tempdir)
        mock.patch('koji.pathinfo', new=self.pathinfo).start()

    def tearDown(self):
        mock.patch.stopall()
        shutil.rmtree(self.tempdir)

    def test_taskid_invalid_path(self):
        self.cursor.fetchone.return_value = None
        self.context.cnx.cursor.return_value = self.cursor
        filepath = '../test/path'
        with self.assertRaises(koji.GenericError) as cm:
            self.exports.getRPMHeaders(taskID=99, filepath=filepath)
        self.assertEqual(f"Invalid filepath: {filepath}", str(cm.exception))
        self.get_rpm.assert_not_called()
        self.get_build.assert_not_called()
        self.get_header_fields.assert_not_called()

    def test_taskid_without_filepath(self):
        self.cursor.fetchone.return_value = None
        self.context.cnx.cursor.return_value = self.cursor
        with self.assertRaises(koji.GenericError) as cm:
            self.exports.getRPMHeaders(taskID=99)
        self.assertEqual("filepath must be specified with taskID", str(cm.exception))
        self.get_rpm.assert_not_called()
        self.get_build.assert_not_called()
        self.get_header_fields.assert_not_called()

    def test_insufficient_args(self):
        with self.assertRaises(koji.GenericError) as cm:
            self.exports.getRPMHeaders(strict=False)
        self.assertEqual("either rpmID or taskID and filepath must be specified",
                         str(cm.exception))
        with self.assertRaises(koji.GenericError) as cm:
            self.exports.getRPMHeaders(filepath='something')
        self.assertEqual("either rpmID or taskID and filepath must be specified",
                         str(cm.exception))
        # we already test taskid without filepath above

    def test_unknown_rpm(self):
        self.get_rpm.return_value = None
        result = self.exports.getRPMHeaders(rpmID='FOO-1-1.noarch', strict=False)
        self.assertEqual(result, {})
        self.get_rpm.assert_called_with('FOO-1-1.noarch', strict=False)

        # again with strict mode
        self.get_rpm.side_effect = koji.GenericError('NO SUCH RPM')
        with self.assertRaises(koji.GenericError) as cm:
            self.exports.getRPMHeaders(rpmID='FOO-1-1.noarch', strict=True)
        self.assertEqual("NO SUCH RPM", str(cm.exception))
        self.get_rpm.assert_called_with('FOO-1-1.noarch', strict=True)
        self.get_build.assert_not_called()
        self.get_header_fields.assert_not_called()

    def test_external_rpm(self):
        rpm_info = {'external_repo_id': 1, 'id': 'RPMID'}
        self.get_rpm.return_value = rpm_info
        result = self.exports.getRPMHeaders(rpmID='FOO-1-1.noarch', strict=False)
        self.assertEqual(result, {})

        # again with strict mode
        with self.assertRaises(koji.GenericError) as cm:
            self.exports.getRPMHeaders(rpmID='FOO-1-1.noarch', strict=True)
        self.assertEqual(f"External rpm: {rpm_info['id']}", str(cm.exception))
        self.get_build.assert_not_called()
        self.get_header_fields.assert_not_called()

    def test_deleted_build(self):
        self.get_rpm.return_value = {'build_id': 'BUILDID', 'external_repo_id': 0}
        build_info = {'nvr': 'NVR', 'state': koji.BUILD_STATES['DELETED']}
        self.get_build.return_value = build_info
        result = self.exports.getRPMHeaders(rpmID='FOO-1-1.noarch', strict=False)
        self.assertEqual(result, {})
        self.get_build.assert_called_with('BUILDID', strict=True)

        # again with strict mode
        with self.assertRaises(koji.GenericError) as cm:
            self.exports.getRPMHeaders(rpmID='FOO-1-1.noarch', strict=True)
        self.assertEqual(f"Build {build_info['nvr']} is deleted", str(cm.exception))
        self.get_build.assert_called_with('BUILDID', strict=True)
        self.get_header_fields.assert_not_called()

    def test_missing_rpm(self):
        self.get_rpm.return_value = {
            'build_id': 'BUILDID',
            'external_repo_id': 0,
            'name': 'pkg',
            'version': '1',
            'release': '2',
            'arch': 'noarch'}
        self.get_build.return_value = {
            'name': 'pkg',
            'version': '1',
            'release': '2',
            'nvr': 'pkg-1-2',
            'state': koji.BUILD_STATES['COMPLETE']}
        rpmpath = '%s/packages/pkg/1/2/noarch/pkg-1-2.noarch.rpm' % self.tempdir

        # rpm does not exist
        result = self.exports.getRPMHeaders(rpmID='pkg-1-2.noarch', strict=False)
        self.assertEqual(result, {})
        self.get_build.assert_called_with('BUILDID', strict=True)

        # again with strict mode
        with self.assertRaises(koji.GenericError) as e:
            self.exports.getRPMHeaders(rpmID='FOO-1-1.noarch', strict=True)
        self.assertEqual(f"Missing rpm file: {rpmpath}", str(e.exception))
        self.get_build.assert_called_with('BUILDID', strict=True)
        self.get_header_fields.assert_not_called()

    def test_rpm_exists(self):
        self.get_rpm.return_value = {
            'build_id': 'BUILDID',
            'external_repo_id': 0,
            'name': 'pkg',
            'version': '1',
            'release': '2',
            'arch': 'noarch'}
        self.get_build.return_value = {
            'name': 'pkg',
            'version': '1',
            'release': '2',
            'nvr': 'pkg-1-2',
            'state': koji.BUILD_STATES['COMPLETE']}
        rpmpath = '%s/packages/pkg/1/2/noarch/pkg-1-2.noarch.rpm' % self.tempdir
        koji.ensuredir(os.path.dirname(rpmpath))
        with open(rpmpath, 'w') as fo:
            fo.write('hello world')
        fakeheaders = {'HEADER': 'SOMETHING'}
        self.get_header_fields.return_value = fakeheaders

        result = self.exports.getRPMHeaders(rpmID='pkg-1-2.noarch', strict=True)
        self.assertEqual(result, fakeheaders)
        self.get_build.assert_called_with('BUILDID', strict=True)
        self.get_header_fields.assert_called_with(rpmpath, None)

    def test_task_rpm_exists(self):
        taskid = 137
        filepath = 'pkg-1-2.noarch.rpm'
        rpmpath = '%s/work/tasks/137/137/pkg-1-2.noarch.rpm' % self.tempdir
        koji.ensuredir(os.path.dirname(rpmpath))
        with open(rpmpath, 'w') as fo:
            fo.write('hello world')
        fakeheaders = {'HEADER': 'SOMETHING'}
        self.get_header_fields.return_value = fakeheaders

        result = self.exports.getRPMHeaders(taskID=taskid, filepath=filepath, strict=False)
        self.assertEqual(result, fakeheaders)
        self.get_rpm.assert_not_called()
        self.get_build.assert_not_called()
        self.get_header_fields.assert_called_with(rpmpath, None)

    def test_task_rpm_missing(self):
        taskid = 137
        filepath = 'pkg-1-2.noarch.rpm'
        rpmpath = '%s/work/tasks/137/137/pkg-1-2.noarch.rpm' % self.tempdir

        result = self.exports.getRPMHeaders(taskID=taskid, filepath=filepath, strict=False)
        self.assertEqual(result, {})

        with self.assertRaises(koji.GenericError) as cm:
            self.exports.getRPMHeaders(taskID=taskid, filepath=filepath, strict=True)
        self.assertEqual(f"Missing rpm file: {rpmpath}", str(cm.exception))
        self.get_rpm.assert_not_called()
        self.get_build.assert_not_called()
        self.get_header_fields.assert_not_called()
