#!/usr/bin/python

# -*- coding: utf-8 -*-

#

# 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 2010 Sun Microsystems, Inc.  All rights reserved.

# Use is subject to license terms.

import testutils

if __name__ == "__main__":

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

import pkg5unittest

import datetime

import errno

import os

import shutil

import stat

import sys

import tempfile

import time

import unittest

import pkg.actions

import pkg.fmri as fmri

import pkg.catalog as catalog

import pkg.client.api_errors as api_errors

import pkg.manifest as manifest

import pkg.misc as misc

import pkg.portable as portable

import pkg.variant as variant

class TestCatalog(pkg5unittest.Pkg5TestCase):

        """Tests for all catalog functionality."""

        def setUp(self):

                pkg5unittest.Pkg5TestCase.setUp(self)

                self.paths = [self.test_root]

                self.c = catalog.Catalog(log_updates=True)

                self.nversions = 0

                stems = {}

                for f in [

                    fmri.PkgFmri("pkg:/[email protected],5.11-1:20000101T120000Z"),

                    fmri.PkgFmri("pkg:/[email protected],5.11-1:20000101T120010Z"),

                    fmri.PkgFmri("pkg:/[email protected],5.11-1.1:20000101T120020Z"),

                    fmri.PkgFmri("pkg:/[email protected],5.11-1.2:20000101T120030Z"),

                    fmri.PkgFmri("pkg:/[email protected],5.11-2:20000101T120040Z"),

                    fmri.PkgFmri("pkg:/[email protected],5.11-1:20000101T120040Z"),

                    fmri.PkgFmri("pkg:/[email protected],5.11-1:20000101T120050Z"),

                    fmri.PkgFmri("pkg:/[email protected],5.11-1.2:20000101T120051Z"),

                    fmri.PkgFmri("pkg:/[email protected],5.11-1.2.3:20000101T120052Z"),

                    fmri.PkgFmri("pkg:/[email protected],5.11-1:20000101T120040Z"),

                    fmri.PkgFmri("pkg:/[email protected],5.11-1:20000101T120040Z"),

                    fmri.PkgFmri("pkg:/[email protected],5.11-1:20000101T120014Z")

                ]:

                        if f.pkg_name == "apkg":

                                f.set_publisher("extra")

                        elif f.pkg_name == "zpkg":

                                f.set_publisher("contrib.opensolaris.org")

                        else:

                                f.set_publisher("opensolaris.org")

                        self.c.add_package(f)

                        self.nversions += 1

                        stems[f.get_pkg_stem()] = None

                # And for good measure, ensure that one of the publishers has

                # a package with the exact same name and version as another

                # publisher's package.

                f = fmri.PkgFmri("pkg://extra/[email protected],5.11-1:20000101T120040Z")

                stems[f.get_pkg_stem()] = None

                self.c.add_package(f)

                self.nversions += 1

                self.npkgs = len(stems)

        def create_test_dir(self, name):

                """Creates a temporary directory with the specified name for

                test usage and returns its absolute path."""

                target = os.path.join(self.test_root, name)

                try:

                        os.makedirs(target, misc.PKG_DIR_MODE)

                except OSError, e:

                        if e.errno != errno.EEXIST:

                                raise e

                return os.path.abspath(target)

        def __gen_manifest(self, f):

                m = manifest.Manifest()

                lines = (

                    "depend [email protected] type=require\n"

                    "set name=facet.devel value=true\n"

                    "set name=info.classification "

                    """value="Desktop (GNOME)/Application" """

                    """value="org.opensolaris.category.2009:GNOME (Desktop)"\n"""

                    "set name=info.classification "

                    """value="Sparc Application" variant.arch=sparc\n"""

                    "set name=info.classification "

                    """value="i386 Application" variant.arch=i386\n"""

                    "set name=variant.arch value=i386 value=sparc\n"

                    "set name=pkg.fmri value=\"%s\"\n"

                    "set name=pkg.summary value=\"Summary %s\"\n"

                    "set name=pkg.summary value=\"Sparc Summary %s\""

                    " variant.arch=sparc\n"

                    "set name=pkg.summary:th value=\"ซอฟต์แวร์ %s\"\n"

                    "set name=pkg.description value=\"Desc %s\"\n" % \

                    (f, f, f, f, f))

                if f.pkg_name == "zpkg":

                        lines += "set name=pkg.renamed value=true\n"

                else:

                        lines += "set name=pkg.obsolete value=true\n"

                m.set_content(lines, signatures=True)

                return m

        def __test_catalog_actions(self, nc, pkg_src_list):

                def expected_dependency(f):

                        expected = [

                            "depend [email protected] type=require",

                            "set name=facet.devel value=true",

                            "set name=variant.arch value=i386 value=sparc",

                        ]

                        if f.pkg_name == "zpkg":

                                expected.append("set name=pkg.renamed "

                                    "value=true")

                        else:

                                expected.append("set name=pkg.obsolete "

                                    "value=true")

                        return expected

                def expected_summary(f):

                        return [

                            ('set name=info.classification '

                            'value="Desktop (GNOME)/Application" '

                            'value="org.opensolaris.category.2009:GNOME (Desktop)"'),

                            ("set name=info.classification "

                            """value="i386 Application" variant.arch=i386"""),

                            "set name=pkg.summary value=\"Summary %s\"" % f,

                            "set name=pkg.description value=\"Desc %s\"" % f,

                        ]

                def expected_all_variant_summary(f):

                        return [

                            ('set name=info.classification '

                            'value="Desktop (GNOME)/Application" '

                            'value="org.opensolaris.category.2009:GNOME (Desktop)"'),

                            ("set name=info.classification "

                            """value="Sparc Application" variant.arch=sparc"""),

                            ("set name=info.classification "

                            """value="i386 Application" variant.arch=i386"""),

                            "set name=pkg.summary value=\"Summary %s\"" % f,

                            ("set name=pkg.summary value=\"Sparc Summary %s\""

                            " variant.arch=sparc" % f),

                            "set name=pkg.description value=\"Desc %s\"" % f,

                        ]

                def expected_all_locale_summary(f):

                        # The comparison has to be sorted for this case.

                        return sorted([

                            "set name=pkg.summary value=\"Summary %s\"" % f,

                            "set name=pkg.description value=\"Desc %s\"" % f,

                            "set name=pkg.summary:th value=\"ซอฟต์แวร์ %s\"" % f,

                            ('set name=info.classification '

                            'value="Desktop (GNOME)/Application" '

                            'value="org.opensolaris.category.2009:GNOME (Desktop)"'),

                            ("set name=info.classification "

                            """value="i386 Application" variant.arch=i386"""),

                        ])

                def expected_categories(f):

                        # The comparison has to be sorted for this case.

                        return set([

                            ("", "Desktop (GNOME)/Application"),

                            ("org.opensolaris.category.2009", "GNOME (Desktop)"),

                            ("", "i386 Application"),

                        ])

                def expected_all_variant_categories(f):

                        # The comparison has to be sorted for this case.

                        return set([

                            ("", "Desktop (GNOME)/Application"),

                            ("org.opensolaris.category.2009", "GNOME (Desktop)"),

                            ("", "i386 Application"),

                            ("", "Sparc Application"),

                        ])

                # Next, ensure its populated.

                def ordered(a, b):

                        rval = cmp(a.pkg_name, b.pkg_name)

                        if rval != 0:

                                return rval

                        rval = cmp(a.publisher, b.publisher)

                        if rval != 0:

                                return rval

                        return cmp(a.version, b.version) * -1

                self.assertEqual([f for f in nc.fmris(ordered=True)],

                    sorted(pkg_src_list, cmp=ordered))

                # This case should raise an AssertionError.

                try:

                        for f, actions in nc.actions([]):

                                break

                except AssertionError:

                        pass

                else:

                        raise RuntimeError("actions() did not raise expected "

                            "exception")

                variants = variant.Variants()

                variants["variant.arch"] = "i386"

                excludes = [variants.allow_action]

                locales = set(("C", "th"))

                # This case should only return the dependency-related actions.

                def validate_dep(f, actions):

                        returned = []

                        for a in actions:

                                self.assertTrue(isinstance(a,

                                    pkg.actions.generic.Action))

                                returned.append(str(a))

                        var = nc.get_entry_variants(f, "variant.arch")

                        vars = nc.get_entry_all_variants(f)

                        if f.pkg_name == "apkg":

                                # No actions should be returned for this case,

                                # as the callback will return an empty manifest.

                                self.assertEqual(returned, [])

                                self.assertEqual(var, None)

                                self.assertEqual([v for v in vars], [])

                                return

                        expected = expected_dependency(f)

                        self.assertEqual(returned, expected)

                        self.assertEqual(var, ["i386", "sparc"])

                        self.assertEqual([(n, vs) for n, vs in vars],

                            [("variant.arch", ["i386", "sparc"])])

                for f, actions in nc.actions([nc.DEPENDENCY]):

                        validate_dep(f, actions)

                latest = [f for f in nc.fmris(last=True)]

                for f, actions in nc.actions([nc.DEPENDENCY], last=True):

                        self.assert_(f in latest)

                        validate_dep(f, actions)

                latest = [

                    (pub, stem, ver)

                    for pub, stem, ver in nc.tuples(last=True)

                ]

                for (pub, stem, ver), entry, actions in nc.entry_actions(

                    [nc.DEPENDENCY], last=True):

                        self.assert_((pub, stem, ver) in latest)

                        f = fmri.PkgFmri("%s@%s" % (stem, ver), publisher=pub)

                        validate_dep(f, actions)

                # This case should only return the summary-related actions (but

                # for all variants).

                for f, actions in nc.actions([nc.SUMMARY]):

                        returned = []

                        for a in actions:

                                self.assertTrue(isinstance(a,

                                    pkg.actions.generic.Action))

                                returned.append(str(a))

                        if f.pkg_name == "apkg":

                                # No actions should be returned for this case,

                                # as the callback will return an empty manifest.

                                self.assertEqual(returned, [])

                                continue

                        expected = expected_all_variant_summary(f)

                        self.assertEqual(returned, expected)

                # This case should only return the summary-related actions (but

                # for 'C' and 'th' locales and without sparc variants).

                for f, actions in nc.actions([nc.SUMMARY], excludes=excludes,

                    locales=locales):

                        returned = []

                        for a in actions:

                                self.assertTrue(isinstance(a,

                                    pkg.actions.generic.Action))

                                returned.append(str(a))

                        if f.pkg_name == "apkg":

                                # No actions should be returned for this case,

                                # as the callback will return an empty manifest.

                                self.assertEqual(returned, [])

                                continue

                        returned.sort()

                        expected = expected_all_locale_summary(f)

                        self.assertEqual(returned, expected)

                # This case should only return the summary-related actions (but

                # without sparc variants).

                for f, actions in nc.actions([nc.SUMMARY], excludes=excludes):

                        returned = []

                        for a in actions:

                                self.assertTrue(isinstance(a,

                                    pkg.actions.generic.Action))

                                returned.append(str(a))

                        if f.pkg_name == "apkg":

                                # No actions should be returned for this case,

                                # as the callback will return an empty manifest.

                                self.assertEqual(returned, [])

                                continue

                        expected = expected_summary(f)

                        self.assertEqual(returned, expected)

                # Verify that retrieving a single entry's actions works as well.

                f = pkg_src_list[0]

                try:

                        for a in nc.get_entry_actions(f, []):

                                break

                except AssertionError:

                        pass

                else:

                        raise RuntimeError("get_entry_actions() did not raise "

                            "expected exception")

                # This case should only return the dependency-related actions.

                returned = [

                    str(a)

                    for a in nc.get_entry_actions(f, [nc.DEPENDENCY])

                ]

                expected = expected_dependency(f)

                self.assertEqual(returned, expected)

                # This case should only return the summary-related actions (but

                # for all variants).

                returned = [

                    str(a)

                    for a in nc.get_entry_actions(f, [nc.SUMMARY])

                ]

                expected = expected_all_variant_summary(f)

                self.assertEqual(returned, expected)

                # This case should only return the summary-related actions (but

                # for 'C' and 'th' locales and without sparc variants).

                returned = sorted([

                    str(a)

                    for a in nc.get_entry_actions(f, [nc.SUMMARY],

                    excludes=excludes, locales=locales)

                ])

                expected = expected_all_locale_summary(f)

                self.assertEqual(returned, expected)

                # This case should only return the summary-related actions (but

                # without sparc variants).

                returned = [

                    str(a)

                    for a in nc.get_entry_actions(f, [nc.SUMMARY],

                    excludes=excludes)

                ]

                expected = expected_summary(f)

                self.assertEqual(returned, expected)

                # This case should return the categories used (but without sparc

                # variants).

                returned = nc.categories(excludes=excludes)

                expected = expected_categories(f)

                self.assertEqual(returned, expected)

        def test_01_attrs(self):

                self.assertEqual(self.npkgs, self.c.package_count)

                self.assertEqual(self.nversions, self.c.package_version_count)

        def test_02_extract_matching_fmris(self):

                """Verify that the filtering logic provided by

                extract_matching_fmris works as expected."""

                cf = fmri.PkgFmri("pkg:/[email protected],5.10-1:20070101T120000Z")

                cl = catalog.extract_matching_fmris(self.c.fmris(),

                    patterns=[cf])[0]

                self.assertEqual(len(cl), 7)

                cf = fmri.PkgFmri("pkg:/[email protected],5.11-1:20061231T120000Z")

                cl = catalog.extract_matching_fmris(self.c.fmris(),

                    patterns=[cf])[0]

                self.assertEqual(len(cl), 7)

                cf = fmri.PkgFmri("pkg:/[email protected],5.11-2")

                cl = catalog.extract_matching_fmris(self.c.fmris(),

                    patterns=[cf])[0]

                self.assertEqual(len(cl), 5)

                cf = fmri.PkgFmri("pkg:/[email protected],5.11-3")

                cl = catalog.extract_matching_fmris(self.c.fmris(),

                    patterns=[cf])[0]

                self.assertEqual(len(cl), 4)

                # First, verify that passing a single version pattern

                # works as expected.

                # Sorts by stem, then version, then publisher.

                def extract_order(a, b):

                        res = cmp(a.pkg_name, b.pkg_name)

                        if res != 0:

                                return res

                        res = cmp(a.version, b.version) * -1

                        if res != 0:

                                return res

                        return cmp(a.publisher, b.publisher)

                # This is a dict containing the set of fmris that are expected

                # to be returned by extract_matching_fmris keyed by version

                # pattern.

                versions = {

                    "*": ["pkg:/[email protected],5.11-1:20000101T120040Z",

                        "pkg:/[email protected],5.11-1:20000101T120000Z",

                        "pkg:/[email protected],5.11-1:20000101T120010Z",

                        "pkg:/[email protected],5.11-1.1:20000101T120020Z",

                        "pkg:/[email protected],5.11-1.2:20000101T120030Z",

                        "pkg:/[email protected],5.11-2:20000101T120040Z",

                        "pkg:/[email protected],5.11-1:20000101T120040Z",

                        "pkg:/[email protected],5.11-1:20000101T120050Z",

                        "pkg:/[email protected],5.11-1.2:20000101T120051Z",

                        "pkg:/[email protected],5.11-1.2.3:20000101T120052Z",

                        "pkg:/[email protected],5.11-1:20000101T120014Z",

                        "pkg:/[email protected],5.11-1:20000101T120040Z",

                        "pkg:/[email protected],5.11-1:20000101T120040Z"],

                    "1.0": ["pkg:/[email protected],5.11-1:20000101T120040Z",

                        "pkg:/[email protected],5.11-1:20000101T120000Z",

                        "pkg:/[email protected],5.11-1:20000101T120010Z",

                        "pkg:/[email protected],5.11-1.1:20000101T120020Z",

                        "pkg:/[email protected],5.11-1.2:20000101T120030Z",

                        "pkg:/[email protected],5.11-2:20000101T120040Z",

                        "pkg:/[email protected],5.11-1:20000101T120014Z",

                        "pkg:/[email protected],5.11-1:20000101T120040Z",

                        "pkg:/[email protected],5.11-1:20000101T120040Z"],

                    "1.1": ["pkg:/[email protected],5.11-1:20000101T120040Z"],

                    "*.1": ["pkg:/[email protected],5.11-1:20000101T120040Z"],

                    "3.*": ["pkg:/[email protected],5.11-1:20000101T120050Z",

                        "pkg:/[email protected],5.11-1.2:20000101T120051Z",

                        "pkg:/[email protected],5.11-1.2.3:20000101T120052Z"],

                    "3.2.*": ["pkg:/[email protected],5.11-1:20000101T120050Z",

                        "pkg:/[email protected],5.11-1.2:20000101T120051Z",

                        "pkg:/[email protected],5.11-1.2.3:20000101T120052Z"],

                    "3.*.*": ["pkg:/[email protected],5.11-1:20000101T120050Z",

                        "pkg:/[email protected],5.11-1.2:20000101T120051Z",

                        "pkg:/[email protected],5.11-1.2.3:20000101T120052Z"],

                    "*,5.11": ["pkg:/[email protected],5.11-1:20000101T120040Z",

                        "pkg:/[email protected],5.11-1:20000101T120000Z",

                        "pkg:/[email protected],5.11-1:20000101T120010Z",

                        "pkg:/[email protected],5.11-1.1:20000101T120020Z",

                        "pkg:/[email protected],5.11-1.2:20000101T120030Z",

                        "pkg:/[email protected],5.11-2:20000101T120040Z",

                        "pkg:/[email protected],5.11-1:20000101T120040Z",

                        "pkg:/[email protected],5.11-1:20000101T120050Z",

                        "pkg:/[email protected],5.11-1.2:20000101T120051Z",

                        "pkg:/[email protected],5.11-1.2.3:20000101T120052Z",

                        "pkg:/[email protected],5.11-1:20000101T120014Z",

                        "pkg:/[email protected],5.11-1:20000101T120040Z",

                        "pkg:/[email protected],5.11-1:20000101T120040Z"],

                    "*,*-*": ["pkg:/[email protected],5.11-1:20000101T120040Z",

                        "pkg:/[email protected],5.11-1:20000101T120000Z",

                        "pkg:/[email protected],5.11-1:20000101T120010Z",

                        "pkg:/[email protected],5.11-1.1:20000101T120020Z",