#!/usr/bin/python

#

# CDDL HEADER START

#

# The contents of this file are subject to the terms of the

# Common Development and Distribution License (the "License").

# You may not use this file except in compliance with the License.

#

# You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE

# or http://www.opensolaris.org/os/licensing.

# See the License for the specific language governing permissions

# and limitations under the License.

#

# When distributing Covered Code, include this CDDL HEADER in each

# file and include the License file at usr/src/OPENSOLARIS.LICENSE.

# If applicable, add the following below this CDDL HEADER, with the

# fields enclosed by brackets "[]" replaced with your own identifying

# information: Portions Copyright [yyyy] [name of copyright owner]

#

# CDDL HEADER END

#

#

# Copyright (c) 2009, 2011, Oracle and/or its affiliates. All rights reserved.

#

import testutils

if __name__ == "__main__":

        testutils.setup_environment("../../../proto")

import pkg5unittest

import calendar

import difflib

import os

import platform

import pprint

import re

import shutil

import unittest

import pkg.client.api as api

import pkg.client.api_errors as api_errors

import pkg.client.progress as progress

import pkg.fmri as fmri

import pkg.misc as misc

import pkg.version as version

CLIENT_API_VERSION = 49

PKG_CLIENT_NAME = "pkg"

class TestApiList(pkg5unittest.ManyDepotTestCase):

        # Only start/stop the depot once (instead of for every test)

        persistent_setup = True

        packages = [

            "apple@1,5.11-0",

            "[email protected],5.11-0",

            "[email protected],5.11-0",

            "[email protected],5.11-0",

            "[email protected],5.11-0",

            "[email protected],5.11-0",

            "[email protected],5.11-1",

            "bat/[email protected],5.11-0",

            "[email protected]",

            "[email protected]",

            "[email protected]",

            "[email protected]",

            "[email protected]",

            "[email protected]",

            "[email protected]",

            "[email protected]",

            "[email protected]",

            "[email protected]",

            "[email protected]",

        ]

        @staticmethod

        def __tuple_order(a, b):

                apub, astem, aver = a

                bpub, bstem, bver = b

                rval = cmp(astem, bstem)

                if rval != 0:

                        return rval

                rval = cmp(apub, bpub)

                if rval != 0:

                        return rval

                aver = version.Version(aver, "5.11")

                bver = version.Version(bver, "5.11")

                return cmp(aver, bver) * -1

        @staticmethod

        def __get_pkg_arch(stem, ver):

                # Attempt to determine current arch and opposite arch.

                # This is so that the tests will see the set of packages

                # they expect to be omitted.

                parch = platform.processor()

                if parch == "i386":

                        oparch = "sparc"

                else:

                        oparch = "i386"

                if stem == "apple":

                        return [parch]

                elif stem in ("entire", "bat/bar", "obsolete"):

                        return

                elif stem in ("corge", "grault", "qux", "quux"):

                        return [parch, oparch]

                return [oparch]

        @staticmethod

        def __get_pkg_cats(stem, ver):

                if stem == "apple":

                        return [

                            "fruit",

                            "org.opensolaris.category.2008:"

                            "Applications/Sound and Video"

                        ]

                elif stem == "bat/bar":

                        return [

                            "food",

                            "org.opensolaris.category.2008:"

                            "Development/Python"

                        ]

                return []

        @staticmethod

        def __get_pkg_summ_desc(stem, ver):

                summ = "Summ. is %s %s" % (stem, ver)

                desc = "Desc. is %s %s" % (stem, ver)

                return summ, desc

        def __get_pkg_states(self, pub, stem, ver, installed=False):

                states = [api.PackageInfo.KNOWN]

                if installed:

                        states.append(api.PackageInfo.INSTALLED)

                if stem == "apple":

                        # Compare with newest version entry for this stem.

                        if ver != str(self.dlist1[6].version):

                                states.append(api.PackageInfo.UPGRADABLE)

                elif stem == "baz":

                        # Compare with newest version entry for this stem.

                        if ver != str(self.dlist1[10].version):

                                states.append(api.PackageInfo.UPGRADABLE)

                elif stem == "corge":

                        # Compare with newest version entry for this stem.

                        nver = str(self.dlist1[12].version)

                        if ver != nver:

                                states.append(api.PackageInfo.UPGRADABLE)

                        if ver == nver:

                                states.append(api.PackageInfo.RENAMED)

                elif stem == "obsolete":

                        states.append(api.PackageInfo.OBSOLETE)

                elif stem == "qux":

                        # Compare with newest version entry for this stem.

                        nver = str(self.dlist1[18].version)

                        if ver != nver:

                                states.append(api.PackageInfo.UPGRADABLE)

                        if ver == nver:

                                states.append(api.PackageInfo.RENAMED)

                return frozenset(states)

        def __get_pub_entry(self, pub, idx, name, ver):

                if pub == "test1":

                        l = self.dlist1

                else:

                        l = self.dlist2

                f = l[idx]

                self.assertEqual(f.pkg_name, name)

                v = str(f.version)

                try:

                        self.assertTrue(v.startswith(ver + ":"))

                except AssertionError:

                        self.debug("\n%s does not start with %s:" % (v, ver))

                        raise

                return f, v

        def __get_exp_pub_entry(self, pub, idx, name, ver, installed=False):

                f, v = self.__get_pub_entry(pub, idx, name, ver)

                return self.__get_expected_entry(pub, name, v,

                    installed=installed)

        def setUp(self):

                pkg5unittest.ManyDepotTestCase.setUp(self, ["test1", "test2",

                    "test3"])

                pkg_data = ""

                for p in self.packages:

                        pkg_data += p

                        stem, ver = p.split("@")

                        # XXX version should not require 5.11

                        sver = version.Version(ver, "5.11")

                        sver = str(sver).split(":", 1)[0]

                        summ, desc = self.__get_pkg_summ_desc(stem, sver)

                        pkg_data += """

open %(stem)s@%(ver)s

add set name=pkg.summary value="%(summ)s"

add set name=pkg.description value="%(desc)s"

""" % { "stem": stem, "ver": ver, "summ": summ, "desc": desc }

                        cats = self.__get_pkg_cats(stem, sver)

                        if cats:

                                pkg_data += "add set name=info.classification"

                                for cat in cats:

                                        pkg_data += ' value="%s"' % cat

                                pkg_data += "\n"

                        arch = self.__get_pkg_arch(stem, sver)

                        if arch:

                                adata = "value="

                                adata += " value=".join(arch)

                                pkg_data += "add set name=variant.arch " \

                                    "%s\n" % adata

                        if stem == "corge" and sver.startswith("1.0"):

                                pkg_data += "add set name=pkg.renamed " \

                                    "value=true\n"

                                pkg_data += "add depend type=require " \

                                    "fmri=grault\n"

                        elif stem == "entire":

                                pkg_data += "add depend type=incorporate " \

                                    "[email protected]\n"

                                pkg_data += "add depend type=incorporate " \

                                    "[email protected]\n"

                                pkg_data += "add depend type=incorporate " \

                                    "[email protected]\n"

                        elif stem == "obsolete":

                                pkg_data += "add set name=pkg.obsolete " \

                                    "value=true\n"

                        elif stem == "qux" and sver.startswith("1.0"):

                                pkg_data += "add set name=pkg.renamed " \

                                    "value=true\n"

                                pkg_data += "add depend type=require " \

                                    "fmri=quux\n"

                        pkg_data += "close\n"

                rurl1 = self.dcs[1].get_repo_url()

                plist = self.pkgsend_bulk(rurl1, pkg_data)

                # Ensure that the second repo's packages have exactly the same

                # timestamps as those in the first ... by copying the repo over.

                d1dir = self.dcs[1].get_repodir()

                d2dir = self.dcs[2].get_repodir()

                self.copy_repository(d1dir, d2dir, { "test1": "test2" })

                self.dlist1 = []

                self.dlist2 = []

                for e in plist:

                        # Unique FMRI object is needed for each list.

                        f = fmri.PkgFmri(str(e))

                        self.dlist1.append(f)

                        f = fmri.PkgFmri(str(e))

                        f.set_publisher("test2")

                        self.dlist2.append(f)

                self.dlist1.sort()

                self.dlist2.sort()

                # The new repository won't have a catalog, so rebuild it.

                self.dcs[2].get_repo(auto_create=True).rebuild()

                # The third repository should remain empty and not be

                # published to.

                # Next, create the image and configure publishers.

                self.image_create(rurl1, prefix="test1")

                rurl2 = self.dcs[2].get_repo_url()

                self.pkg("set-publisher -g " + rurl2 + " test2")

        def assertPrettyEqual(self, actual, expected):

                if actual == expected:

                        return

                actual = pprint.pformat(actual, indent=2)

                expected = pprint.pformat(expected, indent=2)

                self.assertEqual(expected, actual,

                    "Actual output differed from expected output.\n" +

                    "\n".join(difflib.unified_diff(

                        expected.splitlines(), actual.splitlines(),

                        "Expected output", "Actual output", lineterm="")))

                raise AssertionError(output)

        def __get_expected_entry(self, pub, stem, ver, installed=False):

                states = self.__get_pkg_states(pub, stem, ver,

                    installed=installed)

                sver = ver.split(":", 1)[0]

                raw_cats = self.__get_pkg_cats(stem, sver)

                summ, desc = self.__get_pkg_summ_desc(stem, sver)

                scheme = None

                cat = None

                pcats = []

                for e in raw_cats:

                        if e and ":" in e:

                                scheme, cat = e.split(":", 1)

                        else:

                                scheme = ""

                                cat = e

                        pcats.append((scheme, cat))

                return ((pub, stem, ver), summ, pcats, states)

        def __get_expected(self, pkg_list, cats=None, pubs=misc.EmptyI,

            variants=False):

                nlist = {}

                newest = pkg_list == api.ImageInterface.LIST_NEWEST

                if newest:

                        # Get the newest FMRI for each unique package stem on

                        # a per-publisher basis.

                        for plist in (self.dlist1, self.dlist2):

                                for f in plist:

                                        pstem = f.get_pkg_stem()

                                        if pstem not in nlist:

                                                nlist[pstem] = f

                                        elif f.version > nlist[pstem]:

                                                nlist[pstem] = f

                nlist = sorted(nlist.values())

                expected = []

                for plist in (self.dlist1, self.dlist2):

                        for f in plist:

                                pub, stem, ver = f.tuple()

                                ver = str(f.version)

                                if pubs and pub not in pubs:

                                        continue

                                sver = ver.split(":", 1)[0]

                                arch = self.__get_pkg_arch(stem, sver)

                                parch = platform.processor()

                                if not variants and arch and parch not in arch:

                                        continue

                                if newest and f not in nlist:

                                        continue

                                t, summ, pcats, states = \

                                    self.__get_expected_entry(pub, stem, ver)

                                if cats is not None:

                                        if not cats:

                                                if pcats:

                                                        # Want packages with no

                                                        # category.

                                                        continue

                                        elif not \

                                            [sc for sc in cats if sc in pcats]:

                                                # Doesn't match specified

                                                # categories.

                                                continue

                                expected.append((t, summ, pcats, states))

                def pkg_list_order(a, b):

                        at = a[0]

                        bt = b[0]

                        return self.__tuple_order(at, bt)

                expected.sort(cmp=pkg_list_order)

                return expected

        def __get_returned(self, pkg_list, api_obj=None, cats=None,

            num_expected=None, patterns=misc.EmptyI,

            pubs=misc.EmptyI, variants=False):

                if not api_obj:

                        progresstracker = progress.NullProgressTracker()

                        api_obj = api.ImageInterface(self.get_img_path(),

                            CLIENT_API_VERSION, progresstracker, lambda x: False,

                            PKG_CLIENT_NAME)

                # Set of states exposed by the API.

                exp_states = set([api.PackageInfo.FROZEN,

                    api.PackageInfo.INCORPORATED, api.PackageInfo.EXCLUDES,

                    api.PackageInfo.KNOWN, api.PackageInfo.INSTALLED,

                    api.PackageInfo.UPGRADABLE, api.PackageInfo.OBSOLETE,

                    api.PackageInfo.RENAMED])

                # Get ordered list of all packages.

                returned = []

                for entry in api_obj.get_pkg_list(pkg_list, cats=cats,

                    patterns=patterns, pubs=pubs, variants=variants):

                        (pub, stem, ver), summ, pcats, raw_states = entry

                        sver = ver.split(":", 1)[0]

                        # Eliminate states not exposed by the api.

                        states = raw_states.intersection(exp_states)

                        returned.append(((pub, stem, ver), summ, pcats, states))

                return returned

        def __test_list(self, pkg_list, api_obj=None,

            cats=None, num_expected=None, pubs=misc.EmptyI,

            variants=False):

                # Get package list.

                returned = self.__get_returned(pkg_list, api_obj=api_obj,

                    cats=cats, pubs=pubs, variants=variants)

                # Now generate expected list.

                expected = self.__get_expected(pkg_list, cats=cats,

                    pubs=pubs, variants=variants)

                # Compare returned and expected.

                self.assertPrettyEqual(returned, expected)

                if num_expected is not None:

                        self.assertEqual(len(returned), num_expected)

        def test_list_01_full(self):

                """Verify the sort order and content of a full list and

                combinations thereof."""

                progresstracker = progress.NullProgressTracker()

                api_obj = api.ImageInterface(self.get_img_path(), CLIENT_API_VERSION,

                    progresstracker, lambda x: False, PKG_CLIENT_NAME)

                # First check all variants case.

                returned = self.__get_returned(api_obj.LIST_ALL,

                    api_obj=api_obj, variants=True)

                expected = [

                    self.__get_exp_pub_entry("test1", 6, "apple",

                        "1.2.1,5.11-1"),

                    self.__get_exp_pub_entry("test1", 5, "apple",

                        "1.2.1,5.11-0"),

                    self.__get_exp_pub_entry("test1", 4, "apple",

                        "1.2.1,5.11-0"),

                    self.__get_exp_pub_entry("test1", 3, "apple",

                        "1.2.0,5.11-0"),

                    self.__get_exp_pub_entry("test1", 2, "apple", "1.1,5.11-0"),

                    self.__get_exp_pub_entry("test1", 1, "apple", "1.0,5.11-0"),

                    self.__get_exp_pub_entry("test1", 0, "apple", "1,5.11-0"),

                    self.__get_exp_pub_entry("test2", 6, "apple",

                        "1.2.1,5.11-1"),

                    self.__get_exp_pub_entry("test2", 5, "apple",

                        "1.2.1,5.11-0"),

                    self.__get_exp_pub_entry("test2", 4, "apple",

                        "1.2.1,5.11-0"),

                    self.__get_exp_pub_entry("test2", 3, "apple",

                        "1.2.0,5.11-0"),

                    self.__get_exp_pub_entry("test2", 2, "apple", "1.1,5.11-0"),

                    self.__get_exp_pub_entry("test2", 1, "apple", "1.0,5.11-0"),

                    self.__get_exp_pub_entry("test2", 0, "apple", "1,5.11-0"),

                    self.__get_exp_pub_entry("test1", 7, "bat/bar",

                        "1.2,5.11-0"),

                    self.__get_exp_pub_entry("test2", 7, "bat/bar",

                        "1.2,5.11-0"),

                    self.__get_exp_pub_entry("test1", 10, "baz", "1.3,5.11"),

                    self.__get_exp_pub_entry("test1", 9, "baz", "1.0.1,5.11"),

                    self.__get_exp_pub_entry("test1", 8, "baz", "1.0,5.11"),

                    self.__get_exp_pub_entry("test2", 10, "baz", "1.3,5.11"),

                    self.__get_exp_pub_entry("test2", 9, "baz", "1.0.1,5.11"),

                    self.__get_exp_pub_entry("test2", 8, "baz", "1.0,5.11"),

                    self.__get_exp_pub_entry("test1", 12, "corge", "1.0,5.11"),

                    self.__get_exp_pub_entry("test1", 11, "corge", "0.9,5.11"),

                    self.__get_exp_pub_entry("test2", 12, "corge", "1.0,5.11"),

                    self.__get_exp_pub_entry("test2", 11, "corge", "0.9,5.11"),

                    self.__get_exp_pub_entry("test1", 13, "entire", "1.0,5.11"),

                    self.__get_exp_pub_entry("test2", 13, "entire", "1.0,5.11"),

                    self.__get_exp_pub_entry("test1", 14, "grault", "1.0,5.11"),

                    self.__get_exp_pub_entry("test2", 14, "grault", "1.0,5.11"),

                    self.__get_exp_pub_entry("test1", 15, "obsolete",

                        "1.0,5.11"),

                    self.__get_exp_pub_entry("test2", 15, "obsolete",

                        "1.0,5.11"),

                    self.__get_exp_pub_entry("test1", 16, "quux", "1.0,5.11"),

                    self.__get_exp_pub_entry("test2", 16, "quux", "1.0,5.11"),

                    self.__get_exp_pub_entry("test1", 18, "qux", "1.0,5.11"),

                    self.__get_exp_pub_entry("test1", 17, "qux", "0.9,5.11"),

                    self.__get_exp_pub_entry("test2", 18, "qux", "1.0,5.11"),

                    self.__get_exp_pub_entry("test2", 17, "qux", "0.9,5.11"),

                ]

                self.assertPrettyEqual(returned, expected)

                self.assertEqual(len(returned), 38)

                # Next, check no variants case (which has to be done

                # programatically to deal with unit test running on

                # different platforms).

                self.__test_list(api.ImageInterface.LIST_ALL, api_obj=api_obj,

                    num_expected=32, variants=False)

        def test_list_02_newest(self):

                """Verify the sort order and content of a list excluding

                packages not for the current image variant, and all but

                the newest versions of each package for each publisher."""

                self.__test_list(api.ImageInterface.LIST_NEWEST,

                    num_expected=16, variants=False)

                # Verify that LIST_NEWEST will allow version-specific

                # patterns such that the newest version allowed by the

                # pattern is what is listed.

                progresstracker = progress.NullProgressTracker()

                api_obj = api.ImageInterface(self.get_img_path(), CLIENT_API_VERSION,

                    progresstracker, lambda x: False, PKG_CLIENT_NAME)

                returned = self.__get_returned(api_obj.LIST_NEWEST,

                    api_obj=api_obj, patterns=["[email protected]", "bat/bar",

                        "[email protected]"], variants=True)

                expected = [

                    self.__get_exp_pub_entry("test1", 7, "bat/bar",

                        "1.2,5.11-0"),

                    self.__get_exp_pub_entry("test2", 7, "bat/bar",

                        "1.2,5.11-0"),

                    self.__get_exp_pub_entry("test1", 9, "baz", "1.0.1,5.11"),

                    self.__get_exp_pub_entry("test2", 9, "baz", "1.0.1,5.11"),

                    self.__get_exp_pub_entry("test1", 12, "corge", "1.0,5.11"),

                    self.__get_exp_pub_entry("test2", 12, "corge", "1.0,5.11"),

                ]

                self.assertPrettyEqual(returned, expected)

                self.assertEqual(len(returned), 6)

        def test_list_03_cats(self):

                """Verify the sort order and content of a list excluding

                packages not for the current image variant, and packages

                that don't match the specified scheme, category

                combinations."""