#!/usr/bin/env python
#
# Copyright (C) 2012 Red Hat, Inc.  All rights reserved.
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
#
# Authors: Michal Minar <miminar@redhat.com>
#
"""
Unit tests for LMI_SoftwareInstalledPackage provider.
"""

import os
import pywbem
import socket
import stat
import unittest

import common

def make_path_tuple(objpath):
    return tuple(objpath[a] for a in ("SoftwareElementID", "Name", "Version",
        "SoftwareElementState"))

class TestSoftwareInstalledPackage(common.SoftwareBaseTestCase):
    """
    Basic cim operations test.
    """

    CLASS_NAME = "LMI_SoftwareInstalledPackage"
    KEYS = ("Software", "System")

    def make_op(self, pkg, newer=True, ses=2):
        """
        @param ses SoftwareElementState property value
        @return object path of LMI_SoftwareInstaledPackage
        """
        objpath = self.objpath.copy()
        pkg_op = pywbem.CIMInstanceName(
                classname="LMI_SoftwarePackage", namespace="root/cimv2",
                keybindings={
                    "Name" : pkg.name,
                    "SoftwareElementID" : pkg.get_nevra(newer, 'ALWAYS'),
                    "SoftwareElementState" : pywbem.Uint16(ses),
                    "TargetOperatingSystem" : pywbem.Uint16(36),
                    "Version" : getattr(pkg, 'up_ver' if newer else 'ver') })
        system_op = pywbem.CIMInstanceName(
                classname="Linux_ComputerSystem", namespace="root/cimv2",
                keybindings={
                    "CreationClassName" : "Linux_ComputerSystem",
                    "Name" : socket.gethostname() })
        objpath["Software"] = pkg_op
        objpath["System"] = system_op
        return objpath

    @common.mark_dangerous
    def test_get_instance(self):
        """
        Tests GetInstance call on packages from our rpm cache.
        TODO: test this in non-dangerous way
        """
        for pkg in self.pkgdb:
            if not common.is_installed(pkg):
                self.install_pkg(pkg)
            objpath = self.make_op(pkg)
            inst = self.conn.GetInstance(InstanceName=objpath, LocalOnly=False)
            self.assertIsSubclass(inst.path.classname, self.CLASS_NAME)
            self.assertIsSubclass(
                    inst.path['System'].classname, objpath['System'].classname)
            self.assertIsSubclass(inst.path['Software'].classname,
                    objpath['Software'].classname)
            self.assertEqual(2, len(inst.path.keys()))
            for key in self.KEYS:
                self.assertEqual(inst[key], inst.path[key],
                        "Object paths should be the same for %s"%pkg)
            self.assertEqual(inst['Software'], objpath['Software'],
                    "Software key reference should match for %s"%pkg)
            common.remove_pkg(pkg.name)
            objpath['Software']['SoftwareElementState'] = pywbem.Uint16(1)
            with self.assertRaises(pywbem.CIMError) as cmngr:
                self.conn.GetInstance(InstanceName=objpath, LocalOnly=False)
            self.assertEqual(pywbem.CIM_ERR_NOT_FOUND, cmngr.exception.args[0],
                    "Package %s should not be installed"%pkg)

    @common.mark_dangerous
    def test_enum_instance_names(self):
        """
        Tests EnumInstances call.
        TODO: test this in non-dangerous way
        """
        pkg = self.pkgdb[0] if len(self.pkgdb) > 0 else None
        if pkg and common.is_installed(pkg.name):
            common.remove_pkg(pkg.name)
        insts1 = self.conn.EnumerateInstanceNames(ClassName=self.CLASS_NAME)
        if pkg:
            self.install_pkg(pkg)
            insts2 = self.conn.EnumerateInstanceNames(ClassName=self.CLASS_NAME)
            self.assertEqual(len(insts1) + 1, len(insts2))

        if pkg:
            objpath = self.make_op(pkg)
            self.assertIn(make_path_tuple(objpath['Software']),
                    set(make_path_tuple(inst['Software']) for inst in insts2))
        self.assertTrue(all(isinstance(inst, pywbem.CIMInstanceName)
            for inst in insts1))

    @common.mark_dangerous
    def test_enum_instances(self):
        """
        Tests EnumInstances call.
        """
        pkg = self.pkgdb[0] if len(self.pkgdb) > 0 else None
        if pkg and not common.is_installed(pkg.name):
            self.install_pkg(pkg.name)
        insts1 = self.conn.EnumerateInstances(ClassName=self.CLASS_NAME)
        if pkg:
            common.remove_pkg(pkg.name)
            insts2 = self.conn.EnumerateInstances(ClassName=self.CLASS_NAME)
            objpath = self.make_op(pkg)
            self.assertEqual(len(insts2) + 1, len(insts1))
            path_values = set(make_path_tuple(p["Software"]) for p in insts1)
            self.assertIn(make_path_tuple(objpath['Software']), path_values)
            path_values = set(make_path_tuple(p["Software"]) for p in insts2)
            self.assertNotIn(make_path_tuple(objpath['Software']), path_values)

        self.assertTrue(all(inst['Software'] == inst.path['Software']
            for inst in insts1))
        self.assertTrue(all(inst['System'] == inst.path['System']
            for inst in insts1))

    @common.mark_dangerous
    def test_create_instance(self):
        """
        Tests CreateInstance call.
        """
        for pkg in self.pkgdb:
            if common.is_installed(pkg.name):
                common.remove_pkg(pkg.name)
            self.assertFalse(common.is_installed(pkg.name),
                    "Package %s should not be installed"%pkg)
            objpath = self.make_op(pkg, ses=1)
            inst = pywbem.CIMInstance(classname=objpath.classname, path=objpath)
            inst["Software"] = objpath["Software"]
            inst["System"] = objpath["System"]
            iname = self.conn.CreateInstance(NewInstance=inst)
            self.assertIsInstance(iname, pywbem.CIMInstanceName)
            objpath["Software"]["SoftwareElementState"] = pywbem.Uint16(2)
            self.assertEqual(iname["Software"], objpath["Software"],
                    "Software key reference should match for %s"%pkg)
            self.assertIsInstance(iname["System"], pywbem.CIMInstanceName)
            self.assertIsSubclass(iname["System"].classname,
                    objpath["System"].classname)
            self.assertEqual(set(iname.keys()), set(objpath.keys()),
                    "Keys of object paths should be the same for %s"%pkg)
            self.assertTrue(common.is_installed(pkg.name),
                    "Package %s should be installed"%pkg)

            with self.assertRaises(pywbem.CIMError) as cmngr:
                self.conn.CreateInstance(NewInstance=inst)
            self.assertEqual(pywbem.CIM_ERR_ALREADY_EXISTS,
                    cmngr.exception.args[0],
                    "Package %s should already be installed"%pkg)

    @common.mark_dangerous
    def test_delete_instance(self):
        """
        Tests DeleteInstance call.
        """
        for pkg in self.pkgdb:
            if not common.is_installed(pkg.name):
                self.install_pkg(pkg)
            self.assertTrue(common.is_installed(pkg.name),
                    "Package %s must be installed"%pkg)
            objpath = self.make_op(pkg)
            self.conn.DeleteInstance(objpath)
            self.assertFalse(common.is_installed(pkg.name),
                    "Package %s must be uninstalled"%pkg)
            with self.assertRaises(pywbem.CIMError) as cmngr:
                self.conn.DeleteInstance(objpath)
            self.assertIn(cmngr.exception.args[0],
                    [pywbem.CIM_ERR_FAILED, pywbem.CIM_ERR_NOT_FOUND],
                    "Package %s can not be uninstalled again"%pkg)

    @common.mark_dangerous
    def test_check_integrity(self):
        """
        Tests CheckIntegrity call.
        TODO: test this in non-dangerous way
        """
        for pkg in self.pkgdb:
            files = self.pkg_files[pkg.name]
            if (  (   common.is_installed(pkg)
                  and not common.verify_pkg(pkg.name))
               or (   common.is_installed(pkg.name)
                  and not common.is_installed(pkg))) :
                common.remove_pkg(pkg.name)
            if not common.is_installed(pkg.name):
                self.install_pkg(pkg)
            self.assertTrue(common.is_installed(pkg),
                    "Package %s must be installed"%pkg)

            objpath = self.make_op(pkg)
            (rval, oparms) = self.conn.InvokeMethod(
                    MethodName="CheckIntegrity",
                    ObjectName=objpath)
            self.assertEqual(pywbem.Uint32(0), rval,
                    "IntegrityCheck should pass for %s"%pkg.name) # check passed
            self.assertEqual(1, len(oparms))
            self.assertIn("Failed", oparms)
            self.assertEqual(0, len(oparms["Failed"]),
                    "IntegrityCheck should not fail for %s"%pkg.name)

            cnt_bad = 0
            for file_path in files:
                stats = os.lstat(file_path)
                if os.path.islink(file_path):       # modify symbolic link
                    target = os.readlink(file_path)
                    os.remove(file_path)
                    os.symlink(target, file_path) # just touch symlink
                    (rval, oparms) = self.conn.InvokeMethod(
                            MethodName="CheckIntegrity",
                            ObjectName=objpath)
                    # symlink must pass
                    self.assertEqual(cnt_bad, len(oparms["Failed"]),
                            "Symlink %s:%s should pass"%(pkg.name, file_path))
                    os.remove(file_path)
                    # now change target
                    os.symlink("wrong_link_target", file_path)
                elif os.path.isdir(file_path):      # check directory
                    os.chmod(file_path, stats.st_mode)   # just touch dir
                    (rval, oparms) = self.conn.InvokeMethod(
                            MethodName="CheckIntegrity",
                            ObjectName=objpath)
                    # dir must pass
                    self.assertEqual(cnt_bad, len(oparms["Failed"]),
                            "Directory %s:%s should pass"%(pkg.name, file_path))
                    # modify read access of directory
                    os.chmod(file_path, stats.st_mode ^ stat.S_IROTH)
                else:                       # modify regular file
                    # just touch file - this is enough to make it fail
                    with open(file_path, "w+"):
                        pass
                cnt_bad += 1
                (rval, oparms) = self.conn.InvokeMethod(
                        MethodName="CheckIntegrity",
                        ObjectName=objpath)
                self.assertEqual(pywbem.Uint32(1), rval,
                    "File %s:%s should not pass"%(pkg.name, file_path))
                self.assertEqual(1, len(oparms))
                self.assertIn("Failed", oparms)
                self.assertEqual(len(oparms["Failed"]), cnt_bad,
                    "Number of errors not correct. Failed for %s:%s" % (
                        pkg.name, file_path))
                self.assertIn(file_path, (p["Name"] for p in oparms["Failed"]),
                    "File %s:%s should also be in failed"%(pkg.name, file_path))

    @common.mark_dangerous
    def test_method_update(self):
        """
        Tests Update method invocation.
        """
        for pkg in self.pkgdb:
            if (   common.is_installed(pkg.name)
               and not common.is_installed(pkg, False)):
                common.remove_pkg(pkg.name)
            if not common.is_installed(pkg.name):
                self.install_pkg(pkg, False)
            self.assertTrue(common.is_installed(pkg, False),
                "Package %s must be installed"%pkg.get_nevra(False))

            objpath = self.make_op(pkg, False)
            op_up = self.make_op(pkg)
            (rval, oparms) = self.conn.InvokeMethod(
                    MethodName="Update",
                    ObjectName=objpath)
            self.assertEqual(pywbem.Uint16(1), rval,
                "Update should succed for %s"%pkg.get_nevra(False))
            self.assertEqual(1, len(oparms))
            self.assertIn("Installed", oparms)
            self.assertEqual(oparms["Installed"], op_up["Software"],
                "Object paths should match for %s"%pkg)
            self.assertTrue(common.is_installed(pkg),
                "Package %s must be installed"%pkg)

            (rval, oparms) = self.conn.InvokeMethod(
                    MethodName="Update",
                    ObjectName=op_up)
            self.assertEqual(pywbem.Uint16(0), rval,
                    "Package %s should be already updated"%pkg)
            self.assertEqual(1, len(oparms))
            self.assertEqual(oparms["Installed"], op_up["Software"],
                    "Object paths should match for %s"%pkg)
            self.assertTrue(common.is_installed(pkg.name),
                    "Package %s must be installed"%pkg)
            self.assertFalse(common.is_installed(pkg, False),
                    "Older package %s must not be installed" %
                    pkg.get_nevra(False))

            with self.assertRaises(pywbem.CIMError) as cmngr:
                self.conn.InvokeMethod(
                        MethodName="Update",
                        ObjectName=objpath)
            self.assertEqual(pywbem.CIM_ERR_NOT_FOUND, cmngr.exception.args[0],
                    "Older package %s should not be installed" %
                    pkg.get_nevra(False))

            common.remove_pkg(pkg.name)
            self.install_pkg(pkg, False)
            self.assertTrue(common.is_installed(pkg, False))

            (rval, oparms) = self.conn.InvokeMethod(
                    MethodName="Update",
                    ObjectName=objpath,
                    Epoch=pkg.epoch,
                    Version=pkg.ver,
                    Release=pkg.rel)
            self.assertEqual(pywbem.Uint16(0), rval,
                    "Update of package %s should succeed"%pkg)
            self.assertEqual(oparms["Installed"], objpath["Software"],
                    "Object paths should be the same for package %s"%pkg)

            (rval, oparms) = self.conn.InvokeMethod(
                    MethodName="Update",
                    ObjectName=objpath,
                    Epoch=pkg.up_epoch,
                    Version=pkg.up_ver,
                    Release=pkg.up_rel)
            self.assertEqual(pywbem.Uint16(1), rval,
                "Package %s can not be updated twice to highest version"%pkg)
            self.assertEqual(oparms["Installed"], op_up["Software"],
                "Object paths should match for package %s"%pkg)

def suite():
    """For unittest loaders."""
    return unittest.TestLoader().loadTestsFromTestCase(
            TestSoftwareInstalledPackage)

if __name__ == "__main__":
    unittest.main()